Fix pollsession turn queue starving session addition and removal

CMakeLists.txt

@@ -58,16 +58,16 @@ 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
 # Major version is changed with every backward non-compatible API/ABI change in the library, minor version changes
 # 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")

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

examples/server.c

@@ -16,7 +16,6 @@
 #define _GNU_SOURCE
 #include "example.h"
 
-#include <assert.h>
 #include <getopt.h>
 #include <signal.h>
 #include <stdint.h>
@@ -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);
         }
 

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@

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, ...).

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)
 {

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.
  */

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