modctl: redesign based on libmrpc

Redesigned modctl can operate on a hierarchy of dynamically loaded
modules (e.g. comms modules, modules that load modules, etc).
In addition the new modctl:

Fixes flux-framework#57 flux-module list works around lack of sync with usleep
Fixes flux-framework#56 flux-module cannot unload a module it did not load
Fixes flux-framework#94 Request flux module load rank option

modctl speaks RFC 5 to services that implement module extensions.
It implements an "mrpc" based protocol that executes module operations
in parallel on a user-specified nodeset.  The modctl "API" is no
longer exported via libflux-core and is kept private between modctl
and flux-module.

The use of mrpc greatly simplifies the design compared to the previous
modctl.  However, the new design has following caveats:
- does not yet implement scalable .so loading through the KVS
- there is no "smart" data reduction of RFC 5 responses,
  except what libmrpc achieves by KVS object squashing
- some scenarios exist where modctl or the flux-module may hang in
  a kvs_fence(), e.g. node failure or flux-module abort at just he
  wrong place

Some of these scalability and resiliency issues can be solved more
generally through changes to mrpc.  This is likey preferable to building
a more complex, standalone modctl system with these characteristics.

src/cmd/Makefile.am

@@ -12,7 +12,6 @@ AM_CPPFLAGS = \
 
 fluxcmd_ldadd = \
 	$(top_builddir)/src/modules/live/liblive.la \
-	$(top_builddir)/src/modules/modctl/libmodctl.la \
 	$(top_builddir)/src/modules/kvs/libkvs.la \
 	$(top_builddir)/src/modules/api/libapi.la \
 	$(top_builddir)/src/common/libflux/libflux.la \
@@ -60,3 +59,8 @@ flux_mping_LDADD = \
 	$(top_builddir)/src/common/liblsd/liblsd.la \
 	$(fluxcmd_ldadd) \
 	$(LIBUTIL)
+flux_module_LDADD = \
+	$(top_builddir)/src/modules/modctl/libmodctl.la \
+	$(top_builddir)/src/common/libmrpc/libmrpc.la \
+	$(fluxcmd_ldadd) \
+	$(LIBUTIL)

src/cmd/flux-module.c

@@ -31,27 +31,32 @@
 #include <json.h>
 #include <flux/core.h>
 
+#include "src/modules/modctl/modctl.h"
 #include "src/common/libutil/xzmalloc.h"
 #include "src/common/libutil/jsonutil.h"
 #include "src/common/libutil/log.h"
 #include "src/common/libutil/shortjson.h"
 #include "src/common/libutil/readall.h"
 
+const int max_idle = 99;
 
-#define OPTIONS "+hr:"
+
+#define OPTIONS "+hr:an:"
 static const struct option longopts[] = {
     {"help",       no_argument,        0, 'h'},
     {"rank",       required_argument,  0, 'r'},
+    {"all",        no_argument,        0, 'a'},
+    {"nodeset",    required_argument,  0, 'n'},
     { 0, 0, 0, 0 },
 };
 
-void mod_lsmod (flux_t h, uint32_t nodeid, int ac, char **av);
-void mod_rmmod (flux_t h, uint32_t nodeid, int ac, char **av);
-void mod_insmod (flux_t h, uint32_t nodeid, int ac, char **av);
+void mod_lsmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av);
+void mod_rmmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av);
+void mod_insmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av);
 
 typedef struct {
     const char *name;
-    void (*fun)(flux_t h, uint32_t nodeid, int ac, char **av);
+    void (*fun)(flux_t h, uint32_t nodeid, const char *ns, int ac, char **av);
 } func_t;
 
 static func_t funcs[] = {
@@ -76,7 +81,9 @@ void usage (void)
 "       flux-module [OPTIONS] remove module\n"
 "       flux-module [OPTIONS] load module [arg ...]\n"
 "where OPTIONS are:\n"
-"       -r,--rank N    specify nodeid to send request\n"
+"       -r,--rank N      specify nodeid to send request\n"
+"       -n,--nodeset NS  use modctl to distribute request to NS\n"
+"       -a,--all         use modctl to distribute request to entire session\n"
 );
     exit (1);
 }
