static int extract_raw_rdl_alloc (flux_t h, int64_t j, JSON jcb)
{
    int i = 0;
    char k[20];
    int64_t cores = 0;
    JSON ra = Jnew_ar ();
    bool processing = true;

    for (i=0; processing; ++i) {
        snprintf (k, 20, "lwj.%ld.rank.%d.cores", j, i);
        if (kvs_get_int64 (h, k, &cores) < 0) {
            if (errno != EINVAL) 
                flux_log (h, LOG_ERR, "extract %s: %s", k, strerror (errno));
            processing = false; 
        } else {
            JSON elem = Jnew ();       
            JSON o = Jnew ();
            Jadd_int64 (o, JSC_RDL_ALLOC_CONTAINED_NCORES, cores);
            json_object_object_add (elem, JSC_RDL_ALLOC_CONTAINED, o);
            json_object_array_add (ra, elem);
        }
    } 
    json_object_object_add (jcb, JSC_RDL_ALLOC, ra);
    return 0;
}

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;
}

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;
}

static int extract_raw_rdl_alloc (flux_t h, int64_t j, JSON jcb)
{
    int i = 0;
    char *key;
    int64_t cores = 0;
    JSON ra = Jnew_ar ();
    bool processing = true;

    for (i=0; processing; ++i) {
        key = xasprintf ("lwj.%"PRId64".rank.%"PRId32".cores", j, i);
        if (kvs_get_int64 (h, key, &cores) < 0) {
            if (errno != EINVAL)
                flux_log_error (h, "extract %s", key);
            processing = false;
        } else {
            JSON elem = Jnew ();
            JSON o = Jnew ();
            Jadd_int64 (o, JSC_RDL_ALLOC_CONTAINED_NCORES, cores);
            json_object_object_add (elem, JSC_RDL_ALLOC_CONTAINED, o);
            json_object_array_add (ra, elem);
        }
        free (key);
    }
    json_object_object_add (jcb, JSC_RDL_ALLOC, ra);
    return 0;
}

static int extract_raw_rdl_alloc (flux_t *h, int64_t j, json_object *jcb)
{
    int i;
    json_object *ra = Jnew_ar ();
    bool processing = true;

    for (i=0; processing; ++i) {
        char *key = lwj_key (h, j, ".rank.%d.cores", i);
        flux_future_t *f = NULL;
        int64_t cores = 0;
        if (!key || !(f = flux_kvs_lookup (h, 0, key))
                 || flux_kvs_lookup_get_unpack (f, "I", &cores) < 0) {
            if (errno != EINVAL)
                flux_log_error (h, "extract %s", key);
            processing = false;
        } else {
            json_object *elem = Jnew ();
            json_object *o = Jnew ();
            Jadd_int64 (o, JSC_RDL_ALLOC_CONTAINED_NCORES, cores);
            json_object_object_add (elem, JSC_RDL_ALLOC_CONTAINED, o);
            json_object_array_add (ra, elem);
        }
        free (key);
        flux_future_destroy (f);
    }
    json_object_object_add (jcb, JSC_RDL_ALLOC, ra);
    return 0;
}

static int new_job_cb (const char *key, int64_t val, void *arg, int errnum)
{
    int64_t nj = 0;
    int64_t js = 0;
    JSON ss = NULL;
    JSON jcb = NULL;
    char k[20] = {'\0'};
    char path[20] = {'\0'};
    flux_t h = (flux_t) arg;
    jscctx_t *ctx = getctx (h);

    if (ctx->first_time == 1) {
        /* watch is invoked immediately and we shouldn't
         * rely on that event at all.
         */
        ctx->first_time = 0;
        return 0;
    }

    if (chk_errnum (h, errnum) < 0) return 0;

    flux_log (h, LOG_DEBUG, "new_job_cb invoked: key(%s), val(%ld)", key, val);

    js = J_NULL;
    nj = val-1;
    snprintf (k, 20, "%ld", nj);
    snprintf (path, 20, "lwj.%ld", nj);
    if (zhash_insert (ctx->active_jobs, k, (void *)(intptr_t)js) < 0) {
        flux_log (h, LOG_ERR, "new_job_cb: inserting a job to hash failed");
        goto done;
    }

    flux_log (h, LOG_DEBUG, "jobstate_hdlr registered");
    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 (invoke_cbs (h, nj, jcb, errnum) < 0) {
        flux_log (h, LOG_ERR, "new_job_cb: failed to invoke callbacks");
    }
    if (reg_jobstate_hdlr (h, path, job_state_cb) == -1) {
        flux_log (h, LOG_ERR, "new_job_cb: reg_jobstate_hdlr: %s", 
            strerror (errno));
    }

done:
    /* always return 0 so that reactor won't return */
    return 0;
}

