Interprocess Communication (IPC)

Setting Up Two Processes

kdb+ processes talk to each other over TCP. To follow along, start two q sessions in separate terminals. The first is the server; it listens on port 5000:

q -p 5000

In the server session, create a small table to query, and a function that queries it:

trades:([] sym:`AAPL`MSFT`AAPL; price:185.2 410.1 186.0; size:100 200 150)
getTrades:{[s] select from trades where sym=s}

The second session is the client. All the remaining examples run in the client unless stated otherwise.

Opening a Connection

hopen opens a connection and returns a handle — an integer you use to send messages:

h: hopen `::5000                    / this machine, port 5000
h: hopen 5000                       / same thing: a bare port number means this machine
h: hopen `:localhost:5000           / explicit host name
h: hopen ":localhost:5000"          / the same address as a string
h: hopen `:localhost:5000:user:pwd  / with a username and password
h: hopen (`:localhost:5000;2000)    / with a 2000 ms connection timeout

The handle is just an integer, e.g. 4i (each hopen above opens a new connection and returns a new handle). If nothing is listening, hopen signals an error such as 'hop. OS reports: Connection refused.

Synchronous (Blocking) Messages

Applying the handle to a message sends it synchronously: the client waits for the server to evaluate it and send back the result.

h: hopen `::5000

/ Send a string of q code
h "1 + 1"                                   / 2
h "select from trades where sym=`AAPL"      / a 2-row table

/ Send a function and its arguments as a list: (function; arg1; arg2; ...)
h ({select from trades where sym=x}; `AAPL) / the same 2-row table
h (`getTrades; `AAPL)                       / the same again, calling a function defined on the server by name

/ A lambda defined on the client can be sent and run on the server
myFunc: {[x] x * 2}
h (myFunc; 21)                              / 42

Sending a list, rather than a string of code, is generally safer and is what most production code does: the server can check which function is being called before running it (see the IPC practice questions on message handlers).

Asynchronous (Non-Blocking) Messages

Applying the negated handle sends a message asynchronously: the client does not wait, and no result comes back.

neg[h] "update price:price*1.01 from `trades"
neg[h] (insert; `trades; (`GOOG; 142.8; 300))

Async messages are queued and sent when the client returns to its event loop. To push them out immediately, flush the handle with neg[h][] (or neg[h](::)). Messages on one handle are processed by the server in the order they were sent, so a synchronous "chaser" such as h "" returns only after every earlier async message on that handle has run.

Closing a Connection

hclose h

Close handles you no longer need. Each open handle uses a file descriptor on both sides, and before kdb+ 4.1 a process could hold at most 1022 connections.

Message Handlers

When a message arrives, the server calls a handler in the .z namespace. The defaults simply evaluate the message, and you can override them to log, validate or restrict requests:

/ Server side
.z.pg: {[msg] value msg}               / synchronous messages: the return value is sent back
.z.ps: {[msg] value msg}               / asynchronous messages: the return value is discarded
.z.po: {[h] -1 "opened handle ",string h;}   / a connection was opened
.z.pc: {[h] -1 "closed handle ",string h;}   / a connection was closed

Inside a handler, .z.w is the handle of the connection the current message arrived on, .z.u is the caller's username and .z.a its IP address.

Pushing Data to Clients

A server can remember the handles of clients that ask for updates, and later push data to them asynchronously — the pattern a tickerplant uses for its subscribers:

/ Server side
subs: `int$()
sub: {subs,: .z.w;}                       / a client calls this to register
publish: {[data] neg[subs] @\: (`upd; data);}

/ Client side: define the callback, then register
upd: {[data] show data}
h: hopen `::5000
h (`sub; ::)

When the server runs publish x, each subscribed client receives the message (upd; x)and runs its ownupdfunction withx`.

One-Shot Queries

For a single query, you can apply the address directly. q opens a connection, sends the query synchronously, and closes the connection again:

`::5000 "count trades"                    / 4: the 3 rows plus the GOOG row inserted above

If you need to make several calls, open a handle once and close it when you're done:

{[h] r: h "count trades"; hclose h; r} hopen `::5000

Ports and Configuration

/ Start listening on port 5000 from inside a running process
\p 5000

/ Show the current listening port (0 means not listening)
\p

/ Listen only on the loopback interface, so other machines can't connect
\p 127.0.0.1:5000

/ Stop listening (existing connections stay open)
\p 0

The same can be set at start-up on the command line: q -p 5000.

Error Handling Over IPC

An error on the server is signalled back to a synchronous caller, where it can be trapped like any other error:

h: hopen `::5000

/ The server cannot add a number to a symbol, so it signals 'type
@[h; "1 + `a"; {[err] -1 "Remote error: ", err; `error}]

The error handler receives the error message as a string.

A failed hopen can be trapped the same way:

h: @[hopen; (`::5999; 1000); {[err] -1 "Connection failed: ", err; 0Ni}]

Connection Pool Pattern

To spread queries over several connections, open them all up front and hand them out in turn:

.pool.handles: `int$()

.pool.connect: {[host;port;n]
  .pool.handles,: hopen each n # enlist `$":",host,":",string port;
  }

/ Return the first handle and rotate it to the back of the list
.pool.get: {
  h: first .pool.handles;
  .pool.handles: 1 rotate .pool.handles;
  h
  }

.pool.connect["localhost"; 5000; 3]
.pool.get[]

Production tip: Handles go stale if the remote process restarts. Detect closed connections in .z.pc, and wrap hopen in a retry loop (with a growing delay between attempts) to reconnect.