Merge pull request #6049 from chu11/issue5979_kvs_enosys_unload

kvs: return ENOSYS on unfinished requests

Author: mergify[bot]

src/modules/kvs/Makefile.am

@@ -9,6 +9,7 @@ AM_CPPFLAGS = \
 	-I$(top_srcdir) \
 	-I$(top_srcdir)/src/include \
 	-I$(top_srcdir)/src/common/libccan \
+	-I$(top_builddir)/src/common/librouter \
 	-I$(top_builddir)/src/common/libflux \
 	$(JANSSON_CFLAGS)
 

src/modules/kvs/kvs.c

@@ -36,6 +36,7 @@
 #include "src/common/libkvs/kvs_util_private.h"
 #include "src/common/libcontent/content.h"
 #include "src/common/libutil/fsd.h"
+#include "src/common/librouter/msg_hash.h"
 
 #include "waitqueue.h"
 #include "cache.h"
@@ -77,6 +78,7 @@ struct kvs_ctx {
     char *hash_name;
     unsigned int seq;           /* for commit transactions */
     kvs_checkpoint_t *kcp;
+    zhashx_t *requests;         /* track unfinished requests */
     struct list_head work_queue;
 };
 
@@ -111,11 +113,23 @@ static void kvs_ctx_destroy (struct kvs_ctx *ctx)
         flux_watcher_destroy (ctx->idle_w);
         kvs_checkpoint_destroy (ctx->kcp);
         free (ctx->hash_name);
+        zhashx_destroy (&ctx->requests);
         free (ctx);
         errno = saved_errno;
     }
 }
 
+static void request_tracking_add (struct kvs_ctx *ctx, const flux_msg_t *msg)
+{
+    /* ignore if item already tracked */
+    zhashx_insert (ctx->requests, msg, (flux_msg_t *)msg);
+}
+
+static void request_tracking_remove (struct kvs_ctx *ctx, const flux_msg_t *msg)
+{
+    zhashx_delete (ctx->requests, msg);
+}
+
 static void work_queue_check_append_wrapper (struct kvsroot *root,
                                              void *arg)
 {
@@ -164,6 +178,8 @@ static struct kvs_ctx *kvs_ctx_create (flux_t *h)
             goto error;
     }
     ctx->transaction_merge = 1;
+    if (!(ctx->requests = msg_hash_create (MSG_HASH_TYPE_UUID_MATCHTAG)))
+        goto error;
     list_head_init (&ctx->work_queue);
     return ctx;
 error:
@@ -364,6 +380,13 @@ static void getroot_completion (flux_future_t *f, void *arg)
 error:
     if (flux_respond_error (ctx->h, msg, errno, NULL) < 0)
         flux_log_error (ctx->h, "%s: flux_respond_error", __FUNCTION__);
+    /* N.B. getroot request from other requests (e.g. lookup, commit)
+     * may stall and be tracked.  So we need to remove tracking of the
+     * request if there is an error.  We do not remove tracking on
+     * getroot success, as the original request (e.g. lookup, commit)
+     * will deal with the success case.
+     */
+    request_tracking_remove (ctx, msg);
     flux_msg_destroy (msg);
     flux_future_destroy (f);
 }
