2.1. Status Codes

The Ach API functions use return status codes to indicate either successful completion or an error condition. The following codes are defined:

2.2. Creating a channel

You can create a channel by directly calling the ach_create; however, it is generally preferable to call the ach (see also) from the shell as this is less likely to result in races between multiple processes needing to access the channel.

Channels are created using the ach_create function. A channel must be created before it can be opened. This function will create and initialize a POSIX shared memory file for the channel. The channel can then be opened by passing the same channel_name to ach_open.

enum ach_status r = ach_create( "my_channel", 10, 512, NULL );
if( ACH_OK != r ) {
    fprintf( stderr, "Could not create channel: %s\n", ach_result_to_string(r) );
    exit(EXIT_FAILURE);
}
      

Note that the message size given here is not a strict constraint. Individual messages are allowed to be smaller or larger than this value. The only hard constraint is that a total of frame_cnt*frame_size is allocated for all message data in the channel. Thus, the total data required by all buffered messages cannot exceed this value (older messages are overwritten), and no individual message may be larger than this value.

2.3. Opening a channel

Opens a channel for use within this process. The channel must be opened before messages can be sent or received

ach_channel_t channel;
enum ach_status r = ach_open( &channel, "my_channel", NULL );
if( ACH_OK != r ) {
    syslog( LOG_ERR, "Could not open channel: %s\n", ach_result_to_string(r) );
    exit(EXIT_FAILURE);
}
      

Channels, of type ach_channel_t, should always be passed by reference. Copying or passing by value types of ach_channel_t may result in undefined behavior (and will probably not do what you want). Always pass the pointer to ach_channel_t instead.

2.4. Sending Data

The ach_put function writes a new message to the channel. This will go in the next open space in the circular array. If there is insufficient unused space in the circular array, then the oldest entry or entries will be overwritten. If the entire circular array is too small for the new message, the message is not written and ACH_OVERFLOW is returned.

struct my_msg_type my_msg;
/* Fill in my_msg with useful data */
enum ach_status r = ach_put( &channel, &my_msg, sizeof(my_msg) );
if( ACH_OK != r ) {
    syslog( LOG_ERR, "Could not put data: %s\n", ach_result_to_string(r) );
}
      

2.5. Receiving Data

The ach_get is used to received data from a channel. The options parameter controls the behavior of this function and is the bitwise or of the following values.

size_t frame_size;
enum ach_status r;
r  = ach_get( &channel, &my_msg, my_msg_size, &frame_size,
              NULL, ACH_O_NONBLOCK | ACH_O_FIRST );
if( ACH_MISSED_FRAME == r ) {
    printf("Missed a/some messages(s)\n");
} else if( ACH_STALE_FRAMES == r ) {
    printf("No new data\n");
} else if( ACH_OK != r ) {
    syslog( LOG_ERR, "Unable to get a message: %s\n", ach_result_to_string(r) );
} else if( frame_size != sizeof(my_msg) ) {
    syslog( LOG_WARNING, "Unexpected message size %" PRIuPTR ", expecting %" PRIuPTR "\n",
            frame_size, sizeof(my_msg) );
}
      

enum ach_status r = ach_get( &channel, &my_msg, my_msg_size, &frame_size, NULL, ACH_O_WAIT | ACH_O_LAST );
      

struct timespec t;
clockid_t clock;
enum ach_status r;
/* Get the channel clock */
r = ach_channel_clock(&channel,&clock);
if( ACH_OK != r ) {
  fprintf(stderr, "Could not get channel clock: %s\n",
          ach_result_to_string(r));
  exit(EXIT_FAILURE);
}
/* Get current time and set the timeout. */
clock_gettime( clock, &t );
t.tv_sec += 1;
/* Get the message */
enum ach_status r = ach_get( &channel, &my_msg, my_msg_size, &frame_size, &t,
                             ACH_O_WAIT | ACH_O_LAST | ACH_O_ABSTIME );
if( ACH_TIMEOUT == r ) {
  fprintf(stdout, "call to ach_get timed out\n");
}
      

Does anybody know what time it is?

