cmd: use library for scanning treeobj valrefs

Problem: Both flux-dump and flux-fsck have similar code for scanning
a valref treeobj in parallel.

Use the common library function valref_blobrefs() instead.

Author: chu11

src/cmd/builtin/dump.c

@@ -28,25 +28,20 @@
 #include "ccan/str/str.h"
 
 #include "builtin.h"
-
-#define BLOBREF_ASYNC_MAX 1000
+#include "src/cmd/util/util.h"
 
 struct dump_valref_data
 {
     flux_t *h;
     json_t *treeobj;
     const flux_msg_t **msgs;
     const char *path;
-    int total_size;
-    int index;
     int count;
-    int in_flight;
+    int total_size;
     int errorcount;
     int errnum;
 };
 
-static void get_blobref (struct dump_valref_data *dvd);
-
 static void dump_treeobj (struct archive *ar,
                           flux_t *h,
                           const char *path,
@@ -175,70 +170,56 @@ static void dump_write_data (struct archive *ar, const void *data, int size)
                  "assuming non-fatal libarchive write size reporting error");
 }
 
-static void get_blobref_continuation (flux_future_t *f, void *arg)
+static int get_blobref_cb (const flux_msg_t *msg, int index, void *arg)
 {
     struct dump_valref_data *dvd = arg;
-    const flux_msg_t *msg;
     size_t len;
-    int *index;
 
-    index = flux_future_aux_get (f, "index");
-    if (flux_future_get (f, (const void **)&msg) < 0
-        || flux_response_decode_raw (msg, NULL, NULL, &len) < 0) {
-        read_error ("%s: missing blobref %d: %s",
+    if (flux_response_decode_raw (msg, NULL, NULL, &len) < 0) {
+        read_error ("%s: cannot decode blobref %d: %s",
                     dvd->path,
-                    (*index),
-                    future_strerror (f, errno));
-        flux_future_destroy (f);
+                    index,
+                    strerror (errno));
         dvd->errorcount++;
         dvd->errnum = errno;    /* we'll report the last errno */
-        return;
+        return -1;              /* -1 informs no more blobref retrieval */
     }
-    dvd->in_flight--;
     dvd->total_size += len;
-    dvd->msgs[(*index)] = flux_msg_incref (msg);
-
-    /* if an error has occurred, we won't get more blobrefs */
-    if (dvd->index < dvd->count
-        && !dvd->errorcount) {
-        get_blobref (dvd);
-        dvd->in_flight++;
-        dvd->index++;
-    }
-    flux_future_destroy (f);
+    dvd->msgs[index] = flux_msg_incref (msg);
+    return 0;
 }
 
-static void get_blobref (struct dump_valref_data *dvd)
+static int err_blobref_cb (int errnum, int index, void *arg)
 {
-    const char *blobref;
-    flux_future_t *f;
-    int *indexp;
-
-    blobref = treeobj_get_blobref (dvd->treeobj, dvd->index);
+    struct dump_valref_data *dvd = arg;
 
-    if (!(f = content_load_byblobref (dvd->h, blobref, content_flags))
-        || flux_future_then (f, -1, get_blobref_continuation, dvd) < 0) {
-        read_error ("%s: cannot load blobref %d", dvd->path, dvd->index);
-        flux_future_destroy (f);
-    }
-    if (!(indexp = (int *)malloc (sizeof (int))))
-        log_err_exit ("cannot allocate index memory");
-    (*indexp) = dvd->index;
-    if (flux_future_aux_set (f, "index", indexp, free) < 0)
-        log_err_exit ("could not save index value");
+    if (errnum == ENOENT)
+        read_error ("%s: missing blobref %d: %s",
+                    dvd->path,
+                    index,
+                    strerror (errnum));
+    else
+        read_error ("%s: error getting blobref %d: %s",
+                    dvd->path,
+                    index,
+                    strerror (errnum));
+    dvd->errorcount++;
+    dvd->errnum = errnum;    /* we'll report the last errno */
+    return -1;               /* -1 informs no more blobref retrieval*/
 }
 
 static int dump_valref_async (struct dump_valref_data *dvd)
 {
-    while (dvd->in_flight < BLOBREF_ASYNC_MAX
-           && dvd->index < dvd->count) {
-        get_blobref (dvd);
-        dvd->in_flight++;
-        dvd->index++;
-    }
-
-    if (flux_reactor_run (flux_get_reactor (dvd->h), 0) < 0)
-        log_err_exit ("flux_reactor_run");
+    const char *topic = "content.load";
+    if (content_flags & CONTENT_FLAG_CACHE_BYPASS)
+        topic = "content-backing.load";
+    if (valref_blobrefs (dvd->h,
+                         topic,
+                         dvd->treeobj,
+                         get_blobref_cb,
+                         err_blobref_cb,
+                         dvd) < 0)
+        return -1;
 
     if (dvd->errorcount) {
         errno = dvd->errnum;

src/cmd/builtin/fsck.c

@@ -28,6 +28,7 @@
 #include "ccan/str/str.h"
 
 #include "builtin.h"
+#include "src/cmd/util/util.h"
 
 #define BLOBREF_ASYNC_MAX 1000
 
@@ -50,10 +51,6 @@ struct fsck_ctx {
 struct fsck_valref_data
 {
     struct fsck_ctx *ctx;
-    json_t *treeobj;
-    int index;
-    int count;
-    int in_flight;
     const char *path;
     int errorcount;
     int errnum;
@@ -64,8 +61,6 @@ static void fsck_treeobj (struct fsck_ctx *ctx,
                           const char *path,
                           json_t *treeobj);
 
-static void valref_validate (struct fsck_valref_data *fvd);
-
 static void vmsg (struct fsck_ctx *ctx, const char *fmt, va_list ap)
 {
     char buf[128];
@@ -116,65 +111,28 @@ static void save_missing_ref_index (struct fsck_valref_data *fvd, int index)
     zlist_freefn (fvd->missing_indexes, cpy, (zlist_free_fn *) free, true);
 }
 
-static void valref_validate_continuation (flux_future_t *f, void *arg)
+static int err_blobref_cb (int errnum, int index, void *arg)
 {
     struct fsck_valref_data *fvd = arg;
 
-    if (flux_rpc_get (f, NULL) < 0) {
-        int *index = flux_future_aux_get (f, "index");
-        if (fvd->ctx->verbose) {
-            if (errno == ENOENT)
-                errmsg (fvd->ctx,
-                        "%s: missing blobref index=%d",
-                        fvd->path,
-                        (*index));
-            else
-                errmsg (fvd->ctx,
-                        "%s: error retrieving blobref index=%d: %s",
-                        fvd->path,
-                        (*index),
-                        future_strerror (f, errno));
-        }
-        fvd->errorcount++;
-        fvd->errnum = errno;     /* we'll report the last errno */
-        if (fvd->ctx->repair && errno == ENOENT)
-            save_missing_ref_index (fvd, *index);
-    }
-    fvd->in_flight--;
-
-    if (fvd->index < fvd->count) {
-        valref_validate (fvd);
-        fvd->in_flight++;
-        fvd->index++;
+    if (fvd->ctx->verbose) {
+        if (errnum == ENOENT)
+            errmsg (fvd->ctx,
+                    "%s: missing blobref index=%d",
+                    fvd->path,
+                    index);
+        else
+            errmsg (fvd->ctx,
+                    "%s: error retrieving blobref index=%d: %s",
+                    fvd->path,
+                    index,
+                    strerror (errnum));
     }
-
-    flux_future_destroy (f);
-}
-
-static void valref_validate (struct fsck_valref_data *fvd)
-{
-    const char *topic = fvd->ctx->validate_available ?
-        "content-backing.validate" : "content-backing.load";
-    uint32_t hash[BLOBREF_MAX_DIGEST_SIZE];
-    ssize_t hash_size;
-    const char *blobref;
-    flux_future_t *f;
-    int *indexp;
-
-    blobref = treeobj_get_blobref (fvd->treeobj, fvd->index);
-
-    if ((hash_size = blobref_strtohash (blobref, hash, sizeof (hash))) < 0)
-        log_err_exit ("cannot get hash from ref string");
-
-    if (!(f = flux_rpc_raw (fvd->ctx->h, topic, hash, hash_size, 0, 0)))
-        log_err_exit ("failed to validate valref blob");
-    if (flux_future_then (f, -1, valref_validate_continuation, fvd) < 0)
-        log_err_exit ("cannot validate valref blob");
-    if (!(indexp = (int *)malloc (sizeof (int))))
-        log_err_exit ("cannot allocate index memory");
-    (*indexp) = fvd->index;
-    if (flux_future_aux_set (f, "index", indexp, free) < 0)
-        log_err_exit ("could not save index value");
+    fvd->errorcount++;
+    fvd->errnum = errnum;     /* we'll report the last errno */
+    if (fvd->ctx->repair && errnum == ENOENT)
+        save_missing_ref_index (fvd, index);
+    return 0;                 /* 0, continue to scan remaining blobrefs */
 }
 
 static json_t *repair_valref (struct fsck_ctx *ctx,
@@ -336,22 +294,20 @@ static void fsck_valref (struct fsck_ctx *ctx,
                          const char *path,
                          json_t *treeobj)
 {
+    const char *topic = ctx->validate_available ?
+        "content-backing.validate" : "content-backing.load";
     struct fsck_valref_data fvd = {0};
 
     fvd.ctx = ctx;
-    fvd.treeobj = treeobj;
-    fvd.count = treeobj_get_count (treeobj);
     fvd.path = path;
 
-    while (fvd.in_flight < BLOBREF_ASYNC_MAX
-           && fvd.index < fvd.count) {
-        valref_validate (&fvd);
-        fvd.in_flight++;
-        fvd.index++;
-    }
-
-    if (flux_reactor_run (flux_get_reactor (ctx->h), 0) < 0)
-        log_err_exit ("flux_reactor_run");
+    if (valref_blobrefs (ctx->h,
+                         topic,
+                         treeobj,
+                         NULL,
+                         err_blobref_cb,
+                         &fvd) < 0)
+        log_err_exit ("unable to scan valref %s", path);
 
     if (fvd.errorcount) {
         /* each invalid blobref will be output in verbose mode */