session FEATURE session msgs lock with timeout

Otherwise the timeout can easily be ignored
and used only after the msgs lock is held,
which can take some time on its own.

Author: michalvasko

src/session.c

@@ -344,11 +344,6 @@ nc_session_rpc_unlock(struct nc_session *session, int timeout, const char *func)
     return 1;
 }
 
-/*
- * @return 1 - success
- *         0 - timeout
- *        -1 - error
- */
 int
 nc_session_io_lock(struct nc_session *session, int timeout, const char *func)
 {
@@ -395,6 +390,66 @@ nc_session_io_unlock(struct nc_session *session, const char *func)
     return 1;
 }
 
+int
+nc_session_client_msgs_lock(struct nc_session *session, int *timeout, const char *func)
+{
+    int ret;
+    int32_t diff_msec;
+    struct timespec ts_timeout, ts_start, ts_end;
+
+    assert(session->side == NC_CLIENT);
+
+    if (*timeout > 0) {
+        /* get current time */
+        nc_gettimespec_real(&ts_start);
+
+        nc_gettimespec_real(&ts_timeout);
+        nc_addtimespec(&ts_timeout, *timeout);
+
+        ret = pthread_mutex_timedlock(&session->opts.client.msgs_lock, &ts_timeout);
+        if (!ret) {
+            /* update timeout based on what was elapsed */
+            nc_gettimespec_real(&ts_end);
+            diff_msec = nc_difftimespec(&ts_start, &ts_end);
+            *timeout -= diff_msec;
+        }
+    } else if (!*timeout) {
+        ret = pthread_mutex_trylock(&session->opts.client.msgs_lock);
+    } else { /* timeout == -1 */
+        ret = pthread_mutex_lock(&session->opts.client.msgs_lock);
+    }
+
+    if (ret) {
+        if ((ret == EBUSY) || (ret == ETIMEDOUT)) {
+            /* timeout */
+            return 0;
+        }
+
+        /* error */
+        ERR(session, "%s: failed to MSGS lock a session (%s).", func, strerror(ret));
+        return -1;
+    }
+
+    return 1;
+}
+
+int
+nc_session_client_msgs_unlock(struct nc_session *session, const char *func)
+{
+    int ret;
+
+    assert(session->side == NC_CLIENT);
+
+    ret = pthread_mutex_unlock(&session->opts.client.msgs_lock);
+    if (ret) {
+        /* error */
+        ERR(session, "%s: failed to MSGS unlock a session (%s).", func, strerror(ret));
+        return -1;
+    }
+
+    return 1;
+}
+
 API NC_STATUS
 nc_session_get_status(const struct nc_session *session)
 {
@@ -555,7 +610,7 @@ nc_send_msg_io(struct nc_session *session, int io_timeout, struct lyd_node *op)
 API void
 nc_session_free(struct nc_session *session, void (*data_free)(void *))
 {
-    int r, i, rpc_locked = 0, sock = -1;
+    int r, i, rpc_locked = 0, msgs_locked = 0, sock = -1, timeout;
     int connected; /* flag to indicate whether the transport socket is still connected */
     int multisession = 0; /* flag for more NETCONF sessions on a single SSH session */
     struct nc_session *siter;
@@ -601,10 +656,20 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
     }
 
     if (session->side == NC_CLIENT) {
+        timeout = NC_SESSION_FREE_LOCK_TIMEOUT;
+
         /* MSGS LOCK */
-        pthread_mutex_lock(&session->opts.client.msgs_lock);
+        r = nc_session_client_msgs_lock(session, &timeout, __func__);
+        if (r == -1) {
+            return;
+        } else if (r) {
+            msgs_locked = 1;
+        } else {
+            /* else failed to lock it, too bad */
+            ERR(session, "Freeing a session while messages are being received.");
+        }
 
-        /* cleanup message queue*/
+        /* cleanup message queue */
         for (contiter = session->opts.client.msgs; contiter; ) {
             ly_in_free(contiter->msg, 1);
 
@@ -613,8 +678,10 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
             free(p);
         }
 
-        /* MSGS UNLOCK */
-        pthread_mutex_unlock(&session->opts.client.msgs_lock);
+        if (msgs_locked) {
+            /* MSGS UNLOCK */
+            nc_session_client_msgs_unlock(session, __func__);
+        }
 
         /* receive any leftover messages */
         while (nc_read_msg_poll_io(session, 0, &msg) == 1) {

src/session_client.c

@@ -1744,11 +1744,18 @@ recv_msg(struct nc_session *session, int timeout, NC_MSG_TYPE expected, struct l
     *message = NULL;
 
     /* MSGS LOCK */
-    pthread_mutex_lock(&session->opts.client.msgs_lock);
+    r = nc_session_client_msgs_lock(session, &timeout, __func__);
+    if (!r) {
+        ret = NC_MSG_WOULDBLOCK;
+        goto cleanup;
+    } else if (r == -1) {
+        ret = NC_MSG_ERROR;
+        goto cleanup;
+    }
 
     /* Find the expected message in the buffer */
     prev = NULL;
-    for (cont = session->opts.client.msgs; cont && cont->type != expected; cont = cont->next) {
+    for (cont = session->opts.client.msgs; cont && (cont->type != expected); cont = cont->next) {
         prev = cont;
     }
 
@@ -1764,23 +1771,23 @@ recv_msg(struct nc_session *session, int timeout, NC_MSG_TYPE expected, struct l
         ret = cont->type;
         msg = cont->msg;
         free(cont);
-        goto cleanup;
+        goto cleanup_unlock;
     }
 
     /* Read a message from the wire */
     r = nc_read_msg_poll_io(session, timeout, &msg);
     if (!r) {
         ret = NC_MSG_WOULDBLOCK;
-        goto cleanup;
+        goto cleanup_unlock;
     } else if (r == -1) {
         ret = NC_MSG_ERROR;
-        goto cleanup;
+        goto cleanup_unlock;
     }
 
     /* Basic check to determine message type */
     ret = get_msg_type(session, msg);
     if (ret == NC_MSG_ERROR) {
-        goto cleanup;
+        goto cleanup_unlock;
     }
 
     /* If received a message of different type store it in the buffer */
@@ -1793,18 +1800,19 @@ recv_msg(struct nc_session *session, int timeout, NC_MSG_TYPE expected, struct l
         if (!*cont_ptr) {
             ERRMEM;
             ret = NC_MSG_ERROR;
-            goto cleanup;
+            goto cleanup_unlock;
         }
         (*cont_ptr)->msg = msg;
-        (*cont_ptr)->type = ret;
         msg = NULL;
+        (*cont_ptr)->type = ret;
         (*cont_ptr)->next = NULL;
     }
 
-cleanup:
+cleanup_unlock:
     /* MSGS UNLOCK */
-    pthread_mutex_unlock(&session->opts.client.msgs_lock);
+    nc_session_client_msgs_unlock(session, __func__);
 
+cleanup:
     if (ret == expected) {
         *message = msg;
     } else {

src/session_p.h

@@ -534,10 +534,50 @@ int nc_session_rpc_lock(struct nc_session *session, int timeout, const char *fun
 
 int nc_session_rpc_unlock(struct nc_session *session, int timeout, const char *func);
 
+/**
+ * @brief Lock IO lock on a session.
+ *
+ * @param[in] session Session to lock.
+ * @param[in] timeout Timeout in msec to use.
+ * @param[in] func Caller function for logging.
+ * @return 1 on success;
+ * @return 0 on timeout;
+ * @return -1 on error.
+ */
 int nc_session_io_lock(struct nc_session *session, int timeout, const char *func);
 
+/**
+ * @brief Unlock IO lock on a session.
+ *
+ * @param[in] session Session to unlock.
+ * @param[in] func Caller function for logging.
+ * @return 1 on success;
+ * @return -1 on error.
+ */
 int nc_session_io_unlock(struct nc_session *session, const char *func);
 
+/**
+ * @brief Lock MSGS lock on a session.
+ *
+ * @param[in] session Session to lock.
+ * @param[in,out] timeout Timeout in msec to use. If positive and on successful lock, is updated based on what was elapsed.
+ * @param[in] func Caller function for logging.
+ * @return 1 on success;
+ * @return 0 on timeout;
+ * @return -1 on error.
+ */
+int nc_session_client_msgs_lock(struct nc_session *session, int *timeout, const char *func);
+
+/**
+ * @brief Unlock MSGS lock on a session.
+ *
+ * @param[in] session Session to unlock.
+ * @param[in] func Caller function for logging.
+ * @return 1 on success;
+ * @return -1 on error.
+ */
+int nc_session_client_msgs_unlock(struct nc_session *session, const char *func);
+
 int nc_ps_lock(struct nc_pollsession *ps, uint8_t *id, const char *func);
 
 int nc_ps_unlock(struct nc_pollsession *ps, uint8_t id, const char *func);