static int find_all_sockets_cores (resrc_api_ctx_t *rsapi, resrc_t *node,
               int *nsocks, int *ncs)
{
    json_t *reqobj= NULL;
    resrc_reqst_t *req = NULL;
    resrc_tree_t *st = NULL;
    resrc_tree_t *ct = NULL;

    reqobj = Jnew ();
    create_req4allsocks (reqobj);
    req = resrc_reqst_from_json (rsapi, reqobj, NULL);
    *nsocks = resrc_tree_search (rsapi, node, req, &st, false);
    resrc_reqst_destroy (rsapi, req);
    resrc_tree_destroy (rsapi, st, false, false);
    Jput (reqobj);

    reqobj = Jnew ();
    create_req4allcores (reqobj);
    req = resrc_reqst_from_json (rsapi, reqobj, NULL);
    *ncs = resrc_tree_search (rsapi, node, req, &ct, false);
    resrc_reqst_destroy (rsapi, req);
    resrc_tree_destroy (rsapi, ct, false, false);
    Jput (reqobj);

    return (*nsocks > 0 && *ncs > 0) ? 0 : -1;
}

void test_nsrc (flux_t *h, uint32_t nodeid)
{
    flux_future_t *f;
    const int count = 10000;
    json_object *in = Jnew ();
    const char *json_str;
    json_object *out = NULL;
    int i, seq;

    Jadd_int (in, "count", count);
    if (!(f = flux_rpc (h, "req.nsrc", Jtostr (in), FLUX_NODEID_ANY,
                                                      FLUX_RPC_NORESPONSE)))
        log_err_exit ("%s", __FUNCTION__);
    flux_future_destroy (f);
    for (i = 0; i < count; i++) {
        flux_msg_t *msg = flux_recv (h, FLUX_MATCH_ANY, 0);
        if (!msg)
            log_err_exit ("%s", __FUNCTION__);
        if (flux_response_decode (msg, NULL, &json_str) < 0)
            log_msg_exit ("%s: decode %d", __FUNCTION__, i);
        if (!json_str
            || !(out = Jfromstr (json_str))
            || !Jget_int (out, "seq", &seq))
            log_msg_exit ("%s: decode %d payload", __FUNCTION__, i);
        if (seq != i)
            log_msg_exit ("%s: decode %d - seq mismatch %d", __FUNCTION__, i, seq);
        Jput (out);
        flux_msg_destroy (msg);
    }
    Jput (in);
}

// Recevied a reply to a trigger ("sim.reply")
static void reply_cb (flux_t *h,
                      flux_msg_handler_t *w,
                      const flux_msg_t *msg,
                      void *arg)
{
    const char *json_str = NULL;
    json_t *request = NULL;
    ctx_t *ctx = arg;
    sim_state_t *curr_sim_state = ctx->sim_state;
    sim_state_t *reply_sim_state;

    if (flux_msg_get_json (msg, &json_str) < 0 || json_str == NULL
        || !(request = Jfromstr (json_str))) {
        flux_log (h, LOG_ERR, "%s: bad reply message", __FUNCTION__);
        Jput (request);
        return;
    }

    // De-serialize and get new info
    reply_sim_state = json_to_sim_state (request);
    copy_new_state_data (ctx, curr_sim_state, reply_sim_state);

    if (handle_next_event (ctx) < 0) {
        flux_log (h, LOG_DEBUG, "No events remaining");
        if (ctx->exit_on_complete) {
            log_msg_exit ("exit_on_complete is set. Exiting now.");
        } else {
            send_complete_event (h);
        }
    }

    free_simstate (reply_sim_state);
    Jput (request);
}

static void rdl_update_cb (flux_t *h,
                           flux_msg_handler_t *w,
                           const flux_msg_t *msg,
                           void *arg)
{
    json_t *o = NULL;
    const char *json_str = NULL, *rdl_str = NULL;
    ctx_t *ctx = (ctx_t *)arg;

    if (flux_msg_get_json (msg, &json_str) < 0 || json_str == NULL
        || !(o = Jfromstr (json_str))) {
        flux_log (h, LOG_ERR, "%s: bad message", __FUNCTION__);
        Jput (o);
        return;
    }

    Jget_str (o, "rdl_string", &rdl_str);
    if (rdl_str) {
        free (ctx->rdl_string);
        ctx->rdl_string = strdup (rdl_str);
        ctx->rdl_changed = true;
    }

    Jput (o);
}