While ach channels default to using CLOCK_MONOTONIC for timed waits, the issue of determining the correct time is fraught with complications. Traditionally, unix time -- seconds since January 1, 1970 -- has been defined based on UTC. This is given to nanosecond precision with CLOCK_REALTIME (nothing to do with real-time-as-in-low-latency-or-motion-control). However, UTC and thus typically CLOCK_REALTIME is discontinuous. It may be reset by the operator, by the NTP daemon, and it is defined as discontinous when leap seconds occur. This is bad when one wants to use the clock for periodic events as is likely for motion control. CLOCK_MONOTONIC is not discontinous, but even it has issues. The NTP daemon may slew this clock, speeding it up or slowing it down on Linux by 0.5ms per second. Linux does provide the CLOCK_MONOTONIC_RAW which is not slewed, but is also not portable.

2.6. Cancel message wait

To interrupt an in-progress call to ach_get that has blocked waiting for a new message, use the ach_cancel function. This will cause ach_get to return ACH_CANCELED.

ach_channel_t channel;

...

static void sighandler_cancel ( int sig ) {
    enum ach_status r = ach_cancel(&channel, NULL);
    if( ACH_OK != r ) exit(EXIT_FAILURE);
}

...

void setup_sighandler_cancel ( int sig ) {
    struct sigaction act;
    memset(&act, 0, sizeof(act));
    act.sa_handler = &sighandler_cancel;
    if( sigaction(sig, &act, NULL) ) exit(EXIT_FAILURE);
}
      

ach_channel_t channel;

...

ach_cancel_attr_t attr;
ach_cancel_attr_init(&attr);
attr.async_unsafe = 1; /* permit functions that are unsafe in signal handlers */
enum ach_status r = ach_cancel(&channel, &attr);
if( ACH_OK != r ) exit(EXIT_FAILURE);
      

2.7. Closing a channel

2.8. Deleting a channel

2.9. Channel Permissions

Set the POSIX permission bits of the channel to mode. Note that any channel access requires both the read (4) and write (2) bits to be set, because we must write in order to hold the mutex. The executable bit (1) is irrelevant.

      enum ach_status r = ach_get( &channel, 0660 );
    

2.10. Ignore old messages

Updates channel counters to ignore all previously published messages.

3.1. Common Lisp

Ach includes bindings for Common Lisp using CFFI. They can be loaded via ASDF.

3.2. Python

Ach includes bindings for Python using a C extension module and wrapper class.

The module uses Python's Buffer Protocol to obtain a serialized representation for Python objects. This will work with string, bytearray, ctypes.Structure and other types which implement the Buffer Protocol.

#!/usr/bin/env python
import ach
c = ach.Channel('foo')
c.put('bar')
c.close()
    

#!/usr/bin/env python
import ach
c = ach.Channel('foo')
c.flush()
b = bytearray(10)
[status, framesize] = c.get( b, wait=True, last=False )
if status == ach.ACH_OK or status == ach.ACH_MISSED_FRAME:
    print b
else:
    raise ach.AchException( c.result_string(status) )
c.close()
    

5.1. Kernel Module Installation

The most convenient way to build the kernel module will probably be with DKMS (Dynamic Kernel Module System). DKMS enables building and rebuilding of out-of-tree kernel modules (such as ach, device drivers, etc.) when the kernel is upgraded. On Debian and Ubuntu, the package manager automatically invokes DKMS to rebuild the registered modules. To setup ach for DKMS, call configure with ./configure --enable-dkms-build, and the module sources for will be installed and built when you call make install.

./configure --enable-dkms-build
make
sudo make install

Alternatively, you can build the module in the ach source tree directly by calling ./configure --enable-kbuild --disable-dkms. If your kernel headers are not in the typical location and/or do not match the currently running kernel, you may to specify the kernel release and header directory.

./configure --enable-kbuild \
            --disable-dkms \
             KERNELRELEASE=3.13.0-44-lowlatency \
             KDIR=/usr/src/linux-headers-3.13.0-44-lowlatency
make
sudo make install

After the kernel module is build and installed, you need to link it into the kernel. This is done with the modprobe command. The kernel module can be removed with rmmod. The module takes one optional parameter, max_devices which specifies the maximum number of kernel channels that can be created.

sudo modprobe ach
    

sudo modprobe ach max_devices=1024
    

sudo rmmod ach
    

5.2. Creating Kernel Channels

Kernel channels can be created using the ach command by passing the -k flag.

ach mk -k my-channel
    

ach rm my-channel
    

5.3. Kernel Channel Permissions

By default, kernel channels can only be created and accessed by root. However, you can use udev to control kernel channel permissions. The following example will set the group ownership of all ach devices to the ach group. The file needs to be placed in /etc/udev/rules.d/.

