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); +}