Non-blocking Unix domain socket

Question

I've developed quickly two kinds of socket use: the first with blocking mode and the second with non-blocking mode. The sockets are Unix domain sockets. My problem is that the kernel consume a huge amount of CPU (approx: 85 %). My goal is to minimize the kernel CPU usage and to increase the throughput.

I use taskset command to affect each process to a particular CPU core.

The blocking mode Unix socket shows performances of approx 1.3 GB/s. The non-blocking mode Unix socket shows performances of approx 170 MB/s.

The blocking version is faster than the non-blocking (+ epoll) version by approximately 8×.

Blocking version:

client.c

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <stdint.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>

#define SOCK_PATH "echo_socket"


typedef struct proto_t {
    uint32_t        len;
    uint8_t        *data;
} proto_t;

int main(void)
{
    int                 s;
    int                 t;
    int                 len;
    struct sockaddr_un  remote;
    char                buffer[1400];
    proto_t            *frame = (proto_t *)buffer;


    if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
        perror("socket");
        exit(1);
    }

    printf("Trying to connect...\n");

    remote.sun_family = AF_UNIX;
    strcpy(remote.sun_path, SOCK_PATH);
    len = strlen(remote.sun_path) + sizeof(remote.sun_family);

    if (connect(s, (struct sockaddr *)&remote, len) == -1) {
        perror("connect");
        exit(1);
    }

    printf("Connected.\n");

    srand(time(NULL));
    for (;;) {
        len = (rand() % (sizeof(buffer) - sizeof(uint32_t))) + sizeof(uint32_t);

        frame->len = htobe32(len - sizeof(uint32_t));
        if (send(s, frame, len, 0) == -1) {
            perror("send");
            close(s);
            exit(1);
        }
    }

    close(s);

    return 0;
}

server.c

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <stdint.h>
#include <time.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>


#define SOCK_PATH "echo_socket"

int main(void)
{
    int                 s, s2, t, len;
    struct sockaddr_un  local, remote;
    char                str[100];
    int                 stat = 0;
    int last_stat = 0;

    if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
        perror("socket");
        exit(1);
    }

    local.sun_family = AF_UNIX;
    strcpy(local.sun_path, SOCK_PATH);
    unlink(local.sun_path);
    len = strlen(local.sun_path) + sizeof(local.sun_family);

    if (bind(s, (struct sockaddr *)&local, len) == -1) {
        perror("bind");
        close(s);
        exit(1);
    }

    if (listen(s, 5) == -1) {
        perror("listen");
        close(s);
        exit(1);
    }

    for(;;) {
        int n;
        printf("Waiting for a connection...\n");
        t = sizeof(remote);
        if ((s2 = accept(s, (struct sockaddr *)&remote, &t)) == -1) {
            perror("accept");
            close(s);
            exit(1);
        }

        printf("Connected.\n");

        do {
            uint8_t  buffer[1400];
            uint32_t frame_len;
            int now = time(NULL);

            n = recv(s2, &frame_len, sizeof(frame_len), 0);
            frame_len = be32toh(frame_len);

            if (n < sizeof(uint32_t)) {
                break;
            }

            n = recv(s2, buffer, frame_len, 0);

            if (frame_len > 0) {
                if (n < frame_len) {
                    close(s);
                    perror("recv");
                    break;
                }
            }


            stat += frame_len + sizeof(uint32_t);

            if (now - last_stat > 1) {
                last_stat = now;
                printf("received %f MB.\n", ((float)stat / 1024 / 1024));
                stat = 0;
            }
        } while (1);

        close(s2);
    }

    return 0;
}

Non-blocking version:

client.c

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <sys/un.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>

int epollfd;
#define SERVERPORT 8080
#define MAXCONN    5
#define MAXEVENTS  100
#define MAXLEN     1400
#define SOCK_PATH  "echo_socket"

typedef struct event_t {
    int      fd;
    uint32_t event;
    char     data[MAXLEN];
    int      length;
    int      offset;
} event_t;

typedef struct proto_t {
    uint32_t        len;
    uint8_t        *data;
} proto_t;

static void event_set(int epollfd, int op, int fd, uint32_t events, void* data)
{
    struct epoll_event server_ev;

    server_ev.events   = events;
    server_ev.data.ptr = data;

    if(-1 == epoll_ctl(epollfd, op, fd, &server_ev)) {
        printf("Failed to add an event for socket%d Error:%s\n",
               fd, strerror(errno));
        exit(1);
    }

}