# /etc/udev/rules.d/10-ach.rules
# Control permissions for ach character devices

# Set the permissions of the ach control device
KERNEL=="achctrl", GROUP=="ach", MODE=="660"

# Set the permissions of ach channel devices
KERNEL=="ach-*", GROUP=="ach", MODE=="660"

# Create symbolic links for ach channel devices
KERNEL=="ach-*", SYMLINK+="ach/%k"
    

5.4. Waiting on Multiple Kernel Channels

Kernel channels support waiting for multiple events using the poll, select and similar functions. These functions require the file descriptor for the channel.

ach_channel_t channel;
int channel_fd;
enum ach_status r;

/* Open channel */
/* ... */

/* Get Channel File Descriptor */
r = ach_channel_fd( &channel, &channel_fd );
if( ACH_OK != r ) {
    fprintf(stderr, "could not get file descriptor for channel '%s': %s\n",
            names[i], ach_result_to_string(r));
    exit(EXIT_FAILURE);
}
    

The channel file descriptor can then be passed to poll, select, etc. to determine when there is new data to read. Note that channels are always ready to write.

#include <stdlib.h>
#include <pthread.h>
#include <inttypes.h>
#include <ach.h>

#include <stdio.h>
#include <poll.h>
#include <unistd.h>

int main(int argc, char **argv)
{
    const char *names[] = {"channel-0", "channel-1"};
    const size_t n = sizeof(names) / sizeof(names[0]);
    ach_channel_t channel[n];
    struct pollfd pfd[n];

    for( size_t i = 0; i < n; i ++ ) {
        /* Open Channel */
        enum ach_status r = ach_open( &channel[i], names[i], NULL );
        if( ACH_OK != r ) {
            fprintf(stderr, "could not open channel '%s': %s\n",
                    names[i], ach_result_to_string(r));
            exit(EXIT_FAILURE);
        }
        /* Get Channel File Descriptor */
        r = ach_channel_fd( &channel[i], &pfd[i].fd );
        if( ACH_OK != r ) {
            fprintf(stderr, "could not get file descriptor for channel '%s': %s\n",
                    names[i], ach_result_to_string(r));
            exit(EXIT_FAILURE);
        }
        /* Set events to poll for */
        pfd[i].events = POLLIN;
    }

    /* read forever */
    for(;;) {
        /* poll for new messages */
        int r_poll = poll( pfd, n, -1 );
        if( r_poll < 0 ) {
            perror("poll");
            exit(EXIT_FAILURE);
        }
        /* find channels with new data */
        for( size_t i = 0; i < n && r_poll > 0; i++ ) {
            if( (pfd[i].revents & POLLIN) ) {
                /* There's new data on this channel */
                char buf[512];
                size_t frame_size;
                enum ach_status r = ach_get( &channel[i], buf, sizeof(buf), &frame_size,
                                             NULL, ACH_O_NONBLOCK | ACH_O_FIRST );
                switch(r) {
                case ACH_OK:
                case ACH_MISSED_FRAME: {
                    /* this example just writes it to stdout */
                    ssize_t wr = write( STDOUT_FILENO, buf, frame_size );
                    if( wr < 0 ) {
                        perror("write");
                        exit(EXIT_FAILURE);
                    }
                    break;
                }
                default:
                    fprintf( stderr, "Error getting data from '%s': %s\n",
                             names[i], ach_result_to_string(r));
                    exit(EXIT_FAILURE);
                }
                r_poll--;
            }
        }
    }
    return 0;
}
    

The poll function has an advantage over select in that the arguments to poll can be reused over multiple calls while the arguments to select must be recreated for each call. However poll provides only millisecond timing resolution. Instead, the more recent ppoll supports nanosecond timing resolution. For details on these functions, consult the man pages with man -s 2 poll and man -s 2 ppoll.

6.1. Server

achd serve

The achd server should be setup to run from the inetd super-server. Some distributions use the alternative xinetd, which is configured differently.

Inetd and xinetd work by listening for incoming connections on all the ports specified in their configuration. When (x)inetd recieves a new connection, it forks off the process as given in the configuration file, (in this case achd), and hooks the standard input and output of that process to the socket connection.

On Debian-based systems (including Ubuntu and Mint), you can use the openbsd-inetd package, installed with apt-get install openbsd-inetd. Alternatively, you can install xinetd with with apt-get install xinetd.