/* This test is to make sure that deferred responses are handled in order.
 * Arrange for module to source 10K sequenced responses.  Messages 5000-5499
 * are "put back" on the handle using flux_putmsg().  We ensure that
 * the 10K messages are nonetheless received in order.
 */
void test_putmsg (flux_t *h, uint32_t nodeid)
{
    flux_future_t *f;
    const char *json_str;
    const int count = 10000;
    const int defer_start = 5000;
    const int defer_count = 500;
    json_object *in = Jnew ();
    json_object *out = NULL;
    int seq, myseq = 0;
    zlist_t *defer = zlist_new ();
    bool popped = false;
    flux_msg_t *z;

    if (!defer)
        oom ();

    Jadd_int (in, "count", count);
    if (!(f = flux_rpc (h, "req.nsrc", Jtostr (in), FLUX_NODEID_ANY,
                                                      FLUX_RPC_NORESPONSE)))
        log_err_exit ("%s", __FUNCTION__);
    flux_future_destroy (f);
    do {
        flux_msg_t *msg = flux_recv (h, FLUX_MATCH_ANY, 0);
        if (!msg)
            log_err_exit ("%s", __FUNCTION__);
        if (flux_response_decode (msg, NULL, &json_str) < 0)
            log_msg_exit ("%s: decode", __FUNCTION__);
        if (!json_str
            || !(out = Jfromstr (json_str))
            || !Jget_int (out, "seq", &seq))
            log_msg_exit ("%s: decode - payload", __FUNCTION__);
        Jput (out);
        if (seq >= defer_start && seq < defer_start + defer_count && !popped) {
            if (zlist_append (defer, msg) < 0)
                oom ();
            if (seq == defer_start + defer_count - 1) {
                while ((z = zlist_pop (defer))) {
                    if (flux_requeue (h, z, FLUX_RQ_TAIL) < 0)
                        log_err_exit ("%s: flux_requeue", __FUNCTION__);
                    flux_msg_destroy (z);
                }
                popped = true;
            }
            continue;
        }
        if (seq != myseq)
            log_msg_exit ("%s: expected %d got %d", __FUNCTION__, myseq, seq);
        myseq++;
        flux_msg_destroy (msg);
    } while (myseq < count);
    zlist_destroy (&defer);
    Jput (in);
}

/* Proxy ping.
 */
void xping_request_cb (flux_t h, flux_msg_handler_t *w,
                       const flux_msg_t *msg, void *arg)
{
    ctx_t *ctx = arg;
    const char *json_str;
    int saved_errno;
    int rank, seq = ctx->ping_seq++;
    const char *service;
    char *hashkey = NULL;
    JSON in = Jnew ();
    JSON o = NULL;
    flux_msg_t *cpy;

    if (flux_request_decode (msg, NULL, &json_str) < 0) {
        saved_errno = errno;
        goto error;
    }
    if (!(o = Jfromstr (json_str)) || !Jget_int (o, "rank", &rank)
                                   || !Jget_str (o, "service", &service)) {
        saved_errno = errno = EPROTO;
        goto error;
    }
    flux_log (h, LOG_DEBUG, "Rxping rank=%d service=%s", rank, service);

    Jadd_int (in, "seq", seq);
    flux_log (h, LOG_DEBUG, "Tping seq=%d %d!%s", seq, rank, service);

    flux_rpc_t *rpc;
    if (!(rpc = flux_rpc (h, service, Jtostr (in), rank,
                                            FLUX_RPC_NORESPONSE))) {
        saved_errno = errno;
        goto error;
    }
    flux_rpc_destroy (rpc);
    if (!(cpy = flux_msg_copy (msg, true))) {
        saved_errno = errno;
        goto error;
    }
    hashkey = xasprintf ("%d", seq);
    zhash_update (ctx->ping_requests, hashkey, cpy);
    zhash_freefn (ctx->ping_requests, hashkey, (zhash_free_fn *)flux_msg_destroy);
    Jput (o);
    Jput (in);
    if (hashkey)
        free (hashkey);
    return;
error:
    if (flux_respond (h, msg, saved_errno, NULL) < 0)
        flux_log_error (h, "%s: flux_respond", __FUNCTION__);
    Jput (o);
    Jput (in);
}