static void event_handle(void* ptr)
{
    event_t  *ev = ptr;

    if(EPOLLIN == ev->event) {
        return;
    } else
    if(EPOLLOUT == ev->event) {
        int      ret;
        proto_t *frame = (proto_t *)ev->data;

        if (ev->length == 0) {
            /* init send */
            ev->length = (rand() % (sizeof(ev->data)
                                   - sizeof(uint32_t))
                                   + sizeof(uint32_t));

            frame->len = htobe32(ev->length - sizeof(uint32_t));
        }


        ret = write(ev->fd, (ev->data) + (ev->offset), ev->length);
        if( (ret < 0 && EINTR == errno) || ret <= ev->length) {
            /*
             * We either got EINTR or write only sent partial data.
             * Add an write event. We still need to write data.
             */

            if(ret > 0) {
               /*
               * The previous write wrote only partial data to the socket.
               */
                ev->length = ev->length - ret;
                ev->offset = ev->offset + ret;
            }

            if (ev->length == 0) {
                /* write complete */
                ev->offset = 0;
                ev->length = 0;
            }
        } else
        if(ret < 0) {
            /*
             * Some other error occured.
             */
            printf("ERROR: ret < 0");
            close(ev->fd);
            free(ev);
            return;
        }
        if (ret == 0) {
            printf("------------\n");
        }
        event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLOUT, ev);
    }
}


static void socket_set_non_blocking(int fd)
{
    int flags;

    flags = fcntl(fd, F_GETFL, NULL);

    if (flags < 0) {
        printf("fcntl F_GETFL failed.%s", strerror(errno));
        exit(1);
    }

    flags |= O_NONBLOCK;

    if (fcntl(fd, F_SETFL, flags) < 0) {
        printf("fcntl F_SETFL failed.%s", strerror(errno));
        exit(1);
    }
}


int main(int argc, char** argv)
{
    int                 clientfd;
    int                 len    = 0;
    struct sockaddr_un  remote;
    struct epoll_event *events = NULL;
    event_t             ev;

   /*
    * Create server socket. Specify the nonblocking socket option.
    *
    */
    clientfd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);

    if(clientfd < 0)
    {
        printf("Failed to create socket.%s\n", strerror(errno));
        exit(1);
    }

    bzero(&remote, sizeof(remote));

    remote.sun_family = AF_UNIX;
    strcpy(remote.sun_path, SOCK_PATH);
    len = strlen(remote.sun_path) + sizeof(remote.sun_family);

   /*
    * connect to the server
    *
    */
    if (connect(clientfd, (struct sockaddr *)&remote, len) == -1) {
        perror("connect");
        exit(1);
    }

    printf("Connected.\n");

   /*
    * Create epoll context.
    */
    epollfd = epoll_create(1);

    if(epollfd < 0)
    {
        printf("Failed to create epoll context.%s\n", strerror(errno));
        exit(1);
    }

    /*
     * Main loop that listens for event.
     */
    events = calloc(MAXEVENTS, sizeof(struct epoll_event));
    bzero(&ev, sizeof(ev));
    ev.fd = clientfd;
    event_set(epollfd, EPOLL_CTL_ADD, clientfd, EPOLLOUT, &ev);

   while(1) {
        int n = epoll_wait(epollfd, events, MAXEVENTS, -1);

        if(n < 0) {
            printf("Failed to wait.%s\n", strerror(errno));
            exit(1);
        }

        for(int i = 0; i < n; i++) {
            event_t *event = (event_t *)events[i].data.ptr;

            if(events[i].events & EPOLLHUP || events[i].events & EPOLLERR) {
                printf("\nClosing connection socket\n");
                close(event->fd);
            } else
            if(EPOLLIN == events[i].events) {
                event->event = EPOLLOUT;
                event_set(epollfd, EPOLL_CTL_DEL, event->fd, 0, 0);
                printf("ERROR, Cannot receive data\n");
                //event_handle(ev);
            } else
            if(EPOLLOUT == events[i].events) {
                event->event = EPOLLOUT;
                /*
                 * Delete the write event.
                 */
                event_set(epollfd, EPOLL_CTL_DEL, event->fd, 0, 0);
                event_handle(event);
            }
        }
   }

    free(events);
    exit(0);
}

server.c

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <sys/un.h>
#include <unistd.h>
#include <fcntl.h>
#include <time.h>