---

6.1.1. Traditional Inetd Configuration

You should add the following line to /etc/inetd.conf, assuming you have installed ach under /usr/local/:

#/etc/inetd.conf
8076  stream  tcp  nowait  nobody  /usr/local/bin/achd /usr/local/bin/achd serve
    

On Debian-based systems where you have installed ach via the package manger, you can enable achd in inetd by running dpkg-reconfigure ach-utils.

You will probably need to restart inetd after editing the configuration file. How to do this depends on the init system used by your operating system and the particular inetd you are using. Here are some options which may work:

SysV Init (general GNU/Linux)

/etc/init.d/inetd restart

Debian (Ubuntu/Mint) with openbsd-inetd

service openbsd-inetd restart

#/etc/inetd.conf
achd  stream  tcp  nowait  nobody  /usr/local/bin/achd /usr/local/bin/achd serve

#/etc/xinetd.d/achd
service achd
{
	flags           = REUSE
	socket_type     = stream
	protocol        = tcp
	port            = 8076
	wait            = no
	user            = nobody
	server          = /usr/local/bin/achd
	server_args     = serve
	disable         = no
}
    

#/etc/services
achd           8076/tcp
    

---

6.1.3. Testing Server Configuration

Now, you can test this setup by telnetting to port 8076 on the server: telnet SERVER 8076. If you type in some gibberish (e.g. asdf) followed by a newline, then you should see something like this:

Escape character is '^]'.
asdf
status: 14 # ACH_BAD_HEADER
message: malformed header

.
Connection closed by foreign host.
    

Which will mean that achd is properly setup.

If achd is not operating properly, check the log files on your machine for more information. The log output will typically be stored in a file such as /var/log/daemon.log, though this depends on how syslog is configured on your system. You can also add the -vv flags to both server achd (in inetd.conf) the client achd commands to enable debugging output. Syslog may store the debug output in /var/log/debug.log.

For the achd server to access any channel, it must be readable and writable by the user under which achd runs ("nobody" in the above example line for inetd.conf). If you trust all users on the local machine, you can set the channel permissions to "666", but please note that this is not a secure configuration. An alternative is to create a separate user account for achd and ensure that channels are readable and writable by that account or one of its groups.

Currently, achd performs no authentication, access control, or encryption (this may be added in the future). If security is a concern in your application, it must currently be addressed at the network level, e.g. by physical isolation, firewalling, a VPN such as IPSec, and/or SSH tunnelling.

6.2. Client

From the client machine, you can run achd to push or pull message to or from the server.

achd {push | pull} {hostname} {chanel_name} [-t tcp|udp] [-p port] [-z remote_channel_name] [-u microseconds] [-d] [-r] [-q] [-v] [-V] [-?]

To add both encryption and compression, you can also forward achd over SSH. This will tunnel the achd TCP connection through the encrypted and compressed SSH connection.

6.3. Locating Channels via mDNS

Multicast DNS (mDNS) can be used to browse the Ach channels on the local network and to lookup the hostname for an channel. mDNS names are generally in the .local domain To advertise an Ach channel, it is registered as a service with the local mDNS server.

mDNS is a peer-to-peer variation on DNS that reduces the need for prior configuration by automatically discovering services on the local network. Each peer runs an mDNS server, and services, e.g., file shares, printers, Ach channels, are located by sending a multicast query. Popular implementations of mDNS are Bonjour and Avahi. Some communication systems, which were generally developed prior to mDNS, use specialized naming services. For example, ONC RPC uses the port mapper and CORBA has its own naming service.

7.1. Features

• Detach process from terminal and run in background (daemonize), -d flag.

• Lock PID files and write PIDs, -P and -p flags.

• Reconnect if connection is broken, -r flag.

• Redirect standard output and standard error to files, -o and -e flags.

• Restart process if SIGHUP is received

7.2. Signals

Signals are how POSIX user programs are notified of asynchrous events. They are similar in spirit to low-level interrupts and interrupt service routines. When a process receives a signal, its execution is interrupted, and it runs the specified handler for that signal. For example, when you hit Control-C at a terminal, the process in the forground receives the interrupt signal, SIGINT. The default handler for this signal will terminate the process. For more details, see man 7 signal.

Achcop will respond to the following signals. You can send a signal with the kill shell command or the kill C function.

7.3. Examples

