static int
virNetClientCallDispatch(virNetClientPtr client)
{
    PROBE(RPC_CLIENT_MSG_RX,
          "client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u serial=%u",
          client, client->msg.bufferLength,
          client->msg.header.prog, client->msg.header.vers, client->msg.header.proc,
          client->msg.header.type, client->msg.header.status, client->msg.header.serial);

    switch (client->msg.header.type) {
    case VIR_NET_REPLY: /* Normal RPC replies */
    case VIR_NET_REPLY_WITH_FDS: /* Normal RPC replies with FDs */
        return virNetClientCallDispatchReply(client);

    case VIR_NET_MESSAGE: /* Async notifications */
        return virNetClientCallDispatchMessage(client);

    case VIR_NET_STREAM: /* Stream protocol */
        return virNetClientCallDispatchStream(client);

    default:
        virNetError(VIR_ERR_RPC,
                    _("got unexpected RPC call prog %d vers %d proc %d type %d"),
                    client->msg.header.prog, client->msg.header.vers,
                    client->msg.header.proc, client->msg.header.type);
        return -1;
    }
}

virKeepAlivePtr
virKeepAliveNew(int interval,
                unsigned int count,
                void *client,
                virKeepAliveSendFunc sendCB,
                virKeepAliveDeadFunc deadCB,
                virKeepAliveFreeFunc freeCB)
{
    virKeepAlivePtr ka;

    VIR_DEBUG("client=%p, interval=%d, count=%u", client, interval, count);

    if (virKeepAliveInitialize() < 0)
        return NULL;

    if (!(ka = virObjectLockableNew(virKeepAliveClass)))
        return NULL;

    ka->interval = interval;
    ka->count = count;
    ka->countToDeath = count;
    ka->timer = -1;
    ka->client = client;
    ka->sendCB = sendCB;
    ka->deadCB = deadCB;
    ka->freeCB = freeCB;

    PROBE(RPC_KEEPALIVE_NEW,
          "ka=%p client=%p",
          ka, ka->client);

    return ka;
}

void virNetServerClientDispose(void *obj)
{
    virNetServerClientPtr client = obj;

    PROBE(RPC_SERVER_CLIENT_DISPOSE,
          "client=%p", client);

    virObjectUnref(client->identity);

    if (client->privateData &&
        client->privateDataFreeFunc)
        client->privateDataFreeFunc(client->privateData);

#if WITH_SASL
    virObjectUnref(client->sasl);
#endif
    if (client->sockTimer > 0)
        virEventRemoveTimeout(client->sockTimer);
#if WITH_GNUTLS
    virObjectUnref(client->tls);
    virObjectUnref(client->tlsCtxt);
#endif
    virObjectUnref(client->sock);
    virObjectUnlock(client);
}

void virEventPollUpdateTimeout(int timer, int frequency)
{
    unsigned long long now;
    int i;
    PROBE(EVENT_POLL_UPDATE_TIMEOUT,
          "timer=%d frequency=%d",
          timer, frequency);

    if (timer <= 0) {
        VIR_WARN("Ignoring invalid update timer %d", timer);
        return;
    }

    if (virTimeMillisNow(&now) < 0) {
        return;
    }

    virMutexLock(&eventLoop.lock);
    for (i = 0 ; i < eventLoop.timeoutsCount ; i++) {
        if (eventLoop.timeouts[i].timer == timer) {
            eventLoop.timeouts[i].frequency = frequency;
            eventLoop.timeouts[i].expiresAt =
                frequency >= 0 ? frequency + now : 0;
            virEventPollInterruptLocked();
            break;
        }
    }
    virMutexUnlock(&eventLoop.lock);
}

void virEventPollUpdateHandle(int watch, int events) {
    int i;
    bool found = false;
    PROBE(EVENT_POLL_UPDATE_HANDLE,
          "watch=%d events=%d",
          watch, events);

    if (watch <= 0) {
        VIR_WARN("Ignoring invalid update watch %d", watch);
        return;
    }

    virMutexLock(&eventLoop.lock);
    for (i = 0; i < eventLoop.handlesCount; i++) {
        if (eventLoop.handles[i].watch == watch) {
            eventLoop.handles[i].events =
                    virEventPollToNativeEvents(events);
            virEventPollInterruptLocked();
            found = true;
            break;
        }
    }
    virMutexUnlock(&eventLoop.lock);

    if (!found)
        VIR_WARN("Got update for non-existent handle watch %d", watch);
}

