ex_accept_pool_server.c

Accept pool example: three modes of accept + client handling Usage: ./ex_accept_pool_server -p PORT -m single-inline -n 500 ./ex_accept_pool_server -p PORT -m single-pool -n 500 -t 4 ./ex_accept_pool_server -p PORT -m pooled -n 500 -t 4

Modes: single-inline - one thread doing accept + handle client in the same thread. Both accept and handling are fully serialized. Best for: low-traffic servers, simple protocols, debugging.

single-pool - one thread doing accept, dispatching to a thread pool. Accept is serialized, handling is parallelized. Best for: most servers where per-connection work dominates.

pooled - N threads calling accept in parallel (nginx-style), dispatching to a thread pool for client handling. Both accept and handling are parallelized. Best for: high connection rates over real networks, SSL/TLS servers, multi-socket NUMA systems.

Performance notes: On localhost, accept() is near-instant. A single accept thread can drain the kernel queue faster than clients fill it, so single-pool often outperforms pooled mode due to less thread contention. The pooled accept advantage appears over real networks, with SSL handshakes, or at very high connection rates (10k+ conn/sec) where the kernel accept lock becomes the bottleneck.

See the ACCEPT_POOL doxygen module documentation for detailed guidance on choosing the right architecture for your use case. ```c

/*

• Nilorea Library
• Copyright (C) 2005-2026 Castagnier Mickael
• Licensed under the Apache License, Version 2.0 (the "License");
• you may not use this file except in compliance with the License.
• You may obtain a copy of the License at

• http://www.apache.org/licenses/LICENSE-2.0
  

• Unless required by applicable law or agreed to in writing, software
• distributed under the License is distributed on an "AS IS" BASIS,
• WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
• implied. See the License for the specific language governing
• permissions and limitations under the License.
• SPDX-License-Identifier: Apache-2.0 */

#include <stdio.h> #include <stdlib.h> #include <string.h> #include <errno.h> #include <getopt.h> #include <signal.h>

#include "nilorea/n_common.h" #include "nilorea/n_log.h" #include "nilorea/n_network.h" #include "nilorea/n_network_accept_pool.h" #include "nilorea/n_thread_pool.h"

static volatile int server_running = 1; static int total_target = 500; /* Seconds to wait for progress before giving up, 0 disables the deadline.

• Without it a server whose client never showed up sat in its accept loop
• forever and only died when the test harness killed it on its global
• timeout, which surfaced as a bare exit 124 with nothing to diagnose. */ static int idle_timeout = 30;

#define MODE_SINGLE_INLINE 0 #define MODE_SINGLE_POOL 1 #define MODE_POOLED 2

typedef struct CALLBACK_DATA { size_t handled; pthread_mutex_t lock; THREAD_POOL* worker_pool; } CALLBACK_DATA;

static void handle_client(NETWORK* client) { __n_assert(client, return);

netw_start_thr_engine(client);

N_STR* msg = netw_wait_msg(client, 10000, 20000000);
if (msg) {
    netw_add_msg(client, msg);
    u_sleep(50000);
}

netw_close(&client);

}

typedef struct WORKER_ARG { NETWORK* client; CALLBACK_DATA* cb_data; } WORKER_ARG;

static void* worker_handle_client(void* ptr) { WORKER_ARG* arg = (WORKER_ARG*)ptr; __n_assert(arg, return NULL);

handle_client(arg->client);

pthread_mutex_lock(&arg->cb_data->lock);
arg->cb_data->handled++;
size_t h = arg->cb_data->handled;
pthread_mutex_unlock(&arg->cb_data->lock);

if (h % 100 == 0) {
    n_log(LOG_NOTICE, "handled %zu connections", h);
}

Free(arg);
return NULL;

}

static void dispatch_to_pool(NETWORK* client, CALLBACK_DATA* cb_data) { __n_assert(client, return); __n_assert(cb_data, netw_close(&client); return); __n_assert(cb_data->worker_pool, netw_close(&client); return);

WORKER_ARG* arg = NULL;
Malloc(arg, WORKER_ARG, 1);
if (!arg) {
    n_log(LOG_ERR, "failed to allocate worker arg");
    netw_close(&client);
    return;
}
arg->client = client;
arg->cb_data = cb_data;

add_threaded_process(cb_data->worker_pool, &worker_handle_client, (void*)arg, NORMAL_PROC);

}

static void handle_inline_and_count(NETWORK* client, CALLBACK_DATA* cb_data) { __n_assert(cb_data, netw_close(&client); return);

handle_client(client);

pthread_mutex_lock(&cb_data->lock);
cb_data->handled++;
size_t h = cb_data->handled;
pthread_mutex_unlock(&cb_data->lock);

if (h % 100 == 0) {
    n_log(LOG_NOTICE, "handled %zu connections", h);
}

}

static void on_accept_pooled(NETWORK* conn, void* user_data) { CALLBACK_DATA* data = (CALLBACK_DATA*)user_data; __n_assert(data, netw_close(&conn); return); dispatch_to_pool(conn, data); }

static void sighandler(int sig) { (void)sig; server_running = 0; }

static void usage(void) { fprintf(stderr, "Usage: ex_accept_pool_server [options]\n" " -p PORT port to listen on (required)\n" " -a ADDR address to bind to (optional, default: all)\n" " -m MODE 'single-inline', 'single-pool', or 'pooled' (default: single-inline)\n" " -n COUNT number of connections to handle (default: 500)\n" " -t THREADS number of threads for pool/accept (default: 4)\n" " -w SECONDS give up after SECONDS without progress, 0 to wait forever (default: 30)\n" " -V LEVEL log level: LOG_DEBUG, LOG_INFO, LOG_NOTICE, LOG_ERR (default: LOG_NOTICE)\n" " -h show this help\n"); }

int main(int argc, char** argv) { char* addr = NULL; char* port = NULL; char* mode_str = "single-inline"; int nb_threads = 4; int log_level = LOG_NOTICE; int opt;

while ((opt = getopt(argc, argv, "hp:a:m:n:t:w:V:")) != -1) {
    switch (opt) {
        case 'p':
            port = strdup(optarg);
            break;
        case 'a':
            addr = strdup(optarg);
            break;
        case 'm':
            mode_str = optarg;
            break;
        case 'n':
            total_target = atoi(optarg);
            break;
        case 't':
            nb_threads = atoi(optarg);
            break;
        case 'w':
            idle_timeout = atoi(optarg);
            break;
        case 'V':
            if (!strcmp(optarg, "LOG_DEBUG"))
                log_level = LOG_DEBUG;
            else if (!strcmp(optarg, "LOG_INFO"))
                log_level = LOG_INFO;
            else if (!strcmp(optarg, "LOG_NOTICE"))
                log_level = LOG_NOTICE;
            else if (!strcmp(optarg, "LOG_ERR"))
                log_level = LOG_ERR;
            break;
        case 'h':
        default:
            usage();
            exit(1);
    }
}

if (!port) {
    fprintf(stderr, "Error: -p PORT is required\n");
    usage();
    exit(1);
}

int mode = MODE_SINGLE_INLINE;
if (strcmp(mode_str, "single-pool") == 0) {
    mode = MODE_SINGLE_POOL;
} else if (strcmp(mode_str, "pooled") == 0) {
    mode = MODE_POOLED;
} else if (strcmp(mode_str, "single-inline") != 0) {
    fprintf(stderr, "Error: unknown mode '%s'\n", mode_str);
    usage();
    exit(1);
}

set_log_level(log_level);

signal(SIGINT, sighandler);
signal(SIGTERM, sighandler);

#ifdef linux signal(SIGPIPE, SIG_IGN); #endif

/* create listening socket */
NETWORK* server = NULL;
if (netw_make_listening(&server, addr, port, 1024, NETWORK_IPALL) == FALSE) {
    n_log(LOG_ERR, "Failed to create listening socket on %s:%s", addr ? addr : "*", port);
    exit(1);
}
n_log(LOG_NOTICE, "Listening on %s:%s (backlog=1024)", addr ? addr : "*", port);

CALLBACK_DATA cb_data;
cb_data.handled = 0;
cb_data.worker_pool = NULL;
pthread_mutex_init(&cb_data.lock, NULL);

struct timespec t_start, t_end;
clock_gettime(CLOCK_MONOTONIC, &t_start);

const char* mode_names[] = {"SINGLE-INLINE", "SINGLE-POOL", "POOLED"};
n_log(LOG_NOTICE, "=== %s MODE: target %d connections ===", mode_names[mode], total_target);

int exit_code = 0;
/* progress watchdog: get_msec() returns the delta since the previous
 * poll, so accumulating it gives the time since the last connection */
N_TIME idle_timer;
start_HiTimer(&idle_timer);
time_t idle_msec = 0;
size_t seen_progress = 0;

if (mode == MODE_SINGLE_INLINE) {
    /* SINGLE-INLINE MODE
     * One thread doing accept + handle client in the same thread.
     * Both accept and handling are fully serialized. */
    n_log(LOG_NOTICE, "1 thread: accept + handle inline (fully serialized)");

    while (server_running) {
        pthread_mutex_lock(&cb_data.lock);
        size_t h = cb_data.handled;
        pthread_mutex_unlock(&cb_data.lock);
        if ((int)h >= total_target) {
            break;
        }
        idle_msec += get_msec(&idle_timer);
        if (h != seen_progress) {
            seen_progress = h;
            idle_msec = 0;
        }
        if (idle_timeout > 0 && idle_msec > (time_t)idle_timeout * 1000) {
            n_log(LOG_ERR, "no connection for %d seconds, giving up with %zu/%d handled", idle_timeout, h, total_target);
            exit_code = 1;
            break;
        }

        int retval = 0;
        NETWORK* client = netw_accept_from_ex(server, 0, 0, 500, &retval);
        if (client) {
            handle_inline_and_count(client, &cb_data);
        }
    }
} else if (mode == MODE_SINGLE_POOL) {
    /* SINGLE-POOL MODE
     * One thread doing accept, dispatching to a worker thread pool.
     * Accept is serialized, but client handling is parallelized. */
    n_log(LOG_NOTICE, "1 accept thread, %d worker threads", nb_threads);

    cb_data.worker_pool = new_thread_pool((size_t)nb_threads, (size_t)(total_target + 16));
    if (!cb_data.worker_pool) {
        n_log(LOG_ERR, "Failed to create worker thread pool");
        goto cleanup;
    }

    int accepted_count = 0;
    while (server_running && accepted_count < total_target) {
        idle_msec += get_msec(&idle_timer);
        if ((size_t)accepted_count != seen_progress) {
            seen_progress = (size_t)accepted_count;
            idle_msec = 0;
        }
        if (idle_timeout > 0 && idle_msec > (time_t)idle_timeout * 1000) {
            n_log(LOG_ERR, "no connection for %d seconds, giving up with %d/%d accepted", idle_timeout, accepted_count, total_target);
            exit_code = 1;
            break;
        }

        int retval = 0;
        NETWORK* client = netw_accept_from_ex(server, 0, 0, 500, &retval);
        if (client) {
            dispatch_to_pool(client, &cb_data);
            accepted_count++;
        }
    }

    /* wait for all workers to finish */
    n_log(LOG_NOTICE, "all connections accepted, waiting for workers...");
    wait_for_threaded_pool(cb_data.worker_pool);
    destroy_threaded_pool(&cb_data.worker_pool, 500000);
} else {
    /* POOLED MODE
     * N accept threads (accept pool) + worker thread pool.
     * Both accept and handling are parallelized. */
    n_log(LOG_NOTICE, "%d accept threads, %d worker threads", nb_threads, nb_threads);

    cb_data.worker_pool = new_thread_pool((size_t)nb_threads, (size_t)(total_target + 16));
    if (!cb_data.worker_pool) {
        n_log(LOG_ERR, "Failed to create worker thread pool");
        goto cleanup;
    }

    NETW_ACCEPT_POOL* accept_pool = netw_accept_pool_create(
        server, (size_t)nb_threads, 500, on_accept_pooled, &cb_data);
    if (!accept_pool) {
        n_log(LOG_ERR, "Failed to create accept pool");
        destroy_threaded_pool(&cb_data.worker_pool, 500000);
        goto cleanup;
    }

    if (netw_accept_pool_start(accept_pool) == FALSE) {
        n_log(LOG_ERR, "Failed to start accept pool");
        netw_accept_pool_destroy(&accept_pool);
        destroy_threaded_pool(&cb_data.worker_pool, 500000);
        goto cleanup;
    }

    /* wait until we've handled enough connections or user interrupts */
    while (server_running) {
        pthread_mutex_lock(&cb_data.lock);
        size_t h = cb_data.handled;
        pthread_mutex_unlock(&cb_data.lock);

        if ((int)h >= total_target) {
            break;
        }
        idle_msec += get_msec(&idle_timer);
        if (h != seen_progress) {
            seen_progress = h;
            idle_msec = 0;
        }
        if (idle_timeout > 0 && idle_msec > (time_t)idle_timeout * 1000) {
            n_log(LOG_ERR, "no connection for %d seconds, giving up with %zu/%d handled", idle_timeout, h, total_target);
            exit_code = 1;
            break;
        }

        NETW_ACCEPT_POOL_STATS stats;
        netw_accept_pool_get_stats(accept_pool, &stats);
        n_log(LOG_INFO, "stats: accepted=%zu errors=%zu timeouts=%zu active_threads=%zu",
              stats.total_accepted, stats.total_errors, stats.total_timeouts, stats.active_threads);

        u_sleep(100000);
    }

    netw_accept_pool_stop(accept_pool);
    netw_accept_pool_wait(accept_pool, 10);

    NETW_ACCEPT_POOL_STATS final_stats;
    netw_accept_pool_get_stats(accept_pool, &final_stats);
    n_log(LOG_NOTICE, "Pool stats: accepted=%zu errors=%zu timeouts=%zu",
          final_stats.total_accepted, final_stats.total_errors, final_stats.total_timeouts);

    netw_accept_pool_destroy(&accept_pool);

    /* wait for all workers to finish */
    n_log(LOG_NOTICE, "accept pool stopped, waiting for workers...");
    wait_for_threaded_pool(cb_data.worker_pool);
    destroy_threaded_pool(&cb_data.worker_pool, 500000);
}

clock_gettime(CLOCK_MONOTONIC, &t_end);
double elapsed = (double)(t_end.tv_sec - t_start.tv_sec) +
                 (double)(t_end.tv_nsec - t_start.tv_nsec) / 1e9;

n_log(LOG_NOTICE, "=== DONE: %zu connections in %.3f seconds (%.1f conn/sec) ===",
      cb_data.handled, elapsed, (double)cb_data.handled / elapsed);

cleanup: pthread_mutex_destroy(&cb_data.lock); netw_close(&server);

FreeNoLog(addr);
FreeNoLog(port);

netw_unload();
if (exit_code == 0) {
    n_log(LOG_NOTICE, "Server exited cleanly");
} else {
    n_log(LOG_ERR, "Server exited with code %d", exit_code);
}

return exit_code;

}

_Filename: ex_accept_pool_server.c_