void cache_entry_set_json (struct cache_entry *hp, json_object *o)
{
    if (hp) {
        if ((o && hp->o) || (!o && !hp->o)) {
            Jput (o); /* no-op, 'o' is assumed identical to hp->o */
        } else if (o && !hp->o) {
            hp->o = o;
            if (hp->waitlist)
                wait_runqueue (hp->waitlist);
        } else if (!o && hp->o) {
            Jput (hp->o);
            hp->o = NULL;
        }
    }
}

/* Route string will not include the endpoints.
 */
static void ping_cb (flux_t h, flux_msg_handler_t *w,
                     const flux_msg_t *msg, void *arg)
{
    module_t *p = arg;
    const char *json_str;
    JSON o = NULL;
    char *route = NULL;
    char *route_plus_uuid = NULL;
    int rc = -1;

    if (flux_request_decode (msg, NULL, &json_str) < 0)
        goto done;
    if (!(o = Jfromstr (json_str))) {
        errno = EPROTO;
        goto done;
    }
    if (!(route = flux_msg_get_route_string (msg)))
        goto done;
    route_plus_uuid = xasprintf ("%s!%.5s", route, module_get_uuid (p));
    Jadd_str (o, "route", route_plus_uuid);
    rc = 0;
done:
    if (flux_respond (h, msg, rc < 0 ? errno : 0,
                                rc < 0 ? NULL : Jtostr (o)) < 0)
        FLUX_LOG_ERROR (h);
    Jput (o);
    if (route_plus_uuid)
        free (route_plus_uuid);
    if (route)
        free (route);
}

json_object * kvspath_request_json (int64_t id)
{
    json_object *o = json_object_new_object ();
    json_object *ar = json_object_new_array ();
    json_object *v = json_object_new_int64 (id);

    if (!o || !ar || !v) {
        Jput (o);
        Jput (ar);
        Jput (v);
        return (NULL);
    }
    json_object_array_add (ar, v);
    json_object_object_add (o, "ids", ar);
    return (o);
}

static void fixup_newjob_event (flux_t *h, int64_t nj)
{
    json_object *ss = NULL;
    json_object *jcb = NULL;
    int64_t js = J_NULL;
    char *key = xasprintf ("%"PRId64, nj);
    jscctx_t *ctx = getctx (h);

    /* We fix up ordering problem only when new job
       event hasn't been reported through a kvs watch
     */
    jcb = Jnew ();
    ss = Jnew ();
    Jadd_int64 (jcb, JSC_JOBID, nj);
    Jadd_int64 (ss, JSC_STATE_PAIR_OSTATE , (int64_t) js);
    Jadd_int64 (ss, JSC_STATE_PAIR_NSTATE, (int64_t) js);
    json_object_object_add (jcb, JSC_STATE_PAIR, ss);
    if (zhash_insert (ctx->active_jobs, key, (void *)(intptr_t)js) < 0) {
        flux_log (h, LOG_ERR, "new_job_cb: inserting a job to hash failed");
        goto done;
    }
    if (invoke_cbs (h, nj, jcb, 0) < 0) {
        flux_log (h, LOG_ERR,
                     "makeup_newjob_event: failed to invoke callbacks");
        goto done;
    }
done:
    Jput (jcb);
    free (key);
    return;
}

int sequence_request_handler (seqhash_t *s, const flux_msg_t *msg, JSON *outp)
{
    const char *json_str, *topic;
    const char *method;
    JSON in = NULL;
    int rc = -1;

    *outp = NULL;
    if (flux_request_decode (msg, &topic, &json_str) < 0)
        goto done;
    if (!(in = Jfromstr (json_str))) {
        errno = EPROTO;
        goto done;
    }

    method = topic + 8;
    if (strcmp (method, "fetch") == 0)
        rc = handle_seq_fetch (s, in, outp);
    else if (strcmp (method, "set") == 0)
        rc = handle_seq_set (s, in, outp);
    else if (strcmp (method, "destroy") == 0)
        rc = handle_seq_destroy (s, in, outp);
    else
        errno = ENOSYS;
done:
    Jput (in);
    return (rc);
}

static void exit_event_cb (flux_t *h, flux_msg_handler_t *w,
                           const flux_msg_t *msg, void *arg)
{
    ctx_t *ctx = arg;
    barrier_t *b;
    const char *json_str;
    json_object *o = NULL;
    const char *name;
    int errnum;

    if (flux_event_decode (msg, NULL, &json_str) < 0) {
        flux_log_error (h, "%s: decoding event", __FUNCTION__);
        goto done;
    }
    if (!(o = Jfromstr (json_str))
                || !Jget_str (o, "name", &name)
                || !Jget_int (o, "errnum", &errnum)) {
        errno = EPROTO;
        flux_log_error (h, "%s: decoding event", __FUNCTION__);
        goto done;
    }
    if ((b = zhash_lookup (ctx->barriers, name))) {
        b->errnum = errnum;
        zhash_foreach (b->clients, send_enter_response, b);
        zhash_delete (ctx->barriers, name);
    }
done:
    Jput (o);
}

