diff --git a/CMakeLists.txt b/CMakeLists.txt index 82f0e147..3fe59f01 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -58,8 +58,8 @@ set(CMAKE_MACOSX_RPATH TRUE) # minor version changes with added functionality (new tool, functionality of the tool or library, ...) and # micro version is changed with a set of small changes or bugfixes anywhere in the project. set(LIBNETCONF2_MAJOR_VERSION 4) -set(LIBNETCONF2_MINOR_VERSION 5) -set(LIBNETCONF2_MICRO_VERSION 1) +set(LIBNETCONF2_MINOR_VERSION 6) +set(LIBNETCONF2_MICRO_VERSION 0) set(LIBNETCONF2_VERSION ${LIBNETCONF2_MAJOR_VERSION}.${LIBNETCONF2_MINOR_VERSION}.${LIBNETCONF2_MICRO_VERSION}) # Version of the library @@ -67,7 +67,7 @@ set(LIBNETCONF2_VERSION ${LIBNETCONF2_MAJOR_VERSION}.${LIBNETCONF2_MINOR_VERSION # with backward compatible change and micro version is connected with any internal change of the library. set(LIBNETCONF2_MAJOR_SOVERSION 5) set(LIBNETCONF2_MINOR_SOVERSION 4) -set(LIBNETCONF2_MICRO_SOVERSION 18) +set(LIBNETCONF2_MICRO_SOVERSION 19) set(LIBNETCONF2_SOVERSION_FULL ${LIBNETCONF2_MAJOR_SOVERSION}.${LIBNETCONF2_MINOR_SOVERSION}.${LIBNETCONF2_MICRO_SOVERSION}) set(LIBNETCONF2_SOVERSION ${LIBNETCONF2_MAJOR_SOVERSION}) @@ -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/examples/server.c b/examples/server.c index 3834a773..fb669ff1 100644 --- a/examples/server.c +++ b/examples/server.c @@ -16,7 +16,6 @@ #define _GNU_SOURCE #include "example.h" -#include #include #include #include @@ -370,10 +369,8 @@ main(int argc, char **argv) ERR_MSG_CLEANUP("Error polling RPCs\n"); } - /* a session was terminated, so remove it from the ps structure and free it */ + /* a session was terminated, it is no longer in the ps structure and we own it now */ if (r & NC_PSPOLL_SESSION_TERM) { - r = nc_ps_del_session(ps, new_session); - assert(!r); nc_session_free(new_session, NULL); } 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.c b/src/session.c index dbc09ccf..99d8d877 100644 --- a/src/session.c +++ b/src/session.c @@ -500,6 +500,37 @@ nc_rwlock_unlock(pthread_rwlock_t *rwlock, const char *func_name) } } +int +nc_mutex_clocklock(pthread_mutex_t *mutex, const struct timespec *ts_deadline, const char *func_name) +{ + int r; + + if (!mutex) { + ERRINT; + return -1; + } + + if (ts_deadline) { + /* acquire the lock until the deadline, an expired one still locks a free mutex */ + r = pthread_mutex_clocklock(mutex, COMPAT_CLOCK_ID, ts_deadline); + } else { + /* acquire the lock without any timeout */ + r = pthread_mutex_lock(mutex); + } + + if (r) { + if ((r == EBUSY) || (r == ETIMEDOUT)) { + /* timeout, the caller knows what it asked for so it logs the details */ + return 0; + } + + ERR(NULL, "%s: failed to lock mutex (%s).", func_name, strerror(r)); + return -1; + } + + return 1; +} + int nc_mutex_lock(pthread_mutex_t *mutex, int timeout, const char *func_name) { diff --git a/src/session.h b/src/session.h index e1a05878..c8626c53 100644 --- a/src/session.h +++ b/src/session.h @@ -270,6 +270,10 @@ int nc_session_is_callhome(const struct nc_session *session); /** * @brief Free the NETCONF session object. * + * @warning The session must not be in any pollsession structure. A session terminated by + * ::nc_ps_poll() has been removed from it already, any other one has to be removed with + * ::nc_ps_del_session() first. + * * @param[in] session Object to free. * @param[in] data_free Session user data destructor. */ diff --git a/src/session_p.h b/src/session_p.h index 6540ceb8..725120ac 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. + * Only the session array is walked, so waiting this long means the ps queue is jammed. + */ +#define NC_PS_TIMEOUT 500 /** * @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) @@ -1208,28 +1208,21 @@ struct nc_session { */ #define NC_SESSION_TERM_REASON_SET(session, reason) ATOMIC_STORE_RELAXED((session)->term_reason, (uint32_t)(reason)) -enum nc_ps_session_state { - NC_PS_STATE_NONE = 0, /**< session is not being worked with */ - NC_PS_STATE_BUSY, /**< session is being polled or communicated on (and locked) */ - NC_PS_STATE_INVALID /**< session is invalid and was already returned by another poll */ -}; - -struct nc_ps_session { - struct nc_session *session; - enum nc_ps_session_state state; -}; - /* ACCESS locked */ struct nc_pollsession { - struct nc_ps_session **sessions; - 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] */ + struct nc_session **sessions; /**< array of the polled sessions, they are owned by the user */ + uint16_t session_count; /**< number of the polled sessions */ + uint16_t last_event_session; /**< index of the session that had the last event, the next poll + starts on the one after it */ + + 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 { @@ -1495,6 +1488,21 @@ int nc_rwlock_lock(pthread_rwlock_t *rwlock, enum nc_rwlock_mode mode, int timeo */ void nc_rwlock_unlock(pthread_rwlock_t *rwlock, const char *func_name); +/** + * @brief Lock a pthread_mutex until a deadline. + * + * @note Does not log a timeout, the caller knows the deadline it set and can describe it better. + * + * @param[in] mutex Mutex to be acquired. + * @param[in] ts_deadline Absolute time to wait for the lock until, NULL to wait indefinitely. + * An already expired deadline still acquires a free mutex, meaning it behaves as a trylock. + * @param[in] func_name Caller function name for logging purposes. + * @return 1 on success (lock acquired); + * @return 0 on timeout; + * @return -1 on error. + */ +int nc_mutex_clocklock(pthread_mutex_t *mutex, const struct timespec *ts_deadline, const char *func_name); + /** * @brief Lock a pthread_mutex with timeout support. * @@ -1518,9 +1526,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 7bfc3df5..e0587888 100644 --- a/src/session_server.c +++ b/src/session_server.c @@ -1862,154 +1862,281 @@ 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] ts_deadline Absolute time to wait for the turn until, NULL for no timeout. An + * already expired deadline means the turn is only taken if it is free right away. + * @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, const struct timespec *ts_deadline, const char *func) { - int r, rc = 0; - struct timespec ts; + int r; - /* LOCK */ - if (nc_mutex_lock(&ps->lock, NC_PS_LOCK_TIMEOUT, func) != 1) { - return -1; - } + /* 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 (ts_deadline) { + r = pthread_cond_clockwait(&ps->cond, &ps->lock, COMPAT_CLOCK_ID, ts_deadline); + } else { + r = pthread_cond_wait(&ps->cond, &ps->lock); + } - /* 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 (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 waiting for the pollsession turn.", func); + } + return 0; + } + if (r) { + ERR(NULL, "%s: failed to wait for a pollsession condition (%s).", func, strerror(r)); + return -1; + } } - /* 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); + /* take the turn, we are at the beginning of the queue */ + ps->busy = 1; - /* is it our turn? */ - while (ps->queue[ps->queue_begin] != *id) { - nc_timeouttime_get(&ts, NC_PS_QUEUE_TIMEOUT); + return 1; +} - 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)) { - break; - } +/** + * @brief Take the pollsession lock and the turn to work with it. + * + * @note @p ps->lock MUST NOT be held. + * + * @param[in,out] ps Pollsession structure. + * @param[in] preempt Whether this thread preempts the poll thread that has the turn. + * @param[in] ts_deadline Absolute time to wait for the turn until, NULL 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. + */ +static int +_nc_ps_lock(struct nc_pollsession *ps, int preempt, const struct timespec *ts_deadline, const char *func) +{ + int rc; - 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; + /* LOCK */ + rc = nc_mutex_clocklock(&ps->lock, ts_deadline, func); + if (rc != 1) { + if (!rc && preempt) { + ERR(NULL, "%s: timed out waiting for the pollsession lock.", func); } + return rc; } + /* add ourselves into the queue */ + if (nc_ps_queue_add(ps, preempt)) { + rc = -1; + goto cleanup; + } + + /* is it our turn? */ + rc = nc_ps_queue_wait_turn(ps, preempt, ts_deadline, 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, uint8_t id, const char *func) +nc_ps_lock(struct nc_pollsession *ps, int preempt, int timeout_ms, const char *func) { - int r; + struct timespec ts_deadline; - /* LOCK, continue on error */ - r = nc_mutex_lock(&ps->lock, NC_PS_LOCK_TIMEOUT, func); + if (timeout_ms < 0) { + /* no timeout */ + return _nc_ps_lock(ps, preempt, NULL, func); + } - /* 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); - } + /* an already expired deadline means no waiting at all */ + nc_timeouttime_get(&ts_deadline, timeout_ms); + return _nc_ps_lock(ps, preempt, &ts_deadline, func); +} + +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] ts_deadline Absolute time to wait for the turn back until, NULL 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, const struct timespec *ts_deadline, const char *func) +{ + int rc = 1; + + /* LOCK */ + if (nc_mutex_clocklock(&ps->lock, ts_deadline, func) != 1) { + /* the turn was not given up, keep it */ + return 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, ts_deadline, func); + if (rc != 1) { + nc_ps_queue_remove(ps); } - return r == 1 ? 0 : -1; +cleanup: + /* UNLOCK */ + nc_mutex_unlock(&ps->lock, func); + return rc; } API struct nc_pollsession * @@ -2019,8 +2146,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; } @@ -2028,8 +2155,6 @@ nc_ps_new(void) API void nc_ps_free(struct nc_pollsession *ps) { - uint16_t i; - if (!ps) { return; } @@ -2038,13 +2163,10 @@ nc_ps_free(struct nc_pollsession *ps) ERR(NULL, "FATAL: Freeing a pollsession structure that is currently being worked with!"); } - for (i = 0; i < ps->session_count; i++) { - free(ps->sessions[i]); - } - free(ps->sessions); - pthread_mutex_destroy(&ps->lock); + free(ps->queue); pthread_cond_destroy(&ps->cond); + pthread_mutex_destroy(&ps->lock); free(ps); } @@ -2052,38 +2174,43 @@ 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; + struct nc_session **sessions; 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; } - ++ps->session_count; - ps->sessions = nc_realloc(ps->sessions, ps->session_count * sizeof *ps->sessions); - if (!ps->sessions) { + sessions = realloc(ps->sessions, (ps->session_count + 1) * sizeof *ps->sessions); + if (!sessions) { ERRMEM; + /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); - return -1; - } - ps->sessions[ps->session_count - 1] = calloc(1, sizeof **ps->sessions); - if (!ps->sessions[ps->session_count - 1]) { - 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; + ps->sessions = sessions; + ps->sessions[ps->session_count++] = session; /* UNLOCK */ - return nc_ps_unlock(ps, q_id, __func__); + return nc_ps_unlock(ps, __func__); } +/** + * @brief Remove a session from a pollsession structure. + * + * @note The pollsession turn MUST be held. + * + * The session itself is not freed, the caller becomes its owner. + * + * @param[in,out] ps Pollsession structure to remove from. + * @param[in] session NETCONF session to remove, used only if @p index is negative. + * @param[in] index Index of the session to remove, negative to look @p session up. + * @return 0 on success. + * @return -1 if @p session was not found. + */ static int _nc_ps_del_session(struct nc_pollsession *ps, struct nc_session *session, int index) { @@ -2094,13 +2221,10 @@ _nc_ps_del_session(struct nc_pollsession *ps, struct nc_session *session, int in goto remove; } for (i = 0; i < ps->session_count; ++i) { - if (ps->sessions[i]->session == session) { + if (ps->sessions[i] == session) { remove: --ps->session_count; - if (i <= ps->session_count) { - free(ps->sessions[i]); - ps->sessions[i] = ps->sessions[ps->session_count]; - } + ps->sessions[i] = ps->sessions[ps->session_count]; if (!ps->session_count) { free(ps->sessions); ps->sessions = NULL; @@ -2116,20 +2240,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,22 +2260,21 @@ 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; } if (idx < ps->session_count) { - ret = ps->sessions[idx]->session; + ret = ps->sessions[idx]; } /* UNLOCK */ - nc_ps_unlock((struct nc_pollsession *)ps, q_id, __func__); + nc_ps_unlock((struct nc_pollsession *)ps, __func__); return ret; } @@ -2160,26 +2282,25 @@ 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; } for (i = 0; i < ps->session_count; ++i) { - if (match_cb(ps->sessions[i]->session, cb_data)) { - ret = ps->sessions[i]->session; + if (match_cb(ps->sessions[i], cb_data)) { + ret = ps->sessions[i]; break; } } /* UNLOCK */ - nc_ps_unlock((struct nc_pollsession *)ps, q_id, __func__); + nc_ps_unlock((struct nc_pollsession *)ps, __func__); return ret; } @@ -2187,20 +2308,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; } @@ -2642,18 +2762,18 @@ nc_ps_ssh_find_new_channel(struct nc_session *session) * however, hold the session RPC lock and the session must be running. * * @param[in] session Session to use. - * @param[in] io_timeout Timeout to use for acquiring IO lock. * @param[in] now_mono Current monotonic timestamp. * @param[in,out] msg Message to fill in case of an error. * @return NC_PSPOLL_RPC if some application data are available. - * @return NC_PSPOLL_TIMEOUT if a timeout elapsed. + * @return NC_PSPOLL_TIMEOUT if there are no application data or the session IO lock is held + * by another thread, in both cases there is nothing to do with the session right now. * @return NC_PSPOLL_SSH_CHANNEL if a new SSH channel has been created. * @return NC_PSPOLL_SSH_MSG if just an SSH message has been processed. * @return NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR if session has been terminated (@p msg filled). * @return NC_PSPOLL_ERROR on other fatal errors (@p msg filled). */ static int -nc_ps_poll_session_io(struct nc_session *session, int io_timeout, time_t now_mono, char *msg) +nc_ps_poll_session_io(struct nc_session *session, time_t now_mono, char *msg) { struct pollfd pfd; int r, ret = 0; @@ -2675,7 +2795,11 @@ nc_ps_poll_session_io(struct nc_session *session, int io_timeout, time_t now_mon return NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR; } - r = nc_mutex_lock(session->io_lock, io_timeout, __func__); + /* only a non-blocking check follows, so never wait for the IO lock - the pollsession turn is + * held here and ::nc_server_notif_send() may keep the lock for an unbounded time (::nc_write() + * has no deadline, a peer that stops reading blocks it). Skipping costs the session one poll + * interval, waiting would stall the whole pollsession */ + r = nc_mutex_lock(session->io_lock, 0, __func__); if (r < 0) { return NC_PSPOLL_ERROR; } else if (!r) { @@ -2814,80 +2938,47 @@ nc_ps_poll_session_io(struct nc_session *session, int io_timeout, time_t now_mon /** * @brief Poll a single pspoll session. * - * @param[in] ps_session pspoll session to poll. + * @param[in] session Session to poll. * @param[in] now_mono Current monotonic timestamp. * @return NC_PSPOLL_RPC if some application data are available. - * @return NC_PSPOLL_TIMEOUT if a timeout elapsed. + * @return NC_PSPOLL_TIMEOUT if there is nothing to do with the session right now, it is polled + * again the next poll interval. * @return NC_PSPOLL_SSH_CHANNEL if a new SSH channel has been created. * @return NC_PSPOLL_SSH_MSG if just an SSH message has been processed. * @return NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR if session has been terminated. * @return NC_PSPOLL_ERROR on other fatal errors. */ static int -nc_ps_poll_sess(struct nc_ps_session *ps_session, time_t now_mono) +nc_ps_poll_sess(struct nc_session *session, time_t now_mono) { - int ret = NC_PSPOLL_ERROR; + int rc = NC_PSPOLL_ERROR; char msg[256]; - switch (ps_session->state) { - case NC_PS_STATE_NONE: - if (NC_SESSION_STATUS_GET(ps_session->session) == NC_STATUS_RUNNING) { - /* session is fine, work with it, no configuration is accessed */ - ps_session->state = NC_PS_STATE_BUSY; - ret = nc_ps_poll_session_io(ps_session->session, NC_SESSION_LOCK_TIMEOUT, now_mono, msg); - - switch (ret) { - case NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR: - ERR(ps_session->session, "%s.", msg); - ps_session->state = NC_PS_STATE_INVALID; - break; - case NC_PSPOLL_ERROR: - ERR(ps_session->session, "%s.", msg); - ps_session->state = NC_PS_STATE_NONE; - break; - case NC_PSPOLL_TIMEOUT: -#ifdef NC_ENABLED_SSH_TLS - case NC_PSPOLL_SSH_CHANNEL: - case NC_PSPOLL_SSH_MSG: -#endif /* NC_ENABLED_SSH_TLS */ - ps_session->state = NC_PS_STATE_NONE; - break; - case NC_PSPOLL_RPC: - /* let's keep the state busy, we are not done with this session */ - break; - } - } else { - /* session is not fine, let the caller know */ - ret = NC_PSPOLL_SESSION_TERM; - if (NC_SESSION_TERM_REASON_GET(ps_session->session) != NC_SESSION_TERM_CLOSED) { - ret |= NC_PSPOLL_SESSION_ERROR; - } - ps_session->state = NC_PS_STATE_INVALID; + if (NC_SESSION_STATUS_GET(session) != NC_STATUS_RUNNING) { + /* session is not fine, let the caller know, it is removed from the pollsession by it */ + rc = NC_PSPOLL_SESSION_TERM; + if (NC_SESSION_TERM_REASON_GET(session) != NC_SESSION_TERM_CLOSED) { + rc |= NC_PSPOLL_SESSION_ERROR; + } + } else { + /* session is fine, work with it, no configuration is accessed */ + rc = nc_ps_poll_session_io(session, now_mono, msg); + if ((rc == NC_PSPOLL_ERROR) || (rc == (NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR))) { + ERR(session, "%s.", msg); } - break; - case NC_PS_STATE_BUSY: - /* it definitely should not be busy because we have the lock */ - ERRINT; - ret = NC_PSPOLL_ERROR; - break; - case NC_PS_STATE_INVALID: - /* we got it locked, but it will be freed, let it be */ - ret = NC_PSPOLL_TIMEOUT; - break; } - return ret; + return rc; } 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; uint16_t i, j; struct timespec ts_timeout, ts_cur; + const struct timespec *ts_deadline = NULL; struct nc_session *cur_session; - struct nc_ps_session *cur_ps_session; struct nc_server_rpc *rpc = NULL; NC_SESSION_TERM_REASON term_reason; @@ -2897,24 +2988,29 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) *session = NULL; } - /* PS LOCK */ - if (nc_ps_lock(ps, &q_id, __func__)) { - return NC_PSPOLL_ERROR; - } - - if (!ps->session_count) { - nc_ps_unlock(ps, q_id, __func__); - return NC_PSPOLL_NOSESSIONS; - } - - /* fill timespecs */ - nc_timeouttime_get(&ts_cur, 0); + /* fill timespecs, the deadline is kept for the whole call */ if (timeout > -1) { nc_timeouttime_get(&ts_timeout, timeout); + ts_deadline = &ts_timeout; + } + + /* PS LOCK */ + r = _nc_ps_lock(ps, 0, ts_deadline, __func__); + if (r != 1) { + return r ? NC_PSPOLL_ERROR : NC_PSPOLL_TIMEOUT; } /* poll all the sessions one-by-one */ do { + if (!ps->session_count) { + /* there were none to begin with or they were all removed while we did not have the turn */ + nc_ps_unlock(ps, __func__); + return NC_PSPOLL_NOSESSIONS; + } + + /* 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; @@ -2922,8 +3018,7 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) i = j = ps->last_event_session + 1; } do { - cur_ps_session = ps->sessions[i]; - cur_session = cur_ps_session->session; + cur_session = ps->sessions[i]; /* SESSION RPC LOCK */ r = nc_session_rpc_lock(cur_session, 0, __func__); @@ -2931,7 +3026,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_session, ts_cur.tv_sec); /* keep RPC lock in this one case */ if (ret != NC_PSPOLL_RPC) { @@ -2959,18 +3054,35 @@ 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)) { + if (ts_deadline && (nc_timeouttime_cur_diff(ts_deadline) < 1)) { /* final timeout */ break; } + + /* PS CHECK PREEMPT + * let the threads waiting for the pollsession in */ + r = nc_ps_check_preempt(ps, ts_deadline, __func__); + if (r != 1) { + return r ? NC_PSPOLL_ERROR : NC_PSPOLL_TIMEOUT; + } } } while (ret == NC_PSPOLL_TIMEOUT); /* do we want to return the session? */ switch (ret) { - case NC_PSPOLL_RPC: case NC_PSPOLL_SESSION_TERM: case NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR: + /* the session is dead, remove it from ps while we still have the turn so that it is + * reported exactly once, the caller becomes its owner */ + _nc_ps_del_session(ps, NULL, i); + + /* not setting last_event_session as below, index i may be out of bounds after the removal and the + * removal restarted the round-robin anyway */ + if (session) { + *session = cur_session; + } + break; + case NC_PSPOLL_RPC: #ifdef NC_ENABLED_SSH_TLS case NC_PSPOLL_SSH_CHANNEL: case NC_PSPOLL_SSH_MSG: @@ -2985,7 +3097,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) { @@ -2994,14 +3106,8 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) /* error, do not send a reply */ if (NC_SESSION_STATUS_GET(cur_session) != NC_STATUS_RUNNING) { ret |= NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR; - cur_ps_session->state = NC_PS_STATE_INVALID; - } else { - cur_ps_session->state = NC_PS_STATE_NONE; } - } else if (ret & NC_PSPOLL_REPLY_ERROR) { - /* error reply has been sent */ - cur_ps_session->state = NC_PS_STATE_NONE; - } else { + } else if (!(ret & NC_PSPOLL_REPLY_ERROR)) { cur_session->opts.server.last_rpc = ts_cur.tv_sec; /* process RPC and send a reply */ @@ -3012,13 +3118,28 @@ nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session) if ((term_reason != NC_SESSION_TERM_CLOSED) && (term_reason != NC_SESSION_TERM_KILLED)) { ret |= NC_PSPOLL_SESSION_ERROR; } - cur_ps_session->state = NC_PS_STATE_INVALID; - } else { - cur_ps_session->state = NC_PS_STATE_NONE; } } nc_server_rpc_free(rpc); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session died during the RPC, when the turn was not held, take it back and remove + * the session before handing it over to the caller */ + if (nc_ps_lock(ps, 1, timeout, __func__) == 1) { + r = _nc_ps_del_session(ps, cur_session, -1); + + /* PS UNLOCK */ + nc_ps_unlock(ps, __func__); + } else { + r = -1; + } + if (r) { + /* either another thread removed the session and owns it now, or the turn timed out + * and the session is still in ps for a later poll to report, not ours to hand over */ + ret &= ~(NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR); + } + } + /* SESSION RPC UNLOCK */ nc_session_rpc_unlock(cur_session, NC_SESSION_LOCK_TIMEOUT, __func__); } @@ -3029,9 +3150,8 @@ 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; + uint16_t i, count = 0; + struct nc_session **sessions = NULL; if (!ps) { ERRARG(NULL, "ps"); @@ -3039,25 +3159,31 @@ 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; } + if (ps->session_count) { + /* freeing a session takes a while, so only collect them here and free them once the turn is given up */ + sessions = malloc(ps->session_count * sizeof *sessions); + NC_CHECK_ERRMEM_GOTO(!sessions, , cleanup); + } + if (all) { for (i = 0; i < ps->session_count; i++) { - nc_session_free(ps->sessions[i]->session, data_free); - free(ps->sessions[i]); + sessions[i] = ps->sessions[i]; } + count = ps->session_count; + free(ps->sessions); ps->sessions = NULL; ps->session_count = 0; ps->last_event_session = 0; } else { for (i = 0; i < ps->session_count; ) { - if (NC_SESSION_STATUS_GET(ps->sessions[i]->session) != NC_STATUS_RUNNING) { - session = ps->sessions[i]->session; + if (NC_SESSION_STATUS_GET(ps->sessions[i]) != NC_STATUS_RUNNING) { + sessions[count++] = ps->sessions[i]; _nc_ps_del_session(ps, NULL, i); - nc_session_free(session, data_free); continue; } @@ -3065,8 +3191,17 @@ nc_ps_clear(struct nc_pollsession *ps, int all, void (*data_free)(void *)) } } +cleanup: /* UNLOCK */ - nc_ps_unlock(ps, q_id, __func__); + nc_ps_unlock(ps, __func__); + + /* free the sessions only once the turn was given up, ::nc_session_free() waits for the session + * RPC lock, which a poll thread keeps while it no longer has the turn, and which it needs to + * take the turn back in ::nc_ps_poll() */ + for (i = 0; i < count; i++) { + nc_session_free(sessions[i], data_free); + } + free(sessions); } /** diff --git a/src/session_server.h b/src/session_server.h index e9e94b1a..4ddb1da8 100644 --- a/src/session_server.h +++ b/src/session_server.h @@ -361,15 +361,18 @@ uint16_t nc_ps_session_count(struct nc_pollsession *ps); /** * @brief Poll sessions and process any received RPCs. * - * Only one event on one session is handled in one function call. If this event - * is a session termination (::NC_PSPOLL_SESSION_TERM returned), the session - * should be removed from @p ps. + * Only one event on one session is handled in one function call. If this event is a session + * termination (::NC_PSPOLL_SESSION_TERM returned), the session was removed from @p ps and its + * ownership passed to the caller, see @p session. * * @param[in] ps Pollsession structure to use. * @param[in] timeout Poll timeout in milliseconds. 0 for non-blocking call, -1 for * infinite waiting. If ::NC_PSPOLL_NOSESSIONS is returned, no waiting is performed at all. - * @param[in] session Session that was processed and that specific return bits concern. - * Can be NULL. + * It bounds waiting for an event, not the whole call, which may take longer. + * @param[out] session Session that was processed and that specific return bits concern. If + * ::NC_PSPOLL_SESSION_TERM is returned, the session was removed from @p ps and the caller owns + * it, it must be freed with ::nc_session_free(). Can be NULL, but then a terminated session is + * leaked. Otherwise the session is only borrowed and must not be freed. * @return Bitfield of NC_PSPOLL_* macros. */ int nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **session); @@ -378,7 +381,8 @@ int nc_ps_poll(struct nc_pollsession *ps, int timeout, struct nc_session **sessi * @brief Remove sessions from a pollsession structure and * call ::nc_session_free() on them. * - * Calling this function with @p all false makes sense if ::nc_ps_poll() returned ::NC_PSPOLL_SESSION_TERM. + * Sessions terminated by ::nc_ps_poll() are not in @p ps anymore, so @p all false only finds + * those invalidated without a poll, such as by ::nc_session_set_status() from another thread. * * @param[in] ps Pollsession structure to clear. * @param[in] all Whether to free all sessions, or only the invalid ones. diff --git a/src/session_server_ssh.c b/src/session_server_ssh.c index ddab3014..32f03599 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,12 +2107,12 @@ 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; } for (i = 0; i < ps->session_count; ++i) { - cur_session = ps->sessions[i]->session; + cur_session = ps->sessions[i]; if ((NC_SESSION_STATUS_GET(cur_session) == NC_STATUS_RUNNING) && (cur_session->ti_type == NC_TI_SSH) && cur_session->ti.libssh.next) { /* an SSH session with more channels */ @@ -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/ln2_test.c b/tests/ln2_test.c index 0d1ff71b..10ab790e 100644 --- a/tests/ln2_test.c +++ b/tests/ln2_test.c @@ -88,11 +88,12 @@ ln2_glob_test_server_thread(void *arg) /* poll until the session is terminated by the client */ do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); assert(ret & NC_PSPOLL_RPC); } while (!(ret & NC_PSPOLL_SESSION_TERM)); - nc_ps_clear(ps, 1, NULL); + /* the terminated session was removed from ps and we own it now */ + nc_session_free(session, NULL); nc_ps_free(ps); return NULL; } diff --git a/tests/test_authkeys.c b/tests/test_authkeys.c index 25b8862d..138a3c32 100644 --- a/tests/test_authkeys.c +++ b/tests/test_authkeys.c @@ -66,11 +66,12 @@ server_thread(void *arg) assert_int_equal(ret, 0); do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); assert_int_equal(ret & NC_PSPOLL_RPC, NC_PSPOLL_RPC); } while (!(ret & NC_PSPOLL_SESSION_TERM)); - nc_ps_clear(ps, 1, NULL); + /* the terminated session was removed from ps and we own it now */ + nc_session_free(session, NULL); nc_ps_free(ps); return NULL; } diff --git a/tests/test_ch.c b/tests/test_ch.c index 95053200..3822d395 100644 --- a/tests/test_ch.c +++ b/tests/test_ch.c @@ -90,6 +90,7 @@ static void * server_thread_ssh(void *arg) { int ret; + struct nc_session *session; struct nc_pollsession *ps; struct ln2_test_ctx *test_ctx = arg; struct test_ch_data *test_data = test_ctx->test_data; @@ -115,8 +116,11 @@ server_thread_ssh(void *arg) /* poll */ do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); - if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session was removed from ps and we own it now */ + nc_session_free(session, NULL); + } else if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { usleep(500); } } while (!strlen(buffer)); @@ -254,6 +258,7 @@ static void * server_thread_tls(void *arg) { int ret; + struct nc_session *session; struct nc_pollsession *ps; struct ln2_test_ctx *test_ctx = arg; struct test_ch_data *test_data = test_ctx->test_data; @@ -274,8 +279,11 @@ server_thread_tls(void *arg) /* poll */ do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); - if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session was removed from ps and we own it now */ + nc_session_free(session, NULL); + } else if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { usleep(500); } } while (!(ret & NC_PSPOLL_SESSION_TERM)); @@ -419,6 +427,7 @@ static void * server_thread_delete_while_session(void *arg) { int ret; + struct nc_session *session; struct nc_pollsession *ps; struct ln2_test_ctx *test_ctx = arg; struct test_ch_data *test_data = test_ctx->test_data; @@ -441,8 +450,11 @@ server_thread_delete_while_session(void *arg) } do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); - if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session was removed from ps and we own it now */ + nc_session_free(session, NULL); + } else if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { usleep(500); } } while (ret & NC_PSPOLL_RPC); @@ -595,6 +607,7 @@ static void * server_thread_two_ch(void *arg) { int ret; + struct nc_session *session; struct nc_pollsession *ps; struct ch_thread_arg *ch_arg = arg; @@ -615,8 +628,11 @@ server_thread_two_ch(void *arg) } pthread_mutex_unlock(&session_count_mutex); - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); - if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session was removed from ps and we own it now */ + nc_session_free(session, NULL); + } else if (ret & (NC_PSPOLL_TIMEOUT | NC_PSPOLL_NOSESSIONS)) { usleep(10000); } } diff --git a/tests/test_client_monitoring.c b/tests/test_client_monitoring.c index b6a1ac85..c398777d 100644 --- a/tests/test_client_monitoring.c +++ b/tests/test_client_monitoring.c @@ -130,11 +130,19 @@ server_thread(void *arg) /* poll until the client stops sending messages */ do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); + if (ret & NC_PSPOLL_SESSION_TERM) { + /* the session was removed from ps and we own it now */ + break; + } } while ((ret & NC_PSPOLL_RPC)); /* free the session (it will close the socket -> client needs to detect this) */ - nc_ps_clear(ps, 1, NULL); + if (ret & NC_PSPOLL_SESSION_TERM) { + nc_session_free(session, NULL); + } else { + nc_ps_clear(ps, 1, NULL); + } nc_ps_free(ps); return NULL; } diff --git a/tests/test_ps_poll.c b/tests/test_ps_poll.c new file mode 100644 index 00000000..b4afcbb9 --- /dev/null +++ b/tests/test_ps_poll.c @@ -0,0 +1,453 @@ +/** + * @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, *term_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, &term_sess); + 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); + + /* the session was handed over to us, it is no longer in ps */ + assert_ptr_equal(term_sess, sess); + assert_int_equal(nc_ps_session_count(ps), 0); + assert_int_equal(nc_ps_del_session(ps, sess), -1); + 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, &term_sess); + 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_ptr_equal(term_sess, sess); + assert_int_equal(nc_ps_session_count(ps), 0); + test_free_session(sess); + nc_ps_free(ps); +} + +/** + * @brief Terminating the last session in the array must not corrupt the round-robin index. + * + * ::nc_ps_poll() reports a terminated session and removes it from the pollsession in one step, + * so the index it was found on no longer exists. Remembering it as the last event index made + * the next poll start one past the end of the session array. + */ +static void +test_term_last_session(void **state) +{ + struct nc_pollsession *ps; + struct nc_session *sess[3], *term_sess; + struct timespec ts; + uint16_t i; + int ret; + + (void)state; + + ps = nc_ps_new(); + assert_non_null(ps); + + for (i = 0; i < 3; ++i) { + sess[i] = test_new_session(i + 1); + assert_non_null(sess[i]); + } + + /* the last session in the array has been idle for too long, the other two are fine */ + nc_timeouttime_get(&ts, 0); + sess[2]->opts.server.last_rpc = ts.tv_sec - 3; + for (i = 0; i < 3; ++i) { + assert_int_equal(nc_ps_add_session(ps, sess[i]), 0); + } + ATOMIC_STORE_RELAXED(server_opts.idle_timeout, 2); + + /* the last session is terminated, removed and handed over */ + ret = nc_ps_poll(ps, 0, &term_sess); + assert_int_equal(ret, NC_PSPOLL_SESSION_TERM | NC_PSPOLL_SESSION_ERROR); + assert_ptr_equal(term_sess, sess[2]); + assert_int_equal(nc_ps_session_count(ps), 2); + test_free_session(sess[2]); + + /* the next poll must stay inside the session array */ + ret = nc_ps_poll(ps, 0, &term_sess); + assert_int_equal(ret, NC_PSPOLL_TIMEOUT); + assert_int_equal(nc_ps_session_count(ps), 2); + + ATOMIC_STORE_RELAXED(server_opts.idle_timeout, 0); + + for (i = 0; i < 2; ++i) { + assert_int_equal(nc_ps_del_session(ps, sess[i]), 0); + test_free_session(sess[i]); + } + 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), + cmocka_unit_test(test_term_last_session), + }; + + return cmocka_run_group_tests(tests, NULL, NULL); +} diff --git a/tests/test_runtime_changes.c b/tests/test_runtime_changes.c index c00b64c4..7a5cd531 100644 --- a/tests/test_runtime_changes.c +++ b/tests/test_runtime_changes.c @@ -74,10 +74,12 @@ server_thread(void *arg) assert_int_equal(ret, 0); do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); assert_int_equal(ret & NC_PSPOLL_RPC, NC_PSPOLL_RPC); } while (!(ret & NC_PSPOLL_SESSION_TERM)); - nc_ps_clear(ps, 1, NULL); + + /* the terminated session was removed from ps and we own it now */ + nc_session_free(session, NULL); } else { assert_int_equal(msgtype, NC_MSG_ERROR); } diff --git a/tests/test_tls.c b/tests/test_tls.c index ed70e9d5..e21e15bc 100644 --- a/tests/test_tls.c +++ b/tests/test_tls.c @@ -1007,11 +1007,12 @@ server_thread_ctn_san(void *arg) /* poll until the session is terminated by the client */ do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); assert_int_equal(ret & NC_PSPOLL_RPC, NC_PSPOLL_RPC); } while (!(ret & NC_PSPOLL_SESSION_TERM)); - nc_ps_clear(ps, 1, NULL); + /* the terminated session was removed from ps and we own it now */ + nc_session_free(session, NULL); nc_ps_free(ps); return NULL; } diff --git a/tests/test_two_channels.c b/tests/test_two_channels.c index 9d9e97dd..4b8c15a4 100644 --- a/tests/test_two_channels.c +++ b/tests/test_two_channels.c @@ -55,7 +55,7 @@ server_thread(void *arg) ret = nc_ps_poll(ps, 0, &session); if (ret & NC_PSPOLL_SESSION_TERM) { - nc_ps_del_session(ps, session); + /* the session was removed from ps and we own it now */ nc_session_free(session, NULL); del_session_count++; } else if (ret & NC_PSPOLL_SSH_CHANNEL) { diff --git a/tests/test_unix_socket.c b/tests/test_unix_socket.c index 1d888184..eb6cc485 100644 --- a/tests/test_unix_socket.c +++ b/tests/test_unix_socket.c @@ -315,10 +315,12 @@ auth_server_thread(void *arg) ret = nc_ps_add_session(ps, session); assert_int_equal(ret, 0); do { - ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, NULL); + ret = nc_ps_poll(ps, NC_PS_POLL_TIMEOUT, &session); assert_true(ret & NC_PSPOLL_RPC); } while (!(ret & NC_PSPOLL_SESSION_TERM)); - nc_ps_clear(ps, 1, NULL); + + /* the terminated session was removed from ps and we own it now */ + nc_session_free(session, NULL); nc_ps_free(ps); return NULL;