@@ -88,6 +95,8 @@ int main (int argc, char *argv[])
     char *cmd;
     func_t *f;
     uint32_t nodeid = FLUX_NODEID_ANY;
+    char *nodeset = NULL;
+    bool aopt = false;
 
     log_init ("flux-module");
 
@@ -99,6 +108,12 @@ int main (int argc, char *argv[])
             case 'r': /* --rank N */
                 nodeid = strtoul (optarg, NULL, 10);
                 break;
+            case 'a': /* --all */
+                aopt = true;
+                break;
+            case 'n': /* --nodeset NS */
+                nodeset = xstrdup (optarg);
+                break;
             default:
                 usage ();
                 break;
@@ -112,14 +127,20 @@ int main (int argc, char *argv[])
 
     if (!(h = flux_api_open ()))
         err_exit ("flux_api_open");
-    f->fun (h, nodeid, argc - optind, argv + optind);
+    if (!nodeset && aopt) {
+        int size = flux_size (h);
+        nodeset = size > 1 ? xasprintf ("[0-%d]", size - 1) : xstrdup ("0");
+    }
+    f->fun (h, nodeid, nodeset, argc - optind, argv + optind);
     flux_api_close (h);
 
+    if (nodeset)
+        free (nodeset);
     log_fini ();
     return 0;
 }
 
-void mod_insmod (flux_t h, uint32_t nodeid, int ac, char **av)
+void mod_insmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av)
 {
     char *modpath = NULL;
     char *modname = NULL;
@@ -138,20 +159,33 @@ void mod_insmod (flux_t h, uint32_t nodeid, int ac, char **av)
         if (!(modpath = flux_modfind (searchpath, modname)))
             msg_exit ("%s: not found in module search path", modname);
     }
-    if (flux_insmod (h, nodeid, modpath, ac - 1, av + 1) < 0)
-        err_exit ("%s", av[0]);
+    if (ns) {
+        if (flux_modctl_load (h, ns, modpath, ac - 1, av + 1) < 0)
+            err_exit ("%s", av[0]);
+    } else {
+        if (flux_insmod (h, nodeid, modpath, ac - 1, av + 1) < 0)
+            err_exit ("%s", av[0]);
+    }
     if (modpath)
         free (modpath);
     if (modname)
         free (modname);
 }
 
-void mod_rmmod (flux_t h, uint32_t nodeid, int ac, char **av)
+void mod_rmmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av)
 {
+    char *modname = NULL;
+
     if (ac != 1)
         usage ();
-    if (flux_rmmod (h, nodeid, av[0]) < 0)
-        err_exit ("%s", av[0]);
+    modname = av[0];
+    if (ns) {
+        if (flux_modctl_unload (h, ns, modname) < 0)
+            err_exit ("modctl_unload %s", modname);
+    } else {
+        if (flux_rmmod (h, nodeid, modname) < 0)
+            err_exit ("%s", av[0]);
+    }
 }
 
 const char *snip (const char *s, int n)
@@ -179,7 +213,7 @@ static int lsmod_cb (const char *name, int size, const char *digest, int idle,
     return 0;
 }
 
-void mod_lsmod (flux_t h, uint32_t nodeid, int ac, char **av)
+void mod_lsmod (flux_t h, uint32_t nodeid, const char *ns, int ac, char **av)
 {
     char *svc = "cmb";
 
@@ -189,8 +223,13 @@ void mod_lsmod (flux_t h, uint32_t nodeid, int ac, char **av)
         svc = av[0];
     printf ("%-20s %6s %7s %4s %s\n",
             "Module", "Size", "Digest", "Idle", "Nodeset");
-    if (flux_lsmod (h, nodeid, svc, lsmod_cb, NULL) < 0)
-        err_exit ("%s", svc);
+    if (ns) {
+        if (flux_modctl_list (h, svc, ns, lsmod_cb, NULL) < 0)
+            err_exit ("modctl_list");
+    } else {
+        if (flux_lsmod (h, nodeid, svc, lsmod_cb, NULL) < 0)
+            err_exit ("%s", svc);
+    }
 }
 
 

