libflux: add flux_watcher_is_active()

doc/Makefile.am

@@ -208,6 +208,7 @@ MAN3_FILES_SECONDARY = \
 	man3/flux_reactor_active_decref.3 \
 	man3/flux_fd_watcher_get_fd.3 \
 	man3/flux_watcher_stop.3 \
+	man3/flux_watcher_is_active.3 \
 	man3/flux_watcher_destroy.3 \
 	man3/flux_watcher_next_wakeup.3 \
 	man3/flux_handle_watcher_get_flux.3 \

doc/man3/flux_watcher_start.rst

@@ -13,6 +13,8 @@ SYNOPSIS
 
   void flux_watcher_stop (flux_watcher_t *w);
 
+  bool flux_watcher_is_active (flux_watcher_t *w);
+
   void flux_watcher_destroy (flux_watcher_t *w);
 
   double flux_watcher_next_wakeup (flux_watcher_t *w);
@@ -31,6 +33,9 @@ so that it stops receiving events. If :var:`w` is already inactive, the call
 has no effect.  This may be called from within a :type:`flux_watcher_f`
 callback.
 
+:func:`flux_watcher_is_active` returns a true value if the watcher is active
+(i.e. it has been started and not yet stopped) and false otherwise.
+
 :func:`flux_watcher_destroy` destroys a :type:`flux_watcher_t` object :var:`w`,
 after stopping it. It is not safe to destroy a watcher object within a
 :type:`flux_watcher_f` callback.

doc/manpages.py

@@ -283,6 +283,7 @@
     ('man3/flux_timer_watcher_create', 'flux_timer_watcher_reset', 'set/reset a timer', [author], 3),
     ('man3/flux_timer_watcher_create', 'flux_timer_watcher_create', 'set/reset a timer', [author], 3),
     ('man3/flux_watcher_start', 'flux_watcher_stop', 'start/stop/destroy/query reactor watcher', [author], 3),
