From a9cdfeca134f60de5cfacaefb5f2c3f51a2b9a97 Mon Sep 17 00:00:00 2001 From: Roman Janota Date: Thu, 3 Sep 2026 11:11:10 +0200 Subject: [PATCH] session server BUGFIX ps turn queue fairness The pollsession turn queue was strictly FIFO, so nc_ps_add_session() and the other bounded operations queued up behind every poll thread and each of them kept its turn for the whole timeout it was given. A newly established session was therefore not polled for up to thread_count * poll_timeout msec. Threads now queue up with a priority. Every operation except nc_ps_poll() only walks the session array, so it is queued up in front of all the poll threads, and the polling thread hands the turn over. The queue is also allocated and grows on demand instead of failing once MAX_PSPOLL_THREAD_COUNT threads use the same pollsession, which silently dropped the session being added. Waiting longer than NC_PS_QUEUE_TIMEOUT is only an error for the high priority threads now, a poll thread may legitimately wait behind several other poll threads for longer than that. Also fixes the session idle timeout being compared against the poll deadline instead of the current time, which terminated active sessions when the poll timeout was longer than the remaining idle time, and read an uninitialized timespec with an infinite timeout. Fixes #561 --- CMakeLists.txt | 1 - README.md | 12 -- libnetconf2.pc.in | 1 - src/config.h.in | 5 - src/session_p.h | 54 ++++-- src/session_server.c | 396 ++++++++++++++++++++++++++------------- src/session_server_ssh.c | 5 +- tests/CMakeLists.txt | 1 + tests/test_ps_poll.c | 393 ++++++++++++++++++++++++++++++++++++++ 9 files changed, 699 insertions(+), 169 deletions(-) create mode 100644 tests/test_ps_poll.c diff --git a/CMakeLists.txt b/CMakeLists.txt index ae43766a..b3ff55a3 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -100,7 +100,6 @@ option(ENABLE_COMPLY_WITH_ORAN_WG11 "Comply with the O-RAN WG11 security spec (e option(BUILD_SHARED_LIBS "By default, shared libs are enabled. Turn off for a static build." ON) set(READ_INACTIVE_TIMEOUT 20 CACHE STRING "Maximum number of seconds waiting for new data once some data have arrived") set(READ_ACTIVE_TIMEOUT 300 CACHE STRING "Maximum number of seconds for receiving a full message") -set(MAX_PSPOLL_THREAD_COUNT 6 CACHE STRING "Maximum number of threads that could simultaneously access a ps_poll structure") set(TRANSPORT_HANDSHAKE_TIMEOUT 10 CACHE STRING "SSH key exchange and TLS handshake timeout in seconds") set(MESSAGE_MAX_SIZE 1048576 CACHE STRING "Maximum size of a message in kB") set(TIMEOUT_STEP 100 CACHE STRING "Number of microseconds tasks are repeated until timeout elapses") diff --git a/README.md b/README.md index 6308b753..422688b4 100644 --- a/README.md +++ b/README.md @@ -183,18 +183,6 @@ to arrive in its entirety once a beginning is read. The default is 300 (5 minute $ cmake -D READ_ACTIVE_TIMEOUT:String="300" .. ``` -### PSPoll Thread Count - -This value limits the maximum number of threads that can concurrently access -(wait for access) a single pspoll structure. To simplify, how many threads could -simultaneously call a function whose parameter is one and the same pspoll structure. -If using **netopeer2-server**, it will warn that this value needs to be adjusted if -too small. - -``` -$ cmake -D MAX_PSPOLL_THREAD_COUNT:String="6" .. -``` - ### Code Coverage Based on the tests run, it is possible to generate code coverage report. But diff --git a/libnetconf2.pc.in b/libnetconf2.pc.in index 10926a59..f81cf2b2 100644 --- a/libnetconf2.pc.in +++ b/libnetconf2.pc.in @@ -8,5 +8,4 @@ Version: @LIBNETCONF2_VERSION@ Libs: -L${libdir} -lnetconf2 Cflags: -I${includedir} -LN2_MAX_THREAD_COUNT=@MAX_PSPOLL_THREAD_COUNT@ LN2_SCHEMAS_DIR=@YANG_MODULE_DIR@ diff --git a/src/config.h.in b/src/config.h.in index 2e982fe7..90a4356a 100644 --- a/src/config.h.in +++ b/src/config.h.in @@ -70,11 +70,6 @@ */ #define NC_READ_ACT_TIMEOUT @READ_ACTIVE_TIMEOUT@ -/* - * pspoll structure queue size (also found in nc_server.h) - */ -#define NC_PS_QUEUE_SIZE @MAX_PSPOLL_THREAD_COUNT@ - /* * Timeout in msec for transport layer connection handshake/key exchange. * It can be quite a lot on slow machines (waiting for TLS cert-to-name resolution, SSH key cryptography, ...). diff --git a/src/session_p.h b/src/session_p.h index 284cbb54..26a610b0 100644 --- a/src/session_p.h +++ b/src/session_p.h @@ -95,9 +95,15 @@ extern struct nc_server_opts server_opts; #define NC_SESSION_FREE_SSH_POLL_EOF_TIMEOUT 100 /** - * Timeout in msec for a thread to wait for its turn to work with a pollsession structure. + * Initial number of threads the queue of a pollsession structure is allocated for, it grows on demand. */ -#define NC_PS_QUEUE_TIMEOUT 5000 +#define NC_PS_QUEUE_INIT_SIZE 6 + +/** + * Timeout in msec for a thread to get the pollsession lock and its turn. The thread that has the + * turn may be freeing sessions in ::nc_ps_clear(), which blocks for a while on each of them. + */ +#define NC_PS_TIMEOUT 3000 /** * @brief Maximum time (in seconds) to wait for a pending configuration @@ -200,12 +206,6 @@ extern struct nc_server_opts server_opts; */ #define NC_SESSION_CH_LOCK_TIMEOUT 1000 -/** - * @brief Timeout in msec for acquiring the pollsession's lock - * (only O(n) array manipulation, where n is number of sessions (small usually)) - */ -#define NC_PS_LOCK_TIMEOUT 1000 - /** * @brief Timeout in msec for acquiring the notification status lock * (short critical sections for incrementing/decrementing notification status) @@ -1225,11 +1225,14 @@ struct nc_pollsession { uint16_t session_count; uint16_t last_event_session; - pthread_cond_t cond; - pthread_mutex_t lock; - uint8_t queue[NC_PS_QUEUE_SIZE]; /**< round buffer, queue is empty when queue_len == 0 */ - uint8_t queue_begin; /**< queue starts on queue[queue_begin] */ - uint8_t queue_len; /**< queue ends on queue[(queue_begin + queue_len - 1) % NC_PS_QUEUE_SIZE] */ + pthread_cond_t cond; /**< broadcasted whenever a pollsession turn is given up */ + pthread_mutex_t lock; /**< lock for the cond and the queue */ + pthread_t *queue; /**< round buffer, queue is empty when queue_len == 0 */ + uint8_t queue_size; /**< allocated size of queue, 0 until the first thread queues up */ + uint8_t queue_begin; /**< queue starts on queue[queue_begin], that thread gets the turn next */ + uint8_t queue_len; /**< queue ends on queue[(queue_begin + queue_len - 1) % queue_size] */ + int busy; /**< whether a thread is working with the pollsession, the thread at the + beginning of the queue only gets the turn once this is 0 */ }; struct nc_ntf_thread_arg { @@ -1518,9 +1521,30 @@ int nc_mutex_lock(pthread_mutex_t *mutex, int timeout, const char *func_name); */ void nc_mutex_unlock(pthread_mutex_t *mutex, const char *func_name); -int nc_ps_lock(struct nc_pollsession *ps, uint8_t *id, const char *func); +/** + * @brief Wait for the turn of this thread to work with a pollsession. + * + * @param[in,out] ps Pollsession structure. + * @param[in] preempt Whether this thread preempts the poll thread that currently has the turn, + * meaning it is queued up in front of it. Set for every operation that only walks the session + * array, clear for ::nc_ps_poll() which polls for a whole poll interval. + * @param[in] timeout_ms Timeout in msec used for the lock and for waiting for the turn, 0 for no + * waiting, -1 for no timeout. + * @param[in] func Caller function name for logging. + * @return 1 on success and the turn is taken. + * @return 0 on timeout. + * @return -1 on error. + */ +int nc_ps_lock(struct nc_pollsession *ps, int preempt, int timeout_ms, const char *func); -int nc_ps_unlock(struct nc_pollsession *ps, uint8_t id, const char *func); +/** + * @brief Give up the pollsession turn of this thread. + * + * @param[in,out] ps Pollsession structure. + * @param[in] func Caller function name for logging. + * @return 0 on success, -1 on error. + */ +int nc_ps_unlock(struct nc_pollsession *ps, const char *func); int nc_client_session_new_ctx(struct nc_session *session, struct ly_ctx *ctx); diff --git a/src/session_server.c b/src/session_server.c index 688e5dc1..4cc32dd8 100644 --- a/src/session_server.c +++ b/src/session_server.c @@ -1862,154 +1862,273 @@ nc_accept_inout(int fdin, int fdout, const char *username, const struct ly_ctx * return msgtype; } -static void -nc_ps_queue_add_id(struct nc_pollsession *ps, uint8_t *id) +/** + * @brief Add this thread into the pollsession queue. + * + * @note @p ps->lock MUST be held. + * + * @param[in,out] ps Pollsession structure. + * @param[in] preempt Whether this thread preempts the thread that has the turn, meaning it is + * added at the very beginning of the queue instead of at its end. + * @return 0 on success, -1 on error. + */ +static int +nc_ps_queue_add(struct nc_pollsession *ps, int preempt) { - uint8_t q_last; + pthread_t *new_queue; + uint8_t new_size, idx, i; + + if (ps->queue_len == ps->queue_size) { + /* there is no room for another thread, make the queue bigger */ + if (!ps->queue_size) { + new_size = NC_PS_QUEUE_INIT_SIZE; + } else if (ps->queue_size == UINT8_MAX) { + ERR(NULL, "Too many threads (%" PRIu8 ") accessing a single pollsession.", ps->queue_size); + return -1; + } else { + new_size = ps->queue_size + 1; + } - if (ps->queue_len == NC_PS_QUEUE_SIZE) { - ERRINT; - return; + new_queue = malloc(new_size * sizeof *new_queue); + NC_CHECK_ERRMEM_RET(!new_queue, -1); + + /* copy the queue over */ + for (i = 0; i < ps->queue_len; ++i) { + new_queue[i] = ps->queue[(ps->queue_begin + i) % ps->queue_size]; + } + + free(ps->queue); + ps->queue = new_queue; + ps->queue_size = new_size; + ps->queue_begin = 0; } - /* get a unique queue value (by adding 1 to the last added value, if any) */ - if (ps->queue_len) { - q_last = (ps->queue_begin + ps->queue_len - 1) % NC_PS_QUEUE_SIZE; - *id = ps->queue[q_last] + 1; + if (preempt) { + /* queue up in front of everyone, including the thread that has the turn */ + ps->queue_begin = ps->queue_begin ? ps->queue_begin - 1 : ps->queue_size - 1; + idx = ps->queue_begin; } else { - *id = 0; + /* queue up at the very end */ + idx = (ps->queue_begin + ps->queue_len) % ps->queue_size; } - /* add the id into the queue */ + ps->queue[idx] = pthread_self(); ++ps->queue_len; - q_last = (ps->queue_begin + ps->queue_len - 1) % NC_PS_QUEUE_SIZE; - ps->queue[q_last] = *id; + + return 0; } +/** + * @brief Remove this thread from the pollsession queue. + * + * @note @p ps->lock MUST be held. + * + * @param[in,out] ps Pollsession structure. + */ static void -nc_ps_queue_remove_id(struct nc_pollsession *ps, uint8_t id) +nc_ps_queue_remove(struct nc_pollsession *ps) { - uint8_t i, q_idx, found = 0; + uint8_t i, idx; + /* find ourselves */ for (i = 0; i < ps->queue_len; ++i) { - /* get the actual queue idx */ - q_idx = (ps->queue_begin + i) % NC_PS_QUEUE_SIZE; - - if (found) { - if (ps->queue[q_idx] == id) { - /* another equal value, simply cannot be */ - ERRINT; - } - if (found == 2) { - /* move the following values */ - ps->queue[q_idx ? q_idx - 1 : NC_PS_QUEUE_SIZE - 1] = ps->queue[q_idx]; - } - } else if (ps->queue[q_idx] == id) { - /* found our id, there can be no more equal valid values */ - if (i == 0) { - found = 1; - } else { - /* this is not okay, our id is in the middle of the queue */ - found = 2; - } + if (pthread_equal(ps->queue[(ps->queue_begin + i) % ps->queue_size], pthread_self())) { + idx = i; + break; } } - if (!found) { + if (i == ps->queue_len) { ERRINT; return; } - --ps->queue_len; - if (found == 1) { - /* remove the id by moving the queue, otherwise all the values in the queue were moved */ - ps->queue_begin = (ps->queue_begin + 1) % NC_PS_QUEUE_SIZE; + if (!idx) { + /* the very beginning, simply move the queue */ + ps->queue_begin = (ps->queue_begin + 1) % ps->queue_size; + } else { + /* move all the following threads one position forward */ + for (i = idx; i + 1 < ps->queue_len; ++i) { + ps->queue[(ps->queue_begin + i) % ps->queue_size] = ps->queue[(ps->queue_begin + i + 1) % ps->queue_size]; + } } + --ps->queue_len; } -int -nc_ps_lock(struct nc_pollsession *ps, uint8_t *id, const char *func) +/** + * @brief Wait for the turn of this thread to work with a pollsession and take it. + * + * @note @p ps->lock MUST be held and this thread MUST be in the queue. + * + * @param[in,out] ps Pollsession structure. + * @param[in] preempt Whether this thread preempts the turn, timing out is then an error. + * @param[in] timeout_ms Timeout in msec, 0 for no waiting, -1 for no timeout. + * @param[in] func Caller function name for logging. + * @return 1 on success and the turn is taken. + * @return 0 on timeout, the thread is kept in the queue. + * @return -1 on error, the thread is kept in the queue. + */ +static int +nc_ps_queue_wait_turn(struct nc_pollsession *ps, int preempt, int timeout_ms, const char *func) { - int r, rc = 0; + int r; struct timespec ts; - /* LOCK */ - if (nc_mutex_lock(&ps->lock, NC_PS_LOCK_TIMEOUT, func) != 1) { - return -1; - } - - /* check that the queue is long enough */ - if (ps->queue_len == NC_PS_QUEUE_SIZE) { - ERR(NULL, "%s: pollsession queue size (%d) too small.", func, NC_PS_QUEUE_SIZE); - nc_mutex_unlock(&ps->lock, func); - return -1; + if (timeout_ms > 0) { + nc_timeouttime_get(&ts, timeout_ms); } - /* add ourselves into the queue */ - nc_ps_queue_add_id(ps, id); - DBL(NULL, "PS 0x%p TID %lu queue: added %u, head %u, length %u", ps, (long unsigned int)pthread_self(), *id, - ps->queue[ps->queue_begin], ps->queue_len); + /* wait until we are at the beginning of the queue and no one else has the turn */ + while (!pthread_equal(ps->queue[ps->queue_begin], pthread_self()) || ps->busy) { + if (!timeout_ms) { + /* the turn is not free and we are not waiting for it */ + return 0; + } - /* is it our turn? */ - while (ps->queue[ps->queue_begin] != *id) { - nc_timeouttime_get(&ts, NC_PS_QUEUE_TIMEOUT); + if (timeout_ms > 0) { + r = pthread_cond_clockwait(&ps->cond, &ps->lock, COMPAT_CLOCK_ID, &ts); + } else { + r = pthread_cond_wait(&ps->cond, &ps->lock); + } - r = pthread_cond_clockwait(&ps->cond, &ps->lock, COMPAT_CLOCK_ID, &ts); - if (r) { - /** - * This may happen when another thread releases the lock and broadcasts the condition - * and this thread had already timed out. When this thread is scheduled, it returns timed out error - * but when actually this thread was ready for condition. - */ - if ((ETIMEDOUT == r) && (ps->queue[ps->queue_begin] == *id)) { + if (r == ETIMEDOUT) { + /* the deadline may have expired while another thread was giving the turn up */ + if (pthread_equal(ps->queue[ps->queue_begin], pthread_self()) && !ps->busy) { break; } + if (preempt) { + /* the thread with the turn gives it up once per poll interval, so it is jammed */ + ERR(NULL, "%s: timed out after %d ms waiting for the pollsession turn.", func, timeout_ms); + } + return 0; + } + if (r) { ERR(NULL, "%s: failed to wait for a pollsession condition (%s).", func, strerror(r)); - /* remove ourselves from the queue */ - nc_ps_queue_remove_id(ps, *id); - rc = -1; - break; + return -1; } } - /* UNLOCK */ - nc_mutex_unlock(&ps->lock, func); + /* take the turn, we are at the beginning of the queue */ + ps->busy = 1; - return rc; + return 1; } int -nc_ps_unlock(struct nc_pollsession *ps, uint8_t id, const char *func) +nc_ps_lock(struct nc_pollsession *ps, int preempt, int timeout_ms, const char *func) { - int r; + int rc; + struct timespec ts_deadline; - /* LOCK, continue on error */ - r = nc_mutex_lock(&ps->lock, NC_PS_LOCK_TIMEOUT, func); + if (timeout_ms > 0) { + /* the whole call must fit into the timeout, so remember when it expires */ + nc_timeouttime_get(&ts_deadline, timeout_ms); + } - /* we must be the first, it was our turn after all, right? */ - if (ps->queue[ps->queue_begin] != id) { - ERRINT; - /* UNLOCK */ - if (r == 1) { - nc_mutex_unlock(&ps->lock, func); + /* LOCK */ + rc = nc_mutex_lock(&ps->lock, timeout_ms, func); + if (rc != 1) { + return rc; + } + + /* add ourselves into the queue */ + if (nc_ps_queue_add(ps, preempt)) { + rc = -1; + goto cleanup; + } + + if (timeout_ms > 0) { + /* only the rest of the timeout is left for waiting */ + timeout_ms = nc_timeouttime_cur_diff(&ts_deadline); + if (timeout_ms < 0) { + timeout_ms = 0; } + } + + /* is it our turn? */ + rc = nc_ps_queue_wait_turn(ps, preempt, timeout_ms, func); + if (rc != 1) { + /* remove ourselves from the queue */ + nc_ps_queue_remove(ps); + } + +cleanup: + /* UNLOCK */ + nc_mutex_unlock(&ps->lock, func); + return rc; +} + +int +nc_ps_unlock(struct nc_pollsession *ps, const char *func) +{ + /* LOCK */ + if (nc_mutex_lock(&ps->lock, NC_PS_TIMEOUT, func) != 1) { + /* the error was logged, the queue must not be read nor modified without the lock */ + ERR(NULL, "%s: failed to remove a thread from the pollsession queue, it will jam it.", func); return -1; } - /* remove ourselves from the queue */ - nc_ps_queue_remove_id(ps, id); - DBL(NULL, "PS 0x%p TID %lu queue: removed %u, head %u, length %u", ps, (long unsigned int)pthread_self(), id, - ps->queue[ps->queue_begin], ps->queue_len); + assert(ps->busy); - /* broadcast to all other threads that the queue moved */ + /* give up the turn, remove ourselves from the queue and let the next thread in */ + ps->busy = 0; + nc_ps_queue_remove(ps); pthread_cond_broadcast(&ps->cond); /* UNLOCK */ - if (r == 1) { - nc_mutex_unlock(&ps->lock, func); + nc_mutex_unlock(&ps->lock, func); + return 0; +} + +/** + * @brief Give the pollsession turn up if another thread preempted it and take it back afterwards. + * + * A poll thread keeps its turn for as long as its caller asked for, but the other pollsession + * operations only walk the session array, so they queue up in front of it instead of waiting. + * This detects that and waits for them to finish. + * + * @note @p ps->lock MUST NOT be held. + * + * @param[in,out] ps Pollsession structure. + * @param[in] timeout_ms Timeout in msec for getting the turn back, 0 for no waiting, -1 for no timeout. + * @param[in] func Caller function name for logging. + * @return 1 if the turn is held on return. + * @return 0 on timeout, the turn is not held and must not be given up by the caller. + * @return -1 on error, the turn is not held and must not be given up by the caller. + */ +static int +nc_ps_check_preempt(struct nc_pollsession *ps, int timeout_ms, const char *func) +{ + int rc = 1; + + /* LOCK */ + if (nc_mutex_lock(&ps->lock, timeout_ms, func) != 1) { + /* the turn was not given up, keep it */ + return 1; } - return r == 1 ? 0 : -1; + /* it is our turn after all, right? */ + assert(ps->busy); + + if (pthread_equal(ps->queue[ps->queue_begin], pthread_self())) { + /* we are still at the beginning of the queue, no one preempted us */ + goto cleanup; + } + + /* give the turn up but keep our position in the queue */ + ps->busy = 0; + pthread_cond_broadcast(&ps->cond); + + /* wait for the preempting threads to give the turn back */ + rc = nc_ps_queue_wait_turn(ps, 0, timeout_ms, func); + if (rc != 1) { + nc_ps_queue_remove(ps); + } + +cleanup: + /* UNLOCK */ + nc_mutex_unlock(&ps->lock, func); + return rc; } API struct nc_pollsession * @@ -2019,8 +2138,8 @@ nc_ps_new(void) ps = calloc(1, sizeof(struct nc_pollsession)); NC_CHECK_ERRMEM_RET(!ps, NULL); - pthread_cond_init(&ps->cond, NULL); pthread_mutex_init(&ps->lock, NULL); + pthread_cond_init(&ps->cond, NULL); return ps; } @@ -2043,8 +2162,9 @@ nc_ps_free(struct nc_pollsession *ps) } free(ps->sessions); - pthread_mutex_destroy(&ps->lock); + free(ps->queue); pthread_cond_destroy(&ps->cond); + pthread_mutex_destroy(&ps->lock); free(ps); } @@ -2052,12 +2172,11 @@ nc_ps_free(struct nc_pollsession *ps) API int nc_ps_add_session(struct nc_pollsession *ps, struct nc_session *session) { - uint8_t q_id; NC_CHECK_ARG_RET(session, ps, session, -1); /* LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { + if (nc_ps_lock(ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return -1; } @@ -2066,7 +2185,7 @@ nc_ps_add_session(struct nc_pollsession *ps, struct nc_session *session) if (!ps->sessions) { ERRMEM; /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); return -1; } ps->sessions[ps->session_count - 1] = calloc(1, sizeof **ps->sessions); @@ -2074,14 +2193,14 @@ nc_ps_add_session(struct nc_pollsession *ps, struct nc_session *session) ERRMEM; --ps->session_count; /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); return -1; } ps->sessions[ps->session_count - 1]->session = session; ps->sessions[ps->session_count - 1]->state = NC_PS_STATE_NONE; /* UNLOCK */ - return nc_ps_unlock(ps, q_id, __func__); + return nc_ps_unlock(ps, __func__); } static int @@ -2116,20 +2235,19 @@ _nc_ps_del_session(struct nc_pollsession *ps, struct nc_session *session, int in API int nc_ps_del_session(struct nc_pollsession *ps, struct nc_session *session) { - uint8_t q_id; int ret, ret2; NC_CHECK_ARG_RET(session, ps, session, -1); /* LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { + if (nc_ps_lock(ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return -1; } ret = _nc_ps_del_session(ps, session, -1); /* UNLOCK */ - ret2 = nc_ps_unlock(ps, q_id, __func__); + ret2 = nc_ps_unlock(ps, __func__); return ret || ret2 ? -1 : 0; } @@ -2137,13 +2255,12 @@ nc_ps_del_session(struct nc_pollsession *ps, struct nc_session *session) API struct nc_session * nc_ps_get_session(const struct nc_pollsession *ps, uint16_t idx) { - uint8_t q_id; struct nc_session *ret = NULL; NC_CHECK_ARG_RET(NULL, ps, NULL); /* LOCK */ - if (nc_ps_lock((struct nc_pollsession *)ps, &q_id, __func__)) { + if (nc_ps_lock((struct nc_pollsession *)ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return NULL; } @@ -2152,7 +2269,7 @@ nc_ps_get_session(const struct nc_pollsession *ps, uint16_t idx) } /* UNLOCK */ - nc_ps_unlock((struct nc_pollsession *)ps, q_id, __func__); + nc_ps_unlock((struct nc_pollsession *)ps, __func__); return ret; } @@ -2160,14 +2277,13 @@ nc_ps_get_session(const struct nc_pollsession *ps, uint16_t idx) API struct nc_session * nc_ps_find_session(const struct nc_pollsession *ps, nc_ps_session_match_cb match_cb, void *cb_data) { - uint8_t q_id; uint16_t i; struct nc_session *ret = NULL; NC_CHECK_ARG_RET(NULL, ps, NULL); /* LOCK */ - if (nc_ps_lock((struct nc_pollsession *)ps, &q_id, __func__)) { + if (nc_ps_lock((struct nc_pollsession *)ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return NULL; } @@ -2179,7 +2295,7 @@ nc_ps_find_session(const struct nc_pollsession *ps, nc_ps_session_match_cb match } /* UNLOCK */ - nc_ps_unlock((struct nc_pollsession *)ps, q_id, __func__); + nc_ps_unlock((struct nc_pollsession *)ps, __func__); return ret; } @@ -2187,20 +2303,19 @@ nc_ps_find_session(const struct nc_pollsession *ps, nc_ps_session_match_cb match API uint16_t nc_ps_session_count(struct nc_pollsession *ps) { - uint8_t q_id; uint16_t session_count; NC_CHECK_ARG_RET(NULL, ps, 0); /* LOCK (just for memory barrier so that we read the current value) */ - if (nc_ps_lock((struct nc_pollsession *)ps, &q_id, __func__)) { + if (nc_ps_lock((struct nc_pollsession *)ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return 0; } session_count = ps->session_count; /* UNLOCK */ - nc_ps_unlock((struct nc_pollsession *)ps, q_id, __func__); + nc_ps_unlock((struct nc_pollsession *)ps, __func__); return session_count; } @@ -2882,8 +2997,7 @@ nc_ps_poll_sess(struct nc_ps_session *ps_session, time_t now_mono) API int nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) { - int ret = NC_PSPOLL_ERROR, r; - uint8_t q_id; + int ret = NC_PSPOLL_ERROR, r, timeout_left = -1; uint16_t i, j; struct timespec ts_timeout, ts_cur; struct nc_session *cur_session; @@ -2897,24 +3011,27 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) *session = NULL; } + /* fill timespecs */ + if (timeout > -1) { + nc_timeouttime_get(&ts_timeout, timeout); + } + /* PS LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { - return NC_PSPOLL_ERROR; + r = nc_ps_lock(ps, 0, timeout, __func__); + if (r != 1) { + return r ? NC_PSPOLL_ERROR : NC_PSPOLL_TIMEOUT; } if (!ps->session_count) { - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); return NC_PSPOLL_NOSESSIONS; } - /* fill timespecs */ - nc_timeouttime_get(&ts_cur, 0); - if (timeout > -1) { - nc_timeouttime_get(&ts_timeout, timeout); - } - /* poll all the sessions one-by-one */ do { + /* current time, needed for the session idle timeout checks */ + nc_timeouttime_get(&ts_cur, 0); + /* loop from i to j once (all sessions) */ if (ps->last_event_session == ps->session_count - 1) { i = j = 0; @@ -2931,7 +3048,7 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) ret = NC_PSPOLL_ERROR; } else if (r == 1) { /* no one else is currently working with the session, so we can, otherwise skip it */ - ret = nc_ps_poll_sess(cur_ps_session, ts_timeout.tv_sec); + ret = nc_ps_poll_sess(cur_ps_session, ts_cur.tv_sec); /* keep RPC lock in this one case */ if (ret != NC_PSPOLL_RPC) { @@ -2959,9 +3076,25 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) if (ret == NC_PSPOLL_TIMEOUT) { usleep(NC_TIMEOUT_STEP); - if ((timeout > -1) && (nc_timeouttime_cur_diff(&ts_timeout) < 1)) { - /* final timeout */ - break; + if (timeout > -1) { + timeout_left = nc_timeouttime_cur_diff(&ts_timeout); + if (timeout_left < 1) { + /* final timeout */ + break; + } + } + + /* PS CHECK PREEMPT + * let the threads waiting for the pollsession in */ + r = nc_ps_check_preempt(ps, timeout_left, __func__); + if (r != 1) { + return r ? NC_PSPOLL_ERROR : NC_PSPOLL_TIMEOUT; + } + + if (!ps->session_count) { + /* all the sessions were removed while we did not have the turn */ + nc_ps_unlock(ps, __func__); + return NC_PSPOLL_NOSESSIONS; } } } while (ret == NC_PSPOLL_TIMEOUT); @@ -2985,7 +3118,7 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) } /* PS UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); /* we have some data available and the session is RPC locked (but not IO locked) */ if (ret == NC_PSPOLL_RPC) { @@ -3029,7 +3162,6 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) API void nc_ps_clear(struct nc_pollsession *ps, int all, void (*data_free)(void *)) { - uint8_t q_id; uint16_t i; struct nc_session *session; @@ -3039,7 +3171,7 @@ nc_ps_clear(struct nc_pollsession *ps, int all, void (*data_free)(void *)) } /* LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { + if (nc_ps_lock(ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return; } @@ -3066,7 +3198,7 @@ nc_ps_clear(struct nc_pollsession *ps, int all, void (*data_free)(void *)) } /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); } /** diff --git a/src/session_server_ssh.c b/src/session_server_ssh.c index fc97323e..d6bb4d58 100644 --- a/src/session_server_ssh.c +++ b/src/session_server_ssh.c @@ -2099,7 +2099,6 @@ nc_session_accept_ssh_channel(struct nc_session *orig_session, struct nc_session API NC_MSG_TYPE nc_ps_accept_ssh_channel(struct nc_pollsession *ps, struct nc_session **session) { - uint8_t q_id; NC_MSG_TYPE msgtype; struct nc_session *new_session = NULL, *cur_session; struct timespec ts_cur; @@ -2108,7 +2107,7 @@ nc_ps_accept_ssh_channel(struct nc_pollsession *ps, struct nc_session **session) NC_CHECK_ARG_RET(NULL, ps, session, NC_MSG_ERROR); /* LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { + if (nc_ps_lock(ps, 1, NC_PS_TIMEOUT, __func__) != 1) { return NC_MSG_ERROR; } @@ -2135,7 +2134,7 @@ nc_ps_accept_ssh_channel(struct nc_pollsession *ps, struct nc_session **session) } /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); if (!new_session) { ERR(NULL, "No session with a NETCONF SSH channel ready was found."); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 9531b4fb..e1fac8ef 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -64,6 +64,7 @@ libnetconf2_test(NAME test_client_messages) libnetconf2_test(NAME test_client_thread) libnetconf2_test(NAME test_fd_comm) libnetconf2_test(NAME test_io) +libnetconf2_test(NAME test_ps_poll) libnetconf2_test(NAME test_thread_messages) libnetconf2_test(NAME test_unix_socket) diff --git a/tests/test_ps_poll.c b/tests/test_ps_poll.c new file mode 100644 index 00000000..e5665ba2 --- /dev/null +++ b/tests/test_ps_poll.c @@ -0,0 +1,393 @@ +/** + * @file test_ps_poll.c + * @author Roman Janota + * @brief libnetconf2 tests - pollsession queue fairness + * + * @copyright + * Copyright (c) 2026 CESNET, z.s.p.o. + * + * This source code is licensed under BSD 3-Clause License (the "License"). + * You may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://opensource.org/licenses/BSD-3-Clause + */ + +#define _GNU_SOURCE + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +#include "ln2_test.h" + +/* long enough that a poll thread holding its turn for the whole timeout is unmistakable */ +#define TEST_POLL_TIMEOUT 1000 + +/* more threads than the queue was sized for, it has to grow */ +#define TEST_POLLER_COUNT (NC_PS_QUEUE_INIT_SIZE + 4) + +/* a high priority operation must get the turn within one session scan of the polling thread, + * not within one or more poll timeouts */ +#define TEST_ADD_LIMIT 250 + +struct test_state { + struct nc_pollsession *ps; + pthread_t tids[TEST_POLLER_COUNT]; + uint16_t poller_count; + ATOMIC_T stop; +}; + +/* socketpair peer ends of the created sessions, kept open so that the sessions are not + * reported as hung up, closed together with their session */ +static struct { + struct nc_session *sess; + int fd; +} test_peers[8]; + +/** + * @brief Remember the socketpair peer end of a session. + * + * @param[in] sess Session the peer end belongs to. + * @param[in] fd Peer end of the session socketpair. + * @return 0 on success, -1 on error. + */ +static int +test_peer_store(struct nc_session *sess, int fd) +{ + uint32_t i; + + for (i = 0; i < sizeof test_peers / sizeof *test_peers; ++i) { + if (!test_peers[i].sess) { + test_peers[i].sess = sess; + test_peers[i].fd = fd; + return 0; + } + } + + return -1; +} + +/** + * @brief Close the socketpair peer end of a session. + * + * @param[in] sess Session being freed. + */ +static void +test_peer_close(struct nc_session *sess) +{ + uint32_t i; + + for (i = 0; i < sizeof test_peers / sizeof *test_peers; ++i) { + if (test_peers[i].sess == sess) { + close(test_peers[i].fd); + test_peers[i].sess = NULL; + test_peers[i].fd = -1; + return; + } + } + + fail_msg("Session %" PRIu32 " has no stored socketpair peer end.", sess->id); +} + +/** + * @brief Create a bare server session on a socketpair, without any transport handshake. + * + * @param[in] id Session ID to use. + * @return Created session, NULL on error. + */ +static struct nc_session * +test_new_session(uint32_t id) +{ + struct nc_session *sess; + struct timespec ts; + int sock[2]; + + if (socketpair(AF_UNIX, SOCK_STREAM, 0, sock)) { + return NULL; + } + + sess = calloc(1, sizeof *sess); + if (!sess) { + close(sock[0]); + close(sock[1]); + return NULL; + } + + sess->side = NC_SERVER; + pthread_mutex_init(&sess->opts.server.ntf_status_lock, NULL); + pthread_mutex_init(&sess->opts.server.rpc_lock, NULL); + pthread_cond_init(&sess->opts.server.rpc_cond, NULL); + nc_timeouttime_get(&ts, 0); + sess->opts.server.last_rpc = ts.tv_sec; + + sess->io_lock = malloc(sizeof *sess->io_lock); + if (!sess->io_lock) { + free(sess); + close(sock[0]); + close(sock[1]); + return NULL; + } + pthread_mutex_init(sess->io_lock, NULL); + + NC_SESSION_STATUS_SET(sess, NC_STATUS_RUNNING); + sess->id = id; + sess->ti_type = NC_TI_FD; + sess->ti.fd.in = sock[0]; + sess->ti.fd.out = sock[0]; + + /* remember the socketpair peer end so that it can be closed with the session */ + if (test_peer_store(sess, sock[1])) { + pthread_mutex_destroy(&sess->opts.server.ntf_status_lock); + pthread_mutex_destroy(&sess->opts.server.rpc_lock); + pthread_cond_destroy(&sess->opts.server.rpc_cond); + pthread_mutex_destroy(sess->io_lock); + free(sess->io_lock); + free(sess); + close(sock[0]); + close(sock[1]); + return NULL; + } + + return sess; +} + +static void +test_free_session(struct nc_session *sess) +{ + close(sess->ti.fd.in); + test_peer_close(sess); + pthread_mutex_destroy(&sess->opts.server.ntf_status_lock); + pthread_mutex_destroy(&sess->opts.server.rpc_lock); + pthread_cond_destroy(&sess->opts.server.rpc_cond); + pthread_mutex_destroy(sess->io_lock); + free(sess->io_lock); + free(sess); +} + +/** + * @brief Poll the pollsession in a loop, like a server worker thread does. + * + * @param[in] arg Test state. + * @return NULL. + */ +static void * +test_poller_thread(void *arg) +{ + struct test_state *st = arg; + + while (!ATOMIC_LOAD_RELAXED(st->stop)) { + nc_ps_poll(st->ps, TEST_POLL_TIMEOUT, NULL); + } + + return NULL; +} + +static int +setup_f(void **state) +{ + struct test_state *st; + struct nc_session *sess; + uint16_t i; + + st = calloc(1, sizeof *st); + if (!st) { + SETUP_FAIL_LOG; + return 1; + } + ATOMIC_STORE_RELAXED(st->stop, 0); + + st->ps = nc_ps_new(); + if (!st->ps) { + SETUP_FAIL_LOG; + return 1; + } + + /* an already established, idle session, otherwise the pollers would just return no-sessions */ + sess = test_new_session(1); + if (!sess) { + SETUP_FAIL_LOG; + return 1; + } + if (nc_ps_add_session(st->ps, sess)) { + SETUP_FAIL_LOG; + return 1; + } + + /* start the pollers and let them settle into the poll loop */ + for (i = 0; i < TEST_POLLER_COUNT; ++i) { + if (pthread_create(&st->tids[i], NULL, test_poller_thread, st)) { + SETUP_FAIL_LOG; + return 1; + } + ++st->poller_count; + } + usleep(200000); + + *state = st; + return 0; +} + +static int +teardown_f(void **state) +{ + struct test_state *st = *state; + struct nc_session *sess; + uint16_t i; + + ATOMIC_STORE_RELAXED(st->stop, 1); + + /* remove the sessions first, the pollers then return right away instead of each waiting + * for its turn and timing out in it */ + while (nc_ps_session_count(st->ps)) { + sess = nc_ps_get_session(st->ps, 0); + nc_ps_del_session(st->ps, sess); + test_free_session(sess); + } + + for (i = 0; i < st->poller_count; ++i) { + pthread_join(st->tids[i], NULL); + } + + nc_ps_free(st->ps); + free(st); + + return 0; +} + +/** + * @brief Adding a session must not wait for the poll threads to time out. + * + * Session addition used to queue up behind all the poll threads in the same FIFO queue, so a + * newly established session waited poller_count * TEST_POLL_TIMEOUT before it was polled for + * the first time. It now queues up in front of them and the polling thread hands the turn over. + * Also verifies that the queue grows past NC_PS_QUEUE_INIT_SIZE instead of dropping the session. + */ +static void +test_add_session_not_blocked(void **state) +{ + struct test_state *st = *state; + struct nc_session *sess; + struct timespec ts_start; + int32_t elapsed; + + sess = test_new_session(2); + assert_non_null(sess); + + nc_timeouttime_get(&ts_start, 0); + assert_int_equal(nc_ps_add_session(st->ps, sess), 0); + elapsed = -nc_timeouttime_cur_diff(&ts_start); + + assert_int_equal(nc_ps_session_count(st->ps), 2); + assert_true(elapsed < TEST_ADD_LIMIT); +} + +/** + * @brief Removing a session must not be blocked by the idle poll threads either. + */ +static void +test_del_session_not_blocked(void **state) +{ + struct test_state *st = *state; + struct nc_session *sess; + struct timespec ts_start; + int32_t elapsed; + + sess = nc_ps_get_session(st->ps, 0); + assert_non_null(sess); + + nc_timeouttime_get(&ts_start, 0); + assert_int_equal(nc_ps_del_session(st->ps, sess), 0); + elapsed = -nc_timeouttime_cur_diff(&ts_start); + + assert_int_equal(nc_ps_session_count(st->ps), 0); + test_free_session(sess); + assert_true(elapsed < TEST_ADD_LIMIT); +} + +/** + * @brief The session idle timeout must be evaluated against the current time. + * + * It used to be checked against the nc_ps_poll() deadline instead, so a poll timeout longer + * than the remaining idle time terminated a perfectly active session right away. With an + * infinite timeout that timespec was not even initialized. + */ +static void +test_idle_timeout(void **state) +{ + struct nc_pollsession *ps; + struct nc_session *sess; + struct timespec ts, ts_start; + int ret; + int32_t elapsed; + + (void)state; + + ps = nc_ps_new(); + assert_non_null(ps); + + sess = test_new_session(1); + assert_non_null(sess); + assert_int_equal(nc_ps_add_session(ps, sess), 0); + + /* the session was active just now, so with an idle timeout of 2s it must survive for 2s, + * even though the poll deadline is further away than that */ + ATOMIC_STORE_RELAXED(server_opts.idle_timeout, 2); + + nc_timeouttime_get(&ts_start, 0); + ret = nc_ps_poll(ps, 2500, NULL); + elapsed = -nc_timeouttime_cur_diff(&ts_start); + + assert_int_equal(ret, NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR); + assert_int_equal(NC_SESSION_STATUS_GET(sess), NC_STATUS_INVALID); + assert_int_equal(NC_SESSION_TERM_REASON_GET(sess), NC_SESSION_TERM_TIMEOUT); + assert_true(elapsed >= 1000); + + assert_int_equal(nc_ps_del_session(ps, sess), 0); + test_free_session(sess); + + /* a session that really has been idle for too long is terminated right away */ + sess = test_new_session(2); + assert_non_null(sess); + nc_timeouttime_get(&ts, 0); + sess->opts.server.last_rpc = ts.tv_sec - 3; + assert_int_equal(nc_ps_add_session(ps, sess), 0); + + nc_timeouttime_get(&ts_start, 0); + ret = nc_ps_poll(ps, 2500, NULL); + elapsed = -nc_timeouttime_cur_diff(&ts_start); + + assert_int_equal(ret, NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR); + assert_int_equal(NC_SESSION_STATUS_GET(sess), NC_STATUS_INVALID); + assert_int_equal(NC_SESSION_TERM_REASON_GET(sess), NC_SESSION_TERM_TIMEOUT); + assert_true(elapsed < 500); + + ATOMIC_STORE_RELAXED(server_opts.idle_timeout, 0); + + assert_int_equal(nc_ps_del_session(ps, sess), 0); + test_free_session(sess); + nc_ps_free(ps); +} + +int +main(void) +{ + const struct CMUnitTest tests[] = { + cmocka_unit_test_setup_teardown(test_add_session_not_blocked, setup_f, teardown_f), + cmocka_unit_test_setup_teardown(test_del_session_not_blocked, setup_f, teardown_f), + cmocka_unit_test(test_idle_timeout), + }; + + return cmocka_run_group_tests(tests, NULL, NULL); +}