int epollfd;
#define SERVERPORT 8080
#define MAXCONN    5
#define MAXEVENTS  100
#define MAXLEN     1400
#define SOCK_PATH  "echo_socket"

typedef struct event_t {
    int      fd;
    uint32_t event;
    char     data[MAXLEN];
    int      length;
    int      offset;
} event_t;

static void event_set(int epollfd, int op, int fd, uint32_t events, void* data)
{
    struct epoll_event server_ev;

    server_ev.events   = events;
    server_ev.data.ptr = data;

    if(-1 == epoll_ctl(epollfd, op, fd, &server_ev)) {
        printf("Failed to add an event for socket%d Error:%s\n",
               fd, strerror(errno));
        exit(1);
    }

}

static int stat;
static int last_stat;

static void event_handle(void* ptr)
{
    event_t  *ev = ptr;
    uint32_t *len = (uint32_t *)ev->data;

    if(EPOLLIN == ev->event) {
        uint32_t buffer[1400];
        int      n  = 0;
        uint32_t blen;

        if (ev->length < sizeof(uint32_t)) {
            n = read(ev->fd, ev->data + ev->length, sizeof(uint32_t));
        } else {
            uint32_t blen = be32toh(*len);

            if (blen > 0) {
                n = read(ev->fd, ev->data + ev->length,
                         blen - (ev->length - sizeof(uint32_t)));
            } else {
                ev->length = 0;
                return;
            }
        }

        if(n == 0) {
            /*
             * Client closed connection.
             */
            printf("\nClient closed connection.\n");
            close(ev->fd);
            free(ev);
        } else
        if(n < 0) {
            perror("read from socket");
            close(ev->fd);
            free(ev);
        } else {
            int now = time(NULL);

            blen = be32toh(*len);
            ev->length += n;

            if (ev->length >= sizeof(uint32_t)
            &&  blen == ev->length - sizeof(uint32_t))
            {
                /* data complete */
                stat += ev->length;
                ev->length = 0;
            }

            if (now - last_stat >= 1) {
                printf("received %f MB.\n", ((float)stat / 1024 / 1024));
                last_stat = now;
                stat      = 0;
            }

            /*
             * We have read the data. Add an write event so that we can
             * write data whenever the socket is ready to be written.
             */
            /*
            printf("\nAdding write event.\n");
            event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLOUT, ev);
            */
            event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLIN, ev);
        }
    } else
    if(EPOLLOUT == ev->event) {
        int ret;

        printf("EPOLLOUT not handled yet \n");
        return;

        ret = write(ev->fd, (ev->data) + (ev->offset), ev->length);

        if( (ret < 0 && EINTR == errno) || ret < ev->length) {
            /*
             * We either got EINTR or write only sent partial data.
             * Add an write event. We still need to write data.
             */

            event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLOUT, ev);

            if(-1 != ret) {
               /*
               * The previous write wrote only partial data to the socket.
               */
                ev->length = ev->length - ret;
                ev->offset = ev->offset + ret;
            }
        } else
        if(ret < 0) {
            /*
             * Some other error occured.
             */
            close(ev->fd);
            free(ev);
            return;
        } else {

          /*
           * The entire data was written. Add an read event,
           * to read more data from the socket.
           */
            printf("\nAdding Read Event.\n");
            event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLIN, ev);
        }
    }
}


static void socket_set_non_blocking(int fd)
{
    int flags;

    flags = fcntl(fd, F_GETFL, NULL);

    if (flags < 0) {
        printf("fcntl F_GETFL failed.%s", strerror(errno));
        exit(1);
    }

    flags |= O_NONBLOCK;

    if (fcntl(fd, F_SETFL, flags) < 0) {
        printf("fcntl F_SETFL failed.%s", strerror(errno));
        exit(1);
    }
}


