-
Notifications
You must be signed in to change notification settings - Fork 13
.net 8.0 #62
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
00raq00
wants to merge
7
commits into
KxSystems:main
Choose a base branch
from
00raq00:main
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
.net 8.0 #62
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit
Hold shift + click to select a range
6cdc787
.net8.0
c38a4a6
Added ParallelInsertRows
acd3529
fix for type error
f74b795
Merge branch 'main' into main
00raq00 fee18d3
small changes in feed and subscriber demos
1711d6c
Merge branch 'main' of https://github.com/00raq00/csharpkdb into main
d6e5a53
revert in demos
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,101 +1,143 @@ | ||
| using System; | ||
| using System.Security.Cryptography; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using kx; | ||
| using NLog; | ||
|
|
||
| namespace FeedDemo | ||
| { | ||
| static class Program | ||
| { | ||
| private static readonly ILogger Logger = LogManager.GetCurrentClassLogger(); | ||
| static class Program | ||
| { | ||
| private static readonly ILogger Logger = LogManager.GetCurrentClassLogger(); | ||
|
|
||
| private const string QFunc = ".u.upd"; | ||
| private const string TableName = "mytable"; | ||
| private const string QFunc = ".u.upd"; | ||
| private const string TableName = "mytable"; | ||
|
|
||
| static void Main() | ||
| static void Main() | ||
| { | ||
| string host = "localhost"; | ||
| int port = 5001; | ||
| string usernamePassword = $"{Environment.UserName}:mypassword"; | ||
|
|
||
| c connection = null; | ||
| try | ||
| { | ||
| connection = new c(host, port, usernamePassword); | ||
|
|
||
| //Example of 10 single row inserts to a table | ||
| InsertRows(connection); | ||
|
|
||
| //Parallel example of 100 single row inserts to a table | ||
| ParallelInsertRows(host, port, usernamePassword, 100, 10, 4); | ||
|
|
||
| //Parallel example of 1000 single row inserts to a table | ||
| ParallelInsertRows(host, port, usernamePassword, 1000, 100, 300); | ||
|
|
||
| //Parallel example of 1000 single row inserts to a table | ||
| ParallelInsertRows(host, port, usernamePassword, 1000, 1000, 1000); | ||
|
|
||
| //Example of bulk inserts to a table to improve throughput | ||
| BulkInsertRows(connection); | ||
|
|
||
| } | ||
| catch (Exception ex) | ||
| { | ||
| Logger.Error($"Error occurred running Feed-Demo. \r\n{ex}"); | ||
| } | ||
| finally | ||
| { | ||
| if (connection != null) | ||
| { | ||
| string host = "localhost"; | ||
| int port = 5001; | ||
| string usernamePassword = $"{Environment.UserName}:mypassword"; | ||
|
|
||
| c connection = null; | ||
| try | ||
| { | ||
| connection = new c(host, port, usernamePassword); | ||
|
|
||
| //Example of 10 single row inserts to a table | ||
| InsertRows(connection); | ||
|
|
||
| //Example of bulk inserts to a table to improve throughput | ||
| BulkInsertRows(connection); | ||
|
|
||
| } | ||
| catch (Exception ex) | ||
| { | ||
| Logger.Error($"Error occurred running Feed-Demo. \r\n{ex}"); | ||
| } | ||
| finally | ||
| { | ||
| if (connection != null) | ||
| { | ||
| connection.Close(); | ||
| } | ||
| } | ||
| connection.Close(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static void InsertRows(c connection) | ||
| private static void InsertRows(c connection) | ||
| { | ||
| // Single row insert - not as efficient as bulk insert | ||
| Logger.Info("Populating '{0}' table on kdb server with 10 rows...", TableName); | ||
|
|
||
| for (int i = 0; i < 10; i++) | ||
| { | ||
| // Assumes a remote schema of mytable:([]time:`timespan$();sym:`symbol$();price:`float$();size:`long$()) | ||
| object[] row = new object[] | ||
| { | ||
| // Single row insert - not as efficient as bulk insert | ||
| Logger.Info("Populating '{0}' table on kdb server with 10 rows...", TableName); | ||
|
|
||
| for(int i = 0; i < 10; i++) | ||
| { | ||
| // Assumes a remote schema of mytable:([]time:`timespan$();sym:`symbol$();price:`float$();size:`long$()) | ||
| object[] row = new object[] | ||
| { | ||
| new c.KTimespan(100), | ||
| "SYMBOL", | ||
| 93.5, | ||
| 300L | ||
| }; | ||
| }; | ||
|
|
||
| connection.ks(QFunc, TableName, row); | ||
| } | ||
| connection.ks(QFunc, TableName, row); | ||
| } | ||
|
|
||
| Logger.Info("Successfully inserted 10 rows to {0}", TableName); | ||
| } | ||
| Logger.Info("Successfully inserted 10 rows to {0}", TableName); | ||
| } | ||
| private static void ParallelInsertRows(string host, int port, string usernamePassword, int rowCount, int minThreads, int maxDegreeOfParallelism) | ||
| { | ||
| // Single row insert - not as efficient as bulk insert | ||
| Logger.Info("Populating '{0}' table on kdb server with {1} rows...", TableName, rowCount); | ||
| ThreadPool.SetMinThreads(minThreads, minThreads); | ||
|
|
||
| var parallelOptions = new ParallelOptions | ||
| { | ||
| MaxDegreeOfParallelism = maxDegreeOfParallelism | ||
| }; | ||
|
|
||
| private static void BulkInsertRows(c connection) | ||
| Parallel.For(0, rowCount, parallelOptions, i => | ||
| { | ||
| { | ||
| // Bulk row insert - more efficient | ||
| string[] syms = new[] { "ABC", "DEF", "GHI", "JKL" }; | ||
| // Assumes a remote schema of mytable:([]time:`timespan$();sym:`symbol$();price:`float$();size:`long$()) | ||
| object[] row = new object[] | ||
| { | ||
| new c.KTimespan(i), | ||
| "SYMBOL", | ||
| i, | ||
| 300L | ||
| }; | ||
| var c = new c(host, port, usernamePassword); | ||
| c.ks(QFunc, TableName, row); | ||
|
|
||
| c.KTimespan[] times = CreateTestArray(i => new c.KTimespan(i), 10); | ||
| string[] symbols = CreateTestArray(i => syms[RandomNumberGenerator.GetInt32(syms.Length)], 10); | ||
| double[] prices = CreateTestArray(i => i * 1.1, 10); | ||
| long[] sizes = CreateTestArray(i => (long)(i * 100), 10); | ||
| Logger.Info("Successfully inserted {1} row to {0}", TableName, i); | ||
| } | ||
| }); | ||
|
|
||
| Logger.Info("Bulk populating '{0}' table on kdb server without using column names", TableName); | ||
| Logger.Info("Successfully inserted {1} rows to {0}", TableName, rowCount); | ||
| } | ||
|
|
||
| connection.ks(QFunc, TableName, new object[] { times, symbols, prices, sizes }); | ||
| private static void BulkInsertRows(c connection) | ||
| { | ||
| // Bulk row insert - more efficient | ||
| string[] syms = new[] { "ABC", "DEF", "GHI", "JKL" }; | ||
|
|
||
| Logger.Info("Bulk populating '{0}' table on kdb server using column names", TableName); | ||
| c.KTimespan[] times = CreateTestArray(i => new c.KTimespan(i), 10); | ||
| string[] symbols = CreateTestArray(i => syms[RandomNumberGenerator.GetInt32(syms.Length)], 10); | ||
| double[] prices = CreateTestArray(i => i * 1.1, 10); | ||
| long[] sizes = CreateTestArray(i => (long)(i * 100), 10); | ||
|
|
||
| connection.ks(QFunc, TableName, new c.Flip(new c.Dict(new string[] { "time", "sym", "price", "size" }, new object[] { times, symbols, prices, sizes }))); | ||
| Logger.Info("Bulk populating '{0}' table on kdb server without using column names", TableName); | ||
|
|
||
| //block until all messages are processed | ||
| connection.k(string.Empty); | ||
| } | ||
| connection.ks(QFunc, TableName, new object[] { times, symbols, prices, sizes }); | ||
|
|
||
| private static T[] CreateTestArray<T>(Func<int, T> elementBuilder, int arraySize) | ||
| { | ||
| T[] array = new T[arraySize]; | ||
| Logger.Info("Bulk populating '{0}' table on kdb server using column names", TableName); | ||
|
|
||
| for (int i = 0; i < arraySize; i++) | ||
| { | ||
| array[i] = elementBuilder(i); | ||
| } | ||
| return array; | ||
| } | ||
| connection.ks(QFunc, TableName, new c.Flip(new c.Dict(new string[] { "time", "sym", "price", "size" }, new object[] { times, symbols, prices, sizes }))); | ||
|
|
||
| //block until all messages are processed | ||
| connection.k(string.Empty); | ||
| } | ||
|
|
||
| private static T[] CreateTestArray<T>(Func<int, T> elementBuilder, int arraySize) | ||
| { | ||
| T[] array = new T[arraySize]; | ||
|
|
||
| for (int i = 0; i < arraySize; i++) | ||
| { | ||
| array[i] = elementBuilder(i); | ||
| } | ||
| return array; | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,42 +1,42 @@ | ||
| using System; | ||
| using MessagePack; | ||
| using System; | ||
| using System.IO; | ||
| using System.Runtime.Serialization.Formatters.Binary; | ||
| using System.Text.Json; | ||
|
|
||
| namespace kx.Test.TestUtils | ||
| { | ||
| /// <summary> | ||
| /// A helper class for testing serialisation and de-serialisation logic | ||
| /// in unit-tests | ||
| /// </summary> | ||
| internal static class TestSerialisationHelper | ||
| { | ||
| /// <summary> | ||
| /// A helper class for testing serialisation and de-serialisation logic | ||
| /// in unit-tests | ||
| /// Performs binary serialisation and de-serialisation on a specified exception. | ||
| /// </summary> | ||
| internal static class TestSerialisationHelper | ||
| /// <typeparam name="T">The type of exception being tested.</typeparam> | ||
| /// <param name="exception">The exception to be serialised and de-serialised.</param> | ||
| /// <returns> | ||
| /// A de-serialised instance of the exception passed. All serialisable members should match the original. | ||
| /// </returns> | ||
| /// <remarks> | ||
| /// This is primarily intended to confirm custom exceptions within the DeltaApiCore | ||
| /// library comply to ISerialization pattern. | ||
| /// | ||
| /// See https://stackoverflow.com/questions/94488/what-is-the-correct-way-to-make-a-custom-net-exception-serializable | ||
| /// </remarks> | ||
| public static T SerialiseAndDeserialiseException<T>(T exception) | ||
| where T : Exception | ||
| { | ||
| /// <summary> | ||
| /// Performs binary serialisation and de-serialisation on a specified exception. | ||
| /// </summary> | ||
| /// <typeparam name="T">The type of exception being tested.</typeparam> | ||
| /// <param name="exception">The exception to be serialised and de-serialised.</param> | ||
| /// <returns> | ||
| /// A de-serialised instance of the exception passed. All serialisable members should match the original. | ||
| /// </returns> | ||
| /// <remarks> | ||
| /// This is primarily intended to confirm custom exceptions within the DeltaApiCore | ||
| /// library comply to ISerialization pattern. | ||
| /// | ||
| /// See https://stackoverflow.com/questions/94488/what-is-the-correct-way-to-make-a-custom-net-exception-serializable | ||
| /// </remarks> | ||
| public static T SerialiseAndDeserialiseException<T>(T exception) | ||
| where T : Exception | ||
| { | ||
| BinaryFormatter binaryFormatter = new BinaryFormatter(); | ||
| using (var stream = new MemoryStream()) | ||
| { | ||
| MessagePackSerializer.Serialize(stream, exception, MessagePack.Resolvers.ContractlessStandardResolver.Options); | ||
|
|
||
| using (var stream = new MemoryStream()) | ||
| { | ||
| binaryFormatter.Serialize(stream, exception); | ||
| stream.Seek(0, 0); | ||
|
|
||
| stream.Seek(0, 0); | ||
|
|
||
| return (T)binaryFormatter.Deserialize(stream); | ||
| } | ||
| } | ||
| return (T)MessagePackSerializer.Deserialize<T>(stream, MessagePack.Resolvers.ContractlessStandardResolver.Options); | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Re: Demos/FeedDemo/Program.cs
Running
get many `type errors. Just running ParallelInsertRows with the q instance changed to add printing of message received i.e.
see what's causing the errors (mismatching types for the table updates):
The code should prob have been
Though since data is being sent on threads, each thread is interleaving with each other giving different order each time (as expected). Order is the same per thread, but not across threads. e.g. first run, looking at first 5 entries to table when only running parallel inserts:
next run has different order, as expected due to threading
Depending on use-case, this may not be desirable & will leave it off the demo as closer to programming paradigm rather than API example.