session BUGFIX use atomics instead of volatile

For notification thread TID.
Refs #253

Author: michalvasko

src/config.h.in

@@ -36,10 +36,16 @@
 # include <stdatomic.h>
 
 # define ATOMIC_UINT32_T atomic_uint_fast32_t
-# define ATOMIC_INC(x) atomic_fetch_add(x, 1)
+# define ATOMIC_PTR atomic_uintptr_t
+# define ATOMIC_STORE(x, val) atomic_store(&(x), (atomic_uintptr_t)(val))
+# define ATOMIC_LOAD(x) ((void *)atomic_load(&(x)))
+# define ATOMIC_INC(x) atomic_fetch_add(&(x), 1)
 #else
 # define ATOMIC_UINT32_T uint32_t
-# define ATOMIC_INC(x) __sync_add_and_fetch(x, 1)
+# define ATOMIC_PTR void *
+# define ATOMIC_STORE(x, val) (x) = (void *)(val)
+# define ATOMIC_LOAD(x) ((void *)(x))
+# define ATOMIC_INC(x) __sync_add_and_fetch(&(x), 1)
 #endif
 
 /*

src/session.c

@@ -614,8 +614,8 @@ nc_session_free(struct nc_session *session, void (*data_free)(void *))
     }
 
     /* stop notifications loop if any */