int main(int argc, char** argv)
{
    int                 serverfd;
    int                 len    = 0;
    struct sockaddr_un  local;
    struct sockaddr_un  remote;
    struct epoll_event *events = NULL;

   /*
    * Create server socket. Specify the nonblocking socket option.
    *
    */
    serverfd = socket(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0);

    if(serverfd < 0)
    {
        printf("Failed to create socket.%s\n", strerror(errno));
        exit(1);
    }

    bzero(&local, sizeof(local));

    local.sun_family = AF_UNIX;
    strcpy(local.sun_path, SOCK_PATH);
    unlink(local.sun_path);
    len = strlen(local.sun_path) + sizeof(local.sun_family);

   /*
    * Bind the server socket to the required ip-address and port.
    *
    */
    if(bind(serverfd, (struct sockaddr*)&local, len) < 0)
    {
        printf("Failed to bind.%s\n", strerror(errno));
        close(serverfd);
        exit(1);
    }

   /*
    * Mark the server socket has a socket that will be used to .
    * accept incoming connections.
    */
    if(listen(serverfd, MAXCONN) < 0)
    {
        printf("Failed to listen.%s", strerror(errno));
        exit(1);
    }

   /*
    * Create epoll context.
    */
    epollfd = epoll_create(MAXCONN);

    if(epollfd < 0)
    {
        printf("Failed to create epoll context.%s\n", strerror(errno));
        exit(1);
    }

   /*
    * Create read event for server socket.
    */
   event_set(epollfd, EPOLL_CTL_ADD, serverfd, EPOLLIN, &serverfd);

   /*
    * Main loop that listens for event.
    */
   events = calloc(MAXEVENTS, sizeof(struct epoll_event));

   while(1) {
       int n = epoll_wait(epollfd, events, MAXEVENTS, -1);

       if(n < 0) {
           printf("Failed to wait.%s\n", strerror(errno));
           exit(1);
       }

       for(int i = 0; i < n; i++) {
           if(events[i].data.ptr == &serverfd) {
               int connfd;

               if(events[i].events & EPOLLHUP || events[i].events & EPOLLERR) {
                   /*
                    * EPOLLHUP and EPOLLERR are always monitored.
                    */
                   close(serverfd);
                   exit(1);
               }

               /*
                * New client connection is available. Call accept.
                * Make connection socket non blocking.
                * Add read event for the connection socket.
                */
               len = sizeof(remote);
               connfd = accept(serverfd, (struct sockaddr*)&remote, &len);

               if(-1 == connfd) {
                   printf("Accept failed.%s\n", strerror(errno));
                   close(serverfd);
                   exit(1);
               } else {
                   event_t *ev = NULL;

                   socket_set_non_blocking(connfd);

                   printf("Adding a read event\n");
                   ev     = calloc(1, sizeof(event_t));
                   ev->fd = connfd;

                   /*
                    * Add a read event.
                    */
                   event_set(epollfd, EPOLL_CTL_ADD, ev->fd, EPOLLIN, ev);
               }
           } else {
               /*
                *A event has happend for one of the connection sockets.
                *Remove the connection socket from the epoll context.
                * When the event is handled by event_handle() function,
                *it will add the required event to listen for this
                *connection socket again to epoll
                *context
                */

                if(events[i].events & EPOLLHUP || events[i].events & EPOLLERR) {
                    event_t *ev = (event_t *) events[i].data.ptr;

                    printf("\nClosing connection socket\n");
                    close(ev->fd);
                    free(ev);
                } else
                if(EPOLLIN == events[i].events) {
                   event_t* ev = (event_t *) events[i].data.ptr;

                   ev->event = EPOLLIN;
                   /*
                    * Delete the read event.
                    */
                   event_set(epollfd, EPOLL_CTL_DEL, ev->fd, 0, 0);
                   event_handle(ev);
               }
               else if(EPOLLOUT == events[i].events)
               {
                   event_t* ev = (event_t*) events[i].data.ptr;

                   ev->event = EPOLLOUT;

                   /*
                    * Delete the write event.
                    */
                   event_set(epollfd, EPOLL_CTL_DEL, ev->fd, 0, 0);
                   event_handle(ev);
               }
           }

       }
   }

    free(events);
    exit(0);
}

According to what I've read on the Internet, non-blocking mode should be faster than blocking mode. Why am I observing the reverse performance?

• Is there a way to increase the throughput more than 1.3 GB/s?
• Is there a way to minimize the kernel CPU usage?

Programs were compiled using:

gcc -std=gnu99 -O3 {file}.c {bin-name}

I use htop + perf to measure performance.

Answer 1

I think you need to rethink your assumptions. For starters it's not obvious at all that non-blocking would be faster than blocking. The big difference between your two programs is that the non-blocking one can handle multiple clients and your blocking one can't. So it's reasonable to suggest that the support might cost some performance. If you start multiple clients with the non-blocking one, the total throughput may well be higher.

