src/n_network_accept_pool.c

Accept pool for parallel connection acceptance (nginx-style). More...

Functions

	Name
void *	netw_accept_pool_thread_func(void * arg) Thread function for accept pool workers.

Detailed Description

Accept pool for parallel connection acceptance (nginx-style).

Author: Castagnier Mickael

Version: 1.0

Date: 11/03/2026

Functions Documentation

function netw_accept_pool_thread_func

static void * netw_accept_pool_thread_func(
    void * arg
)

Thread function for accept pool workers.

Parameters:

• arg pointer to the NETW_ACCEPT_POOL

Return: NULL

Each thread loops calling accept on the shared listening socket and invokes the user callback for each accepted connection.

Source code

/*
 * 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 "nilorea/n_network_accept_pool.h"

#include <errno.h>
#include <string.h>

static void* netw_accept_pool_thread_func(void* arg) {
    NETW_ACCEPT_POOL* pool = (NETW_ACCEPT_POOL*)arg;

    __n_assert(pool, return NULL);
    __n_assert(pool->server, return NULL);
    __n_assert(pool->callback, return NULL);

    pthread_mutex_lock(&pool->stats_lock);
    pool->stats.active_threads++;
    pthread_mutex_unlock(&pool->stats_lock);

    while (netw_accept_pool_atomic_read_state(pool) == NETW_ACCEPT_POOL_RUNNING) {
        int retval = 0;
        NETWORK* accepted = netw_accept_from_ex(pool->server, 0, 0, pool->accept_timeout, &retval);

        /* check if we should exit */
        if (netw_accept_pool_atomic_read_state(pool) != NETW_ACCEPT_POOL_RUNNING) {
            if (accepted) {
                netw_close(&accepted);
            }
            break;
        }

        if (accepted) {
            pthread_mutex_lock(&pool->stats_lock);
            pool->stats.total_accepted++;
            pthread_mutex_unlock(&pool->stats_lock);

            pool->callback(accepted, pool->user_data);
        } else {
            if (retval == EINTR) {
                /* interrupted, just retry */
                continue;
            }
            if (retval == EAGAIN || retval == EWOULDBLOCK || retval == 0) {
                /* timeout or no connection available */
                pthread_mutex_lock(&pool->stats_lock);
                pool->stats.total_timeouts++;
                pthread_mutex_unlock(&pool->stats_lock);
                /* small sleep to avoid busy-looping on non-blocking accept */
                if (pool->accept_timeout == -1) {
                    u_sleep(1000);
                }
            } else {
                /* actual error */
                pthread_mutex_lock(&pool->stats_lock);
                pool->stats.total_errors++;
                pthread_mutex_unlock(&pool->stats_lock);
                n_log(LOG_ERR, "accept pool thread: accept error, retval=%d", retval);
            }
        }
    }

    pthread_mutex_lock(&pool->stats_lock);
    if (pool->stats.active_threads > 0) {
        pool->stats.active_threads--;
    }
    pthread_mutex_unlock(&pool->stats_lock);

    return NULL;
}

NETW_ACCEPT_POOL* netw_accept_pool_create(NETWORK* server, size_t nb_threads, int accept_timeout, netw_accept_callback_t callback, void* user_data) {
    __n_assert(server, return NULL);
    __n_assert(callback, return NULL);

    if (nb_threads == 0) {
        n_log(LOG_ERR, "netw_accept_pool_create: nb_threads must be > 0");
        return NULL;
    }

    NETW_ACCEPT_POOL* pool = NULL;
    Malloc(pool, NETW_ACCEPT_POOL, 1);
    __n_assert(pool, return NULL);

    pool->server = server;
    pool->nb_accept_threads = nb_threads;
    pool->accept_timeout = accept_timeout;
    pool->callback = callback;
    pool->user_data = user_data;

    netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_IDLE);

    memset(&pool->stats, 0, sizeof(NETW_ACCEPT_POOL_STATS));

    if (pthread_mutex_init(&pool->stats_lock, NULL) != 0) {
        n_log(LOG_ERR, "netw_accept_pool_create: failed to init stats mutex");
        Free(pool);
        return NULL;
    }

    Malloc(pool->accept_threads, pthread_t, nb_threads);
    if (!pool->accept_threads) {
        n_log(LOG_ERR, "netw_accept_pool_create: failed to allocate thread array");
        pthread_mutex_destroy(&pool->stats_lock);
        Free(pool);
        return NULL;
    }

    return pool;
}

