kvs: return ENOSYS on unfinished requests

Problem: When the KVS module is unloaded, any unfinished requests
(get, put, wait-version, etc.) should get an ENOSYS response.  That
is presently not handled.

Solution: Track all requests sent to the kvs module that will not be
returned immediately.  Upon module exit, respond to all unfinished
requests with ENOSYS.

Fixes #5979

Author: chu11

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);
 }
@@ -1437,8 +1460,10 @@ static void lookup_request_cb (flux_t *h, flux_msg_handler_t *mh,
 
     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);
 }
 
@@ -1476,8 +1503,10 @@ static void lookup_plus_request_cb (flux_t *h, flux_msg_handler_t *mh,
 
     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;
     }
 
@@ -1502,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);
 }
 
 
@@ -1526,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;
 }
 
@@ -1624,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;
 
@@ -1666,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;
     }
 
@@ -1724,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);
 }
 
 
@@ -1809,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:
@@ -1854,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;
     }
 
@@ -1935,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;
 }
@@ -1961,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;
     }
 
@@ -1977,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 */
     }
 
@@ -1985,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,
@@ -2021,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;
         }
     }
@@ -2035,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,
@@ -3026,6 +3081,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.
      */