Merge pull request #76 from ali-ince/1.7-valgrind-fixes

Fixes for concurrency issues

Author: ali-ince

CMakeLists.txt

@@ -7,7 +7,7 @@ project(seabolt
 IF (DEFINED ENV{SEABOLT_VERSION})
     set(_SEABOLT_VERSION $ENV{SEABOLT_VERSION})
 ELSE ()
-    set(_SEABOLT_VERSION "1.7.3-dev")
+    set(_SEABOLT_VERSION "1.7.4-dev")
 ENDIF ()
 
 set(SEABOLT_VERSION ${_SEABOLT_VERSION} CACHE STRING "The version of seabolt being built")

run_tests.ps1

@@ -186,7 +186,7 @@ try
     CheckBoltKit
     Compile
 
-    $Neo4jVersion = "-e 3.4"
+    $Neo4jVersion = "-e 3.5"
     If (Test-Path "env:NEOCTRLARGS") {
         $Neo4jVersion = $env:NEOCTRLARGS
     }

run_tests.sh

@@ -20,7 +20,7 @@ if [[ -z "${PYTHON}" ]]; then
 fi
 
 if [[ -z "${NEOCTRLARGS}" ]]; then
-    NEO4J_VERSION="-e 3.4"
+    NEO4J_VERSION="-e 3.5"
 else
     NEO4J_VERSION="${NEOCTRLARGS}"
 fi

src/seabolt/src/CMakeLists.txt

@@ -180,8 +180,13 @@ foreach (target ${SEABOLT_SHARED} ${SEABOLT_STATIC})
         if (ON_WINDOWS)
             target_link_libraries(${target}
                     PUBLIC
-                    crypt32
-                    legacy_stdio_definitions)
+                    crypt32)
+
+            if (MSVC)
+                target_link_libraries(${target}
+                        PUBLIC
+                        legacy_stdio_definitions)
+            endif ()
         endif ()
     endif ()
 

src/seabolt/src/bolt/communication-plain-posix.c

@@ -19,6 +19,7 @@
 #include "communication-plain.h"
 #include "status-private.h"
 
+#include <pthread.h>
 #include <errno.h>
 
 int socket_last_error(BoltCommunication* comm)