int netw_accept_pool_start(NETW_ACCEPT_POOL* pool) {
    __n_assert(pool, return FALSE);

    uint32_t current = netw_accept_pool_atomic_read_state(pool);
    if (current != NETW_ACCEPT_POOL_IDLE && current != NETW_ACCEPT_POOL_STOPPED) {
        n_log(LOG_ERR, "netw_accept_pool_start: pool is not idle or stopped (state=%u)", current);
        return FALSE;
    }

    /* reset stats */
    pthread_mutex_lock(&pool->stats_lock);
    memset(&pool->stats, 0, sizeof(NETW_ACCEPT_POOL_STATS));
    clock_gettime(CLOCK_MONOTONIC, &pool->stats.start_time);
    pthread_mutex_unlock(&pool->stats_lock);

    netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_RUNNING);

    for (size_t i = 0; i < pool->nb_accept_threads; i++) {
        int err = pthread_create(&pool->accept_threads[i], NULL, netw_accept_pool_thread_func, pool);
        if (err != 0) {
            n_log(LOG_ERR, "netw_accept_pool_start: failed to create thread %zu: %s", i, strerror(err));
            /* stop already-created threads */
            netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_STOPPING);
            for (size_t j = 0; j < i; j++) {
                pthread_join(pool->accept_threads[j], NULL);
            }
            netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_STOPPED);
            return FALSE;
        }
    }

    n_log(LOG_NOTICE, "accept pool started with %zu threads", pool->nb_accept_threads);
    return TRUE;
}

int netw_accept_pool_stop(NETW_ACCEPT_POOL* pool) {
    __n_assert(pool, return FALSE);

    uint32_t current = netw_accept_pool_atomic_read_state(pool);
    if (current != NETW_ACCEPT_POOL_RUNNING) {
        n_log(LOG_WARNING, "netw_accept_pool_stop: pool is not running (state=%u)", current);
        return FALSE;
    }

    n_log(LOG_NOTICE, "accept pool: stop requested");
    netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_STOPPING);

    /* shutdown the listening socket to unblock any threads stuck in select()/accept() */
    if (pool->server && pool->server->link.sock != INVALID_SOCKET) {
        shutdown(pool->server->link.sock, SHUT_RDWR);
    }

    return TRUE;
}

int netw_accept_pool_wait(NETW_ACCEPT_POOL* pool, int timeout_sec) {
    __n_assert(pool, return FALSE);

    uint32_t current = netw_accept_pool_atomic_read_state(pool);
    if (current == NETW_ACCEPT_POOL_IDLE || current == NETW_ACCEPT_POOL_STOPPED) {
        return TRUE;
    }

    if (current == NETW_ACCEPT_POOL_RUNNING) {
        n_log(LOG_WARNING, "netw_accept_pool_wait: pool still running, calling stop first");
        netw_accept_pool_stop(pool);
    }

    (void)timeout_sec;
    for (size_t i = 0; i < pool->nb_accept_threads; i++) {
        pthread_join(pool->accept_threads[i], NULL);
    }

    netw_accept_pool_atomic_write_state(pool, NETW_ACCEPT_POOL_STOPPED);
    n_log(LOG_NOTICE, "accept pool: all threads joined");

    return TRUE;
}

int netw_accept_pool_get_stats(NETW_ACCEPT_POOL* pool, NETW_ACCEPT_POOL_STATS* stats) {
    __n_assert(pool, return FALSE);
    __n_assert(stats, return FALSE);

    pthread_mutex_lock(&pool->stats_lock);
    memcpy(stats, &pool->stats, sizeof(NETW_ACCEPT_POOL_STATS));
    pthread_mutex_unlock(&pool->stats_lock);

    return TRUE;
}

int netw_accept_pool_destroy(NETW_ACCEPT_POOL** pool) {
    __n_assert(pool, return FALSE);
    __n_assert((*pool), return FALSE);

    uint32_t current = netw_accept_pool_atomic_read_state(*pool);
    if (current == NETW_ACCEPT_POOL_RUNNING || current == NETW_ACCEPT_POOL_STOPPING) {
        n_log(LOG_WARNING, "netw_accept_pool_destroy: pool still active, stopping and waiting");
        netw_accept_pool_stop(*pool);
        netw_accept_pool_wait(*pool, 0);
    }

    pthread_mutex_destroy(&(*pool)->stats_lock);
    Free((*pool)->accept_threads);
    Free(*pool);

    return TRUE;
}