Multi-threaded application performance in C

Question

I'm programming a multithreaded application (server-client UNIX socket) implemented for the server-side. I have two thread check_cmd_thr and server_thr.

I use check_cmd_thr to check data input from keyboard, if the input is start then the server is allowed to receive messages, if the input is stop then the server is not allowed to receive messages.

The purpose of this code is show the reliability of the UNIX datagram socket.

The code is working well 90%, it's just about the performance of the code.

• When the user sends stop, the server doesn't stop receiving messages immediately.
• After the server stop receiving message, the client still sends successfully 11 messages before stopping to wait for the server. The point is, at that moment, the server doesn't show those 11 messages to the console log.
• When the user sends start, the server "pop up" those 11 messages and continues receiving new messages from the client.

What I expect is

• When the user sends stop, the server stop receiving messages immediately, and the client can not sends any messages to the server after that.

I think that the design of the code implemented for the server-side is not good. Please show me some suggestions to improve. Thank you.

---

Server's code

#include "urd_hdr.h"
#include <pthread.h>

#define SERVER_LOCK()       pthread_mutex_lock(&mtx)
#define SERVER_UNLOCK()     pthread_mutex_unlock(&mtx);

static u_int8_t rx_is_allowed = 0;
static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;

static int init_server(char *svr_path);
static void *check_cmd_thr(void *arg);
static void *server_thr(void *arg);

/**
 * @brief Initialize server
 * @param svr_path socket pathname of server
 * @return Bound socket of server, -1 for errors
*/
static int init_server(char *svr_path)
{
    SOCK_ADDR_UN_t  svr_addr;
    int             sfd;

    // Create socket:
    sfd = socket(AF_UNIX, SOCK_DGRAM, 0);
    if (sfd == -1) {
        printf("Error line[%d]: %s\n", __LINE__, strerror(errno));
        return -1;
    }

    // Verify socket's namepath:
    if (strlen(svr_path) > (sizeof(svr_addr.sun_path) - 1)) {
        printf("Error line[%d]: invalid socket name\n", __LINE__);
        return -1;
    }

    // Remove any exitsting file having the same namepath with server's:
    if ((remove(svr_path) == -1) && (errno != ENOENT)) {
        printf("Error line[%d]: %s\n", __LINE__, strerror(errno));
        return -1;
    }

    // Bind the socket to desired addr:
    memset(&svr_addr, 0, sizeof(SOCK_ADDR_UN_t));
    svr_addr.sun_family = AF_UNIX;
    strncpy(svr_addr.sun_path, svr_path, sizeof(svr_addr.sun_path) - 1);

    if (bind(sfd, (SOCK_ADDR_t *)&svr_addr, sizeof(SOCK_ADDR_UN_t)) == -1) {
        printf("Error line[%d]: %s\n", __LINE__, strerror(errno));
        return -1;
    }

    return sfd;
}

static void *check_cmd_thr(void *arg)
{
    char in_buf[10];
    int8_t len;

    printf ("Start %s\n", __func__);
    while (1) {
        len = read(STDIN_FILENO, in_buf, sizeof(in_buf)); 
        if (len == -1) {
            printf("Error line[%d]: fails to read from stdin, %s\n", __LINE__, strerror(errno));
            continue;
        }

        if (len > sizeof(in_buf)) {
            printf("The length of input buffer is too long\n");
            continue;
        }

        in_buf[len - 1] = 0x0;

        SERVER_LOCK();
        if (strcmp(in_buf, "start") == 0) {
            printf("=> Start receiving packets\n");
            rx_is_allowed = 1;
        } else if (strcmp(in_buf, "stop") == 0) {
            printf("=> Stop receiving packets\n");
            rx_is_allowed = 0;
        } else {
            printf("Invalid input buffer\n");
        }

        SERVER_UNLOCK();
    }

    return 0;
}

static void *server_thr(void *arg)
{
    int16_t     sfd;
    int8_t      buf[SOCKET_BUF_SIZE];
    ssize_t     num_rw;

    printf ("Start %s\n", __func__);

    sfd = init_server(SOCKET_SV_PATH);
    if (sfd == -1) {
        printf("Failed to initialize server\n");
    } else {
        printf("Initialize server successfully\n");
    }

    while (1) {
        SERVER_LOCK();

        if (rx_is_allowed) {    
            // Receive data from client:
            memset(buf, 0, sizeof(buf));
            num_rw = recvfrom(sfd, buf, SOCKET_BUF_SIZE, 0, NULL, NULL);

            if (num_rw == -1) {
                printf("Error line[%d]: %s\n", __LINE__, strerror(errno));
                SERVER_UNLOCK();
                continue;
            }
            
            printf("Server received %ld bytes, ret_buf=[%.*s]\n", (long)num_rw, (int)num_rw, buf);
        }

        SERVER_UNLOCK();
        usleep(100);
    }

    return 0;
}

int main(int argc, char **argv)
{
    pthread_t tid_1, tid_2;
    int ret;

    rx_is_allowed = 1;

    ret = pthread_create(&tid_1, NULL, server_thr, NULL);
    if (ret != 0) {
        printf("Fail to create server_thr\n");
    } else {
        printf("Create server_thr succesfully\n");
    }

    ret = pthread_create(&tid_2, NULL, check_cmd_thr, NULL);
    if (ret != 0) {
        printf("Fail to create check_cmd_thr\n");
    } else {
        printf("Create check_cmd_thr succesfully\n");
    }

    ret = pthread_join(tid_1, NULL);
    if (ret != 0) {
        printf("Fail to join server_thr\n");
    } else {
        printf("Join server_thr succesfully\n");
    }

    ret = pthread_join(tid_2, NULL);
    if (ret != 0) {
        printf("Fail to join check_cmd_thr\n");
    } else {
        printf("Join check_cmd_thr succesfully\n");
    }

    return 0;
}