/*
 * Unregister a callback from a file handle
 * NB, it *must* be safe to call this from within a callback
 * For this reason we only ever set a flag in the existing list.
 * Actual deletion will be done out-of-band
 */
int virEventPollRemoveHandle(int watch) {
    int i;
    PROBE(EVENT_POLL_REMOVE_HANDLE,
          "watch=%d",
          watch);

    if (watch <= 0) {
        VIR_WARN("Ignoring invalid remove watch %d", watch);
        return -1;
    }

    virMutexLock(&eventLoop.lock);
    for (i = 0; i < eventLoop.handlesCount; i++) {
        if (eventLoop.handles[i].deleted)
            continue;

        if (eventLoop.handles[i].watch == watch) {
            EVENT_DEBUG("mark delete %d %d", i, eventLoop.handles[i].fd);
            eventLoop.handles[i].deleted = 1;
            virEventPollInterruptLocked();
            virMutexUnlock(&eventLoop.lock);
            return 0;
        }
    }
    virMutexUnlock(&eventLoop.lock);
    return -1;
}

/*
 * Unregister a callback for a timer
 * NB, it *must* be safe to call this from within a callback
 * For this reason we only ever set a flag in the existing list.
 * Actual deletion will be done out-of-band
 */
int virEventPollRemoveTimeout(int timer)
{
    size_t i;
    PROBE(EVENT_POLL_REMOVE_TIMEOUT,
          "timer=%d",
          timer);

    if (timer <= 0) {
        VIR_WARN("Ignoring invalid remove timer %d", timer);
        return -1;
    }

    virMutexLock(&eventLoop.lock);
    for (i = 0; i < eventLoop.timeoutsCount; i++) {
        if (eventLoop.timeouts[i].deleted)
            continue;

        if (eventLoop.timeouts[i].timer == timer) {
            eventLoop.timeouts[i].deleted = 1;
            virEventPollInterruptLocked();
            virMutexUnlock(&eventLoop.lock);
            return 0;
        }
    }
    virMutexUnlock(&eventLoop.lock);
    return -1;
}

int
virKeepAliveStart(virKeepAlivePtr ka,
                  int interval,
                  unsigned int count)
{
    int ret = -1;
    time_t delay;
    int timeout;
    time_t now;

    virKeepAliveLock(ka);

    if (ka->timer >= 0) {
        VIR_DEBUG("Keepalive messages already enabled");
        ret = 0;
        goto cleanup;
    }

    if (interval > 0) {
        if (ka->interval > 0) {
            virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
                           _("keepalive interval already set"));
            goto cleanup;
        }
        ka->interval = interval;
        ka->count = count;
        ka->countToDeath = count;
    }

    if (ka->interval <= 0) {
        VIR_DEBUG("Keepalive messages disabled by configuration");
        ret = 0;
        goto cleanup;
    }

    PROBE(RPC_KEEPALIVE_START,
          "ka=%p client=%p interval=%d count=%u",
          ka, ka->client, interval, count);

    now = time(NULL);
    delay = now - ka->lastPacketReceived;
    if (delay > ka->interval)
        timeout = 0;
    else
        timeout = ka->interval - delay;
    ka->intervalStart = now - (ka->interval - timeout);
    ka->timer = virEventAddTimeout(timeout * 1000, virKeepAliveTimer,
                                   ka, virKeepAliveTimerFree);
    if (ka->timer < 0)
        goto cleanup;

    /* the timer now has another reference to this object */
    ka->refs++;
    ret = 0;

cleanup:
    virKeepAliveUnlock(ka);
    return ret;
}

void virNetClientRef(virNetClientPtr client)
{
    virNetClientLock(client);
    client->refs++;
    PROBE(RPC_CLIENT_REF,
          "client=%p refs=%d",
          client, client->refs);
    virNetClientUnlock(client);
}

/*
 * Run a single iteration of the event loop, blocking until
 * at least one file handle has an event, or a timer expires
 */