static inline void create_req4allnodes (json_t *reqobj)
{
    json_t *req1;
    json_t *req2;
    Jadd_str (reqobj, "type", "node");
    Jadd_int (reqobj, "req_qty", 1);
    req1 = Jnew ();
    Jadd_str (req1, "type", "socket");
    Jadd_int (req1, "req_qty", 1);
    req2 = Jnew ();
    Jadd_str (req2, "type", "core");
    Jadd_int (req2, "req_qty", 1);
    json_object_set_new (req1, "req_child", req2);
    json_object_set_new (reqobj, "req_child", req1);
}

static int handle_seq_fetch (seqhash_t *s, JSON in, JSON *outp)
{
    const char *name;
    bool create = false;
    bool created = false;
    int64_t v, pre, post, *valp;

    if (!Jget_str (in, "name", &name)
        || !Jget_bool (in, "create", &create)
        || !Jget_int64 (in, "preincrement", &pre)
        || !Jget_int64 (in, "postincrement", &post)) {
        errno = EPROTO;
        return (-1);
    }
    if (seq_fetch_and_add (s, name, pre, post, &v) < 0) {
        if (!create || (errno != ENOENT))
            return (-1);
        /*  Create and initialize
         */
        valp = seq_create (s, name);
        *valp += pre;
        v = *valp;
        *valp += post;
        created = true;
    }

    *outp = Jnew ();
    Jadd_str (*outp, "name", name);
    Jadd_int64 (*outp, "value", v);
    if (create && created)
        Jadd_bool (*outp, "created", true);
    return (0);
}

static int64_t next_jobid (flux_t h)
{
    int64_t ret = (int64_t) -1;
    const char *json_str;
    json_object *req, *resp;
    flux_rpc_t *rpc;

    req = Jnew ();
    Jadd_str (req, "name", "lwj");
    Jadd_int64 (req, "preincrement", 1);
    Jadd_int64 (req, "postincrement", 0);
    Jadd_bool (req, "create", true);
    rpc = flux_rpc (h, "cmb.seq.fetch",
                    json_object_to_json_string (req), 0, 0);
    json_object_put (req);

    if ((flux_rpc_get (rpc, NULL, &json_str) < 0)
        || !(resp = json_tokener_parse (json_str))) {
        flux_log_error (h, "rpc_get");
        goto out;
    }

    Jget_int64 (resp, "value", &ret);
    json_object_put (resp);
out:
    flux_rpc_destroy (rpc);
    return ret;
}

static int query_rdesc (flux_t h, int64_t j, JSON *jcb)
{
    JSON o = NULL;
    int64_t nnodes = -1;
    int64_t ntasks = -1;
    
    if (extract_raw_nnodes (h, j, &nnodes) < 0) return -1;
    if (extract_raw_ntasks (h, j, &ntasks) < 0) return -1;

    *jcb = Jnew ();
    o = Jnew ();
    Jadd_int64 (o, JSC_RDESC_NNODES, nnodes);
    Jadd_int64 (o, JSC_RDESC_NTASKS, ntasks);
    json_object_object_add (*jcb, JSC_RDESC, o);
    return 0;
}

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);
}

void send_ping (struct ping_ctx *ctx)
{
    struct timespec t0;
    json_object *in = Jnew ();
    flux_rpc_t *rpc;
    struct ping_data *pdata = xzmalloc (sizeof (*pdata));

    pdata->tstat = xzmalloc (sizeof (*(pdata->tstat)));
    pdata->seq = 0;
    pdata->route = NULL;
    pdata->rpc_count = 0;

    Jadd_int (in, "seq", ctx->send_count);
    monotime (&t0);
    Jadd_int64 (in, "time.tv_sec", t0.tv_sec);
    Jadd_int64 (in, "time.tv_nsec", t0.tv_nsec);
    Jadd_str (in, "pad", ctx->pad);

    if (ctx->rank)
        rpc = flux_rpc_multi (ctx->h, ctx->topic, Jtostr (in), ctx->rank, 0);
    else
        rpc = flux_rpc (ctx->h, ctx->topic, Jtostr (in), ctx->nodeid, 0);
    if (!rpc)
        log_err_exit ("flux_rpc");
    flux_rpc_aux_set (rpc, pdata, ping_data_free);
    if (flux_rpc_then (rpc, ping_continuation, ctx) < 0)
        log_err_exit ("flux_rpc_then");

    Jput (in);

    ctx->send_count++;
}

