自学教程：C++ zhash_new函数代码示例

static agent_t *s_agent_new (zctx_t *ctx, void *pipe){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    if (!self)        return NULL;    self->ctx = ctx;    self->pipe = pipe;    self->whitelist = zhash_new ();    if (self->whitelist)        self->blacklist = zhash_new ();    //  Create ZAP handler and get ready for requests    if (self->blacklist)        self->handler = zsocket_new (self->ctx, ZMQ_REP);    if (self->handler) {        if (zsocket_bind (self->handler, ZAP_ENDPOINT) == 0)            zstr_send (self->pipe, "OK");        else            zstr_send (self->pipe, "ERROR");    }    else        s_agent_destroy (&self);    return self;}

static zyre_node_t *zyre_node_new (zsock_t *pipe, void *args){    zyre_node_t *self = (zyre_node_t *) zmalloc (sizeof (zyre_node_t));    self->inbox = zsock_new (ZMQ_ROUTER);    if (self->inbox == NULL) {        free (self);        return NULL;            //  Could not create new socket    }    //  Use ZMQ_ROUTER_HANDOVER so that when a peer disconnects and    //  then reconnects, the new client connection is treated as the    //  canonical one, and any old trailing commands are discarded.    zsock_set_router_handover (self->inbox, 1);        self->pipe = pipe;    self->outbox = (zsock_t *) args;    self->poller = zpoller_new (self->pipe, NULL);    self->beacon_port = ZRE_DISCOVERY_PORT;    self->interval = 0;         //  Use default    self->uuid = zuuid_new ();    self->peers = zhash_new ();    self->peer_groups = zhash_new ();    self->own_groups = zhash_new ();    self->headers = zhash_new ();    zhash_autofree (self->headers);    //  Default name for node is first 6 characters of UUID:    //  the shorter string is more readable in logs    self->name = (char *) zmalloc (7);    memcpy (self->name, zuuid_str (self->uuid), 6);    return self;}

static zyre_node_t *zyre_node_new (zctx_t *ctx, void *pipe){    zyre_node_t *self = (zyre_node_t *) zmalloc (sizeof (zyre_node_t));    self->ctx = ctx;    self->pipe = pipe;    self->inbox = zsocket_new (ctx, ZMQ_ROUTER);    if (self->inbox == NULL) {        free (self);        return NULL;            //  Interrupted 0MQ call    }    self->port = zsocket_bind (self->inbox, "tcp://*:*");    if (self->port < 0) {        free (self);        return NULL;            //  Interrupted 0MQ call    }    self->beacon = zbeacon_new (self->ctx, ZRE_DISCOVERY_PORT);    if (!self->beacon) {        free (self);        return NULL;            //  Exhausted process sockets    }    self->uuid = zuuid_new ();    self->peers = zhash_new ();    self->peer_groups = zhash_new ();    self->own_groups = zhash_new ();    self->headers = zhash_new ();    zhash_autofree (self->headers);    return self;}

static agent_t *s_agent_new (zctx_t *ctx, void *control){    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    self->ctx = ctx;    self->control = control;    self->router = zsocket_new (ctx, ZMQ_ROUTER);    //  Connect our data socket to caller's endpoint    self->data = zsocket_new (ctx, ZMQ_PAIR);    char *endpoint = zstr_recv (self->control);    int rc = zsocket_connect (self->data, "%s", endpoint);    assert (rc != -1);    free (endpoint);    //  Create new client codec using cert from API    byte public_key [32];    byte secret_key [32];    rc = zmq_recv (self->control, public_key, 32, 0);    assert (rc == 32);    rc = zmq_recv (self->control, secret_key, 32, 0);    assert (rc == 32);    self->cert = zcert_new_from (public_key, secret_key);            self->metadata = zhash_new ();    zhash_autofree (self->metadata);    self->clients = zhash_new ();    self->max_clients = 100;    self->max_pending = 10;    self->client_ttl = 3600;    //  60 minutes    self->pending_ttl = 60;     //  60 seconds    return self;}

static zsync_node_t *zsync_node_new (){    int rc;    zsync_node_t *self = (zsync_node_t *) zmalloc (sizeof (zsync_node_t));       self->ctx = zctx_new ();    assert (self->ctx);    self->zyre = zyre_new (self->ctx);      assert (self->zyre);        // Obtain permanent UUID    self->own_uuid = zuuid_new ();    if (zsys_file_exists (UUID_FILE)) {        // Read uuid from file        zfile_t *uuid_file = zfile_new (".", UUID_FILE);        int rc = zfile_input (uuid_file);    // open file for reading                assert (rc == 0);         zchunk_t *uuid_chunk = zfile_read (uuid_file, 16, 0);        assert (zchunk_size (uuid_chunk) == 16);    // make sure read succeeded        zuuid_set (self->own_uuid, zchunk_data (uuid_chunk));        zfile_destroy (&uuid_file);    } else {        // Write uuid to file        zfile_t *uuid_file = zfile_new (".", UUID_FILE);        rc = zfile_output (uuid_file); // open file for writing        assert (rc == 0);        zchunk_t *uuid_bin = zchunk_new ( zuuid_data (self->own_uuid), 16);        rc = zfile_write (uuid_file, uuid_bin, 0);        assert (rc == 0);        zfile_destroy (&uuid_file);    }        // Obtain peers and states    self->peers = zlist_new ();    if (zsys_file_exists (PEER_STATES_FILE)) {        zhash_t *peer_states = zhash_new ();        int rc = zhash_load (peer_states, PEER_STATES_FILE);        assert (rc == 0);        zlist_t *uuids = zhash_keys (peer_states);        char *uuid = zlist_first (uuids);        while (uuid) {            char * state_str = zhash_lookup (peer_states, uuid);            uint64_t state;            sscanf (state_str, "%"SCNd64, &state);            zlist_append (self->peers, zsync_peer_new (uuid, state));             uuid = zlist_next (uuids);        }    }        self->zyre_peers = zhash_new ();    self->terminated = false;    return self;}

static intserver_initialize (server_t *self){    //  Construct properties here    self->services = zhash_new();    self->workers = zhash_new();    self->waiting = zlist_new();    s_server_t *server = (s_server_t *) self;    self->router = server->router;    return 0;}

static agent_t *agent_new (zctx_t *ctx, void *pipe){    void *inbox = zsocket_new (ctx, ZMQ_ROUTER);    if (!inbox)                 //  Interrupted        return NULL;    agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));    self->ctx = ctx;    self->pipe = pipe;    self->udp = zre_udp_new (ZRE_DISCOVERY_PORT);    self->inbox = inbox;    self->host = zre_udp_host (self->udp);    self->port = zsocket_bind (self->inbox, "tcp://*:*");    sprintf (self->endpoint, "%s:%d", self->host, self->port);    if (self->port < 0) {       //  Interrupted        zre_udp_destroy (&self->udp);        free (self);        return NULL;    }    self->uuid = zre_uuid_new ();    self->identity = strdup (zre_uuid_str (self->uuid));    self->peers = zhash_new ();    self->peer_groups = zhash_new ();    self->own_groups = zhash_new ();    self->headers = zhash_new ();    zhash_autofree (self->headers);    self->log = zre_log_new (self->endpoint);    //  Set up content distribution network: Each server binds to an    //  ephemeral port and publishes a temporary directory that acts    //  as the outbox for this node.    //    sprintf (self->fmq_outbox, "%s/send/%s", s_tmpdir (), self->identity);    zfile_mkdir (self->fmq_outbox);    sprintf (self->fmq_inbox, "%s/recv/%s", s_tmpdir (), self->identity);    zfile_mkdir (self->fmq_inbox);    self->fmq_server = fmq_server_new ();    self->fmq_service = fmq_server_bind (self->fmq_server, "tcp://*:*");    fmq_server_publish (self->fmq_server, self->fmq_outbox, "/");    fmq_server_set_anonymous (self->fmq_server, true);    char publisher [32];    sprintf (publisher, "tcp://%s:%d", self->host, self->fmq_service);    zhash_update (self->headers, "X-FILEMQ", publisher);    //  Client will connect as it discovers new nodes    self->fmq_client = fmq_client_new ();    fmq_client_set_inbox (self->fmq_client, self->fmq_inbox);    fmq_client_set_resync (self->fmq_client, true);    fmq_client_subscribe (self->fmq_client, "/");        return self;}

static broker_t *s_broker_new (int verbose){    broker_t *self = (broker_t *) zmalloc (sizeof (broker_t));    self->ctx = zctx_new ();    self->socket = zsocket_new (self->ctx, ZMQ_ROUTER);    self->verbose = verbose;    self->services = zhash_new ();    self->workers = zhash_new ();    self->waiting = zlist_new ();    self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;    return self;}

static vocket_t *vocket_new (driver_t *driver, int socktype, char *vtxname){    assert (driver);    vocket_t *self = (vocket_t *) zmalloc (sizeof (vocket_t));    self->driver = driver;    self->vtxname = strdup (vtxname);    self->binding_hash = zhash_new ();    self->peering_hash = zhash_new ();    self->peering_list = zlist_new ();    self->live_peerings = zlist_new ();    self->socktype = socktype;    uint index;    for (index = 0; index < tblsize (s_vocket_config); index++)        if (socktype == s_vocket_config [index].socktype)            break;    if (index < tblsize (s_vocket_config)) {        self->routing = s_vocket_config [index].routing;        self->nomnom = s_vocket_config [index].nomnom;        self->min_peerings = s_vocket_config [index].min_peerings;        self->max_peerings = s_vocket_config [index].max_peerings;    }    else {        zclock_log ("E: invalid vocket type %d", socktype);        exit (1);    }    //  Create msgpipe vocket and connect over inproc to vtxname    self->msgpipe = zsocket_new (driver->ctx, ZMQ_PAIR);    assert (self->msgpipe);    zsocket_connect (self->msgpipe, "inproc://%s", vtxname);    //  If we drop on no peerings, start routing input now    if (self->min_peerings == 0) {        //  Ask reactor to start monitoring vocket's msgpipe pipe        zmq_pollitem_t item = { self->msgpipe, 0, ZMQ_POLLIN, 0 };        zloop_poller (driver->loop, &item, s_vocket_input, self);    }    //  Store this vocket per driver so that driver can cleanly destroy    //  all its vockets when it is destroyed.    zlist_push (driver->vockets, self);    //* Start transport-specific work    self->inbuf_max = VTX_TCP_INBUF_MAX;    self->outbuf_max = VTX_TCP_OUTBUF_MAX;    //* End transport-specific work    return self;}

static int extract_raw_pdescs (flux_t *h, int64_t j, int64_t n, json_object *jcb){    int rc = -1;    int64_t i = 0;    char *hnm;    const char *cmd = NULL;    zhash_t *eh = NULL; /* hash holding a set of unique exec_names */    zhash_t *hh = NULL; /* hash holding a set of unique host_names */    json_object *o = NULL;    json_object *po = NULL;    json_object *pa = Jnew_ar ();    json_object *hns = Jnew_ar ();    json_object *ens = Jnew_ar ();    if (!(eh = zhash_new ()) || !(hh = zhash_new ()))        oom ();    for (i=0; i < (int) n; i++) {        int64_t eix = 0, hix = 0;        int64_t pid = 0, nid = 0;        if (extract_raw_pdesc (h, j, i, &o) != 0)            goto done;        if (!fetch_rank_pdesc (o, &pid, &nid, &cmd))            goto done;        eix = build_name_array (eh, cmd, ens);        /* FIXME: we need a hostname service */        hnm = xasprintf ("%"PRId64, nid);        hix = build_name_array (hh, hnm, hns);        po = build_parray_elem (pid, eix, hix);        json_object_array_add (pa, po);        po = NULL;        Jput (o);        o = NULL;        free (hnm);    }    add_pdescs_to_jcb (&hns, &ens, &pa, jcb);    rc = 0;done:    if (o) Jput (o);    if (po) Jput (po);    if (pa) Jput (pa);    if (hns) Jput (hns);    if (ens) Jput (ens);    zhash_destroy (&eh);    zhash_destroy (&hh);    return rc;}

static broker_t *s_broker_new (int verbose){    broker_t *self = (broker_t *) calloc (1, sizeof (broker_t));    //  Initialize broker state    self->context = zmq_init (1);    self->socket = zmq_socket (self->context, ZMQ_XREP);    self->verbose = verbose;    self->services = zhash_new ();    self->workers = zhash_new ();    self->waiting = zlist_new ();    self->heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;    return self;}

int main (void){    //  Prepare our context and publisher socket    zctx_t *ctx = zctx_new ();    void *publisher = zsocket_new (ctx, ZMQ_PUB);    zsocket_set_hwm (publisher, 0);    zsocket_bind (publisher, "tcp://*:5556");    zclock_sleep (200);    zhash_t *kvmap = zhash_new ();    int64_t sequence = 0;    srandom ((unsigned) time (NULL));    while (!zctx_interrupted) {        //  Distribute as key-value message        kvmsg_t *kvmsg = kvmsg_new (++sequence);        kvmsg_fmt_key  (kvmsg, "%d", randof (10000));        kvmsg_fmt_body (kvmsg, "%d", randof (1000000));        kvmsg_send     (kvmsg, publisher);        kvmsg_store   (&kvmsg, kvmap);    }    printf (" Interrupted/n%d messages out/n", (int) sequence);    zhash_destroy (&kvmap);    zctx_destroy (&ctx);    return 0;}

struct cache *cache_create (void){    struct cache *cache = xzmalloc (sizeof (*cache));    if (!(cache->zh = zhash_new ()))        oom ();    return cache;}

int mod_main (flux_t *h, int argc, char **argv){    int saved_errno;    flux_msg_handler_t **handlers = NULL;    if (argc == 1 && !strcmp (argv[0], "--init-failure")) {        flux_log (h, LOG_INFO, "aborting during init per test request");        errno = EIO;        goto error;    }    if (!(modules = zhash_new ())) {        errno = ENOMEM;        goto error;    }    if (flux_get_rank (h, &rank) < 0)        goto error;    if (flux_msg_handler_addvec (h, htab, NULL, &handlers) < 0)        goto error;    if (flux_reactor_run (flux_get_reactor (h), 0) < 0) {        flux_log_error (h, "flux_reactor_run");        goto error;    }    zhash_destroy (&modules);    return 0;error:    saved_errno = errno;    flux_msg_handler_delvec (handlers);    zhash_destroy (&modules);    errno = saved_errno;    return -1;}

static intserver_initialize (server_t *self){    zsys_notice ("starting zpipes service");    self->pipes = zhash_new ();    return 0;}

int main (void){    clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));    self->port = 5556;    self->ctx = zctx_new ();    self->kvmap = zhash_new ();    self->loop = zloop_new ();    zloop_set_verbose (self->loop, FALSE);    //  Set up our clone server sockets    self->snapshot  = zsocket_new (self->ctx, ZMQ_ROUTER);    self->publisher = zsocket_new (self->ctx, ZMQ_PUB);    self->collector = zsocket_new (self->ctx, ZMQ_PULL);    zsocket_bind (self->snapshot,  "tcp://*:%d", self->port);    zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1);    zsocket_bind (self->collector, "tcp://*:%d", self->port + 2);    //  Register our handlers with reactor    zmq_pollitem_t poller = { self->snapshot, 0, ZMQ_POLLIN };    zloop_poller (self->loop, &poller, s_snapshots, self);    poller.socket = self->collector;    zloop_poller (self->loop, &poller, s_collector, self);    zloop_timer  (self->loop, 1000, 0, s_flush_ttl, self);    //  Run reactor until process interrupted    zloop_start (self->loop);    zloop_destroy (&self->loop);    zhash_destroy (&self->kvmap);    zctx_destroy (&self->ctx);    free (self);    return 0;}

attr_t *attr_create (void){    attr_t *attrs = xzmalloc (sizeof (*attrs));    if (!(attrs->hash = zhash_new ()))        oom ();    return attrs;}

void* collabclient_new( char* address, int port ){#ifdef BUILD_COLLAB    printf("collabclient_new() address:%s port:%d/n", address, port );        cloneclient_t *cc = 0;    cc = (cloneclient_t *) zmalloc (sizeof (cloneclient_t));    cc->magic_number = MAGIC_VALUE;    cc->ctx   = obtainMainZMQContext();    cc->loop  = 0;    cc->address = copy( address );    cc->port  = port;    cc->kvmap = zhash_new();    cc->publisher_sendseq = 1;    cc->sequence = 0;    cc->preserveUndo = 0;    cc->roundTripTimerWaitingSeq = 0;    collabclient_remakeSockets( cc );    int32 roundTripTimerMS = pref_collab_roundTripTimerMS;    cc->roundTripTimer = BackgroundTimer_new( roundTripTimerMS, 					      collabclient_roundTripTimer,					      cc );        return cc;#endif    return 0;}

zcert_t *zcert_new_from (byte *public_key, byte *secret_key){    zcert_t *self = (zcert_t *) zmalloc (sizeof (zcert_t));    if (!self)        return NULL;    assert (public_key);    assert (secret_key);    self->metadata = zhash_new ();    if (self->metadata) {        zhash_autofree (self->metadata);        memcpy (self->public_key, public_key, 32);        memcpy (self->secret_key, secret_key, 32);#if (ZMQ_VERSION_MAJOR == 4)        zmq_z85_encode (self->public_txt, self->public_key, 32);        zmq_z85_encode (self->secret_txt, self->secret_key, 32);#else        strcpy (self->public_txt, FORTY_ZEROES);        strcpy (self->secret_txt, FORTY_ZEROES);#endif    }    else        zcert_destroy (&self);    return self;}

void mod_lsmod (flux_t h, opt_t opt){    char *service = "cmb";    if (opt.argc > 1)        usage ();    if (opt.argc == 1)        service = opt.argv[0];    printf ("%-20s %6s %7s %4s %s/n",            "Module", "Size", "Digest", "Idle", "Nodeset");    if (opt.direct) {        zhash_t *mods = zhash_new ();        if (!mods)            oom ();        char *topic = xasprintf ("%s.lsmod", service);        flux_rpc_t *r = flux_rpc_multi (h, topic, NULL, opt.nodeset, 0);        if (!r)            err_exit ("%s", topic);        while (!flux_rpc_completed (r)) {            const char *json_str;            uint32_t nodeid;            if (flux_rpc_get (r, &nodeid, &json_str) < 0)                err_exit ("%s", topic);            if (lsmod_hash_cb (nodeid, json_str, mods) < 0)                err_exit ("%s[%u]", topic, nodeid);        }        flux_rpc_destroy (r);        lsmod_map_hash (mods, lsmod_print_cb, NULL);        zhash_destroy (&mods);        free (topic);    } else {        if (flux_modctl_list (h, service, opt.nodeset, lsmod_print_cb, NULL) < 0)            err_exit ("modctl_list");    }}

static ctx_t *getctx (flux_t *h){    ctx_t *ctx = (ctx_t *)flux_aux_get (h, "flux::barrier");    if (!ctx) {        ctx = xzmalloc (sizeof (*ctx));        if (!(ctx->barriers = zhash_new ())) {            errno = ENOMEM;            goto error;        }        if (flux_get_rank (h, &ctx->rank) < 0) {            flux_log_error (h, "flux_get_rank");            goto error;        }        if (!(ctx->timer = flux_timer_watcher_create (flux_get_reactor (h),                       barrier_reduction_timeout_sec, 0., timeout_cb, ctx) )) {            flux_log_error (h, "flux_timer_watacher_create");            goto error;        }        ctx->h = h;        flux_aux_set (h, "flux::barrier", ctx, freectx);    }    return ctx;error:    freectx (ctx);    return NULL;}

static ctx_t *getctx (flux_t h){    int saved_errno;    ctx_t *ctx = (ctx_t *)flux_aux_get (h, "req");    if (!ctx) {        ctx = xzmalloc (sizeof (*ctx));        ctx->h = h;        ctx->ping_requests = zhash_new ();        ctx->clog_requests = zlist_new ();        if (!ctx->clog_requests || !ctx->ping_requests) {            saved_errno = ENOMEM;            goto error;         }        if (flux_get_rank (h, &ctx->rank) < 0) {            saved_errno = errno;            goto error;        }        flux_aux_set (h, "req", ctx, freectx);    }    return ctx;error:    freectx (ctx);    errno = saved_errno;    return NULL;}

static zhash_t *determine_all_min_bandwidth (struct rdl *rdl,                                             zlist_t *running_jobs){    double root_bw;    double *curr_value = NULL;    struct resource *root = NULL;    job_t *curr_job = NULL;    char job_id_str[100];    zhash_t *job_hash = zhash_new ();    root = rdl_resource_get (rdl, "default");    root_bw = get_max_bandwidth (root);    curr_job = zlist_first (running_jobs);    while (curr_job != NULL) {        curr_value = (double *)malloc (sizeof (double));        *curr_value = root_bw;        sprintf (job_id_str, "%d", curr_job->id);        zhash_insert (job_hash, job_id_str, curr_value);        zhash_freefn (job_hash, job_id_str, free);        curr_job = zlist_next (running_jobs);    }    determine_all_min_bandwidth_helper (root, root_bw, job_hash);    return job_hash;}

voidzyre_peer_test (bool verbose){    printf (" * zyre_peer: ");    zsock_t *mailbox = zsock_new_dealer ("@tcp://127.0.0.1:5551");    zhash_t *peers = zhash_new ();    zuuid_t *you = zuuid_new ();    zuuid_t *me = zuuid_new ();    zyre_peer_t *peer = zyre_peer_new (peers, you);    assert (!zyre_peer_connected (peer));    zyre_peer_connect (peer, me, "tcp://127.0.0.1:5551");    assert (zyre_peer_connected (peer));    zyre_peer_set_name (peer, "peer");    assert (streq (zyre_peer_name (peer), "peer"));    zre_msg_t *msg = zre_msg_new (ZRE_MSG_HELLO);    zre_msg_set_endpoint (msg, "tcp://127.0.0.1:5552");    int rc = zyre_peer_send (peer, &msg);    assert (rc == 0);    msg = zre_msg_recv (mailbox);    assert (msg);    if (verbose)        zre_msg_print (msg);    zre_msg_destroy (&msg);    //  Destroying container destroys all peers it contains    zhash_destroy (&peers);    zuuid_destroy (&me);    zuuid_destroy (&you);    zsock_destroy (&mailbox);    printf ("OK/n");}

static self_t *s_self_new (zsock_t *pipe){    self_t *self = (self_t *) zmalloc (sizeof (self_t));    self->pipe = pipe;    self->whitelist = zhash_new ();    self->blacklist = zhash_new ();    //  Create ZAP handler and get ready for requests    self->handler = zsock_new (ZMQ_REP);    assert (self->handler);    int rc = zsock_bind (self->handler, "inproc://zeromq.zap.01");    assert (rc == 0);    self->poller = zpoller_new (self->pipe, self->handler, NULL);    return self;}

zhash_t *zhash_dup (zhash_t *self){    if (!self)        return NULL;    zhash_t *copy = zhash_new ();    if (copy) {        copy->destructor = self->destructor;        copy->duplicator = self->duplicator;        uint index;        size_t limit = primes [self->prime_index];        for (index = 0; index < limit; index++) {            item_t *item = self->items [index];            while (item) {                if (zhash_insert (copy, item->key, item->value)) {                    zhash_destroy (&copy);                    break;                }                item = item->next;            }        }    }    return copy;}

modhash_t *modhash_create (void){    modhash_t *mh = xzmalloc (sizeof (*mh));    if (!(mh->zh_byuuid = zhash_new ()))        oom ();    return mh;}

zhash_t *get_command_list_hash (const char *pattern){    int rc;    size_t i;    glob_t gl;    zhash_t *zh = NULL;    rc = glob (pattern, GLOB_ERR, NULL, &gl);    switch (rc) {        case 0:            break; /* have results, fall-through. */        case GLOB_ABORTED:            /* No help.d directory? */            goto out;            break;        case GLOB_NOMATCH:            goto out;            break;        default:            fprintf (stderr, "glob: unknown error %d/n", rc);            break;    }    zh = zhash_new ();    //zhash_set_destructor (zh, (czmq_destructor *) zlist_destroy);    for (i = 0; i < gl.gl_pathc; i++) {        const char *file = gl.gl_pathv[i];        if (command_list_read (zh, file) < 0)            log_err_exit ("%s: failed to read content/n", file);    }    globfree (&gl);out:    return (zh);}

voidzdir_watch (zsock_t *pipe, void *unused){    zdir_watch_t *watch = s_zdir_watch_new (pipe);    assert (watch);    watch->loop = zloop_new ();    assert (watch->loop);    watch->subs = zhash_new ();    assert (watch->subs);    zloop_reader (watch->loop, pipe, s_on_command, watch);    zloop_reader_set_tolerant (watch->loop, pipe); // command pipe needs to be tolerant, otherwise we'd have a hard time shutting down    s_zdir_watch_timeout (watch, 250); // default poll time of 250ms    //  Signal initialization    zsock_signal (pipe, 0);    // Dispatch above handlers    zloop_start (watch->loop);    if (watch->verbose)        zsys_info ("zdir_watch: Complete");    // signal destruction    zsock_signal (watch->pipe, 0);    // Done - cleanup and exit    s_zdir_watch_destroy (&watch);}