/* Note: return value of this function is ignored by libsubprocess.
 * Also: zio_json_decode() returns -1 on error, 0 on eof, strlen(s) on
 * success; caller must free 's'.
 */
static int subprocess_io_cb (struct subprocess *p, const char *json_str)
{
    runlevel_t *r;
    json_object *o = NULL;
    const char *name;
    int len;
    bool eof;
    char *s = NULL, *argz = NULL, *line = NULL;
    size_t argz_len;

    r = subprocess_get_context (p, "runlevel_t");
    assert (r != NULL);

    if (!r->io_cb)
        goto done;
    if (!(o = Jfromstr (json_str)) || !Jget_str (o, "name", &name))
        goto done;
    len = zio_json_decode (json_str, (void **)&s, &eof);
    if (len <= 0 || !s || !*s || s[len] != '\0')
        goto done;
    if (argz_create_sep (s, '\n', &argz, &argz_len) != 0)
        goto done;
    while ((line = argz_next (argz, argz_len, line)) && *line)
        r->io_cb (r, name, line, r->io_cb_arg);
done:
    if (s)
        free (s);
    if (argz)
        free (argz);
    Jput (o);
    return 0;
}

// builds the trigger request and sends it to "mod_name"
// converts sim_state to JSON, formats request tag based on "mod_name"
static int send_trigger (flux_t *h, const char *mod_name, sim_state_t *sim_state)
{
    int rc = 0;
    flux_future_t *future = NULL;
    json_t *o = NULL;
    char *topic = NULL;

    // Reset the next timer for "mod_name" to -1 before we trigger
    double *next_time = zhash_lookup (sim_state->timers, mod_name);
    *next_time = -1;

    o = sim_state_to_json (sim_state);

    topic = xasprintf ("%s.trigger", mod_name);
    future = flux_rpc (h, topic, Jtostr (o), FLUX_NODEID_ANY, FLUX_RPC_NORESPONSE);
    if (!future) {
        flux_log (h, LOG_ERR, "failed to send trigger to %s", mod_name);
        rc = -1;
    }

    Jput (o);
    free (topic);
    flux_future_destroy (future);
    return rc;
}

/* Return 'n' sequenced responses.
 */
void nsrc_request_cb (flux_t h, flux_msg_handler_t *w,
                      const flux_msg_t *msg, void *arg)
{
    const char *json_str;
    int saved_errno;
    JSON o = Jnew ();
    int i, count;
    int rc = -1;

    if (flux_request_decode (msg, NULL, &json_str) < 0) {
        saved_errno = errno;
        goto done;
    }
    if (!(o = Jfromstr (json_str)) || !Jget_int (o, "count", &count)) {
        saved_errno = errno = EPROTO;
        goto done;
    }
    for (i = 0; i < count; i++) {
        Jadd_int (o, "seq", i);
        if (flux_respond (h, msg, 0, Jtostr (o)) < 0) {
            saved_errno = errno;
            flux_log_error (h, "%s: flux_respond", __FUNCTION__);
            goto done;
        }
    }
    rc = 0;
done:
    if (rc < 0) {
        if (flux_respond (h, msg, saved_errno, NULL) < 0)
            flux_log_error (h, "%s: flux_respond", __FUNCTION__);
    }
    Jput (o);
}

static int job_status_cb (const char *jcbstr, void *arg, int errnum)
{
    int64_t os = 0;
    int64_t ns = 0;
    int64_t j = 0;
    jstatctx_t *ctx = NULL;
    flux_t h = (flux_t)arg;
    JSON jcb = NULL;

    ctx = getctx (h);
    if (errnum > 0) {
        flux_log (ctx->h, LOG_ERR, "job_status_cb: errnum passed in");
        return -1;
    }

    if (!(jcb = Jfromstr (jcbstr))) {
        flux_log (ctx->h, LOG_ERR, "job_status_cb: error parsing JSON string");
        return -1;
    }
    get_jobid (jcb, &j);
    get_states (jcb, &os, &ns);
    Jput (jcb);

    fprintf (ctx->op, "%s->%s\n", 
    jsc_job_num2state ((job_state_t)os), 
    jsc_job_num2state ((job_state_t)ns));
    fflush (ctx->op);

    return 0;
}