Now as to your non-blocking code, it seems to be somewhat inefficient. For example you spend a lot of time adding and removing FDs from your epoll() file descriptor. That shouldn't be necessary in a properly coded program. When the descriptor is readable you read it and then go back to epoll(). Add the descriptor once, then leave it there.

Secondly you spend time working out exactly how many bytes to read for the framing. Don't do that. Have a buffer of say 64KB where you read the bytes into and then parse the framing out of that. Look up buffering.

An important tip for network performance is to think about how many syscalls are you doing per message? ISTM you're doing at least five but with the parameters you have you should be able to write a version doing a maximum of two syscalls per packet on average (the epoll and a single read) and less if there are lots of packets (read more than one packet per read()). You know the maximum frame size upfront, which simplifies coding considerably.

Also, debug print statements cost performance.

Answer 2

I'll breakdown the question in two:

1. Blocking sockets getting faster than nonblocking sockets:

   This is my implementation and I've got very different results from yours. Please note that there are two binaries generated for asyncserver/client. The "w" versions do idle wait and the "plain async" do busy wait.

   Here are the test results on my PC:

   asyncserverw/syncserverw: 3008000000 bytes xferred in 6.26287 avg 4.80291e+08 (480MB/s)
   syncserver/syncclient: 3008000000 bytes xferred in 4.97424 avg 6.04715e+08 (604MB/s)
   asyncserver/asyncclient: 3008000000 bytes xferred in 2.70924 avg 1.11027e+09 (1.11GB/s)
   

   As pointed in the previous answer, there is some cost to do asynchronous I/O, as shown in the idle wait results (blocking vs "w"). The busy wait boosted the throughput to almost double and you can share the CPU among several clients at once.

2. Lower CPU usage while keeping the throughput:

   As already pointed out, you don't need to do more than one recv() to get a packet. You can either receive a large buffer and parse all the packets in there or you can do like I did: use SOCK_SEQPACKET. It's a connection oriented socket that can preserve packet boundaries. So, you one write = one packet and when you call recv(), you'll receive exactly one packet if you provide a buffer that's large enough.

Reference code:

Makefile:

CCFLAGS=-std=c++11 -O3

all: sync async

sync: syncclient syncserver

async: asyncclient asyncserver


syncclient: syncclient.cc
    g++ $(CCFLAGS) -o syncclient syncclient.cc


syncserver: syncserver.cc
    g++ $(CCFLAGS) -o syncserver syncserver.cc

asyncclient: asyncclient.cc
    g++ $(CCFLAGS) -o asyncclient asyncclient.cc
    g++ $(CCFLAGS) -DWAIT_TIME=-1 -o asyncclientw asyncclient.cc


asyncserver: asyncserver.cc
    g++ $(CCFLAGS) -o asyncserver asyncserver.cc
    g++ $(CCFLAGS) -DWAIT_TIME=-1 -o asyncserverw asyncserver.cc

syncserver.cc

#include <iostream>
#include <cstddef>
#include <cerrno>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include "message.hh"

int main()
{
    Message message;
    int fd = socket(AF_UNIX, SOCK_SEQPACKET, 0);
    int cfd = -1;
    sockaddr_un addr = {.sun_family=AF_UNIX};
    memcpy(addr.sun_path, address, address_len);
    if (!bind(fd, reinterpret_cast<sockaddr*>(&addr), offsetof(sockaddr_un, sun_path) + address_len)
        && !listen(fd, 1)
        && (cfd = accept(fd, nullptr, nullptr)) > -1)
    {
        int recvd = 1;
        while (recvd > 0)
        {
            recvd = recv(cfd, &message, sizeof message, 0);
            send(cfd, &message, recvd, MSG_NOSIGNAL);
        }
        std::cout << recvd << ' ' << errno << std::endl;
    }
    close(cfd);
    close(fd);
}

syncclient.cc

#include <iostream>
#include <chrono>
#include <cstddef>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include "message.hh"

int main()
{
    Message message;
    int fd = socket(AF_UNIX, SOCK_SEQPACKET, 0);
    sockaddr_un addr = {.sun_family=AF_UNIX};
    memcpy(addr.sun_path, address, address_len);
    if (!connect(fd, reinterpret_cast<sockaddr*>(&addr), offsetof(sockaddr_un, sun_path) + address_len))
    {
        size_t sent = 0;
        size_t recvd = 0;
        auto begin = std::chrono::steady_clock::now();
        for (size_t i=0; i<attempts; ++i)
        {
            message.count = i + 1;
            sent += send(fd, &message, sizeof message, MSG_NOSIGNAL);
            recvd += recv(fd, &message, sizeof message, 0);
        }
        auto end = std::chrono::steady_clock::now();
        sent += recvd;
        std::chrono::duration<double> deltat = end - begin;
        std::cout << sent << " bytes xferred in " << deltat.count() << " avg " << (sent / deltat.count()) << std::endl;
    }
    close(fd);
}