-    if ((session->side == NC_CLIENT) && session->opts.client.ntf_tid) {
-        session->opts.client.ntf_tid = NULL;
+    if ((session->side == NC_CLIENT) && ATOMIC_LOAD(session->opts.client.ntf_tid)) {
+        ATOMIC_STORE(session->opts.client.ntf_tid, NULL);
         /* the thread now knows it should quit */
     }
 

src/session_client.c

@@ -1493,7 +1493,7 @@ get_msg(struct nc_session *session, int timeout, uint64_t msgid, struct lyxml_el
 
     /* we read notif, want a rpc-reply */
     if (msgid && (msgtype == NC_MSG_NOTIF)) {
-        if (!session->opts.client.ntf_tid) {
+        if (!ATOMIC_LOAD(session->opts.client.ntf_tid)) {
             ERR("Session %u: received a <notification> but session is not subscribed.", session->id);
             lyxml_free(session->ctx, xml);
             return NC_MSG_ERROR;
@@ -2059,7 +2059,7 @@ nc_session_ntf_thread_running(const struct nc_session *session)
         return 0;
     }
 
-    return session->opts.client.ntf_tid ? 1 : 0;
+    return ATOMIC_LOAD(session->opts.client.ntf_tid) ? 1 : 0;
 }
 
 API void
@@ -2214,9 +2214,9 @@ nc_recv_notif_thread(void *arg)
     free(ntarg);
 
     /* remember our allocated tid, we will be freeing it */
-    ntf_tid = (pthread_t *)session->opts.client.ntf_tid;
+    ntf_tid = ATOMIC_LOAD(session->opts.client.ntf_tid);
 
-    while (session->opts.client.ntf_tid) {
+    while (ATOMIC_LOAD(session->opts.client.ntf_tid)) {
         msgtype = nc_recv_notif(session, NC_CLIENT_NOTIF_THREAD_SLEEP / 1000, &notif);
         if (msgtype == NC_MSG_NOTIF) {
             notif_clb(session, notif);
@@ -2235,7 +2235,7 @@ nc_recv_notif_thread(void *arg)
     }
 
     VRB("Session %u: notification thread exit.", session->id);
-    session->opts.client.ntf_tid = NULL;
+    ATOMIC_STORE(session->opts.client.ntf_tid, NULL);
     free(ntf_tid);
     return NULL;
 }
@@ -2244,6 +2244,7 @@ API int
 nc_recv_notif_dispatch(struct nc_session *session, void (*notif_clb)(struct nc_session *session, const struct nc_notif *notif))
 {
     struct nc_ntf_thread_arg *ntarg;
+    pthread_t *tid;
     int ret;
 
     if (!session) {
@@ -2255,7 +2256,7 @@ nc_recv_notif_dispatch(struct nc_session *session, void (*notif_clb)(struct nc_s
     } else if ((session->status != NC_STATUS_RUNNING) || (session->side != NC_CLIENT)) {
         ERR("Session %u: invalid session to receive Notifications.", session->id);
         return -1;
-    } else if (session->opts.client.ntf_tid) {
+    } else if (ATOMIC_LOAD(session->opts.client.ntf_tid)) {
         ERR("Session %u: separate notification thread is already running.", session->id);
         return -1;
     }
@@ -2268,20 +2269,21 @@ nc_recv_notif_dispatch(struct nc_session *session, void (*notif_clb)(struct nc_s
     ntarg->session = session;
     ntarg->notif_clb = notif_clb;
 
-    /* just so that nc_recv_notif_thread() does not immediately exit, the value does not matter */
-    session->opts.client.ntf_tid = malloc(sizeof *session->opts.client.ntf_tid);
-    if (!session->opts.client.ntf_tid) {
+    tid = malloc(sizeof *tid);
+    if (!tid) {
         ERRMEM;
         free(ntarg);
         return -1;
     }
+    /* just so that nc_recv_notif_thread() does not immediately exit, the value does not matter */
+    ATOMIC_STORE(session->opts.client.ntf_tid, tid);
 
-    ret = pthread_create((pthread_t *)session->opts.client.ntf_tid, NULL, nc_recv_notif_thread, ntarg);
+    ret = pthread_create(tid, NULL, nc_recv_notif_thread, ntarg);
     if (ret) {
         ERR("Session %u: failed to create a new thread (%s).", strerror(errno));
         free(ntarg);
-        free((pthread_t *)session->opts.client.ntf_tid);
-        session->opts.client.ntf_tid = NULL;
+        free(tid);
+        ATOMIC_STORE(session->opts.client.ntf_tid, NULL);
         return -1;
     }
 

src/session_p.h

@@ -426,7 +426,7 @@ struct nc_session {
             char **cpblts;                 /**< list of server's capabilities on client side */
             struct nc_msg_cont *replies;   /**< queue for RPC replies received instead of notifications */
             struct nc_msg_cont *notifs;    /**< queue for notifications received instead of RPC reply */
-            volatile pthread_t *ntf_tid;   /**< running notifications receiving thread */
+            ATOMIC_PTR ntf_tid;            /**< running notifications receiving thread */
 
             /* client flags */
             /* some server modules failed to load so the data from them will be ignored - not use strict flag for parsing */

src/session_server.c

@@ -773,7 +773,7 @@ nc_accept_inout(int fdin, int fdout, const char *username, struct nc_session **s
     (*session)->ctx = server_opts.ctx;
 
     /* assign new SID atomically */
-    (*session)->id = ATOMIC_INC(&server_opts.new_session_id);
+    (*session)->id = ATOMIC_INC(server_opts.new_session_id);
 
     /* NETCONF handshake */
     msgtype = nc_handshake_io(*session);
@@ -2349,7 +2349,7 @@ nc_accept(int timeout, struct nc_session **session)
     pthread_rwlock_unlock(&server_opts.endpt_lock);
 
     /* assign new SID atomically */
-    (*session)->id = ATOMIC_INC(&server_opts.new_session_id);
+    (*session)->id = ATOMIC_INC(server_opts.new_session_id);
 
     /* NETCONF handshake */
     msgtype = nc_handshake_io(*session);
@@ -2497,7 +2497,7 @@ nc_server_ch_add_client(const char *name)
     client = &server_opts.ch_clients[server_opts.ch_client_count - 1];
 
     client->name = lydict_insert(server_opts.ctx, name, 0);
-    client->id = ATOMIC_INC(&server_opts.new_client_id);
+    client->id = ATOMIC_INC(server_opts.new_client_id);
     client->ch_endpts = NULL;
     client->ch_endpt_count = 0;
     client->conn_type = 0;
@@ -3134,7 +3134,7 @@ nc_connect_ch_endpt(struct nc_ch_endpt *endpt, struct nc_session **session)
     }
 
     /* assign new SID atomically */
-    (*session)->id = ATOMIC_INC(&server_opts.new_session_id);
+    (*session)->id = ATOMIC_INC(server_opts.new_session_id);
 
     /* NETCONF handshake */
     msgtype = nc_handshake_io(*session);

src/session_server_ssh.c

@@ -1501,7 +1501,7 @@ nc_session_accept_ssh_channel(struct nc_session *orig_session, struct nc_session
     }
 
     /* assign new SID atomically */
-    new_session->id = ATOMIC_INC(&server_opts.new_session_id);
+    new_session->id = ATOMIC_INC(server_opts.new_session_id);
 
     /* NETCONF handshake */
     msgtype = nc_handshake_io(new_session);
@@ -1572,7 +1572,7 @@ nc_ps_accept_ssh_channel(struct nc_pollsession *ps, struct nc_session **session)
     }
 
     /* assign new SID atomically */
-    new_session->id = ATOMIC_INC(&server_opts.new_session_id);
+    new_session->id = ATOMIC_INC(server_opts.new_session_id);
 
     /* NETCONF handshake */
     msgtype = nc_handshake_io(new_session);