int virEventPollRunOnce(void)
{
    struct pollfd *fds = NULL;
    int ret, timeout, nfds;

    virMutexLock(&eventLoop.lock);
    eventLoop.running = 1;
    virThreadSelf(&eventLoop.leader);

    virEventPollCleanupTimeouts();
    virEventPollCleanupHandles();

    if (!(fds = virEventPollMakePollFDs(&nfds)) ||
        virEventPollCalculateTimeout(&timeout) < 0)
        goto error;

    virMutexUnlock(&eventLoop.lock);

 retry:
    PROBE(EVENT_POLL_RUN,
          "nhandles=%d timeout=%d",
          nfds, timeout);
    ret = poll(fds, nfds, timeout);
    if (ret < 0) {
        EVENT_DEBUG("Poll got error event %d", errno);
        if (errno == EINTR || errno == EAGAIN) {
            goto retry;
        }
        virReportSystemError(errno, "%s",
                             _("Unable to poll on file handles"));
        goto error_unlocked;
    }
    EVENT_DEBUG("Poll got %d event(s)", ret);

    virMutexLock(&eventLoop.lock);
    if (virEventPollDispatchTimeouts() < 0)
        goto error;

    if (ret > 0 &&
        virEventPollDispatchHandles(nfds, fds) < 0)
        goto error;

    virEventPollCleanupTimeouts();
    virEventPollCleanupHandles();

    eventLoop.running = 0;
    virMutexUnlock(&eventLoop.lock);
    VIR_FREE(fds);
    return 0;

 error:
    virMutexUnlock(&eventLoop.lock);
 error_unlocked:
    VIR_FREE(fds);
    return -1;
}

void
virKeepAliveRef(virKeepAlivePtr ka)
{
    virKeepAliveLock(ka);
    ka->refs++;
    PROBE(RPC_KEEPALIVE_REF,
          "ka=%p client=%p refs=%d",
          ka, ka->client, ka->refs);
    virKeepAliveUnlock(ka);
}

void
virKeepAliveDispose(void *obj)
{
    virKeepAlivePtr ka = obj;

    PROBE(RPC_KEEPALIVE_DISPOSE,
          "ka=%p", ka);

    ka->freeCB(ka->client);
}

static virNetClientPtr virNetClientNew(virNetSocketPtr sock,
                                       const char *hostname)
{
    virNetClientPtr client = NULL;
    int wakeupFD[2] = { -1, -1 };

    if (pipe2(wakeupFD, O_CLOEXEC) < 0) {
        virReportSystemError(errno, "%s",
                             _("unable to make pipe"));
        goto error;
    }

    if (VIR_ALLOC(client) < 0)
        goto no_memory;

    client->refs = 1;

    if (virMutexInit(&client->lock) < 0)
        goto error;

    client->sock = sock;
    client->wakeupReadFD = wakeupFD[0];
    client->wakeupSendFD = wakeupFD[1];
    wakeupFD[0] = wakeupFD[1] = -1;

    if (hostname &&
        !(client->hostname = strdup(hostname)))
        goto no_memory;

    /* Set up a callback to listen on the socket data */
    client->refs++;
    if (virNetSocketAddIOCallback(client->sock,
                                  VIR_EVENT_HANDLE_READABLE,
                                  virNetClientIncomingEvent,
                                  client,
                                  virNetClientEventFree) < 0) {
        client->refs--;
        VIR_DEBUG("Failed to add event watch, disabling events");
    }

    PROBE(RPC_CLIENT_NEW,
          "client=%p refs=%d sock=%p",
          client, client->refs, client->sock);

    return client;

no_memory:
    virReportOOMError();
error:
    VIR_FORCE_CLOSE(wakeupFD[0]);
    VIR_FORCE_CLOSE(wakeupFD[1]);
    virNetClientFree(client);
    return NULL;
}

static virNetServerClientPtr
virNetServerClientNewInternal(unsigned long long id,
                              virNetSocketPtr sock,
                              int auth,
#ifdef WITH_GNUTLS
                              virNetTLSContextPtr tls,
#endif
                              bool readonly,
                              size_t nrequests_max,
                              long long timestamp)
{
    virNetServerClientPtr client;

    if (virNetServerClientInitialize() < 0)
        return NULL;

    if (!(client = virObjectLockableNew(virNetServerClientClass)))
        return NULL;

    client->id = id;
    client->sock = virObjectRef(sock);
    client->auth = auth;
    client->readonly = readonly;
#ifdef WITH_GNUTLS
    client->tlsCtxt = virObjectRef(tls);
#endif
    client->nrequests_max = nrequests_max;
    client->conn_time = timestamp;

    client->sockTimer = virEventAddTimeout(-1, virNetServerClientSockTimerFunc,
                                           client, NULL);
    if (client->sockTimer < 0)
        goto error;

    /* Prepare one for packet receive */
    if (!(client->rx = virNetMessageNew(true)))
        goto error;
    client->rx->bufferLength = VIR_NET_MESSAGE_LEN_MAX;
    if (VIR_ALLOC_N(client->rx->buffer, client->rx->bufferLength) < 0)
        goto error;
    client->nrequests = 1;

    PROBE(RPC_SERVER_CLIENT_NEW,
          "client=%p sock=%p",
          client, client->sock);

    return client;

 error:
    virObjectUnref(client);
    return NULL;
}