asyncserver.cc

#include <iostream>
#include <cstddef>
#include <cerrno>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/epoll.h>
#include <unistd.h>
#include "message.hh"

#ifndef WAIT_TIME
#define WAIT_TIME 0
#endif

int main()
{
    Message message;
    int fd = socket(AF_UNIX, SOCK_SEQPACKET, 0);
    int cfd = -1;
    int efd = epoll_create1(0);
    epoll_event events[0x100];
    sockaddr_un addr = {.sun_family=AF_UNIX};
    memcpy(addr.sun_path, address, address_len);
    if (!bind(fd, reinterpret_cast<sockaddr*>(&addr), offsetof(sockaddr_un, sun_path) + address_len)
        && !listen(fd, 1)
        && (cfd = accept4(fd, nullptr, nullptr, SOCK_NONBLOCK)) > -1)
    {
        auto & event = events[0];
        event.events = EPOLLIN | EPOLLET;
        epoll_ctl(efd, EPOLL_CTL_ADD, cfd, &event);
        int recvd = 1;
        int r = 1;
        int n;
        while (recvd > 0)
        {
            do
            {
                n = epoll_wait(efd, events, 0x100, WAIT_TIME);
            } while (!n);
            for (int e=0; e<n; ++e)
            {
                recvd = recv(cfd, &message, sizeof message, 0);
                if (recvd > 0)
                {
                    send(cfd, &message, recvd, MSG_NOSIGNAL);
                    --e;
                }
                else if (recvd < 0 && errno == EAGAIN)
                    recvd = 1;
            }
        }
    }
    close(cfd);
    close(fd);
}

asyncclient.cc

#include <iostream>
#include <chrono>
#include <cstddef>
#include <cerrno>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <sys/epoll.h>
#include "message.hh"

#ifndef WAIT_TIME
#define WAIT_TIME 0
#endif

int main()
{
    Message message;
    int fd = socket(AF_UNIX, SOCK_SEQPACKET | SOCK_NONBLOCK, 0);
    int efd = epoll_create1(0);
    epoll_event events[0x100];
    auto & event = events[0];
    event.data.u32 = 0;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl(efd, EPOLL_CTL_ADD, fd, &event);

    sockaddr_un addr = {.sun_family=AF_UNIX};
    memcpy(addr.sun_path, address, address_len);
    int n;
    if (!connect(fd, reinterpret_cast<sockaddr*>(&addr), offsetof(sockaddr_un, sun_path) + address_len))
    {
        size_t sent = 0;
        size_t recvd = 0;
        auto begin = std::chrono::steady_clock::now();
        for (size_t i=0; i<attempts; ++i)
        {
            message.count = i + 1;
            sent += send(fd, &message, sizeof message, MSG_NOSIGNAL);
            //std::cout << "s:" << sent << ' ' << errno << std::endl;
            do
            {
                n = epoll_wait(efd, events, 0x100, WAIT_TIME);
            } while (!n);
            for (int e=0; e<n; ++e)
            {
                int recvd2 = recv(fd, &message, sizeof message, 0);
                //std::cout << recvd << ' ' << recvd2 << ' ' << errno << std::endl;
                if (recvd2 > 0)
                {
                    recvd += recvd2;
                    --e;
                }
            }
        }
        auto end = std::chrono::steady_clock::now();
        sent += recvd;
        std::chrono::duration<double> deltat = end - begin;
        std::cout << sent << " bytes xferred in " << deltat.count() << " avg " << (sent / deltat.count()) << std::endl;
    }
    else
        std::cout << "connect failed" << std::endl;
    close(fd);
    close(efd);
}

message.hh

#ifndef MESSAGE_HH_
#define MESSAGE_HH_

#include <sys/socket.h>

struct Message
{
    size_t count;
    char bytes[1492];
};

size_t attempts = 1000000;

constexpr 
const char address[]="\0echo_server";
const socklen_t address_len = sizeof address;

#endif