---

Header file

#ifndef _URD_HDR_H_
#define _URD_HDR_H_

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>

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

#include <errno.h>
#include <unistd.h>

#define SOCKET_SV_PATH      "/tmp/dxduc/ud_reliable"
#define SOCKET_BUF_SIZE     20

typedef struct sockaddr     SOCK_ADDR_t;
typedef struct sockaddr_un  SOCK_ADDR_UN_t;
typedef socklen_t           SOCKLEN_t;

#endif

---

Client's code

#include "urd_hdr.h"

#define MAX_MSG_SIZE     200

static void send_msg(int sd, const SOCK_ADDR_UN_t *addr, SOCKLEN_t addr_len);

int main(int argc, char **argv)
{
    SOCK_ADDR_UN_t      svr_addr;
    int                 sfd;
    SOCKLEN_t           svr_addr_len;
    
    // Create socket:
    sfd = socket(AF_UNIX, SOCK_DGRAM, 0);
    if (sfd == -1) {
        printf("%s, %s-line %d\n", strerror(errno), __func__, __LINE__);
        exit(EXIT_FAILURE);
    }

    // Construct server's address:
    memset(&svr_addr, 0, sizeof(SOCK_ADDR_UN_t));
    svr_addr.sun_family = AF_UNIX;
    strncpy(svr_addr.sun_path, SOCKET_SV_PATH, sizeof(svr_addr.sun_path) - 1);
    svr_addr_len = sizeof(SOCK_ADDR_UN_t);

    printf("Client is ready to run\n");

    while (1) {
        send_msg(sfd, &svr_addr, svr_addr_len);

        sleep(1);
    }
}

static void send_msg(int sd, const SOCK_ADDR_UN_t *addr, SOCKLEN_t addr_len)
{
    static int      idx = 0;
    static char     str[] = "message ";
    static char     msg[MAX_MSG_SIZE];

    // Format message before sending:
    memset(&msg, 0, sizeof(msg));
    sprintf(msg, "%s%d", str, idx++);

    // Send msg to server:
    if (sendto(sd, msg, strlen(msg), 0, (SOCK_ADDR_t *)addr, addr_len) != strlen(msg)) {
        printf("%s, %s-line %d\n", strerror(errno), __func__, __LINE__);
        exit(EXIT_FAILURE);
    } else {
        printf("Message is sent successfully: [%s]\n", msg);
    }
}

Answer 1

First of all, get rid of your secret macro language so that other programmers who don't know your what your secret names mean can read the code. This means replace the mysterious, unreadable SERVER_LOCK with the super-readable pthread_mutex_lock(&mtx) and so on.

Second, grabbing a mutex and holding on to it while the program does I/O or unrelated string comparisons etc is very bad for performance. You do this all over, so it's a massive performance killer.

What you should do is this:

    if (strcmp(in_buf, "start") == 0) {
        printf("=> Start receiving packets\n");

        pthread_mutex_lock(&mtx)
          rx_is_allowed = 1;
        pthread_mutex_unlock(&mtx);
    } 
    else if (strcmp(in_buf, "stop") == 0) {
        printf("=> Stop receiving packets\n");
        
        pthread_mutex_lock(&mtx)
          rx_is_allowed = 0;
        pthread_mutex_unlock(&mtx);
    } 
    else {
        printf("Invalid input buffer\n");
    }

That is, hold on to the mutex as short a time as possible and only for the purpose of protecting variables shared between threads. In this case only when you modify the rx_is_allowed variable (or any such variable shared between threads). Fix this and I'm pretty certain all the mentioned problems will disappear.

Answer 2

#define SERVER_LOCK()       pthread_mutex_lock(&mtx)
#define SERVER_UNLOCK()     pthread_mutex_unlock(&mtx);

I'm not sure why you're defining macros for these simple calls (it mainly just serves to obfuscate your code), but that second one (with a ;) is liable to give surprises when it's expanded.

static u_int8_t rx_is_allowed = 0;

I don't see where that type gets defined. Did you mean uint8_t (from <stdint.h>)? For a variable with _is_ in the name like that, I would expect to see a bool.

typedef struct sockaddr     SOCK_ADDR_t;
typedef struct sockaddr_un  SOCK_ADDR_UN_t;
typedef socklen_t           SOCKLEN_t;

Careful there - POSIX lays claim to type names ending in _t! Again, I don't see any value in these definitions, compared to using the type name that's better known to your readers.

sfd = socket(AF_UNIX, SOCK_DGRAM, 0);
if (sfd == -1) {
    printf("Error line[%d]: %s\n", __LINE__, strerror(errno));
    return -1;
}

Good! I like to see errors properly checked. To make this even better, write the message to stderr rather than stdout (consider perror() for that), and return a positive value, preferably EXIT_FAILURE.

memset(&svr_addr, 0, sizeof(SOCK_ADDR_UN_t));

While not wrong, the correctness of this would be more obvious if we used sizeof svr_addr, so we didn't need to check if the type name agrees with the variable.

---

Sorry, that's as far as I got before I had to yield to other demands on my time. I hope the partial review is useful.