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

voidzmsg_remove (zmsg_t *self, zframe_t *frame){    assert (self);    self->content_size -= zframe_size (frame);    zlist_remove (self->frames, frame);}

static voidsub_patch_add (sub_t *self, fmq_patch_t *patch){    //  Skip file creation if client already has identical file    fmq_patch_digest_set (patch);    if (fmq_patch_op (patch) == patch_create) {        char *digest = zhash_lookup (self->cache, fmq_patch_virtual (patch));        if (digest && strcasecmp (digest, fmq_patch_digest (patch)) == 0)            return;             //  Just skip patch for this client    }    //  Remove any previous patches for the same file    fmq_patch_t *existing = (fmq_patch_t *) zlist_first (self->client->patches);    while (existing) {        if (streq (fmq_patch_virtual (patch), fmq_patch_virtual (existing))) {            zlist_remove (self->client->patches, existing);            fmq_patch_destroy (&existing);            break;        }        existing = (fmq_patch_t *) zlist_next (self->client->patches);    }    if (fmq_patch_op (patch) == patch_create)        zhash_insert (self->cache,            fmq_patch_digest (patch), fmq_patch_virtual (patch));    //  Track that we've queued patch for client, so we don't do it twice    zlist_append (self->client->patches, fmq_patch_dup (patch));}

static void s_worker_delete(worker_t *self, int disconnect){    assert (self);    if (disconnect) {        s_worker_send (self, MDPW_DISCONNECT, NULL); // ! Расширить до отключение через mmi	}    if (self->service) {        zlist_remove (self->service->waiting, self);        self->service->workers--;    }        zlist_remove (self->broker->waiting, self);    //  This implicitly calls s_worker_destroy    zhash_delete (self->broker->workers, self->id_string);}

int wait_destroy_msg (waitqueue_t *q, wait_test_msg_f cb, void *arg){    zlist_t *tmp = NULL;    wait_t *w;    int rc = -1;    int count = 0;    assert (q->magic == WAITQUEUE_MAGIC);    w = zlist_first (q->q);    while (w) {        if (w->hand.msg && cb != NULL && cb (w->hand.msg, arg)) {            if (!tmp && !(tmp = zlist_new ()))                oom ();            if (zlist_append (tmp, w) < 0)                oom ();            w->hand.cb = NULL; // prevent wait_runone from restarting handler            count++;        }        w = zlist_next (q->q);    }    if (tmp) {        while ((w = zlist_pop (tmp))) {            zlist_remove (q->q, w);            if (--w->usecount == 0)                wait_destroy (w);        }    }    rc = count;    zlist_destroy (&tmp);    return rc;}

static void subprocess_free (struct subprocess *p){    assert (p != NULL);    if (p->sm)        zlist_remove (p->sm->processes, (void *) p);    zhash_destroy (&p->zhash);    hooks_table_free (p);    fda_closeall (p->child_fda);    if (p->argz)        free (p->argz);    if (p->envz)        free (p->envz);    if (p->cwd)        free (p->cwd);    zio_destroy (p->zio_in);    zio_destroy (p->zio_out);    zio_destroy (p->zio_err);    if (p->parentfd != -1)        close (p->parentfd);    if (p->childfd != -1)        close (p->childfd);    flux_watcher_destroy (p->child_watcher);    memset (p, 0, sizeof (*p));    free (p);}

static voids_engine_handle_buy_request (engine_t *self, zframe_t *price,        zframe_t *volume, zframe_t *reply_to){    char *price_str = zframe_strdup (price);    char *volume_str = zframe_strdup (volume);    order_t *buy_order = s_order_new (self->worker,            atoi (price_str), atoi (volume_str), reply_to);    free (price_str);    free (volume_str);    while (buy_order->volume) {      order_t *sell_order = s_engine_match_sell_order (self);      if (sell_order == NULL || sell_order->price > buy_order->price)          break;      int volume = MIN (buy_order->volume, sell_order->volume);      s_order_update (buy_order, volume);      s_order_update (sell_order, volume);      if (sell_order->volume == 0) {          zlist_remove (self->sell_orders, sell_order);          s_order_destroy (&sell_order);      }    }    if (buy_order->volume == 0)        s_order_destroy (&buy_order);    else        zlist_append (self->buy_orders, buy_order);}

voidzctx__socket_destroy (zctx_t *self, void *zocket){    assert (self);    assert (zocket);    zsocket_set_linger (zocket, self->linger);    zmq_close (zocket);    zlist_remove (self->sockets, zocket);}

static voids_service_enable_command (service_t *self, const char *command){    char *item = (char *) zlist_first (self->blacklist);    while (item && !streq (item, command))        item = (char *) zlist_next (self->blacklist);    if (item) {        zlist_remove (self->blacklist, item);        free (item);    }}

static voids_worker_delete(worker_t *self, int disconnect){    assert(self);    if (disconnect) {        mdp_msg_t *msg = mdp_msg_new();        assert(msg);        mdp_msg_set_id(msg, MDP_MSG_DISCONNECT);        mdp_msg_set_routing_id(msg, self->address);        mdp_msg_send(msg, self->broker->router);    }        if (self->service) {        zlist_remove(self->service->waiting, self);        self->service->workers--;    }    zlist_remove(self->broker->waiting, self);    // This implicitly calls s_worker_destroy.    zhash_delete(self->broker->workers, self->identity);}

static voids_timer_remove (zloop_t *self, int timer_id){    s_timer_t *timer = (s_timer_t *) zlist_first (self->timers);    while (timer) {        if (timer->timer_id == timer_id) {            zlist_remove (self->timers, timer);            free (timer);            break;        }        timer = (s_timer_t *) zlist_next (self->timers);    }}

static voids_workers_purge (zlist_t *workers){    worker_t *worker = (worker_t *) zlist_first (workers);    while (worker) {        if (zclock_time () < worker->expiry)            break;              //  Worker is alive, we're done here        zlist_remove (workers, worker);        s_worker_destroy (&worker);        worker = (worker_t *) zlist_first (workers);    }}

static voids_worker_ready (worker_t *self, zlist_t *workers){    worker_t *worker = (worker_t *) zlist_first (workers);    while (worker) {        if (streq (self->identity, worker->identity)) {            zlist_remove (workers, worker);            s_worker_destroy (&worker);            break;        }        worker = (worker_t *) zlist_next (workers);    }    zlist_append (workers, self);}

//  If message was already on pending list, remove it and return true,//  else return false.static ints_was_pending (clonesrv_t *self, kvmsg_t *kvmsg){    kvmsg_t *held = (kvmsg_t *) zlist_first (self->pending);    while (held) {        if (memcmp (kvmsg_uuid (kvmsg),                    kvmsg_uuid (held), sizeof (uuid_t)) == 0) {            zlist_remove (self->pending, held);            return true;        }        held = (kvmsg_t *) zlist_next (self->pending);    }    return false;}

static voidmount_sub_purge (mount_t *self, client_t *client){    sub_t *sub = (sub_t *) zlist_first (self->subs);    while (sub) {        if (sub->client == client) {            sub_t *next = (sub_t *) zlist_next (self->subs);            zlist_remove (self->subs, sub);            sub_destroy (&sub);            sub = next;        }        else            sub = (sub_t *) zlist_next (self->subs);    }}

static void s_service_dispatch(service_t *self, zmsg_t *msg){    assert(self);    if (msg) {                    //  Queue message if any        zlist_append(self->requests, msg);	}    s_broker_purge(self->broker);    while (zlist_size(self->waiting) && zlist_size(self->requests)) {        worker_t *worker = zlist_pop(self->waiting);        zlist_remove(self->broker->waiting, worker);        zmsg_t *msg = zlist_pop(self->requests);        s_worker_send(worker, MDPW_REQUEST, msg);        zmsg_destroy (&msg);    }}

static voidpeering_lower (peering_t *self){    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    if (driver->verbose)        zclock_log ("I: (tcp) take down peering to %s", self->address);    if (self->alive) {        self->alive = FALSE;        zlist_remove (vocket->live_peerings, self);        if (zlist_size (vocket->live_peerings) < vocket->min_peerings) {            //  Ask reactor to stop monitoring vocket's msgpipe pipe            zmq_pollitem_t item = { vocket->msgpipe, 0, ZMQ_POLLIN, 0 };            zloop_poller_end (driver->loop, &item);        }    }}

static voidvocket_destroy (vocket_t **self_p){    assert (self_p);    if (*self_p) {        vocket_t *self = *self_p;        driver_t *driver = self->driver;        if (self->min_peerings == 0) {            //  Ask reactor to stop monitoring vocket's msgpipe            zmq_pollitem_t item = { self->msgpipe, 0, ZMQ_POLLIN, 0 };            zloop_poller_end (driver->loop, &item);        }        //  Close message msgpipe socket        zsocket_destroy (driver->ctx, self->msgpipe);        //  Destroy all bindings for this vocket        zhash_destroy (&self->binding_hash);        //  Destroy all peerings for this vocket        zhash_destroy (&self->peering_hash);        zlist_destroy (&self->peering_list);        zlist_destroy (&self->live_peerings);        //  Remove vocket from driver list of vockets        zlist_remove (driver->vockets, self);#ifdef VOCKET_STATS        char *type_name [] = {            "PAIR", "PUB", "SUB", "REQ", "REP",            "DEALER", "ROUTER", "PULL", "PUSH",            "XPUB", "XSUB"        };        printf ("I: type=%s sent=%d recd=%d outp=%d inp=%d drop=%d/n",            type_name [self->socktype],            self->outgoing, self->incoming,            self->outpiped, self->inpiped,            self->dropped);#endif        free (self->vtxname);        free (self);        *self_p = NULL;    }}

voidzloop_reader_end (zloop_t *self, zsock_t *sock){    assert (self);    assert (sock);    s_reader_t *reader = (s_reader_t *) zlist_first (self->readers);    while (reader) {        if (reader->sock == sock) {            zlist_remove (self->readers, reader);            free (reader);            self->need_rebuild = true;        }        reader = (s_reader_t *) zlist_next (self->readers);    }    if (self->verbose)        zsys_debug ("zloop: cancel %s reader", zsock_type_str (sock));}

intzpoller_remove (zpoller_t *self, void *reader){    assert (self);    assert (reader);    int rc = 0;#ifdef ZMQ_HAVE_POLLER    void *socket = zsock_resolve (reader);    if (socket)        rc = zmq_poller_remove (self->zmq_poller, socket);    else        rc = zmq_poller_remove_fd (self->zmq_poller, *(SOCKET *) reader);#else    zlist_remove (self->reader_list, reader);    self->need_rebuild = true;#endif    return rc;}

static void client_write_cb (flux_t h, flux_fd_watcher_t *w, int fd,                             int revents, void *arg){    client_t *c = arg;    if (revents & FLUX_POLLERR)        goto disconnect;    if (revents & FLUX_POLLOUT) {        if (client_send_try (c) < 0)            goto disconnect;        //flux_log (h, LOG_DEBUG, "send: client ready");    }    if (zlist_size (c->outqueue) == 0)        flux_fd_watcher_stop (h, w);    return;disconnect:    zlist_remove (c->ctx->clients, c);    client_destroy (c);}

static voidmount_sub_store (mount_t *self, client_t *client, fmq_msg_t *request){    assert (self);    assert (self->subs);        //  Store subscription along with any previous ones    //  Coalesce subscriptions that are on same path    char *path = fmq_msg_path (request);    sub_t *sub = (sub_t *) zlist_first (self->subs);    while (sub) {        if (client == sub->client) {            //  If old subscription is superset/same as new, ignore new            if (strncmp (path, sub->path, strlen (sub->path)) == 0)                return;            else            //  If new subscription is superset of old one, remove old            if (strncmp (sub->path, path, strlen (path)) == 0) {                zlist_remove (self->subs, sub);                sub_destroy (&sub);                sub = (sub_t *) zlist_first (self->subs);            }            else                sub = (sub_t *) zlist_next (self->subs);        }        else            sub = (sub_t *) zlist_next (self->subs);    }    //  New subscription for this client, append to our list    sub = sub_new (client, path, fmq_msg_cache (request));    zlist_append (self->subs, sub);    //  If client requested resync, send full mount contents now    if (fmq_msg_options_number (client->request, "RESYNC", 0) == 1) {        zlist_t *patches = zdir_resync (self->dir, self->alias);        while (zlist_size (patches)) {            zdir_patch_t *patch = (zdir_patch_t *) zlist_pop (patches);            sub_patch_add (sub, patch);            zdir_patch_destroy (&patch);        }        zlist_destroy (&patches);    }}

//  The dispatch method sends request to the worker.static voids_service_dispatch (service_t *self){    assert (self);    s_broker_purge (self->broker);    if (zlist_size (self->waiting) == 0)        return;    while (zlist_size (self->requests) > 0) {        worker_t *worker = (worker_t*)zlist_pop (self->waiting);        zlist_remove (self->waiting, worker);        zmsg_t *msg = (zmsg_t*)zlist_pop (self->requests);        s_worker_send (worker, MDPW_REQUEST, NULL, msg);        //  Workers are scheduled in the round-robin fashion        zlist_append (self->waiting, worker);        zmsg_destroy (&msg);    }}

voidzctx_destroy (zctx_t **self_p){    assert (self_p);    if (*self_p) {        zctx_t *self = *self_p;        while (zlist_size (self->sockets)) {            void *socket = zlist_first (self->sockets);            zsockopt_set_linger (socket, self->linger);            zmq_close (socket);            zlist_remove (self->sockets, socket);        }        zlist_destroy (&self->sockets);        if (self->main && self->context)            zmq_term (self->context);        free (self);        *self_p = NULL;    }}

static voids_service_destroy (void *argument){    service_t *service = (service_t *) argument;    while (zlist_size (service->requests)) {        zmsg_t *msg = (zmsg_t*)zlist_pop (service->requests);        zmsg_destroy (&msg);    }    //  Free memory keeping  blacklisted commands.    char *command = (char *) zlist_first (service->blacklist);    while (command) {        zlist_remove (service->blacklist, command);        free (command);    }    zlist_destroy (&service->requests);    zlist_destroy (&service->waiting);    zlist_destroy (&service->blacklist);    free (service->name);    free (service);}

voidzloop_poller_end (zloop_t *self, zmq_pollitem_t *item){    assert (self);    assert (item->socket || item->fd);    s_poller_t *poller = (s_poller_t *) zlist_first (self->pollers);    while (poller) {        if ((item->socket && item->socket == poller->item.socket)        ||  (item->fd     && item->fd     == poller->item.fd)) {            zlist_remove (self->pollers, poller);            free (poller);            self->dirty = true;        }        poller = (s_poller_t *) zlist_next (self->pollers);    }    if (self->verbose)        zclock_log ("I: zloop: cancel %s poller (%p, %d)",            item->socket? zsocket_type_str (item->socket): "FD",            item->socket, item->fd);}

static voidpeering_delete (void *argument){    peering_t *self = (peering_t *) argument;    vocket_t *vocket = self->vocket;    driver_t *driver = self->driver;    if (driver->verbose)        zclock_log ("I: (tcp) delete peering %s", self->address);    //* Start transport-specific work    s_close_handle (self->handle, driver);    //* End transport-specific work    vtx_codec_destroy (&self->input);    vtx_codec_destroy (&self->output);    peering_lower (self);    zlist_remove (vocket->peering_list, self);    zloop_timer_end (driver->loop, self);    free (self->address);    free (self);    vocket->peerings--;}

static voids_service_dispatch(service_t *self){    while ((zlist_size(self->requests) > 0) &&           (zlist_size(self->waiting) > 0)) {        worker_t *worker = (worker_t *) zlist_pop(self->waiting);        zlist_remove(self->broker->waiting, worker);        mdp_msg_t *msg = (mdp_msg_t *) zlist_pop(self->requests);        mdp_msg_t *worker_msg = mdp_msg_new();        mdp_msg_set_id(worker_msg, MDP_MSG_WORKER_REQUEST);        mdp_msg_set_routing_id(worker_msg, worker->address);        zframe_t *address = zframe_dup(mdp_msg_routing_id(msg));        mdp_msg_set_address(worker_msg, &address);        zmsg_t *body = mdp_msg_get_body(msg);        mdp_msg_set_body(worker_msg, &body);        mdp_msg_send(worker_msg, self->broker->router);        mdp_msg_destroy(&worker_msg);        mdp_msg_destroy(&msg);    }}

int module_unsubscribe (modhash_t *mh, const char *uuid, const char *topic){    module_t *p = zhash_lookup (mh->zh_byuuid, uuid);    char *s;    int rc = -1;    if (!p) {        errno = ENOENT;        goto done;    }    s = zlist_first (p->subs);    while (s) {        if (!strcmp (topic, s)) {            zlist_remove (p->subs, s);            break;        }        s = zlist_next (p->subs);    }    rc = 0;done:    return rc;}

voidzloop_poller_end (zloop_t *self, zmq_pollitem_t *item){    assert (self);    assert (item->socket || item->fd);    s_poller_t *poller = (s_poller_t *) zlist_first (self->pollers);    while (poller) {        if ((item->socket && item->socket == poller->item.socket)        ||  (item->fd     && item->fd     == poller->item.fd)) {            zlist_remove (self->pollers, poller);            free (poller);            //  Force rebuild to avoid reading from freed poller            self->need_rebuild = true;        }        poller = (s_poller_t *) zlist_next (self->pollers);    }    if (self->verbose)        zsys_debug ("zloop: cancel %s poller (%p, %d)",            item->socket? zsocket_type_str (item->socket): "FD",            item->socket, item->fd);}

intzsys_close (void *handle, const char *filename, size_t line_nbr){    ZMUTEX_LOCK (s_mutex);    //  It's possible atexit() has already happened if we're running under    //  a debugger that redirects the main thread exit.    if (filename && s_sockref_list) {        s_sockref_t *sockref = (s_sockref_t *) zlist_first (s_sockref_list);        while (sockref) {            if (sockref->handle == handle) {                zlist_remove (s_sockref_list, sockref);                free (sockref);                break;            }            sockref = (s_sockref_t *) zlist_next (s_sockref_list);        }    }    s_open_sockets--;    zmq_close (handle);    ZMUTEX_UNLOCK (s_mutex);    return 0;}