static virNetServerClientPtr
virNetServerClientNewInternal(virNetSocketPtr sock,
                              int auth,
                              bool readonly,
                              size_t nrequests_max,
                              virNetTLSContextPtr tls)
{
    virNetServerClientPtr client;

    if (virNetServerClientInitialize() < 0)
        return NULL;

    if (!(client = virObjectNew(virNetServerClientClass)))
        return NULL;

    if (virMutexInit(&client->lock) < 0) {
        VIR_FREE(client);
        return NULL;
    }

    client->sock = virObjectRef(sock);
    client->auth = auth;
    client->readonly = readonly;
    client->tlsCtxt = virObjectRef(tls);
    client->nrequests_max = nrequests_max;

    client->sockTimer = virEventAddTimeout(-1, virNetServerClientSockTimerFunc,
                                           client, NULL);
    if (client->sockTimer < 0)
        goto error;

    /* Prepare one for packet receive */
    if (!(client->rx = virNetMessageNew(true)))
        goto error;
    client->rx->bufferLength = VIR_NET_MESSAGE_LEN_MAX;
    if (VIR_ALLOC_N(client->rx->buffer, client->rx->bufferLength) < 0) {
        virReportOOMError();
        goto error;
    }
    client->nrequests = 1;

    PROBE(RPC_SERVER_CLIENT_NEW,
          "client=%p sock=%p",
          client, client->sock);

    return client;

error:
    virObjectUnref(client);
    return NULL;
}

int virNetClientSend(virNetClientPtr client,
                     virNetMessagePtr msg,
                     bool expectReply)
{
    virNetClientCallPtr call;
    int ret = -1;

    PROBE(RPC_CLIENT_MSG_TX_QUEUE,
          "client=%p len=%zu prog=%u vers=%u proc=%u type=%u status=%u serial=%u",
          client, msg->bufferLength,
          msg->header.prog, msg->header.vers, msg->header.proc,
          msg->header.type, msg->header.status, msg->header.serial);

    if (expectReply &&
        (msg->bufferLength != 0) &&
        (msg->header.status == VIR_NET_CONTINUE)) {
        virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
                    _("Attempt to send an asynchronous message with a synchronous reply"));
        return -1;
    }

    if (VIR_ALLOC(call) < 0) {
        virReportOOMError();
        return -1;
    }

    virNetClientLock(client);

    if (virCondInit(&call->cond) < 0) {
        virNetError(VIR_ERR_INTERNAL_ERROR, "%s",
                    _("cannot initialize condition variable"));
        goto cleanup;
    }

    msg->donefds = 0;
    if (msg->bufferLength)
        call->mode = VIR_NET_CLIENT_MODE_WAIT_TX;
    else
        call->mode = VIR_NET_CLIENT_MODE_WAIT_RX;
    call->msg = msg;
    call->expectReply = expectReply;

    ret = virNetClientIO(client, call);

cleanup:
    ignore_value(virCondDestroy(&call->cond));
    VIR_FREE(call);
    virNetClientUnlock(client);
    return ret;
}

bool
virKeepAliveCheckMessage(virKeepAlivePtr ka,
                         virNetMessagePtr msg,
                         virNetMessagePtr *response)
{
    bool ret = false;

    VIR_DEBUG("ka=%p, client=%p, msg=%p",
              ka, ka ? ka->client : "(null)", msg);

    *response = NULL;
    if (!ka)
        return false;

    virObjectLock(ka);

    ka->countToDeath = ka->count;
    ka->lastPacketReceived = ka->intervalStart = time(NULL);

    if (msg->header.prog == KEEPALIVE_PROGRAM &&
        msg->header.vers == KEEPALIVE_PROTOCOL_VERSION &&
        msg->header.type == VIR_NET_MESSAGE) {
        PROBE(RPC_KEEPALIVE_RECEIVED,
              "ka=%p client=%p prog=%d vers=%d proc=%d",
              ka, ka->client, msg->header.prog,
              msg->header.vers, msg->header.proc);
        ret = true;
        switch (msg->header.proc) {
        case KEEPALIVE_PROC_PING:
            VIR_DEBUG("Got keepalive request from client %p", ka->client);
            *response = virKeepAliveMessage(ka, KEEPALIVE_PROC_PONG);
            break;

        case KEEPALIVE_PROC_PONG:
            VIR_DEBUG("Got keepalive response from client %p", ka->client);
            break;

        default:
            VIR_DEBUG("Ignoring unknown keepalive message %d from client %p",
                      msg->header.proc, ka->client);
        }
    }

    if (ka->timer >= 0)
        virEventUpdateTimeout(ka->timer, ka->interval * 1000);

    virObjectUnlock(ka);

    return ret;
}