+    ('man3/flux_watcher_start', 'flux_watcher_is_active', 'start/stop/destroy/query reactor watcher', [author], 3),
     ('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),

src/common/libflux/attr.c

@@ -14,16 +14,14 @@
 #include <errno.h>
 #include <stdbool.h>
 #include <jansson.h>
+#include <flux/core.h>
 
 #include "ccan/str/str.h"
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libhostlist/hostlist.h"
 #include "src/common/libidset/idset.h"
 #include "src/common/libutil/errprintf.h"
 
-#include "attr.h"
-#include "rpc.h"
-
 enum {
     FLUX_ATTRFLAG_IMMUTABLE = 1,
 };

src/common/libflux/attr.h

@@ -25,8 +25,6 @@
  * attributes programmatically.
  */
 
-#include "handle.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/common/libflux/barrier.h

@@ -11,8 +11,6 @@
 #ifndef _FLUX_CORE_BARRIER_H
 #define _FLUX_CORE_BARRIER_H
 
-#include "handle.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/common/libflux/composite_future.c

@@ -11,11 +11,10 @@
 #if HAVE_CONFIG_H
 #include "config.h"
 #endif
+#include <flux/core.h>
 
 #include "src/common/libczmqcontainers/czmq_containers.h"
 
-#include "future.h"
-
 /*  Type-specific data for a composite future:
  */
 struct composite_future {

src/common/libflux/conf.c

@@ -31,7 +31,6 @@
 #include "src/common/libutil/errprintf.h"
 #include "ccan/str/str.h"
 
-#include "conf.h"
 #include "conf_private.h"
 
 struct builtin {

src/common/libflux/conf.h

@@ -11,7 +11,6 @@
 #ifndef _FLUX_CORE_CONF_H
 #define _FLUX_CORE_CONF_H
 
-#include <flux/core.h>
 #include <stdarg.h>
 
 #ifdef __cplusplus

src/common/libflux/connector_interthread.c

@@ -27,13 +27,13 @@
 #include <unistd.h>
 #include <poll.h>
 #include <pthread.h>
-
 #include <flux/core.h>
 
 #include "src/common/libutil/errprintf.h"
 #include "src/common/libutil/aux.h"
 #include "ccan/list/list.h"
 #include "ccan/str/str.h"
+
 #include "message_private.h" // for access to msg->aux
 #include "msg_deque.h"
 

src/common/libflux/connector_loop.c

@@ -15,7 +15,6 @@
 #endif
 #include <unistd.h>
 #include <poll.h>
-
 #include <flux/core.h>
 
 #include "ccan/str/str.h"

src/common/libflux/disconnect.h

@@ -11,8 +11,6 @@
 #ifndef _FLUX_CORE_DISCONNECT_H
 #define _FLUX_CORE_DISCONNECT_H
 
-#include "message.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/common/libflux/ev_flux.c

@@ -13,8 +13,7 @@
 #endif
 #include <stddef.h>
 #include <stdbool.h>
-
-#include "handle.h"
+#include <flux/core.h>
 
 #include "src/common/libev/ev.h"
 
@@ -97,6 +96,11 @@ void ev_flux_stop (struct ev_loop *loop, struct ev_flux *w)
     ev_idle_stop (loop, &w->idle_w);
 }
 
+bool ev_flux_is_active (struct ev_flux *w)
+{
+    return ev_is_active (&w->prepare_w);
+}
+
 /*
  * vi:tabstop=4 shiftwidth=4 expandtab
  */

src/common/libflux/ev_flux.h

@@ -32,5 +32,6 @@ struct ev_flux {
 int ev_flux_init (struct ev_flux *w, ev_flux_f cb, flux_t *h, int events);
 void ev_flux_start (struct ev_loop *loop, struct ev_flux *w);
 void ev_flux_stop (struct ev_loop *loop, struct ev_flux *w);
+bool ev_flux_is_active (struct ev_flux *w);
 
 #endif /* !_EV_FLUX_H */

src/common/libflux/event.c

@@ -15,13 +15,10 @@
 #include <stdbool.h>
 #include <stdarg.h>
 #include <jansson.h>
+#include <flux/core.h>
 
 #include "ccan/base64/base64.h"
 
-#include "event.h"
-#include "rpc.h"
-#include "message.h"
-
 flux_future_t *flux_event_subscribe_ex (flux_t *h,
                                         const char *topic,
                                         int flags)

src/common/libflux/event.h

@@ -11,9 +11,6 @@
 #ifndef _FLUX_CORE_EVENT_H
 #define _FLUX_CORE_EVENT_H
 
-#include "message.h"
-#include "future.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/common/libflux/flog.c

@@ -20,12 +20,7 @@
 #include <unistd.h>
 #include <assert.h>
 #include <inttypes.h>
-
-#include "flog.h"
-#include "attr.h"
-#include "message.h"
-#include "request.h"
-#include "rpc.h"
+#include <flux/core.h>
 
 #include "src/common/libutil/wallclock.h"
 #include "src/common/libutil/stdlog.h"

src/common/libflux/flog.h

@@ -16,8 +16,6 @@
 #include <stdlib.h>
 #include <errno.h>
 
-#include "handle.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/common/libflux/flux.h

@@ -12,11 +12,11 @@
 #define _FLUX_CORE_FLUX_H
 
 #include "types.h"
+#include "message.h"
 #include "handle.h"
 #include "reactor.h"
 #include "msg_handler.h"
 #include "connector.h"
-#include "message.h"
 #include "msglist.h"
 #include "request.h"
 #include "response.h"

src/common/libflux/fripp.c

@@ -21,11 +21,7 @@
 #include <syslog.h>
 #include <time.h>
 #include <unistd.h>
-
-#include "attr.h"
-#include "flog.h"
-#include "fripp.h"
-#include "reactor.h"
+#include <flux/core.h>
 
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libutil/iterators.h"

src/common/libflux/future.c

@@ -16,13 +16,11 @@
 #include <stdlib.h>
 #include <errno.h>
 #include <assert.h>
+#include <flux/core.h>
 
 #include "src/common/libczmqcontainers/czmq_containers.h"
 #include "src/common/libutil/aux.h"
 
-#include "future.h"
-#include "flog.h"
-
 struct now_context {
     flux_t *h;              // (optional) cloned flux_t handle
     flux_reactor_t *r;      // reactor created for this get/check

src/common/libflux/future.h

@@ -11,12 +11,6 @@
 #ifndef _FLUX_CORE_FUTURE_H
 #define _FLUX_CORE_FUTURE_H
 
-#include "reactor.h"
-#include "types.h"
-#include "handle.h"
-#include "msg_handler.h"
-#include "flog.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -25,8 +19,6 @@ extern "C" {
  * See flux_future_then(3).
  */
 
-typedef struct flux_future flux_future_t;
-
 typedef void (*flux_continuation_f)(flux_future_t *f, void *arg);
 
 int flux_future_then (flux_future_t *f, double timeout,

src/common/libflux/handle.c

@@ -24,6 +24,7 @@
 #include <caliper/cali.h>
 #include <sys/syscall.h>
 #endif
+#include <flux/core.h>
 
 #include "src/common/libflux/plugin_private.h"
 #include "src/common/librouter/rpc_track.h"
@@ -38,13 +39,6 @@
 #include "ccan/array_size/array_size.h"
 #include "ccan/str/str.h"
 
-#include "handle.h"
-#include "reactor.h"
-#include "connector.h"
-#include "message.h"
-#include "msg_handler.h" // for flux_sleep_on ()
-#include "flog.h"
-#include "conf.h"
 #include "msg_deque.h"
 #include "message_private.h" // to check msg refcount in flux_send_new ()
 
@@ -404,6 +398,34 @@ void flux_close (flux_t *h)
     flux_handle_destroy (h);
 }
 
+int flux_set_reactor (flux_t *h, flux_reactor_t *r)
+{
+    if (flux_aux_get (h, "flux::reactor")) {
+        errno = EEXIST;
+        return -1;
+    }
+    if (flux_aux_set (h, "flux::reactor", r, NULL) < 0)
+        return -1;
+    return 0;
+}
+
+flux_reactor_t *flux_get_reactor (flux_t *h)
+{
+    flux_reactor_t *r = flux_aux_get (h, "flux::reactor");
+    if (!r) {
+        if ((r = flux_reactor_create (0))) {
+            if (flux_aux_set (h,
+                              "flux::reactor",
+                              r,
+                              (flux_free_f)flux_reactor_destroy) < 0) {
+                flux_reactor_destroy (r);
+                r = NULL;
+            }
+        }
+    }
+    return r;
+}
+
 /* Create an idset that is configured as an allocator per idset_alloc(3) to
  * be used as a matchtag allocator.  Remove FLUX_MATCHTAG_NONE from the pool.
  */

src/common/libflux/handle.h

@@ -15,15 +15,10 @@
 #include <stdbool.h>
 #include <string.h>
 
-#include "types.h"
-#include "message.h"
-
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-typedef struct flux_handle flux_t;
-
 typedef struct {
     int request_tx;
     int request_rx;
@@ -100,6 +95,13 @@ void flux_decref(flux_t *h);
  */
 flux_t *flux_clone (flux_t *orig);
 
+/* Get/set the reactor associated with the handle.
+ * A reactor is created with flags=0 if 'get' is called before one is set.
+ * 'set' fails if there is already a reactor.
+ */
+flux_reactor_t *flux_get_reactor (flux_t *h);
+int flux_set_reactor (flux_t *h, flux_reactor_t *r);
+
 /* Drop connection to broker and re-establish, if supported by connector.
  */
 int flux_reconnect (flux_t *h);

src/common/libflux/message.c

@@ -35,14 +35,13 @@
 #include <fnmatch.h>
 #include <inttypes.h>
 #include <jansson.h>
+#include <flux/core.h>
 
 #include "src/common/libutil/aux.h"
 #include "src/common/libutil/errno_safe.h"
 #include "ccan/array_size/array_size.h"
 #include "ccan/str/str.h"
 
-#include "message.h"
-
 #include "message_private.h"
 #include "message_iovec.h"
 #include "message_route.h"

src/common/libflux/message.h

@@ -17,14 +17,11 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <sys/types.h>
-#include "types.h"
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-typedef struct flux_msg flux_msg_t;
-
 enum {
     FLUX_MSGTYPE_REQUEST    = 0x01,
     FLUX_MSGTYPE_RESPONSE   = 0x02,

src/common/libflux/message_iovec.c

@@ -17,8 +17,8 @@
 #include <arpa/inet.h>
 #include <assert.h>
 #include <inttypes.h>
+#include <flux/core.h>
 
-#include "message.h"
 #include "message_private.h"
 #include "message_route.h"
 #include "message_proto.h"