@@ -66,16 +67,31 @@ int socket_transform_error(BoltCommunication* comm, int error_code)
 int socket_ignore_sigpipe(void** replaced_action)
 {
 #if !defined(SO_NOSIGPIPE)
-    if (*replaced_action==NULL) {
-        *replaced_action = malloc(sizeof(struct sigaction));
+    sigset_t sig_block, sig_restore, sig_pending;
+
+    sigemptyset(&sig_block);
+    sigaddset(&sig_block, SIGPIPE);
+
+    int result = pthread_sigmask(SIG_BLOCK, &sig_block, &sig_restore);
+    if (result!=0) {
+        return result;
+    }
+
+    int sigpipe_pending = -1;
+    if (sigpending(&sig_pending)!=-1) {
+        sigpipe_pending = sigismember(&sig_pending, SIGPIPE);
+    }
+
+    if (sigpipe_pending==-1) {
+        pthread_sigmask(SIG_SETMASK, &sig_restore, NULL);
+        return -1;
     }
 
-    struct sigaction ignore_act;
-    memset(&ignore_act, '\0', sizeof(ignore_act));
-    memset(*replaced_action, '\0', sizeof(struct sigaction));
-    ignore_act.sa_handler = SIG_IGN;
-    ignore_act.sa_flags = 0;
-    return sigaction(SIGPIPE, &ignore_act, *replaced_action);
+    if (*replaced_action==NULL) {
+        *replaced_action = malloc(sizeof(sigset_t));
+    }
+    memcpy(*replaced_action, &sig_restore, sizeof(sigset_t));
+    return 0;
 #endif
     UNUSED(replaced_action);
     return BOLT_SUCCESS;
@@ -85,10 +101,26 @@ int socket_restore_sigpipe(void** action_to_restore)
 {
 #if !defined(SO_NOSIGPIPE)
     if (action_to_restore!=NULL) {
-        int result = sigaction(SIGPIPE, *action_to_restore, NULL);
+        sigset_t sig_block;
+
+        sigemptyset(&sig_block);
+        sigaddset(&sig_block, SIGPIPE);
+
+        struct timespec ts;
+        ts.tv_sec = 0;
+        ts.tv_nsec = 0;
+
+        while (sigtimedwait(&sig_block, 0, &ts)==-1) {
+            if (errno!=EINTR) {
+                break;
+            }
+        }
+
+        pthread_sigmask(SIG_SETMASK, (sigset_t*) *action_to_restore, NULL);
+
         free(*action_to_restore);
         *action_to_restore = NULL;
-        return result;
+        return 0;
     }
 #endif
     UNUSED(action_to_restore);

src/seabolt/src/bolt/communication-secure-openssl.c

@@ -214,9 +214,19 @@ void BoltSecurityContext_destroy(BoltSecurityContext* context)
 
 static mutex_t* locks;
 
-unsigned long secure_openssl_id_callback(void);
+struct CRYPTO_dynlock_value {
+    mutex_t mutex;
+};
 
-void secure_openssl_locking_callback(int mode, int type, const char* file, int line);
+static unsigned long secure_openssl_id_callback(void);
+
+static void secure_openssl_locking_callback(int mode, int type, const char* file, int line);
+
+static struct CRYPTO_dynlock_value* secure_openssl_create_mutex(const char* file, int line);
+
+static void secure_openssl_lock_mutext(int mode, struct CRYPTO_dynlock_value* l, const char* file, int line);
+
+static void secure_openssl_destroy_mutex(struct CRYPTO_dynlock_value* l, const char* file, int line);
 
 void CRYPTO_thread_setup(void)
 {
@@ -230,21 +240,31 @@ void CRYPTO_thread_setup(void)
         BoltSync_mutex_create(&locks[i]);
     }
 
-    CRYPTO_set_id_callback(&secure_openssl_id_callback);
-    CRYPTO_set_locking_callback(&secure_openssl_locking_callback);
+    CRYPTO_set_id_callback(secure_openssl_id_callback);
+    CRYPTO_set_locking_callback(secure_openssl_locking_callback);
+
+    CRYPTO_set_dynlock_create_callback(secure_openssl_create_mutex);
+    CRYPTO_set_dynlock_lock_callback(secure_openssl_lock_mutext);
+    CRYPTO_set_dynlock_destroy_callback(secure_openssl_destroy_mutex);
 }
 
-static void CRYPTO_thread_cleanup(void)
+void CRYPTO_thread_cleanup(void)
 {
-    int i;
+    CRYPTO_set_id_callback(NULL);
+    CRYPTO_set_locking_callback(NULL);
+
+    CRYPTO_set_dynlock_create_callback(NULL);
+    CRYPTO_set_dynlock_lock_callback(NULL);
+    CRYPTO_set_dynlock_destroy_callback(NULL);
 
     CRYPTO_set_locking_callback(NULL);
-    for (i = 0; i<CRYPTO_num_locks(); i++)
+    for (int i = 0; i<CRYPTO_num_locks(); i++) {
         BoltSync_mutex_destroy(&locks[i]);
+    }
     OPENSSL_free(locks);
 }
 
-void secure_openssl_locking_callback(int mode, int type, const char* file, int line)
+static void secure_openssl_locking_callback(int mode, int type, const char* file, int line)
 {
     UNUSED(file);
     UNUSED(line);
@@ -257,18 +277,52 @@ void secure_openssl_locking_callback(int mode, int type, const char* file, int l
     }
 }
 
-unsigned long secure_openssl_id_callback(void)
+static unsigned long secure_openssl_id_callback(void)
 {
     return BoltThread_id();
 }
 
+static struct CRYPTO_dynlock_value* secure_openssl_create_mutex(const char* file, int line)
+{
+    UNUSED(file);
+    UNUSED(line);
+
+    struct CRYPTO_dynlock_value* value;
+    value = (struct CRYPTO_dynlock_value*) malloc(sizeof(struct CRYPTO_dynlock_value));
+    if (!value) return NULL;
+    BoltSync_mutex_create(&value->mutex);
+    return value;
+}
+
+static void secure_openssl_lock_mutext(int mode, struct CRYPTO_dynlock_value* l, const char* file, int line)
+{
+    UNUSED(file);
+    UNUSED(line);
+
+    if (mode & CRYPTO_LOCK) {
+        BoltSync_mutex_lock(&l->mutex);
+    }
+    else {
+        BoltSync_mutex_unlock(&l->mutex);
+    }
+}
+
+static void secure_openssl_destroy_mutex(struct CRYPTO_dynlock_value* l, const char* file, int line)
+{
+    UNUSED(file);
+    UNUSED(line);
+
+    BoltSync_mutex_destroy(&l->mutex);
+    free(l);
+}
+
 #endif
 
 int BoltSecurityContext_startup()
 {
 #if OPENSSL_VERSION_NUMBER<0x10100000L
-    SSL_library_init();
     CRYPTO_thread_setup();
+    SSL_library_init();
 #else
     OPENSSL_init_ssl(0, NULL);
 #endif