/* only send signal if running under achcop */
if( my_parent_is_achcop ) {
    int sig;
    /* determine signal to send */
    if( initialized_correctly ) {
        sig = SIGUSR1; /* success */
    } else {
        sig = SIGUSR2; /* failure */
    }
    /* send the signal */
    int r = kill( getppid(), sig );
    /* check that we sent the signal */
    if( r ) {
        perror("Couldn't signal parent");
        exit(EXIT_FAILURE);
    }
}
    

8.1. Log Format

The Log format is similar to the TCP network protocol for achpipe. The log files begins with a sequence of ASCII key-value headers, followed by a single "." character, followed by a sequence of messages.

ACHLOG
channel-name: foo
log-version: 0
log-time-ach: 6226133.801454417
log-time-real: 1371238788.956337009 # Fri Jun 14 15:39:48 2013
local-host: daneel
user: ntd # Neil T. Dantam
.

Messages are framed in the file with the following C struct:

struct {
    uint8_t reserved[8];   /* reserved for future use */
    uint8_t size_bytes[8]; /* size of data, stored little endian */
    uint8_t data[1];       /* flexible array containing size_bytes of data */
};
    

8.2. Usage

9.1. Disable CPU frequency scaling

Many current CPUs will lower their frequency when idle in order to reduce consumption. This is great for extending your laptop's battery life but poor for minimizing latency. On an x86/amd64 PC, you can likely disable frequency scaling from the BIOS, which will reduce your system's latency. In the author's experience, simply changing to a "performance" CPU governer from within the operating system did NOT provide the same latency reduction; you need to disable freqency scaling from the BIOS.

9.2. Use a Real-Time Operating System

General purpose operating systems (e.g. GNU/Linux, FreeBSD) are designed to maximize throughput and perhaps to reduce latency to "human-tolerable" (0.1s) levels. The worst-case performance may be sacrificed in order to improve average-case performance. For a real-time application, it is generally worst-case performance that matters rather than average case, and latency is more important than throughput.

---

9.2.2. Linux / Xenomai

Xenomai is a dual-kernel real-time operating system. A hard real-time kernel, Adeos, handles interrupts and a full Linux kernel runs along-side, permitting normal Linux processes to run directly. Real-time processes must use a specific Xenomai API skin. One of these skins is POSIX, and this POSIX interface is sufficient to use Ach for real-time Xenomai processes.

There is one important caveat when using Ach on Xenomai. Because of the dual-kernel approach, Xenomai scheduling primitives -- mutexes and condition variables -- are different from the Linux primitives; a Linux process and a Xenomai process cannot synchronize on the same condition variable. Therefore, Ach compiled for Xenomai cannot directly share channels with Ach compiled for Linux. If you want a Linux process and a Xenomai process on the same machine to communicate via Ach, you can use achd to forward the messages between the two sides. This issue does not affect messages forwarded between different machines.

9.3. Benchmark your system

You can use the included achbench command to test performance on your system while you try different configurations. This program will fork a specified number of publishers and subscribers and measure the latency of message passing.

10.1. Data Structure

The core data structure of an Ach channel is a pair of circular arrays: a data array and an index array. The data array contains variable sized entries which store the actual message frames sent through the Ach channel. The index array contains fixed size entries where each entry has both an offset into the data array and the length of that data entry. A head offset into each array indicates both the place to insert the next data and the location of the most recent message frame. This pair of circular arrays allows us to find the variable sized message frames by first looking at a known offset in the fixed-sized index array.

Access to the channel is synchronized using a mutex and a queue of waiting readers. This allows readers to either periodically poll the channel for new data or to wait on the condition variable until a writer has posted a new message. Using a read/write lock instead would have allowed only polling. Additionally, synchronization using a mutex prevents starvation and enables proper priority inheritance between processes, important to maintaining real-time performance.

10.2. Core Procedures

Two procedures compose the core of ach: ach_put and ach_get.

10.3. Channel Mappings

Ach supports multiple back-end mappings for the channel data structure: (1) the process-local heap, (2) a shared-memory file, and (3) a kernel-space buffer on Linux.

10.4. Serialization

Ach is orthogonal to the issue of serialization: it only handles raw byte arrays. Thus, you can choose whatever type of serialization is most appropriate for you application. For maximum performance, it may be appropriate to define message types as simple C structures. If more flexibility is required, there are a variety of other options to choose from: XDR, ASN.1, Protocol Buffers, etc.