void
virKeepAliveStop(virKeepAlivePtr ka)
{
    virObjectLock(ka);

    PROBE(RPC_KEEPALIVE_STOP,
          "ka=%p client=%p",
          ka, ka->client);

    if (ka->timer > 0) {
        virEventRemoveTimeout(ka->timer);
        ka->timer = -1;
    }

    virObjectUnlock(ka);
}

value arc_hash_insert(arc *c, value hash, value key, value val)
{
  unsigned int hv, index, i;
  value e;

  index = 0;
  /* First of all, look for the key if a binding already exists for it */
  e = hash_lookup(c, hash, key, &index);
  if (BOUND_P(e)) {
    /* if we are already bound, overwrite the old value */
    e = VINDEX(HASH_TABLE(hash), index);
    BVALUE(e) = val;
    return(val);
  }
  /* Not yet bound.  Look for a slot where we can put it */
  if (HASH_NENTRIES(hash)+1 > HASH_LLIMIT(hash))
    hashtable_expand(c, hash);
  SET_NENTRIES(hash, HASH_NENTRIES(hash)+1);
  hv = arc_hash(c, key);
  index = hv & TABLEMASK(hash);
  for (i=0;; i++) {
    e = VINDEX(HASH_TABLE(hash), index);
    /* If we see an empty bucket in our search, or if we see a bucket
       whose key is the same as the key specified, we have found the
       place where the element should go. This second case should never
       happen, based on what we did above, but hey, belt and suspenders. */
    if (EMPTYP(e) || arc_is2(c, BKEY(e), key) == CTRUE)
      break;
    /* We found a bucket, but it is occupied by some other key. Continue
       probing. */
    index = (index + PROBE(i)) & TABLEMASK(hash);
  }

  if (EMPTYP(e)) {
    /* No such key in the hash table yet.  Create a bucket and
       assign it to the table. */
    e = mkhashbucket(c, key, val, index, hash, INT2FIX(hv));
    SVINDEX(HASH_TABLE(hash), index, e);
  } else {
    /* The key already exists.  Use the current bucket but change the
       value to the value specified. */
    BVALUE(e) = val;
  }
  return(val);
}

/* Used post dispatch to actually remove any handles that
 * were previously marked as deleted. This asynchronous
 * cleanup is needed to make dispatch re-entrant safe.
 */
static void virEventPollCleanupHandles(void)
{
    size_t i;
    size_t gap;
    VIR_DEBUG("Cleanup %zu", eventLoop.handlesCount);

    /* Remove deleted entries, shuffling down remaining
     * entries as needed to form contiguous series
     */
    for (i = 0; i < eventLoop.handlesCount;) {
        if (!eventLoop.handles[i].deleted) {
            i++;
            continue;
        }

        PROBE(EVENT_POLL_PURGE_HANDLE,
              "watch=%d",
              eventLoop.handles[i].watch);
        if (eventLoop.handles[i].ff) {
            virFreeCallback ff = eventLoop.handles[i].ff;
            void *opaque = eventLoop.handles[i].opaque;
            virMutexUnlock(&eventLoop.lock);
            ff(opaque);
            virMutexLock(&eventLoop.lock);
        }

        if ((i+1) < eventLoop.handlesCount) {
            memmove(eventLoop.handles+i,
                    eventLoop.handles+i+1,
                    sizeof(struct virEventPollHandle)*(eventLoop.handlesCount
                                                   -(i+1)));
        }
        eventLoop.handlesCount--;
    }

    /* Release some memory if we've got a big chunk free */
    gap = eventLoop.handlesAlloc - eventLoop.handlesCount;
    if (eventLoop.handlesCount == 0 ||
        (gap > eventLoop.handlesCount && gap > EVENT_ALLOC_EXTENT)) {
        EVENT_DEBUG("Found %zu out of %zu handles slots used, releasing %zu",
                    eventLoop.handlesCount, eventLoop.handlesAlloc, gap);
        VIR_SHRINK_N(eventLoop.handles, eventLoop.handlesAlloc, gap);
    }
}