Added mutex locking to zctx socket list

Author: ulikoehler

src/zctx.c

@@ -66,6 +66,7 @@ struct _zctx_t {
     int pipehwm;                //  Send/receive HWM for pipes
     int sndhwm;                 //  ZMQ_SNDHWM for normal sockets
     int rcvhwm;                 //  ZMQ_RCVHWM for normal sockets
+    zmutex_t* socketsMutex;     //  Synchronizes access to socket zlist
 };
 
 
@@ -105,6 +106,7 @@ zctx_new (void)
     self->sndhwm = 1000;
     self->rcvhwm = 1000;
     self->main = true;
+    self->socketsMutex = zmutex_new ();
     zsys_handler_set (s_signal_handler);
     return self;
 }
@@ -119,11 +121,14 @@ zctx_destroy (zctx_t **self_p)
     assert (self_p);
     if (*self_p) {
         zctx_t *self = *self_p;
+        zmutex_lock (self->socketsMutex);
         while (zlist_size (self->sockets))
             zctx__socket_destroy (self, zlist_first (self->sockets));
         zlist_destroy (&self->sockets);
+        zmutex_unlock (self->socketsMutex);
         if (self->main && self->context)
             zmq_term (self->context);
+        zmutex_destroy (&self->socketsMutex);
         free (self);
         *self_p = NULL;
     }
@@ -263,10 +268,13 @@ zctx__socket_new (zctx_t *self, int type)
     zsocket_set_sndhwm (zocket, self->sndhwm);
     zsocket_set_rcvhwm (zocket, self->rcvhwm);
 #endif
+    zmutex_lock (self->socketsMutex);
     if (zlist_push (self->sockets, zocket)) {
         zmq_close (zocket);
+        zmutex_unlock (self->socketsMutex);
         return NULL;
     }
+    zmutex_unlock (self->socketsMutex);
     return zocket;
 }
 
@@ -296,7 +304,9 @@ zctx__socket_destroy (zctx_t *self, void *zocket)
         zsocket_set_linger (zocket, self->linger);
         zmq_close (zocket);
     }
+    zmutex_lock (self->socketsMutex);
     zlist_remove (self->sockets, zocket);
+    zmutex_unlock (self->socketsMutex);
 }