Merge pull request #3797 from chu11/issue3617_zmqutil

Move libzmq dependent functions out of libflux-core into libzmqutil

configure.ac

@@ -502,6 +502,7 @@ AC_CONFIG_FILES( \
   src/common/librlist/Makefile \
   src/common/libczmqcontainers/Makefile \
   src/common/libccan/Makefile \
+  src/common/libzmqutil/Makefile \
   src/bindings/Makefile \
   src/bindings/lua/Makefile \
   src/bindings/python/Makefile \

doc/Makefile.am

@@ -80,7 +80,6 @@ MAN3_FILES_PRIMARY = \
 	man3/flux_reactor_create.3 \
 	man3/flux_fd_watcher_create.3 \
 	man3/flux_watcher_start.3 \
-	man3/flux_zmq_watcher_create.3 \
 	man3/flux_handle_watcher_create.3 \
 	man3/flux_timer_watcher_create.3 \
 	man3/flux_periodic_watcher_create.3 \
@@ -163,7 +162,6 @@ MAN3_FILES_SECONDARY = \
 	man3/flux_watcher_stop.3 \
 	man3/flux_watcher_destroy.3 \
 	man3/flux_watcher_next_wakeup.3 \
-	man3/flux_zmq_watcher_get_zsock.3 \
 	man3/flux_handle_watcher_get_flux.3 \
 	man3/flux_timer_watcher_reset.3 \
 	man3/flux_periodic_watcher_reset.3 \

doc/conf.py

@@ -332,8 +332,6 @@ def setup(app):
     ('man3/flux_watcher_start', 'flux_watcher_destroy', 'start/stop/destroy/query reactor watcher', [author], 3),
     ('man3/flux_watcher_start', 'flux_watcher_next_wakeup', 'start/stop/destroy/query reactor watcher', [author], 3),
     ('man3/flux_watcher_start', 'flux_watcher_start', 'start/stop/destroy/query reactor watcher', [author], 3),
-    ('man3/flux_zmq_watcher_create', 'flux_zmq_watcher_get_zsock', 'create ZeroMQ watcher', [author], 3),
-    ('man3/flux_zmq_watcher_create', 'flux_zmq_watcher_create', 'create ZeroMQ watcher', [author], 3),
     ('man3/idset_create', 'idset_create', 'Manipulate numerically sorted sets of non-negative integers', [author], 3),
     ('man3/idset_create', 'idset_destroy', 'Manipulate numerically sorted sets of non-negative integers', [author], 3),
     ('man3/idset_create', 'idset_set', 'Manipulate numerically sorted sets of non-negative integers', [author], 3),

doc/man3/flux_reactor_create.rst

@@ -116,7 +116,6 @@ SEE ALSO
 ========
 
 flux_fd_watcher_create(3), flux_handle_watcher_create(3),
-flux_zmq_watcher_create(3), flux_timer_watcher_create(3),
-flux_watcher_start(3)
+flux_timer_watcher_create(3), flux_watcher_start(3)
 
 `libev home page <http://software.schmorp.de/pkg/libev.html>`__

doc/man3/index.rst

@@ -70,7 +70,6 @@ man3
    flux_sync_create
    flux_timer_watcher_create
    flux_watcher_start
-   flux_zmq_watcher_create
    idset_create
    idset_encode
    flux_jobtap_get_flux

src/broker/Makefile.am

@@ -66,6 +66,7 @@ libbroker_la_SOURCES = \
 flux_broker_LDADD = \
 	$(builddir)/libbroker.la \
 	$(top_builddir)/src/common/libflux-core.la \
+	$(top_builddir)/src/common/libzmqutil/libzmqutil.la \
 	$(top_builddir)/src/common/libpmi/libpmi_client.la \
 	$(top_builddir)/src/common/libflux-internal.la \
 	$(top_builddir)/src/common/libflux-optparse.la \
@@ -89,6 +90,7 @@ test_ldadd = \
 	$(builddir)/libbroker.la \
 	$(top_builddir)/src/common/libtestutil/libtestutil.la \
 	$(top_builddir)/src/common/libflux-core.la \
+	$(top_builddir)/src/common/libzmqutil/libzmqutil.la \
 	$(top_builddir)/src/common/libpmi/libpmi_client.la \
 	$(top_builddir)/src/common/libflux-internal.la \
 	$(top_builddir)/src/common/libtap/libtap.la \

src/broker/module.c

@@ -22,6 +22,8 @@
 #include <sys/syscall.h>
 #endif
 
+#include "src/common/libzmqutil/msg_zsock.h"
+#include "src/common/libzmqutil/reactor.h"
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libutil/log.h"
 #include "src/common/libutil/iterators.h"
@@ -253,7 +255,7 @@ flux_msg_t *module_recvmsg (module_t *p)
     int type;
     struct flux_msg_cred cred;
 
-    if (!(msg = flux_msg_recvzsock (p->sock)))
+    if (!(msg = zmqutil_msg_recv (p->sock)))
         goto error;
     if (flux_msg_get_type (msg, &type) < 0)
         goto error;
@@ -314,7 +316,7 @@ int module_sendmsg (module_t *p, const flux_msg_t *msg)
                 goto done;
             if (flux_msg_route_push (cpy, p->modhash->uuid_str) < 0)
                 goto done;
-            if (flux_msg_sendzsock (p->sock, cpy) < 0)
+            if (zmqutil_msg_send (p->sock, cpy) < 0)
                 goto done;
             break;
         }
