kdb+ IPC for Beginners

In this blog we will go over the basics of IPC (Inter process communication) in kdb+ (v4.0). Processes can communicate with each other through TCP/IP (networking protocols that allow computers to communicate) and all the tools we need are already bundled into q.

Connecting to another q process

For process 1 to be able to connect to process 2, process 2 must be listening on a port. This can be done either from the command line or within the process itself:

// command line
$ q -p 1234

// within process
q)\p 1234

Process 1 can then connect to process 2 using the q keyword hopen. The most simple form of this is when both processes are running on the same host and the only argument required is the port:

q)hopen 1234
4i

//alternatively
q)hopen `::1234
5i

The integer value returned from hopen is the handle, which we can then use to communicate with process 2. It is common practice to assign the handle to a variable h when using hopen which we will see later. You can open multiple handles to the same process; a dictionary of open handles on process 1 is stored in the variable .z.W. The key is the handle and the value corresponds to the number of bytes waiting in their output queues (more on this later).

q).z.W
4|
5|

If we only want to see what handles are currently open/active we can use .z.H (introduced in v4.0) which returns an integer list:

q).z.H
4 5i
q).z.H~key .z.W
1b

For processes that are on different hosts, or that need credentials to access, hopen can also be written in the following format:

q)hopen `:host:port:user:password

//optional timeout for hopen in ms
q)hopen (`:host:port:user:password;1000)

A handle is closed using hclose:

q)hclose 5
q).z.W
4|

Sending requests to the remote process

//process 1              //process 2
           
q)h:hopen 1234           
q)h"2+2"
4
q)h"x:3"                 q)x
                         3
q)h"show 4"              q)
                         4   

Another way to send requests over a handle is in the form of a function and a list of args:

//process 1              //process 2
           
q)h:hopen 1234           
q)h(+;2;2)
4
                         q)f:{x+y}
                         
q)h(`f;4;5)             
9                         

Sync vs Async

So far we have only seen synchronous requests, which will block the handle until a response is received. To send an asynchronous request we use the negative version of the handle:

q)neg[h]"y:5"

This serializes and puts a message on the output queue for handle h, and does not block the client. To ensure an async message is sent immediately, you can flush the pending outgoing queue for handle h:

q)neg[h][]

To ensure an async message has been processed, you can send a sync “chaser”:

q)h""

As it is a sync message it sends any pending outgoing (async) messages on h, sends the sync request message, and processes any pending incoming messages on h until a response (or error) message is received. It is good practice to send an empty chaser message immediately before applying hclose to a handle you have been sending async messages on. This is to ensure any buffered messages are sent before the connection is closed.

We can look at this example to demonstrate what the output queue looks like after making an async call:

q)h:hopen`::1234
q)neg[h]"2+2";.z.W
4| 17

//we can confirm this by checking the uncompressed length of the query
q)-22!"2+2"
17

Here is an example of how flushing the handle affects the bytes in the output queue:

q)sleep:{now:.z.p; waittime:x * 00:00:01; while[waittime>.z.p - now]}
q)neg[h]"2+2";sleep[5];.z.W
4| 17
q)neg[h]"2+2";neg[h][];sleep[5];.z.W
4|

There are still bytes in .z.W at the end of the first example as when you send a request async then kdb+ will just queue it until it finishes whatever it is it’s currently doing. In the second example it is forced to flush before the sleep statement so there’s nothing in the output queue after.

Message Handlers

There are several message handlers in the .z namespace associated with IPC. They have default behaviours, but can be customised by users.

.z.po (open)

This is called when a connection to a kdb+ session has been successfully opened and after it has been validated against any -u/-U file and .z.pw checks (explained later). The argument to this function is the connection handle of the sending process and is typically used to build a dictionary/table of handles to session information, for example .z.a (ip address) and .z.u (username).

//process 1

//create table to track connections
q)connections:([]handle:`int$();ip:`int$();user:`$())

//set .z.po to capture connection details
q).z.po:{`connections upsert (.z.w;.z.a;.z.u)}

//process 2

q)h:hopen 1234

//process 1

q)connections
handle ip         user
------------------------
6      2130706433 user123

You can reset .z.po (and any other message handlers) to their default behavior using \x

q)\x .z.po

.z.pc (close)

Called when a connection to a kdb+ session is closed. Unlike .z.po, by the time this handler is invoked, any variables associated with the connection are lost and the only information passed to the function is the handle that was closed. This allows us to clean up any connection related data in our process, for example from the connections table created above.

//process 1

//set .z.pc to delete connection details
q).z.pc:{delete from `connections where handle=x}

//process 2

q)hclose h

//process 1

q)connections
handle ip         user
------------------------

.z.pw (password)

Evaluated before the .z.po handler, but after any -u/U command line checks. This can be used to implement custom authorisation. The arguments should be the user (symbol) and password (string) and the result should be a boolean atom. If 1b is returned the connection has been successful, otherwise 0b is returned and the process trying to connect will get an ‘access error. Here is an example of a custom implementation that denies access to any user in a pre-defined blocklist:

//process 1

//set .z.pw to deny user from blacklist 
q)blacklist:`user123
q).z.pw:{[user;pass] $[user in blacklist;0b;1b]}

//process 2 (user123)

q)hopen 1234
'access

.z.pg (get — sync)

This function is invoked when a synchronous request is sent to the process. The default behaviour is the equivalent of running value on the object passed via the synchronous request. The return value (if any) will be returned to the process that called it. A simple customisation could be inserting details of the request into a table that tracks users and their queries:

//process 1

//create table to track connections
q)query_tracker:([]time:`timestamp$();user:`$();query:())

//set .z.pg to capture query details
q).z.pg:{[x]`query_tracker upsert (.z.p;.z.u;.z.u;x);value x}

//process 2

q)h"2+2"
4

//process 1

q)query_tracker
time                          user    query
-------------------------------------------
2024.02.01D15:58:08.762149010 user123 "2+2"

.z.ps (set — async)

This is similar to .z.pg, but instead is invoked when an asynchronous request is sent to the process. It has the same default behaviour (value); although in this case there will be no return value.

Interesting Examples

Dictionaries and .z.pg’s value behaviour

An interesting example to look at is a dictionary and how .z.pg defaults to value. Depending on how we send a dictionary across a handle (with .z.pg set to the default behaviour) we can see different results:

//process 2

q)h"`a`b`c!1 2 3"
a|1
b|2
c|3

In the above example the full dictionary is returned, as is expected in q when you run value on a string containing a dictionary. However the following syntax gives us a different result:

//process 2

q)h`a`b`c!1 2 3
1 2 3

This is because if you call value on a dictionary without stringing it, you will get the values of the dictionary itself.

What happens if you send a request to a process then set your own process to sleep?

This depends if you send a sync or an async request. Let’s look at a sync example:

//process 2                    //process 1

q)h"c:10";system "sleep 10"    q)c
                               10   

We can immediately see that c has been set to 10 on process 1, before process 2 starts to sleep. However if this was an async request…

//process 2                         //process 1

q)neg[h]"c:11";system "sleep 10"    q)c
                                      'c
                                        [0] c
                                            ^
                                    //wait 10 seconds
                                    q)c
                                    11

This is because when you send a sync message, the process waits for a result before continuing execution. In the async example, the message gets put into the outgoing queue and continues execution immediately, meaning that the process sleeps for 10 seconds before the message can be flushed from the queue.

About the Author:
Brooke Hopley is a Senior Software Engineer at Version 1.