refactor: make memory allocator and atomic operations definitions public

This also replaces the liburcu atomic operations with native gcc/clang
operations. In the long run, liburcu should be removed from the core lib
and only used in nodestore plugins.

include/ua_config.h.in

@@ -20,13 +20,6 @@
 extern "C" {
 #endif
 
-#ifndef _XOPEN_SOURCE
-# define _XOPEN_SOURCE 500
-#endif
-#ifndef _DEFAULT_SOURCE
-# define _DEFAULT_SOURCE
-#endif
-
 #define UA_LOGLEVEL ${UA_LOGLEVEL}
 #define UA_GIT_COMMIT_ID "${GIT_COMMIT_ID}"
 #cmakedefine UA_ENABLE_MULTITHREADING
@@ -42,6 +35,18 @@ extern "C" {
 #cmakedefine UA_ENABLE_NONSTANDARD_UDP
 #cmakedefine UA_ENABLE_NONSTANDARD_STATELESS
 
+/**
+ * Standard Includes
+ * ----------------- */
+#ifndef _XOPEN_SOURCE
+# define _XOPEN_SOURCE 500
+#endif
+#ifndef _DEFAULT_SOURCE
+# define _DEFAULT_SOURCE
+#endif
+#include <stddef.h>
+#include <stdint.h>
+
 /**
  * Function Export
  * --------------- */
@@ -107,6 +112,94 @@ extern "C" {
 # define UA_FUNC_ATTR_WARN_UNUSED_RESULT
 #endif
 
+/**
+ * Memory Management
+ * -----------------
+ * Replace the macros for custom memory allocators if necessary */
+#include <stdlib.h>
+#ifdef _WIN32
+# include <malloc.h>
+#endif
+
+#define UA_free(ptr) free(ptr)
+#define UA_malloc(size) malloc(size)
+#define UA_calloc(num, size) calloc(num, size)
+#define UA_realloc(ptr, size) realloc(ptr, size)
+
+#ifndef NO_ALLOCA
+# if defined(__GNUC__) || defined(__clang__)
+#  define UA_alloca(size) __builtin_alloca (size)
+# elif defined(_WIN32)
+#  define UA_alloca(SIZE) _alloca(SIZE)
+# else
+#  include <alloca.h>
+#  define UA_alloca(SIZE) alloca(SIZE)
+# endif
+#endif
+
+/**
+ * Atomic Operations
+ * -----------------
+ * Atomic operations that synchronize across processor cores (for
+ * multithreading). Only the inline-functions defined next are used. Replace
+ * with architecture-specific operations if necessary. */
+
+#ifndef UA_ENABLE_MULTITHREADING
+# define UA_atomic_sync()
+#else
+# ifdef _MSC_VER /* Visual Studio */
+#  define UA_atomic_sync() _ReadWriteBarrier()
+# else /* GCC/Clang */
+#  define UA_atomic_sync() __sync_synchronize()
+# endif
+#endif
+
+static UA_INLINE void *
+UA_atomic_xchg(void * volatile * addr, void *new) {
+#ifndef UA_ENABLE_MULTITHREADING
+    void *old = *addr;
+    *addr = new;
+    return old;
+#else
+# ifdef _MSC_VER /* Visual Studio */
+    return _InterlockedExchangePointer(addr, new);
+# else /* GCC/Clang */
+    return __sync_lock_test_and_set(addr, new);
+# endif
+#endif
+}
+
+static UA_INLINE void *
+UA_atomic_cmpxchg(void * volatile * addr, void *expected, void *new) {
+#ifndef UA_ENABLE_MULTITHREADING
+    void *old = *addr;
+    if(old == expected) {
+        *addr = new;
+    }
+    return old;
+#else
+# ifdef _MSC_VER /* Visual Studio */
+    return _InterlockedCompareExchangePointer(addr, expected, new);
+# else /* GCC/Clang */
+    return __sync_val_compare_and_swap(addr, expected, new);
+# endif
+#endif
+}
+
+static UA_INLINE uint32_t
+UA_atomic_add(volatile uint32_t *addr, uint32_t increase) {
+#ifndef UA_ENABLE_MULTITHREADING
+    *addr += increase;
+    return *addr;
+#else
+# ifdef _MSC_VER /* Visual Studio */
+    return _InterlockedExchangeAdd(addr, increase) + increase;
+# else /* GCC/Clang */
+    return __sync_add_and_fetch(addr, increase);
+# endif
+#endif
+}
+
 /**
  * Binary Encoding Overlays
  * ------------------------
@@ -179,7 +272,6 @@ extern "C" {
 /**
  * Embed unavailable libc functions
  * -------------------------------- */
-#include <stddef.h>
 #ifdef UA_ENABLE_EMBEDDED_LIBC
   void *memcpy(void *UA_RESTRICT dest, const void *UA_RESTRICT src, size_t n);
   void *memset(void *dest, int c, size_t n);

include/ua_types.h

@@ -20,7 +20,6 @@ extern "C" {
 
 #include "ua_config.h"
 #include "ua_constants.h"
-#include <stdint.h>
 #include <stdbool.h>
 
 /**

src/server/ua_nodestore_concurrent.c

@@ -3,6 +3,7 @@
 #include "ua_server_internal.h"
 
 #ifdef UA_ENABLE_MULTITHREADING /* conditional compilation */
+#include <urcu/rculfhash.h>
 
 struct nodeEntry {
     struct cds_lfht_node htn; ///< Contains the next-ptr for urcu-hashmap

src/server/ua_securechannel_manager.c

@@ -26,12 +26,11 @@ void UA_SecureChannelManager_deleteMembers(UA_SecureChannelManager *cm) {
 
 static void removeSecureChannel(UA_SecureChannelManager *cm, channel_list_entry *entry){
     LIST_REMOVE(entry, pointers);
+    UA_atomic_add(&cm->currentChannelCount, (UA_UInt32)-1);
     UA_SecureChannel_deleteMembersCleanup(&entry->channel);
 #ifndef UA_ENABLE_MULTITHREADING
-    cm->currentChannelCount--;
     UA_free(entry);
 #else
-    cm->currentChannelCount = uatomic_add_return(&cm->currentChannelCount, -1);
     UA_Server_delayedFree(cm->server, entry);
 #endif
 }
@@ -108,11 +107,7 @@ UA_SecureChannelManager_open(UA_SecureChannelManager *cm, UA_Connection *conn,
     /* Set all the pointers internally */
     UA_Connection_attachSecureChannel(conn, &entry->channel);
     LIST_INSERT_HEAD(&cm->channels, entry, pointers);
-#ifndef UA_ENABLE_MULTITHREADING
-    cm->currentChannelCount++;
-#else
-    cm->currentChannelCount = uatomic_add_return(&cm->currentChannelCount, 1);
-#endif
+    UA_atomic_add(&cm->currentChannelCount, 1);
     return UA_STATUSCODE_GOOD;
 }
 
@@ -151,7 +146,8 @@ UA_SecureChannelManager_renew(UA_SecureChannelManager *cm, UA_Connection *conn,
     return UA_STATUSCODE_GOOD;
 }
 
-UA_SecureChannel * UA_SecureChannelManager_get(UA_SecureChannelManager *cm, UA_UInt32 channelId) {
+UA_SecureChannel *
+UA_SecureChannelManager_get(UA_SecureChannelManager *cm, UA_UInt32 channelId) {
     channel_list_entry *entry;
     LIST_FOREACH(entry, &cm->channels, pointers) {
         if(entry->channel.securityToken.channelId == channelId)
@@ -160,7 +156,8 @@ UA_SecureChannel * UA_SecureChannelManager_get(UA_SecureChannelManager *cm, UA_U
     return NULL;
 }
 
-UA_StatusCode UA_SecureChannelManager_close(UA_SecureChannelManager *cm, UA_UInt32 channelId) {
+UA_StatusCode
+UA_SecureChannelManager_close(UA_SecureChannelManager *cm, UA_UInt32 channelId) {
     channel_list_entry *entry;
     LIST_FOREACH(entry, &cm->channels, pointers) {
         if(entry->channel.securityToken.channelId == channelId)
@@ -168,15 +165,6 @@ UA_StatusCode UA_SecureChannelManager_close(UA_SecureChannelManager *cm, UA_UInt
     }
     if(!entry)
         return UA_STATUSCODE_BADINTERNALERROR;
-
-    LIST_REMOVE(entry, pointers);
-    UA_SecureChannel_deleteMembersCleanup(&entry->channel);
-#ifndef UA_ENABLE_MULTITHREADING
-    cm->currentChannelCount--;
-    UA_free(entry);
-#else
-    cm->currentChannelCount = uatomic_add_return(&cm->currentChannelCount, -1);
-    UA_Server_delayedFree(cm->server, entry);
-#endif
+    removeSecureChannel(cm, entry);
     return UA_STATUSCODE_GOOD;
 }

src/server/ua_securechannel_manager.h

@@ -13,7 +13,7 @@ typedef struct channel_list_entry {
 
 typedef struct UA_SecureChannelManager {
     LIST_HEAD(channel_list, channel_list_entry) channels; // doubly-linked list of channels
-    size_t currentChannelCount;
+    UA_UInt32 currentChannelCount;
     UA_UInt32 lastChannelId;
     UA_UInt32 lastTokenId;
     UA_Server *server;

src/server/ua_server_binary.c

@@ -307,11 +307,7 @@ processOPN(UA_Server *server, UA_Connection *connection,
 
     /* Encode the message after the secureconversationmessageheader */
     size_t tmpPos = 12; /* skip the header */
-#ifndef UA_ENABLE_MULTITHREADING
-    seqHeader.sequenceNumber = ++channel->sendSequenceNumber;
-#else
-    seqHeader.sequenceNumber = uatomic_add_return(&channel->sendSequenceNumber, 1);
-#endif
+    seqHeader.sequenceNumber = UA_atomic_add(&channel->sendSequenceNumber, 1);
     retval |= UA_AsymmetricAlgorithmSecurityHeader_encodeBinary(&asymHeader, &resp_msg, &tmpPos); // just mirror back
     retval |= UA_SequenceHeader_encodeBinary(&seqHeader, &resp_msg, &tmpPos);
     UA_NodeId responseType = UA_NODEID_NUMERIC(0, UA_TYPES[UA_TYPES_OPENSECURECHANNELRESPONSE].binaryEncodingId);

src/server/ua_server_internal.h

@@ -12,6 +12,36 @@
 #define ANONYMOUS_POLICY "open62541-anonymous-policy"
 #define USERNAME_POLICY "open62541-username-policy"
 
+/* liburcu includes */
+#ifdef UA_ENABLE_MULTITHREADING
+# define _LGPL_SOURCE
+# include <urcu.h>
+# include <urcu/lfstack.h>
+# ifdef NDEBUG
+#  define UA_RCU_LOCK() rcu_read_lock()
+#  define UA_RCU_UNLOCK() rcu_read_unlock()
+#  define UA_ASSERT_RCU_LOCKED()
+#  define UA_ASSERT_RCU_UNLOCKED()
+# else
+   extern UA_THREAD_LOCAL bool rcu_locked;
+#   define UA_ASSERT_RCU_LOCKED() assert(rcu_locked)
+#   define UA_ASSERT_RCU_UNLOCKED() assert(!rcu_locked)
+#   define UA_RCU_LOCK() do {                     \
+        UA_ASSERT_RCU_UNLOCKED();                 \
+        rcu_locked = true;                        \
+        rcu_read_lock(); } while(0)
+#   define UA_RCU_UNLOCK() do {                   \
+        UA_ASSERT_RCU_LOCKED();                   \
+        rcu_locked = false;                       \
+        rcu_read_lock(); } while(0)
+# endif
+#else
+# define UA_RCU_LOCK()
+# define UA_RCU_UNLOCK()
+# define UA_ASSERT_RCU_LOCKED()
+# define UA_ASSERT_RCU_UNLOCKED()
+#endif
+
 #ifdef UA_ENABLE_EXTERNAL_NAMESPACES
 /** Mapping of namespace-id and url to an external nodestore. For namespaces
     that have no mapping defined, the internal nodestore is used by default. */

src/server/ua_server_worker.c

@@ -107,7 +107,7 @@ workerLoop(UA_Worker *worker) {
             /* nothing to do. sleep until a job is dispatched (and wakes up all worker threads) */
             pthread_cond_wait(&server->dispatchQueue_condition, &mutex);
         }
-        uatomic_inc(counter);
+        UA_atomic_add(counter, 1);
     }
 
     pthread_mutex_unlock(&mutex);
@@ -443,25 +443,15 @@ dispatchDelayedJobs(UA_Server *server, void *_) {
         dw = dw->next;
     }
 
-#if (__GNUC__ <= 4 && __GNUC_MINOR__ <= 6)
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wextra"
-#pragma GCC diagnostic ignored "-Wcast-qual"
-#pragma GCC diagnostic ignored "-Wunused-value"
-#endif
     /* process and free all delayed jobs from here on */
     while(dw) {
         for(size_t i = 0; i < dw->jobsCount; i++)
             processJob(server, &dw->jobs[i]);
-        struct DelayedJobs *next = uatomic_xchg(&beforedw->next, NULL);
+        struct DelayedJobs *next = UA_atomic_xchg((void**)&beforedw->next, NULL);
         UA_free(dw->workerCounters);
         UA_free(dw);
         dw = next;
     }
-#if (__GNUC__ <= 4 && __GNUC_MINOR__ <= 6)
-#pragma GCC diagnostic pop
-#endif
-
 }
 
 #endif

src/server/ua_session_manager.c

@@ -18,6 +18,18 @@ void UA_SessionManager_deleteMembers(UA_SessionManager *sm) {
     }
 }
 
+static void
+removeSessionEntry(UA_SessionManager *sm, session_list_entry *sentry) {
+    LIST_REMOVE(sentry, pointers);
+    UA_atomic_add(&sm->currentSessionCount, (UA_UInt32)-1);
+    UA_Session_deleteMembersCleanup(&sentry->session, sm->server);
+#ifndef UA_ENABLE_MULTITHREADING
+    UA_free(sentry);
+#else
+    UA_Server_delayedFree(sm->server, sentry);
+#endif
+}
+
 void UA_SessionManager_cleanupTimedOut(UA_SessionManager *sm, UA_DateTime nowMonotonic) {
     session_list_entry *sentry, *temp;
     LIST_FOREACH_SAFE(sentry, &sm->sessions, pointers, temp) {
@@ -25,15 +37,7 @@ void UA_SessionManager_cleanupTimedOut(UA_SessionManager *sm, UA_DateTime nowMon
             UA_LOG_DEBUG(sm->server->config.logger, UA_LOGCATEGORY_SESSION,
                          "Session with token %i has timed out and is removed",
                          sentry->session.sessionId.identifier.numeric);
-            LIST_REMOVE(sentry, pointers);
-            UA_Session_deleteMembersCleanup(&sentry->session, sm->server);
-#ifndef UA_ENABLE_MULTITHREADING
-            sm->currentSessionCount--;
-            UA_free(sentry);
-#else
-            sm->currentSessionCount = uatomic_add_return(&sm->currentSessionCount, -1);
-            UA_Server_delayedFree(sm->server, sentry);
-#endif
+            removeSessionEntry(sm, sentry);
         }
     }
 }
@@ -58,7 +62,7 @@ UA_SessionManager_getSession(UA_SessionManager *sm, const UA_NodeId *token) {
     return NULL;
 }
 
-/** Creates and adds a session. But it is not yet attached to a secure channel. */
+/* Creates and adds a session. But it is not yet attached to a secure channel. */
 UA_StatusCode
 UA_SessionManager_createSession(UA_SessionManager *sm, UA_SecureChannel *channel,
                                 const UA_CreateSessionRequest *request, UA_Session **session) {
@@ -69,7 +73,7 @@ UA_SessionManager_createSession(UA_SessionManager *sm, UA_SecureChannel *channel
     if(!newentry)
         return UA_STATUSCODE_BADOUTOFMEMORY;
 
-    sm->currentSessionCount++;
+    UA_atomic_add(&sm->currentSessionCount, 1);
     UA_Session_init(&newentry->session);
     newentry->session.sessionId = UA_NODEID_GUID(1, UA_Guid_random());
     newentry->session.authenticationToken = UA_NODEID_GUID(1, UA_Guid_random());
@@ -93,18 +97,8 @@ UA_SessionManager_removeSession(UA_SessionManager *sm, const UA_NodeId *token) {
         if(UA_NodeId_equal(&current->session.authenticationToken, token))
             break;
     }
-
     if(!current)
         return UA_STATUSCODE_BADSESSIONIDINVALID;
-
-    LIST_REMOVE(current, pointers);
-    UA_Session_deleteMembersCleanup(&current->session, sm->server);
-#ifndef UA_ENABLE_MULTITHREADING
-    sm->currentSessionCount--;
-    UA_free(current);
-#else
-    sm->currentSessionCount = uatomic_add_return(&sm->currentSessionCount, -1);
-    UA_Server_delayedFree(sm->server, current);
-#endif
+    removeSessionEntry(sm, current);
     return UA_STATUSCODE_GOOD;
 }

src/ua_connection.c

@@ -113,44 +113,22 @@ UA_Connection_completeMessages(UA_Connection *connection, UA_ByteString * UA_RES
     return retval;
 }
 
-#if (__GNUC__ >= 4 && __GNUC_MINOR__ >= 6)
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wextra"
-#pragma GCC diagnostic ignored "-Wcast-qual"
-#pragma GCC diagnostic ignored "-Wunused-value"
-#endif
-
 void UA_Connection_detachSecureChannel(UA_Connection *connection) {
-#ifdef UA_ENABLE_MULTITHREADING
     UA_SecureChannel *channel = connection->channel;
     if(channel)
-        uatomic_cmpxchg(&channel->connection, connection, NULL);
-    uatomic_set(&connection->channel, NULL);
-#else
-    if(connection->channel)
-        connection->channel->connection = NULL;
-    connection->channel = NULL;
-#endif
+        /* only replace when the channel points to this connection */
+        UA_atomic_cmpxchg((void**)&channel->connection, connection, NULL);
+    UA_atomic_xchg((void**)&connection->channel, NULL);
 }
 
+// TODO: Return an error code
 void
 UA_Connection_attachSecureChannel(UA_Connection *connection,
                                   UA_SecureChannel *channel) {
-#ifdef UA_ENABLE_MULTITHREADING
-    if(uatomic_cmpxchg(&channel->connection, NULL, connection) == NULL)
-        uatomic_set((void**)&connection->channel, (void*)channel);
-#else
-    if(channel->connection != NULL)
-        return;
-    channel->connection = connection;
-    connection->channel = channel;
-#endif
+    if(UA_atomic_cmpxchg((void**)&channel->connection, NULL, connection) == NULL)
+        UA_atomic_xchg((void**)&connection->channel, (void*)channel);
 }
 
-#if (__GNUC__ >= 4 && __GNUC_MINOR__ >= 6)
-#pragma GCC diagnostic pop
-#endif
-
 UA_StatusCode
 UA_EndpointUrl_split_ptr(const char *endpointUrl, char *hostname,
                          const char ** port, const char **path) {

src/ua_securechannel.c

@@ -67,18 +67,10 @@ void UA_SecureChannel_attachSession(UA_SecureChannel *channel, UA_Session *sessi
     if(!se)
         return;
     se->session = session;
-#ifdef UA_ENABLE_MULTITHREADING
-    if(uatomic_cmpxchg(&session->channel, NULL, channel) != NULL) {
+    if(UA_atomic_cmpxchg((void**)&session->channel, NULL, channel) != NULL) {
         UA_free(se);
         return;
     }
-#else
-    if(session->channel != NULL) {
-        UA_free(se);
-        return;
-    }
-    session->channel = channel;
-#endif
     LIST_INSERT_HEAD(&channel->sessions, se, pointers);
 }
 
@@ -171,11 +163,7 @@ UA_SecureChannel_sendChunk(UA_ChunkInfo *ci, UA_ByteString *dst, size_t offset)
     symSecHeader.tokenId = channel->securityToken.tokenId;
     UA_SequenceHeader seqHeader;
     seqHeader.requestId = ci->requestId;
-#ifndef UA_ENABLE_MULTITHREADING
-    seqHeader.sequenceNumber = ++channel->sendSequenceNumber;
-#else
-    seqHeader.sequenceNumber = uatomic_add_return(&channel->sendSequenceNumber, 1);
-#endif
+    seqHeader.sequenceNumber = UA_atomic_add(&channel->sendSequenceNumber, 1);
     size_t offset_header = 0;
     UA_SecureConversationMessageHeader_encodeBinary(&respHeader, dst, &offset_header);
     UA_SymmetricAlgorithmSecurityHeader_encodeBinary(&symSecHeader, dst, &offset_header);

src/ua_util.h

@@ -3,50 +3,15 @@
 
 #include "ua_config.h"
 
+/* Assert */
 #include <assert.h>
 #define UA_assert(ignore) assert(ignore)
 
-/*********************/
-/* Memory Management */
-/*********************/
-
-/* Replace the macros with functions for custom allocators if necessary */
-#include <stdlib.h> // malloc, free
-#ifdef _WIN32
-# include <malloc.h>
-#endif
-
-#ifndef UA_free
-# define UA_free(ptr) free(ptr)
-#endif
-#ifndef UA_malloc
-# define UA_malloc(size) malloc(size)
-#endif
-#ifndef UA_calloc
-# define UA_calloc(num, size) calloc(num, size)
-#endif
-#ifndef UA_realloc
-# define UA_realloc(ptr, size) realloc(ptr, size)
-#endif
-
-#ifndef NO_ALLOCA
-# if defined(__GNUC__) || defined(__clang__)
-#  define UA_alloca(size) __builtin_alloca (size)
-# elif defined(_WIN32)
-#  define UA_alloca(SIZE) _alloca(SIZE)
-# else
-#  include <alloca.h>
-#  define UA_alloca(SIZE) alloca(SIZE)
-# endif
-#endif
-
+/* container_of */
 #define container_of(ptr, type, member) \
     (type *)((uintptr_t)ptr - offsetof(type,member))
 
-/************************/
 /* Thread Local Storage */
-/************************/
-
 #ifdef UA_ENABLE_MULTITHREADING
 # if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
 #  define UA_THREAD_LOCAL _Thread_local /* C11 */
@@ -55,7 +20,7 @@
 # elif defined(_MSC_VER)
 #  define UA_THREAD_LOCAL __declspec(thread) /* MSVC extension */
 # else
-#  warning The compiler does not allow thread-local variables. The library can be built, but will not be thread safe.
+#  warning The compiler does not allow thread-local variables. The library can be built, but will not be thread-safe.
 # endif
 #endif
 
@@ -63,41 +28,7 @@
 # define UA_THREAD_LOCAL
 #endif
 
-/*************************/
-/* External Dependencies */
-/*************************/
+/* BSD Queue Macros */
 #include "queue.h"
 
-#ifdef UA_ENABLE_MULTITHREADING
-# define _LGPL_SOURCE
-# include <urcu.h>
-# include <urcu/wfcqueue.h>
-# include <urcu/uatomic.h>
-# include <urcu/rculfhash.h>
-# include <urcu/lfstack.h>
-# ifdef NDEBUG
-#  define UA_RCU_LOCK() rcu_read_lock()
-#  define UA_RCU_UNLOCK() rcu_read_unlock()
-#  define UA_ASSERT_RCU_LOCKED()
-#  define UA_ASSERT_RCU_UNLOCKED()
-# else
-   extern UA_THREAD_LOCAL bool rcu_locked;
-#   define UA_ASSERT_RCU_LOCKED() assert(rcu_locked)
-#   define UA_ASSERT_RCU_UNLOCKED() assert(!rcu_locked)
-#   define UA_RCU_LOCK() do {                     \
-        UA_ASSERT_RCU_UNLOCKED();                 \
-        rcu_locked = true;                        \
-        rcu_read_lock(); } while(0)
-#   define UA_RCU_UNLOCK() do {                   \
-        UA_ASSERT_RCU_LOCKED();                   \
-        rcu_locked = false;                       \
-        rcu_read_lock(); } while(0)
-# endif
-#else
-# define UA_RCU_LOCK()
-# define UA_RCU_UNLOCK()
-# define UA_ASSERT_RCU_LOCKED()
-# define UA_ASSERT_RCU_UNLOCKED()
-#endif
-
 #endif /* UA_UTIL_H_ */