Skip to content

How to defer synchronous IPC responses

This page explains how to defer a synchronous interprocess communication (IPC) response so that a gateway can do asynchronous work while the client waits for an ordinary synchronous reply.

Asynchronous messaging gives you concurrency, but synchronous messaging is the more convenient model for a client application: send a query, get an answer. A deferred response gives you both. The gateway accepts a synchronous request, suspends the reply, dispatches the work asynchronously, and answers later — while the client sees nothing but a normal synchronous exchange.

Suspend the automatic response

-30! unblocks the main event loop during .z.pg, so you can dispatch asynchronous work and respond once it finishes.

Capture the client handle from .z.w, then call -30!(::). That call stops q from sending the response it would otherwise send when the function exits.

.z.pg: {[query]
  / 1. Dispatch async work to the workers
  remoteFunction: {[clntHandle; query]
    neg[.z.w](`callback; clntHandle; @[(0b;)value@; query; {[errorString](1b; errorString)}])
  };
  neg[workerHandles]@\:(remoteFunction; .z.w; query);

  / 2. Suspend the automatic response
  -30!(::);
 }

Two details matter here. You can call -30!(::) at any point in the execution path of .z.pg; and you must keep the client handle from .z.w somewhere you can reach later, because that is the only way back to the waiting client.

Send the deferred result

When the asynchronous work finishes, call the ternary form of -30! to complete the request:

-30!(handle; isError; result)
  • handle is the waiting client handle you captured earlier
  • isError is 0b for data, or 1b to raise an error on the client
  • result is the payload, or the error text when isError is 1b
-30!(handle; 0b; tableData)   / respond with data
-30!(handle; 1b; "limit")     / respond with an error, signaling 'limit to the client

Responding to a handle that is not waiting raises an error

A deferred response sent to a handle that is no longer waiting signals 'Handle n was not expecting a response msg.

A multi-worker gateway

Use a deferred response to fan a request out to several workers, collect their results, and return a single synchronous response to the client. The gateway stays free to serve other clients while the workers work, and answers once every worker has replied.

This example runs the gateway and both workers on one host for convenience. In practice each can run on its own machine.

The workers

Start two worker processes, each holding a small table with different data, so you can tell which rows came from which worker. In practice each worker's table would hold a much larger dataset.

Start the first worker, listening on port 6000:

q -p 6000

q)t: ([] a: 1 2 3; b: 4 5 6)

Start the second worker, listening on port 6001:

q -p 6001

q)t: ([] a: 11 12 13; b: 14 15 16)

The gateway script

workerHandles: hopen each 6000 6001   / handles to the worker processes

pending: ()!()                        / results received so far, keyed by client handle

/ joins the results once all of them have arrived
reduceFunction: raze

/ each worker calls this with (0b; result) or (1b; errorString)
callback: {[clientHandle; result]
 pending[clientHandle],: enlist result;              / store the result
 / respond once every worker has replied for this client
 if[count[workerHandles] = count pending clientHandle;
   isError: 0 < sum pending[clientHandle][;0];       / did any worker report an error?
   result: pending[clientHandle][;1];                / the results, or the error strings
   / send the first error, or the reduced result
   r: $[isError; {first x where 10h = type each x}; reduceFunction] result;
   -30!(clientHandle; isError; r);
   pending[clientHandle]: ();                        / clear the stored results
 ]
 }

.z.pg: {[query]
  remoteFunction: {[clntHandle; query]
    neg[.z.w](`callback; clntHandle; @[(0b;)value@; query; {[errorString](1b; errorString)}])
  };
  neg[workerHandles]@\:(remoteFunction; .z.w; query);   / send the query to each worker
  -30!(::);                                             / defer the response
 }

Run the gateway script, listening on port 5000 for client requests:

q gateway.q -p 5000

What the client sees

From the client's side this is ordinary synchronous IPC. The asynchronous fan-out is hidden: the gateway forwards the query to both workers and, once both have replied, applies raze to combine their results into one table before returning it.

q)h: hopen 5000
q)h"select from t"
a  b
-----
1  4
2  5
3  6
11 14
12 15
13 16

Row order follows reply order

raze joins the results in the order the workers reply, which varies from run to run. Sort the result in the gateway, or in the client, if you need a predictable order.

A worker error reaches the client as an ordinary q error:

q)h"select from nosuchtable"
'nosuchtable

Limits of this pattern

A deferred response solves blocking for a gateway: a q process that fans a query out to back ends, deserializes their replies, and combines them before answering the client. It doesn't help you build a transparent, protocol-aware load balancer in front of a pool of back ends. That component has to stream bytes between clients and servers without fully deserializing them, which q can't express — see why you cannot write this one in q.

Prepare for production

To run deferred responses in production, handle worker errors and clean up abandoned requests. Both keep the gateway stable and stop a client from waiting forever.

Handle worker errors

If a worker fails, the client mustn't hang. Have the worker catch the error and send it back to the gateway, which passes it on to the client.

  1. In the worker, use trap (@[f;fx;e] or .[g;enlist gx;e]) to catch errors, and return a list of a success flag and either the result or the error message.
  2. In the gateway, check the flag. If it reports an error, send that error to the client with -30!(handle; 1b; errorString).

This is the expression remoteFunction uses in the examples above:

/ (0b; result) on success, (1b; "error string") on failure
result: @[(0b;) value@; query; {[errorString](1b; errorString)}]

Clean up abandoned requests

Remove pending requests when a client disconnects or a worker stops responding.

When a client disconnects

If a client disconnects while waiting, drop its request from the tracking dictionary in the .z.pc handler:

.z.pc: {[handle]
  pending _: handle;   / remove the handle from the pending dictionary
}

When a request times out

Use the .z.ts timer to find requests that have waited too long, send the client an error, and clear the request:

/ called from .z.ts
checkTimeouts: {
  / ... identify stale handles ...
  -30!(staleHandle; 1b; "timeout");
  pending _: staleHandle;
}

Next steps