src/include/flux/core.h

@@ -9,7 +9,6 @@
 #include "src/modules/api/api.h"
 #include "src/modules/kvs/kvs.h"
 #include "src/modules/live/live.h"
-#include "src/modules/modctl/modctl.h"
 #include "src/modules/barrier/barrier.h"
 
 #endif /* FLUX_CORE_H */

src/lib/libcore/Makefile.am

@@ -9,7 +9,6 @@ libflux_core_la_LIBADD = \
 	$(top_builddir)/src/modules/kvs/libkvs.la \
 	$(top_builddir)/src/modules/api/libapi.la \
 	$(top_builddir)/src/modules/live/liblive.la \
-	$(top_builddir)/src/modules/modctl/libmodctl.la \
 	$(top_builddir)/src/modules/barrier/libbarrier.la \
 	$(top_builddir)/src/common/libflux/libflux.la \
 	$(top_builddir)/src/common/libutil/libutil.la \

src/lib/libcore/core.h

@@ -34,7 +34,6 @@
 #include "core/api.h"
 #include "core/kvs.h"
 #include "core/live.h"
-#include "core/modctl.h"
 #include "core/barrier.h"
 
 #endif /* !_FLUX_CORE_H */

src/modules/modctl/Makefile.am

@@ -10,6 +10,7 @@ AM_CPPFLAGS = \
 fluxmod_LTLIBRARIES = modctl.la
 
 fluxmod_libadd = \
+	$(top_builddir)/src/common/libmrpc/libmrpc.la \
 	$(top_builddir)/src/modules/kvs/libkvs.la \
 	$(top_builddir)/src/common/libflux/libflux.la \
 	$(top_builddir)/src/common/libutil/libutil.la \
@@ -19,13 +20,43 @@ fluxmod_libadd = \
 fluxmod_ldflags = -avoid-version -module -shared -export-dynamic \
         -export-symbols-regex '^mod_(main|name)$$'
 
-modctl_la_SOURCES = modctl.c
+modctl_la_SOURCES = \
+	modctl.c \
+	proto.c \
+	proto.h
 modctl_la_LDFLAGS = $(fluxmod_ldflags)
 modctl_la_LIBADD = $(fluxmod_libadd)
 
 #
 # API for module
 #
-fluxcoreinclude_HEADERS = modctl.h
+noinst_HEADERS = modctl.h
 noinst_LTLIBRARIES = libmodctl.la
-libmodctl_la_SOURCES = libmodctl.c
+libmodctl_la_SOURCES = libmodctl.c proto.c
+
+#
+# Unit tests
+#
+TESTS = \
+	test_proto.t
+
+test_ldadd = \
+        $(top_builddir)/src/common/libutil/libutil.la \
+        $(top_builddir)/src/common/libtap/libtap.la \
+        $(top_builddir)/src/common/liblsd/liblsd.la \
+        $(JSON_LIBS) $(LIBZMQ) $(LIBCZMQ) $(LIBPTHREAD)
+
+test_cppflags = \
+        -DTEST_MAIN \
+        -I$(top_srcdir)/src/common/libtap \
+        $(AM_CPPFLAGS)
+
+check_PROGRAMS = $(TESTS)
+
+TEST_EXTENSIONS = .t
+T_LOG_DRIVER = env AM_TAP_AWK='$(AWK)' $(SHELL) \
+       $(top_srcdir)/config/tap-driver.sh
+
+test_proto_t_SOURCES = proto.c
+test_proto_t_CPPFLAGS = $(test_cppflags)
+test_proto_t_LDADD = $(test_ldadd)