ex_accept_pool_server.c

Go to the documentation of this file.

/*
 * 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;
}