int rpctest_nodeid_cb (flux_t h, int type, zmsg_t **zmsg, void *arg)
{
    int errnum = 0;
    uint32_t nodeid;
    JSON o = NULL;
    int flags;

    if (flux_request_decode (*zmsg, NULL, NULL) < 0
            || flux_msg_get_nodeid (*zmsg, &nodeid, &flags) < 0) {
        errnum = errno;
        goto done;
    }
    if (nodeid == nodeid_fake_error) {
        nodeid_fake_error = -1;
        errnum = EPERM; /* an error not likely to be seen */
        goto done;
    }
    o = Jnew ();
    Jadd_int (o, "nodeid", nodeid);
    Jadd_int (o, "flags", flags);
done:
    (void)flux_respond (h, *zmsg, errnum, Jtostr (o));
    Jput (o);
    zmsg_destroy (zmsg);
    return 0;
}

static void request_hwloc_reload (flux_t h, const char *nodeset,
                                  const char *walk_topology)
{
    flux_rpc_t *rpc;
    JSON o = NULL;
    const char *json_str = NULL;

    if ((o = hwloc_reload_json_create (walk_topology)))
        json_str = Jtostr (o);
    if (!(rpc = flux_rpc_multi (h, "resource-hwloc.reload", json_str,
                                                        nodeset, 0)))
        log_err_exit ("flux_rpc_multi");
    while (!flux_rpc_completed (rpc)) {
        const char *json_str;
        uint32_t nodeid = FLUX_NODEID_ANY;
        if (flux_rpc_get (rpc, &nodeid, &json_str) < 0) {
            if (nodeid == FLUX_NODEID_ANY)
                log_err ("flux_rpc_get");
            else
                log_err ("rpc(%"PRIu32")", nodeid);
        }
    }
    flux_rpc_destroy (rpc);
    Jput (o);
}

/* Return 'n' sequenced responses.
 */
static int nsrc_request_cb (flux_t h, int typemask, zmsg_t **zmsg, void *arg)
{
    JSON o = Jnew ();
    int i, count;

    if (flux_json_request_decode (*zmsg, &o) < 0) {
        if (flux_err_respond (h, errno, zmsg) < 0)
            flux_log (h, LOG_ERR, "%s: flux_err_respond: %s", __FUNCTION__,
                      strerror (errno));
        goto done;
    }
    if (!Jget_int (o, "count", &count)) {
        if (flux_err_respond (h, EPROTO, zmsg) < 0)
            flux_log (h, LOG_ERR, "%s: flux_err_respond: %s", __FUNCTION__,
                      strerror (errno));
        goto done;
    }
    for (i = 0; i < count; i++) {
        zmsg_t *cpy = zmsg_dup (*zmsg);
        if (!cpy)
            oom ();
        Jadd_int (o, "seq", i);
        if (flux_json_respond (h, o, &cpy) < 0)
            flux_log (h, LOG_ERR, "%s: flux_json_respond: %s", __FUNCTION__,
                      strerror (errno));
        zmsg_destroy (&cpy);
    }
    zmsg_destroy (zmsg);
done:
    Jput (o);
    return 0;
}

static int get_rlist_result (zhash_t *mods, flux_mrpc_t *mrpc,
                             uint32_t nodeid, int *ep)
{
    JSON o = NULL;
    int rc = -1;
    const char *name, *digest;
    int i, len, errnum, size, idle;
    module_t *m;

    if (flux_mrpc_get_outarg_obj (mrpc, nodeid, &o) < 0)
        goto done;
    if (modctl_rlist_dec (o, &errnum, &len) < 0)
        goto done;
    for (i = 0; i < len; i++) {
        if (modctl_rlist_dec_nth (o, i, &name, &size, &digest, &idle) < 0)
            goto done;
        if (!(m = zhash_lookup (mods, digest))) {
            if (!(m = module_create (name, size, digest, idle, nodeid)))
                goto done;
            zhash_update (mods, digest, m);
            zhash_freefn (mods, digest, (zhash_free_fn *)module_destroy);
        } else if (module_update (m, idle, nodeid) < 0)
            goto done;
    }
    *ep = errnum;
    rc = 0;
done:
    Jput (o);
    return rc;
}