POSIX message queue latency benchmarking

mq_lat.c

#define _GNU_SOURCE
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>
#include <stdio.h>
#include <time.h>
#include <signal.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
#include <sched.h>

#define MQ_NAME "/mq_bench"
#define NR_REC 2000000

/* per mq_overview(7), the minimum is 128, whereas the ceiling is
 * HARD_MSGSIZEMAX for recent kernels.
 */
#define MSG_SIZE 128

volatile int should_run = 1;
volatile size_t nr_msg;
struct timespec start;
char buf[MSG_SIZE];

struct {
    struct timespec start;
} __attribute__ ((aligned(32))) payload;

void* sender_wrk(void *data)
{
    int fd = mq_open(MQ_NAME, O_WRONLY | O_NONBLOCK);
    if (fd < 0)
        perror("sender: mq_open failed");

    while(should_run) {
        clock_gettime(CLOCK_MONOTONIC, &payload.start);

        mq_send(fd, (char*) &payload, sizeof(payload), 0);
    }

    mq_close(fd);
    pthread_exit(NULL);
}

void sigalrm_handler(int sig)
{
    should_run = 0;
}

void sigint_handler(int sig)
{
    should_run = 0;
}

int main(int ac, char **av)
{
    pthread_t sender;
    cpu_set_t cpu_affi;
    int has_affi = 1;
    int idx = 0;

    if (ac != 3)
        has_affi = 0;

    // default values, ignored for root processes, but there still exists a hard limit
    struct mq_attr mq_attr = {.mq_maxmsg = 10, .mq_msgsize = MSG_SIZE};

    if (has_affi) {
        CPU_ZERO(&cpu_affi);
        CPU_SET(0, &cpu_affi); // reveiver has affinity of CPU 0
        if (sched_setaffinity(getpid(), sizeof(cpu_set_t), &cpu_affi)) {
            perror("Error setting CPU affinity");
            exit(-1);
        }
    }

    signal(SIGINT, sigint_handler);
    signal(SIGALRM, sigalrm_handler);

    if (ac > 1) {
        alarm(strtol(av[1], NULL, 10));
    }

    mqd_t fd = mq_open(MQ_NAME, O_RDONLY | O_CREAT, 0644, &mq_attr);
    if (fd < 0) {
        perror("receiver: mq_open failed");
        exit(-1);
    }
    if (mq_getattr(fd, &mq_attr)) {
        perror("mq_getattr failed");
        exit(-1);
    }
    printf("mq_msgsize: %ld\nmq_curmsgs: %ld\nmq_maxmsg: %ld\n",
           mq_attr.mq_msgsize, mq_attr.mq_curmsgs, mq_attr.mq_maxmsg);

    if (pthread_create(&sender, NULL, sender_wrk, NULL)) {
        puts("Failed creating sender thread");
        exit(-1);
    }
    if (has_affi) {
        CPU_ZERO(&cpu_affi);
        CPU_SET(1, &cpu_affi);
        if (pthread_setaffinity_np(sender, sizeof(cpu_set_t), &cpu_affi)) {
            perror("Error setting CPU affinity");
            exit(-1);
        }
    }

    printf("payload sz: %u\n", sizeof(payload));

    while(should_run) {
        ssize_t nr_recv;
        struct timespec now;

        mq_receive(fd, buf, sizeof(buf), NULL);

        clock_gettime(CLOCK_MONOTONIC, &now);
        if (idx++ < NR_REC) {
          fprintf(stderr,"%ld\n", ((now.tv_sec * 1000000000) + now.tv_nsec) -
                  ((((typeof(payload)*)buf)->start.tv_sec * 1000000000) +
                  ((typeof(payload)*)buf)->start.tv_nsec));
        } else
            break;
    }

    if (mq_close(fd)) {
        perror("mq close failed");
        exit(-1);
    }
    if (mq_unlink(MQ_NAME)) {
        perror("mq unlink failed");
        exit(-1);
    }
    return 0;
}