@@ -1273,11 +1296,10 @@ static void lookup_wait_error_cb (wait_t *w, int errnum, void *arg)
 }
 
 static lookup_t *lookup_common (flux_t *h, flux_msg_handler_t *mh,
-                                const flux_msg_t *msg, void *arg,
+                                const flux_msg_t *msg, struct kvs_ctx *ctx,
                                 flux_msg_handler_f replay_cb,
                                 bool *stall)
 {
-    struct kvs_ctx *ctx = arg;
     int flags;
     const char *ns = NULL;
     const char *key;
@@ -1431,14 +1453,17 @@ static lookup_t *lookup_common (flux_t *h, flux_msg_handler_t *mh,
 static void lookup_request_cb (flux_t *h, flux_msg_handler_t *mh,
                                const flux_msg_t *msg, void *arg)
 {
+    struct kvs_ctx *ctx = arg;
     lookup_t *lh;
     json_t *val;
     bool stall = false;
 
-    if (!(lh = lookup_common (h, mh, msg, arg, lookup_request_cb,
+    if (!(lh = lookup_common (h, mh, msg, ctx, lookup_request_cb,
                               &stall))) {
-        if (stall)
+        if (stall) {
+            request_tracking_add (ctx, msg);
             return;
+        }
         goto error;
     }
 
@@ -1450,10 +1475,12 @@ static void lookup_request_cb (flux_t *h, flux_msg_handler_t *mh,
         flux_log_error (h, "%s: flux_respond_pack", __FUNCTION__);
     lookup_destroy (lh);
     json_decref (val);
+    request_tracking_remove (ctx, msg);
     return;
 error:
     if (flux_respond_error (h, msg, errno, NULL) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
     lookup_destroy (lh);
 }
 
@@ -1467,16 +1494,19 @@ static void lookup_request_cb (flux_t *h, flux_msg_handler_t *mh,
 static void lookup_plus_request_cb (flux_t *h, flux_msg_handler_t *mh,
                                     const flux_msg_t *msg, void *arg)
 {
+    struct kvs_ctx *ctx = arg;
     lookup_t *lh;
     json_t *val = NULL;
     const char *root_ref;
     int root_seq;
     bool stall = false;
 
-    if (!(lh = lookup_common (h, mh, msg, arg, lookup_plus_request_cb,
+    if (!(lh = lookup_common (h, mh, msg, ctx, lookup_plus_request_cb,
                               &stall))) {
-        if (stall)
+        if (stall) {
+            request_tracking_add (ctx, msg);
             return;
+        }
         goto error;
     }
 
@@ -1501,10 +1531,12 @@ static void lookup_plus_request_cb (flux_t *h, flux_msg_handler_t *mh,
     }
     lookup_destroy (lh);
     json_decref (val);
+    request_tracking_remove (ctx, msg);
     return;
 error:
     if (flux_respond_error (h, msg, errno, NULL) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
 }
 
 
@@ -1525,6 +1557,7 @@ static int finalize_transaction_req (treq_t *tr,
             flux_log_error (cbd->ctx->h, "%s: flux_respond_pack", __FUNCTION__);
     }
 
+    request_tracking_remove (cbd->ctx, req);
     return 0;
 }
 
@@ -1623,6 +1656,9 @@ static void relaycommit_request_cb (flux_t *h, flux_msg_handler_t *mh,
         goto error;
     }
 
+    /* N.B. no request tracking for relay.  The relay does not get a
+     * response, only the original via finalize_transaction_bynames().
+     */
     work_queue_check_append (ctx, root);
     return;
 
@@ -1665,8 +1701,10 @@ static void commit_request_cb (flux_t *h, flux_msg_handler_t *mh,
     }
 
     if (!(root = getroot (ctx, ns, mh, msg, NULL, &stall))) {
-        if (stall)
+        if (stall) {
+            request_tracking_add (ctx, msg);
             return;
+        }
         goto error;
     }
 
@@ -1723,11 +1761,13 @@ static void commit_request_cb (flux_t *h, flux_msg_handler_t *mh,
         }
         flux_future_destroy (f);
     }
+    request_tracking_add (ctx, msg);
     return;
 
 error:
     if (flux_respond_error (h, msg, errno, errmsg) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
 }
 
 
@@ -1808,6 +1848,9 @@ static void relayfence_request_cb (flux_t *h, flux_msg_handler_t *mh,
         work_queue_check_append (ctx, root);
     }
 
+    /* N.B. no request tracking for relay.  The relay does not get a
+     * response, only the original via finalize_transaction_bynames().
+     */
     return;
 
 error:
@@ -1853,8 +1896,10 @@ static void fence_request_cb (flux_t *h, flux_msg_handler_t *mh,
     }
 
     if (!(root = getroot (ctx, ns, mh, msg, NULL, &stall))) {
-        if (stall)
+        if (stall) {
+            request_tracking_add (ctx, msg);
             goto stall;
+        }
         goto error;
     }
 
@@ -1934,11 +1979,13 @@ static void fence_request_cb (flux_t *h, flux_msg_handler_t *mh,
         }
         flux_future_destroy (f);
     }
+    request_tracking_add (ctx, msg);
     return;
 
 error:
     if (flux_respond_error (h, msg, errno, errmsg) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
 stall:
     return;
 }
@@ -1960,8 +2007,10 @@ static void wait_version_request_cb (flux_t *h, flux_msg_handler_t *mh,
     }
 
     if (!(root = getroot (ctx, ns, mh, msg, NULL, &stall))) {
-        if (stall)
+        if (stall) {
+            request_tracking_add (ctx, msg);
             return;
+        }
         goto error;
     }
 
@@ -1976,6 +2025,7 @@ static void wait_version_request_cb (flux_t *h, flux_msg_handler_t *mh,
             flux_log_error (h, "%s: kvs_wait_version_add", __FUNCTION__);
             goto error;
         }
+        request_tracking_add (ctx, msg);
         return; /* stall */
     }
 
@@ -1984,11 +2034,13 @@ static void wait_version_request_cb (flux_t *h, flux_msg_handler_t *mh,
                            "rootref", root->ref) < 0)
         flux_log_error (h, "%s: flux_respond_pack", __FUNCTION__);
 
+    request_tracking_remove (ctx, msg);
     return;
 
 error:
     if (flux_respond_error (h, msg, errno, NULL) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
 }
 
 static void getroot_request_cb (flux_t *h, flux_msg_handler_t *mh,
@@ -2020,8 +2072,10 @@ static void getroot_request_cb (flux_t *h, flux_msg_handler_t *mh,
          */
         bool stall = false;
         if (!(root = getroot (ctx, ns, mh, msg, NULL, &stall))) {
-            if (stall)
+            if (stall) {
+                request_tracking_add (ctx, msg);
                 return;
+            }
             goto error;
         }
     }
@@ -2034,10 +2088,12 @@ static void getroot_request_cb (flux_t *h, flux_msg_handler_t *mh,
                            "flags", root->flags,
                            "namespace", root->ns_name) < 0)
         flux_log_error (h, "%s: flux_respond_pack", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
     return;
 error:
     if (flux_respond_error (h, msg, errno, NULL) < 0)
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+    request_tracking_remove (ctx, msg);
 }
 
 static void error_event_cb (flux_t *h, flux_msg_handler_t *mh,
@@ -2257,9 +2313,10 @@ static void stats_get_cb (flux_t *h, flux_msg_handler_t *mh,
     }
 
     if (flux_respond_pack (h, msg,
-                           "{ s:O s:O }",
+                           "{ s:O s:O s:i }",
                            "cache", cstats,
-                           "namespace", nsstats) < 0)
+                           "namespace", nsstats,
+                           "pending_requests", zhashx_size (ctx->requests)) < 0)
         flux_log_error (h, "%s: flux_respond_pack", __FUNCTION__);
     json_decref (tstats);
     json_decref (cstats);
@@ -3025,6 +3082,19 @@ int mod_main (flux_t *h, int argc, char **argv)
         flux_log_error (h, "flux_reactor_run");
         goto done;
     }
+    if (zhashx_size (ctx->requests) > 0) {
+        /* anything that has not yet completed gets an ENOSYS */
+        const flux_msg_t *msg = zhashx_first (ctx->requests);
+        while (msg) {
+            const char *topic = "unknown";
+            if (flux_msg_get_topic (msg, &topic) < 0)
+                flux_log_error (ctx->h, "%s: flux_msg_get_topic", __FUNCTION__);
+            if (flux_respond_error (ctx->h, msg, ENOSYS, NULL) < 0)
+                flux_log_error (ctx->h, "%s: flux_respond_error", __FUNCTION__);
+            flux_log (ctx->h, LOG_ERR, "failing pending '%s' request", topic);
+            msg = zhashx_next (ctx->requests);
+        }
+    }
     /* Checkpoint the KVS root to the content backing store.
      * If backing store is not loaded, silently proceed without checkpoint.
      */

t/t1000-kvs.t

@@ -1273,6 +1273,15 @@ test_expect_success 'setroot-unpause request with empty payload fails with EPROT
 	${RPC} kvs.setroot-unpause 71 </dev/null
 '
 
+#
+# ensure no lingering pending requests
+#
+
+test_expect_success 'kvs: no pending requests at end of tests before module removal' '
+	pendingcount=$(flux module stats -p pending_requests kvs) &&
+	test $pendingcount -eq 0
+'
+
 #
 # module corner cases
 #