-
Notifications
You must be signed in to change notification settings - Fork 1
Add async module #73
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
davidjedw
wants to merge
3
commits into
main
Choose a base branch
from
add_async_lib
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
Add async module #73
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| ## KDB+/Q Asynchronous Communication Library | ||
|
|
||
| This library provides kdb+/q functions for sending either deferred synchronous or asynchronous postback requests from a client process over a handle or list of handles, with error trapping at various points. Either type of request can be sent via conventional kdb+/q IPC or asynchronous broadcast. | ||
|
|
||
| Each of the library functions have no dependencies on the server-side code. | ||
|
|
||
| --- | ||
|
|
||
| ### Core Concepts | ||
|
|
||
| kdb+ processes can communicate with each using either synchronous or asynchronous calls. Synchronous calls expect a response and so the server must process the request when it is received to generate the result and return it to the waiting client. Asynchronous calls do not expect a response so allow for greater flexibility. The effect of synchronous calls can be replicated with asynchronous calls in one of two ways: | ||
|
|
||
| - deferred synchronous: the client sends an asynchronous request, then blocks on the handle waiting for the result. This allows the server more flexibility as to how and when the query is processed. | ||
|
|
||
| - asynchronous postback: the client sends an asynchronous request which is wrapped in a function to be posted back to the client when the result is ready. This allows the server flexibility as to how and when the query is processed, and allows the client to continue processing while the server is generating the result. | ||
|
|
||
| If either of these are carried out via asynchronous broadcast, the request will only be serialized once across a list of handles as opposed to convetional kdb+/q IPC where the request is serialised for each handle. For sending a larger message across multiple handles, this can reduce latency as well as memory/CPU overhead. | ||
|
|
||
| --- | ||
|
|
||
| ### Package Use | ||
|
|
||
| Note, in each of the examples below handles is a list of two handles to different server processes | ||
|
|
||
| ##### async.deferred | ||
| Can be used to make deferred synchronous calls via conventional kdb+/q IPC. It will send the query down each of the handles, then block and wait on the handles | ||
| The result set is of the form (successvector each handle; result vector) | ||
| ```q | ||
| // async.deferred[handles;query] | ||
| q)async.deferred[handles;"2+2"] | ||
| 1 1 | ||
| 4 4 | ||
| ``` | ||
|
|
||
| ##### async.broadcastdeferred | ||
| As above, except the query will be sent via asynchronous broadcast. | ||
| Note, that if there is an issue with any of the handles, the query won't be sent down any handle | ||
| ```q | ||
| // async.broadcastdeferred[handles;query] | ||
| q)async.broadcastdeferred[handles;"2+2"] | ||
| 1 1 | ||
| 4 4 | ||
|
|
||
| ``` | ||
|
|
||
| ##### async.postback | ||
| Can be used to make asynchronous postback calls via conventional kdb+/q IPC. | ||
| Wrap the supplied query in a postback function | ||
| Don't block the handle when waiting | ||
| Success vector is returned that it has been sent correctly | ||
| The result is then returned once executed by the server, although it is not wrapped in the status | ||
| ```q | ||
| // async.postback[handles;query;postback] | ||
| q)async.postback[handles;"2+2";{show x}] | ||
| 11b | ||
| 4 | ||
| 4 | ||
| ``` | ||
|
|
||
| ##### async.broadcastpostback | ||
| As above, except the query will be sent via asynchronous broadcast. | ||
| Similar to async.broadcast_deferred, if there is an issue with any of the handles, the query won't be sent down any handle | ||
| ```q | ||
| // async.broadcastpostback[handles;query;postback] | ||
| q)async.broadcastpostback[handles;"2+2";{show x}] | ||
| 11b | ||
| 4 | ||
| 4 | ||
| ``` |
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 |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| / library for sending async messages from a client process | ||
|
|
||
| send:{[w;h;q] | ||
| / build the query to send | ||
| tosend:$[w; ({[q] @[neg .z.w;@[{[q] (1b;value q)};q;{(0b;"error: server fail:",x)}];()]};q); | ||
| ({[q] @[neg .z.w;@[{[q] value q};q;{"error: server fail:",x}];()]};q)]; | ||
| / error trapping sending the query down the handle followed by an async flush | ||
| .[{x@y; x(::);1b};(h;tosend);0b]} | ||
|
|
||
| deferred:{[handles;query] | ||
| / for sending deferred synchronous message to a list of handles | ||
| sent:send[1b;;query] each handles:neg abs handles,(); | ||
| / block and wait for the results | ||
| res:{$[y;@[x;(::);(0b;"error: comm fail: handle closed while waiting for result")];(0b;"error: comm fail: failed to send query")]}'[abs handles;sent]; | ||
| / return results | ||
| (res[;0];res[;1])} | ||
|
|
||
|
|
||
| postback:{[handles;query;postback] | ||
| / for sending asynchronous postback message to a list of handles where the message is wrapped in the function postback | ||
| send[0b;;({[q;p] (p;@[value;q;{"error: server fail: ",x}])};query;postback)] each handles:neg abs handles,()} | ||
|
|
||
| broadcastdeferred:{[handles;query] | ||
| / for sending deferred synchronous message to a list of handles via async broadcast | ||
| tosend:({[q] @[neg .z.w;@[{[q] (1b;value q)};q;{(0b;"error: server fail:",x)}];()]};query); | ||
| sent:.[{-25!(x;y); x(::);1b};(handles;tosend);{(0b;"error: ",x)}]; | ||
| if[not first sent;:sent]; | ||
| / block and wait for the results | ||
| res:{$[y;@[x;(::);(0b;"error: comm fail: handle closed while waiting for result")];(0b;"error: comm fail: failed to send query")]}'[abs handles;sent]; | ||
| / return results | ||
| (res[;0];res[;1])} | ||
|
|
||
| broadcastpostback:{[handles;query;postback] | ||
| / for sending asynchronous postback message to a list of handles via async broadcast where the message is wrapped in the function postback | ||
| q:({[q;p] (p;@[value;q;{"error: server fail: ",x}])};query;postback); | ||
| tosend:({[q] @[neg .z.w;@[{[q] value q};q;{"error: server fail: ",x}];()]};q); | ||
| / error trapping sending the query down the handle followed by an async flush | ||
| .[{-25!(x;y); x(::);(count x)#1b};(handles;tosend);{(y#0b;"error: ",x)}[;count handles]]} | ||
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 |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| \l ::async.q | ||
|
|
||
| export:([deferred;postback;broadcastdeferred;broadcastpostback]) |
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 |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| action,ms,bytes,lang,code,repeat,minver,comment | ||
| before,0,0,q,async:use`di.async,1,1,load module into session | ||
|
|
||
| comment,,,,,,,Testing async.deferred | ||
| run,0,0,q,system"q -p 1234 -q &",1,,start server in background | ||
| run,0,0,q,system"q -p 1235 -q &",1,,start server in background | ||
| run,0,0,q,system"sleep 1",1,,wait for server to initialize | ||
| run,0,0,q,h:raze @[hopen;;()]each(`::1234;`::1235),1,,open handles stored in h on client for testing | ||
| true,0,0,q,all h>0,1,,verify client connected to server | ||
| run,0,0,q,h[0]"f:{x+1}",1,,define function f on one handle | ||
| run,0,0,q,.test.defres1:async.deferred[h;({x+1};1)],1,,testing async.deferred call that should execute on both servers | ||
| true,0,0,q,.test.defres1~((1b;1b);(2;2)),1,,successful async.deferred call with expected result | ||
| run,0,0,q,.test.defres2:async.deferred[h;({x+`a};1)],1,,testing async.deferred call that should fail | ||
| true,0,0,q,not any .test.defres2[0;],1,,async.deferred call failed on both servers with expected result | ||
| run,0,0,q,.test.defres3:async.deferred[h;(`f;1)],1,,testing async.deferred call that should fail on one server | ||
| true,0,0,q,10b ~ .test.defres3[0;],1,,async.deferred call failed on one server with expected result | ||
| run,0,0,q,.test.defres4:async.deferred[h[1];({exit 0};())],1,,server exits while client is waiting for result | ||
| true,0,0,q,(first .test.defres4[1])~"error: comm fail: handle closed while waiting for result",1,,async.deferred call failed with expected message received | ||
| run,0,0,q,.test.defres5:async.deferred[h;({x+1};1)],1,,testing async.deferred call that should fail on one closed handle and be successful on the other | ||
| true,0,0,q,(.test.defres5[0];.test.defres5[1])~(10b;(2;"error: comm fail: failed to send query")),1,,async.deferred failed to send query to one handle successful on the other | ||
|
|
||
| comment,,,,,,,Testing async.postback | ||
| run,0,0,q,system"q -p 1235 -q &",1,,start server in background | ||
| run,0,0,q,system"sleep 1",1,,wait for server to initialize | ||
| run,0,0,q,h:raze @[hopen;;()]each(`::1234;`::1235),1,,open handles stored in h on client for testing | ||
| true,0,0,q,all h>0,1,,verify client connected to server | ||
| run,0,0,q,".test.pbrestab:([]handle:();time:();result:())",1,,define table .test.pbrestab to store results from async.postback | ||
| run,0,0,q,".test.storeresult:{`.test.pbrestab upsert (.z.w;.z.t;enlist x)}",1,,define a function to store the posted back results in table .test.prestab | ||
| run,0,0,q,async.postback[h;({x+1};2);`.test.storeresult],1,,testing async.postback that should execute on both servers | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,(enlist 3;enlist 3) ~ -2#exec result from .test.pbrestab,1,,successful async.postback with expected result | ||
| run,0,0,q,async.postback[h;({x+`a};2);`.test.storeresult],1,,testing async.postback call that should fail | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,all all (exec result from -2#.test.pbrestab) like\: "error*",1,,async.postback call failed on both servers with expected result | ||
| run,0,0,q,async.postback[h;(`f;1);`.test.storeresult],1,,testing async.postback call that should fail on one server | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,(enlist 2;enlist "error: server fail: f") ~ exec result from -2#.test.pbrestab,1,,async.postback call failed on one server with expected result | ||
|
|
||
| comment,,,,,,,Testing async.broadcastdeferred | ||
| run,0,0,q,.test.bdefres1:async.broadcastdeferred[h;({x+1};1)],1,,testing async.broadcastdeferred call that should execute on both servers | ||
| true,0,0,q,.test.bdefres1~((1b;1b);(2;2)),1,,successful async.broadcastdeferred call with expected result | ||
| run,0,0,q,.test.bdefres2:async.broadcastdeferred[h;({x+`a};1)],1,,testing async.broadcastdeferred call that should fail | ||
| true,0,0,q,not any .test.bdefres2[0;],1,,async.broadcastdeferred call failed on both servers with expected result | ||
| run,0,0,q,.test.bdefres3:async.broadcastdeferred[h;(`f;1)],1,,testing async.broadcastdeferred call that should fail on one server | ||
| true,0,0,q,10b ~ .test.bdefres3[0;],1,,async.broadcastdeferred call failed on one server with expected result | ||
| run,0,0,q,.test.bdefres3:async.broadcastdeferred[h;(`f;1)],1,,testing async.broadcastdeferred call that should fail on one server | ||
| true,0,0,q,10b ~ .test.bdefres3[0;],1,,async.broadcastdeferred call failed on one server with expected result | ||
|
|
||
| comment,,,,,,,Testing async.broadcastpostback | ||
| run,0,0,q,".test.bpbrestab:([]handle:();time:();result:())",1,,define table .test.bpbrestab to store results from async.broadcastpostback | ||
| run,0,0,q,".test.storeresult:{`.test.bpbrestab upsert (.z.w;.z.t;enlist x)}",1,,define a function .test.storeresult to store the posted back results in .test.bpbrestab | ||
| run,0,0,q,async.broadcastpostback[h;({x+1};2);`.test.storeresult],1,,testing async.broadcastpostback that should execute on both servers | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,(enlist 3;enlist 3) ~ -2#exec result from .test.bpbrestab,1,,successful async.broadcastpostback with expected result | ||
| run,0,0,q,async.broadcastpostback[h;({x+`a};2);`.test.storeresult],1,,testing async.broadcastpostback call that should fail on both servers | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,all all (exec result from -2#.test.bpbrestab) like\: "error*",1,,async.broadcastpostback call failed on both servers with expected result | ||
| run,0,0,q,async.broadcastpostback[h;(`f;1);`.test.storeresult],1,,testing async.broadcastpostback call that should fail on one server | ||
| run,0,0,q,@[;"";()] each h,1,,wait for server response | ||
| true,0,0,q,(enlist 2;enlist "error: server fail: f") ~ exec result from -2#.test.bpbrestab,1,,async.broadcastpostback call failed on one server with expected result | ||
|
|
||
| run,0,0,q,@[;"exit 0";()] each neg h,,1,,closing remaining server process |
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.
is there ever a reason not to use broadcast? i.e. should we just get rid of the non-broadcast versions and rename the broadcast ones?
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.
Only scenario I can think of is if there's an issue with any of the handles - with broadcast, one invalid handle causes none of the messages to be sent, whereas for the non-broadcast versions only the message to the invalid handle will fail. Bit of a rare case though, and ideally some connection management code would prevent that situation in the first place. So to tidy things up, it's probably not worth keeping the non-broadcast versions in that case