@@ -323,12 +325,12 @@ int module_sendmsg (module_t *p, const flux_msg_t *msg)
                 goto done;
             if (flux_msg_route_delete_last (cpy) < 0)
                 goto done;
-            if (flux_msg_sendzsock (p->sock, cpy) < 0)
+            if (zmqutil_msg_send (p->sock, cpy) < 0)
                 goto done;
             break;
         }
         default:
-            if (flux_msg_sendzsock (p->sock, msg) < 0)
+            if (zmqutil_msg_send (p->sock, msg) < 0)
                 goto done;
             break;
     }
@@ -612,13 +614,13 @@ module_t *module_add (modhash_t *mh, const char *path)
         log_err ("zsock_bind inproc://%s", module_get_uuid (p));
         goto cleanup;
     }
-    if (!(p->broker_w = flux_zmq_watcher_create (
+    if (!(p->broker_w = zmqutil_watcher_create (
                                         flux_get_reactor (p->modhash->broker_h),
                                         p->sock,
                                         FLUX_POLLIN,
                                         module_cb,
                                         p))) {
-        log_err ("flux_zmq_watcher_create");
+        log_err ("zmqutil_watcher_create");
         goto cleanup;
     }
     /* Set creds for connection.

src/broker/overlay.c

@@ -19,6 +19,8 @@
 #include <jansson.h>
 #include <uuid.h>
 
+#include "src/common/libzmqutil/msg_zsock.h"
+#include "src/common/libzmqutil/reactor.h"
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libutil/log.h"
 #include "src/common/libutil/kary.h"
@@ -349,7 +351,7 @@ static int overlay_sendmsg_parent (struct overlay *ov, const flux_msg_t *msg)
         errno = EHOSTUNREACH;
         goto done;
     }
-    rc = flux_msg_sendzsock (ov->parent.zsock, msg);
+    rc = zmqutil_msg_send (ov->parent.zsock, msg);
     if (rc == 0)
         ov->parent.lastsent = flux_reactor_now (ov->reactor);
 done:
@@ -501,7 +503,7 @@ static int overlay_sendmsg_child (struct overlay *ov, const flux_msg_t *msg)
         errno = EHOSTUNREACH;
         goto done;
     }
-    rc = flux_msg_sendzsock_ex (ov->bind_zsock, msg, true);
+    rc = zmqutil_msg_send_ex (ov->bind_zsock, msg, true);
 done:
     return rc;
 }
@@ -569,7 +571,7 @@ static void child_cb (flux_reactor_t *r, flux_watcher_t *w,
     struct child *child;
     int status;
 
-    if (!(msg = flux_msg_recvzsock (ov->bind_zsock)))
+    if (!(msg = zmqutil_msg_recv (ov->bind_zsock)))
         return;
     if (flux_msg_get_type (msg, &type) < 0
         || !(sender = flux_msg_route_last (msg)))
@@ -630,7 +632,7 @@ static void parent_cb (flux_reactor_t *r, flux_watcher_t *w,
     int type;
     const char *topic = NULL;
 
-    if (!(msg = flux_msg_recvzsock (ov->parent.zsock)))
+    if (!(msg = zmqutil_msg_recv (ov->parent.zsock)))
         return;
     if (flux_msg_get_type (msg, &type) < 0) {
         goto drop;
@@ -770,11 +772,11 @@ static int overlay_zap_init (struct overlay *ov)
         log_err ("could not bind to %s", ZAP_ENDPOINT);
         return -1;
     }
-    if (!(ov->zap_w = flux_zmq_watcher_create (ov->reactor,
-                                               ov->zap,
-                                               FLUX_POLLIN,
-                                               overlay_zap_cb,
-                                               ov)))
+    if (!(ov->zap_w = zmqutil_watcher_create (ov->reactor,
+                                              ov->zap,
+                                              FLUX_POLLIN,
+                                              overlay_zap_cb,
+                                              ov)))
         return -1;
     flux_watcher_start (ov->zap_w);
     return 0;
@@ -943,12 +945,12 @@ int overlay_connect (struct overlay *ov)
         zsock_set_identity (ov->parent.zsock, ov->uuid);
         if (zsock_connect (ov->parent.zsock, "%s", ov->parent.uri) < 0)
             goto nomem;
-        if (!(ov->parent.w = flux_zmq_watcher_create (ov->reactor,
-                                                      ov->parent.zsock,
-                                                      FLUX_POLLIN,
-                                                      parent_cb,
-                                                      ov)))
-        return -1;
+        if (!(ov->parent.w = zmqutil_watcher_create (ov->reactor,
+                                                     ov->parent.zsock,
+                                                     FLUX_POLLIN,
+                                                     parent_cb,
+                                                     ov)))
+            return -1;
         flux_watcher_start (ov->parent.w);
         if (hello_request_send (ov, ov->rank, FLUX_CORE_VERSION_HEX) < 0)
             return -1;
@@ -982,11 +984,11 @@ int overlay_bind (struct overlay *ov, const char *uri)
      */
     if (!(ov->bind_uri = zsock_last_endpoint (ov->bind_zsock)))
         return -1;
-    if (!(ov->bind_w = flux_zmq_watcher_create (ov->reactor,
-                                                ov->bind_zsock,
-                                                FLUX_POLLIN,
-                                                child_cb,
-                                                ov)))
+    if (!(ov->bind_w = zmqutil_watcher_create (ov->reactor,
+                                               ov->bind_zsock,
+                                               FLUX_POLLIN,
+                                               child_cb,
+                                               ov)))
         return -1;
     flux_watcher_start (ov->bind_w);
     /* Ensure that ipc files are removed when the broker exits.

src/broker/test/overlay.c

@@ -18,6 +18,7 @@
 #include <czmq.h>
 
 #include "src/common/libtap/tap.h"
+#include "src/common/libzmqutil/msg_zsock.h"
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libtestutil/util.h"
 #include "src/common/libutil/stdlog.h"
@@ -436,7 +437,7 @@ void trio (flux_t *h)
     zsock_set_identity (zsock_none, "2");
     ok (zsock_connect (zsock_none, "%s", parent_uri) == 0,
         "none-2: zsock_connect %s (no security) works", parent_uri);
-    ok (flux_msg_sendzsock (zsock_none, msg) == 0,
+    ok (zmqutil_msg_send (zsock_none, msg) == 0,
         "none-2: zsock_msg_sendzsock works");
 
     /* 2) Curve, and correct server publc key, but client public key
@@ -453,8 +454,8 @@ void trio (flux_t *h)
     zcert_destroy (&cert);
     ok (zsock_connect (zsock_curve, "%s", parent_uri) == 0,
         "curve-2: zsock_connect %s works", parent_uri);
-    ok (flux_msg_sendzsock (zsock_curve, msg) == 0,
-        "curve-2: flux_msg_sendzsock works");
+    ok (zmqutil_msg_send (zsock_curve, msg) == 0,
+        "curve-2: zmqutil_msg_send works");
 
     /* Neither of the above attempts should have gotten a message through.
      */

src/common/Makefile.am

@@ -23,7 +23,8 @@ SUBDIRS = \
 	libhostlist \
 	librlist \
 	libczmqcontainers \
-	libccan
+	libccan \
+	libzmqutil
 
 AM_CFLAGS = $(WARNING_CFLAGS) $(CODE_COVERAGE_CFLAGS)
 AM_LDFLAGS = $(CODE_COVERAGE_LIBS)
@@ -34,6 +35,7 @@ fluxinclude_HEADERS = core.h schedutil.h
 noinst_LTLIBRARIES = libflux-internal.la
 libflux_internal_la_SOURCES =
 libflux_internal_la_LIBADD = \
+	$(builddir)/libflux/libmessage.la \
 	$(builddir)/liblsd/liblsd.la \
 	$(builddir)/libccan/libccan.la \
 	$(builddir)/libutil/libutil.la \
@@ -47,7 +49,6 @@ libflux_internal_la_LIBADD = \
 	$(builddir)/libhostlist/libhostlist.la \
 	$(builddir)/libczmqcontainers/libczmqcontainers.la \
 	$(JANSSON_LIBS) \
-	$(ZMQ_LIBS) \
 	$(LIBUUID_LIBS) \
 	$(LIBPTHREAD) \
 	$(LIBDL) \