/* kvs.put
 */
JSON kp_tput_enc (const char *key, const char *json_str, bool link, bool dir)
{
    JSON o = NULL;
    JSON val = NULL;

    if (!key) {
        errno = EINVAL;
        goto error;
    }
    o = Jnew ();
    if (json_str) {
        if (!(val = Jfromstr (json_str))) {
            errno = EINVAL;
            goto error;
        }
    }
    json_object_object_add (o, key, val);
    if (dir)
        Jadd_bool (o, ".flag_mkdir", true);
    if (link)
        Jadd_bool (o, ".flag_symlink", true);
    return o;
error:
    Jput (o);
    return NULL;
}

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;
}

void cmd_copy_tokvs (flux_t h, int argc, char **argv)
{
    char *file, *key;
    int fd, len;
    uint8_t *buf;
    JSON o;

    if (argc != 2)
        msg_exit ("copy-tokvs: specify key and filename");
    key = argv[0];
    file = argv[1];
    if (!strcmp (file, "-")) {
        if ((len = read_all (STDIN_FILENO, &buf)) < 0)
            err_exit ("stdin");
    } else {
        if ((fd = open (file, O_RDONLY)) < 0)
            err_exit ("%s", file);
        if ((len = read_all (fd, &buf)) < 0)
            err_exit ("%s", file);
        (void)close (fd);
    }
    o = Jnew ();
    util_json_object_add_data (o, "data", buf, len);
    if (kvs_put (h, key, Jtostr (o)) < 0)
        err_exit ("%s", key);
    if (kvs_commit (h) < 0)
        err_exit ("kvs_commit");
    Jput (o);
    free (buf);
}

/* 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);
}

/* 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;
}

int getrusage_json (int who, json_object **op)
{
    struct rusage ru;
    json_object *o;

    if (getrusage (who, &ru) < 0)
        return -1;
    o = Jnew ();
    Jadd_double (o, "utime",
            (double)ru.ru_utime.tv_sec + 1E-6 * ru.ru_utime.tv_usec);
    Jadd_double (o, "stime",
            (double)ru.ru_stime.tv_sec + 1E-6 * ru.ru_stime.tv_usec);
    Jadd_int64 (o, "maxrss",        ru.ru_maxrss);
    Jadd_int64 (o, "ixrss",         ru.ru_ixrss);
    Jadd_int64 (o, "idrss",         ru.ru_idrss);
    Jadd_int64 (o, "isrss",         ru.ru_isrss);
    Jadd_int64 (o, "minflt",        ru.ru_minflt);
    Jadd_int64 (o, "majflt",        ru.ru_majflt);
    Jadd_int64 (o, "nswap",         ru.ru_nswap);
    Jadd_int64 (o, "inblock",       ru.ru_inblock);
    Jadd_int64 (o, "oublock",       ru.ru_oublock);
    Jadd_int64 (o, "msgsnd",        ru.ru_msgsnd);
    Jadd_int64 (o, "msgrcv",        ru.ru_msgrcv);
    Jadd_int64 (o, "nsignals",      ru.ru_nsignals);
    Jadd_int64 (o, "nvcsw",         ru.ru_nvcsw);
    Jadd_int64 (o, "nivcsw",        ru.ru_nivcsw);
    *op = o;
    return 0;
}

static json_object *build_parray_elem (int64_t pid, int64_t eix, int64_t hix)
{
    json_object *po = Jnew ();
    Jadd_int64 (po, JSC_PDESC_RANK_PDARRAY_PID, pid);
    Jadd_int64 (po, JSC_PDESC_RANK_PDARRAY_EINDX, eix);
    Jadd_int64 (po, JSC_PDESC_RANK_PDARRAY_HINDX, hix);
    return po;
}

static int query_pdesc (flux_t *h, int64_t j, json_object **jcb)
{
    int64_t ntasks = 0;
    if (extract_raw_ntasks (h, j, &ntasks) < 0) return -1;
    *jcb = Jnew ();
    Jadd_int64 (*jcb, JSC_PDESC_SIZE, ntasks);
    return extract_raw_pdescs (h, j, ntasks, *jcb);
}