c - 如何使用Libevent检测客户端已连接到服务器

Question

我正在编写使用libevent设置本地主机服务器和本地客户端的程序。而且我需要知道客户端何时连接到服务器。 Libevent为事件回调提供了六个事件信号，分别是BEV_EVENT_READING，BEV_EVENT_WRITING，BEV_EVENT_ERROR，BEV_EVENT_TIMEOUT，BEV_EVENT_EOF和BEV_EVENT_CONNECTED。基本上，我想让客户端进入事件循环，并保持等待服务器在线。如果我只是在没有服务器在线的情况下运行客户端程序，则有时会同时获得BEV_EVENT_READING和BEV_EVENT_CONNECTED标志。 BEV_EVENT_READING的错误是connection refused，这是有道理的，因为服务器处于脱机状态。但是，这种情况下的BEV_EVENT_CONNECTED标志也会被设置，即使我调度它，程序也会立即离开。下面是代码片段

address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(7777);
buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);

bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, NULL);
bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);

bufferevent_socket_connect(buffev_ptr, (sockaddr_pt)&address, sizeof(address));

因此，对于 mysignal回调，我应该确切写些什么来处理客户端连接的情况？

void_t Client::mysignal(buffev_pt bev, short flag, void_pt user_data) {

  // An event occurred during a read operation on the bufferevent.
  if(flag & BEV_EVENT_READING) {
  }

  // An event occurred during a write operation on the bufferevent.
  if(flag & BEV_EVENT_WRITING) {
  }

  // An error occurred during a bufferevent operation.
  if(flag & BEV_EVENT_ERROR) {
  }

  // A timeout expired on the bufferevent.
  if(flag & BEV_EVENT_TIMEOUT) {
  }

  // We got an end-of-line indication on the bufferevent.
  if(flag & BEV_EVENT_EOF) {
  }

  // We finished a requested connection on the bufferevent.
  if(flag & BEV_EVENT_CONNECTED) {
  }
}

如果我在没有服务器在线的情况下运行客户端程序，则 BEV_EVENT_CONNECTED和 BEV_EVENT_ERROR可能同时发生。我以为 BEV_EVENT_CONNECTED仅在客户端连接到服务器时设置，但似乎没有。

Answer 1

这是一个很好的示例页面:http://www.wangafu.net/~nickm/libevent-book/Ref6_bufferevent.html

有几件事:bufferevent_socket_connect可能立即失败[附带说明:套接字已设为非阻塞状态]。处理不正确。它是可能同时获得CONNECT和ERROR。我在代码中添加了一些内容并对其进行了注释(请原谅免费的样式清理)。

您的代码只是在处理事件而不跟踪任何“状态”信息，因此我还添加了一个“状态”结构，您可以使用该结构来确定如何处理传入的事件，甚至可以确定连接是否发生以及是否发生连接。可用。远非完美，但可能会有所帮助:

// connection state control
struct privdata {
    u32 priv_evseen;                    // events previously seen
    struct trace_buffer *priv_trace;    // trace buffer to aid debug (e.g.)
    ...
};

int
start_connection(...)
{
    struct privdata *data;
    int err;

    address.sin_family = AF_INET;
    address.sin_addr.s_addr = inet_addr("127.0.0.1");
    address.sin_port = htons(7777);
    buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);

    data = calloc(1,sizeof(struct privdata));

    bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, data);
    bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);

    // NOTE: this can fail immediately
    err = bufferevent_socket_connect(buffev_ptr,
        (sockaddr_pt) &address,sizeof(address));

    // NOTE: if it fails above, the connection attempt will _not_ be retried by
    // libevent -- we _must_ tear down and recreate from scratch (i.e. libevent
    // has already released the socket it tried to connect with)
    //
    // we'll have to decide how/when/if we want to reenter this function to
    // try again (e.g. server is offline and we requeue externally to call us
    // [say] every 5 minutes, with message "server offline -- will retry in
    // 5 minutes")
    if (err < 0) {
        err = errno;

        event_base_free(_evbase_ptr);
        free(data);

        errno = err;
        return -1;
    }

    event_base_dispatch(base);

    return 0;
}

void_t
Client::mysignal(buffev_pt bev, short flag, void_pt user_data)
{
    struct privdata *priv;

    priv = user_data;

    // NOTE: it _is_ possible to get an event that has _both_ CONNNECTED and
    // ERROR because when the connection occurs, libevent issues a getsockopt
    // to check for errors
    //
    // we can use our private data to determine if an ERROR occurs at the
    // connection point (e.g. priv_evseen does _not_ have CONNECTED set) or
    // later during normal socket operation

    do {
        // wait for connection
        // NOTE: this should also handle other errors like timeout
        if ((priv->priv_evseen & CONNECTED) == 0) {
            switch (flag & (ERROR | CONNECTED)) {
            case CONNECTED: // got clean connection -- puppy is happy!
                break;

            case ERROR: // got error first -- CONNECTED event unlikely
                break;

            case (CONNECTED | ERROR):  // connected with error -- how bad is it?
                break;

            case 0:  // huh? -- the only excuse ...
                if (flag & TIMEOUT)
                    ...
                break;
            }
            break;
        }

        // events that occur during normal operation ...

        // An error occurred during a bufferevent operation.
        if (flag & ERROR) {
        }

        // A timeout expired on the bufferevent.
        if (flag & TIMEOUT) {
        }

        // We got an end-of-line indication on the bufferevent.
        if (flag & EOF) {
        }

        // We finished a requested connection on the bufferevent.
        // NOTE: this should _never_ happen now
        if (flag & CONNECTED) {
        }

        // An event occurred during a read operation on the bufferevent.
        if (flag & READING) {
        }

        // An event occurred during a write operation on the bufferevent.
        if (flag & WRITING) {
        }
    } while (0);

    // remember types we've seen before
    priv->priv_seen |= flag;
}

以下是一些相关的libevent来源:

int
bufferevent_socket_connect(struct bufferevent *bev, struct sockaddr *sa, int socklen)
{
    struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bev, struct bufferevent_private, bev);

    evutil_socket_t fd;
    int r = 0;
    int result = -1;
    int ownfd = 0;

    _bufferevent_incref_and_lock(bev);

    if (!bufev_p)
        goto done;

    fd = bufferevent_getfd(bev);
    if (fd < 0) {
        if (!sa)
            goto done;
        fd = socket(sa->sa_family, SOCK_STREAM, 0);
        if (fd < 0)
            goto done;
        if (evutil_make_socket_nonblocking(fd) < 0)
            goto done;
        ownfd = 1;
    }
    if (sa) {
#ifdef WIN32
        if (bufferevent_async_can_connect(bev)) {
            bufferevent_setfd(bev, fd);
            r = bufferevent_async_connect(bev, fd, sa, socklen);
            if (r < 0)
                goto freesock;
            bufev_p->connecting = 1;
            result = 0;
            goto done;
        }
        else
#endif
            r = evutil_socket_connect(&fd, sa, socklen);
        if (r < 0)
            goto freesock;
    }
#ifdef WIN32
    /* ConnectEx() isn't always around, even when IOCP is enabled. Here, we borrow the socket object's write handler to fall back on a non-blocking connect() when ConnectEx() is unavailable. */
    if (BEV_IS_ASYNC(bev)) {
        event_assign(&bev->ev_write, bev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bev);
    }
#endif
    bufferevent_setfd(bev, fd);
    if (r == 0) {
        if (!be_socket_enable(bev, EV_WRITE)) {
            bufev_p->connecting = 1;
            result = 0;
            goto done;
        }
    }
    else if (r == 1) {
        /* The connect succeeded already. How very BSD of it. */
        result = 0;
        bufev_p->connecting = 1;
        event_active(&bev->ev_write, EV_WRITE, 1);
    }
    else {
        /* The connect failed already.  How very BSD of it. */
        bufev_p->connection_refused = 1;
        bufev_p->connecting = 1;
        result = 0;
        event_active(&bev->ev_write, EV_WRITE, 1);
    }

    goto done;

  freesock:
    _bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
    if (ownfd)
        evutil_closesocket(fd);
    /* do something about the error? */
  done:
    _bufferevent_decref_and_unlock(bev);
    return result;
}

/* XXX we should use an enum here. */
/* 2 for connection refused, 1 for connected, 0 for not yet, -1 for error. */
int
evutil_socket_connect(evutil_socket_t * fd_ptr, struct sockaddr *sa, int socklen)
{
    int made_fd = 0;

    if (*fd_ptr < 0) {
        if ((*fd_ptr = socket(sa->sa_family, SOCK_STREAM, 0)) < 0)
            goto err;
        made_fd = 1;
        if (evutil_make_socket_nonblocking(*fd_ptr) < 0) {
            goto err;
        }
    }

    if (connect(*fd_ptr, sa, socklen) < 0) {
        int e = evutil_socket_geterror(*fd_ptr);

        if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
            return 0;
        if (EVUTIL_ERR_CONNECT_REFUSED(e))
            return 2;
        goto err;
    }
    else {
        return 1;
    }

  err:
    if (made_fd) {
        evutil_closesocket(*fd_ptr);
        *fd_ptr = -1;
    }
    return -1;
}

/* Check whether a socket on which we called connect() is done
   connecting. Return 1 for connected, 0 for not yet, -1 for error.  In the
   error case, set the current socket errno to the error that happened during
   the connect operation. */
int
evutil_socket_finished_connecting(evutil_socket_t fd)
{
    int e;
    ev_socklen_t elen = sizeof(e);

    if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void *) &e, &elen) < 0)
        return -1;

    if (e) {
        if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
            return 0;
        EVUTIL_SET_SOCKET_ERROR(e);
        return -1;
    }

    return 1;
}

struct bufferevent *
bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)
{
    struct bufferevent_private *bufev_p;
    struct bufferevent *bufev;

#ifdef WIN32
    if (base && event_base_get_iocp(base))
        return bufferevent_async_new(base, fd, options);
#endif

    if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private))) == NULL)
        return NULL;

    if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket, options) < 0) {
        mm_free(bufev_p);
        return NULL;
    }
    bufev = &bufev_p->bev;
    evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);

    event_assign(&bufev->ev_read, bufev->ev_base, fd, EV_READ | EV_PERSIST, bufferevent_readcb, bufev);
    event_assign(&bufev->ev_write, bufev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bufev);

    evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);

    evbuffer_freeze(bufev->input, 0);
    evbuffer_freeze(bufev->output, 1);

    return bufev;
}

static void
bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
{
    struct bufferevent *bufev = arg;
    struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
    int res = 0;
    short what = BEV_EVENT_WRITING;
    int connected = 0;
    ev_ssize_t atmost = -1;

    _bufferevent_incref_and_lock(bufev);

    if (event == EV_TIMEOUT) {
        /* Note that we only check for event==EV_TIMEOUT. If event==EV_TIMEOUT|EV_WRITE, we can safely ignore the timeout, since a read has occurred */
        what |= BEV_EVENT_TIMEOUT;
        goto error;
    }
    if (bufev_p->connecting) {
        int c = evutil_socket_finished_connecting(fd);

        /* we need to fake the error if the connection was refused immediately - usually connection to localhost on BSD */
        if (bufev_p->connection_refused) {
            bufev_p->connection_refused = 0;
            c = -1;
        }

        if (c == 0)
            goto done;

        bufev_p->connecting = 0;
        if (c < 0) {
            event_del(&bufev->ev_write);
            event_del(&bufev->ev_read);
            _bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);
            goto done;
        }
        else {
            connected = 1;
#ifdef WIN32
            if (BEV_IS_ASYNC(bufev)) {
                event_del(&bufev->ev_write);
                bufferevent_async_set_connected(bufev);
                _bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
                goto done;
            }
#endif
            _bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
            if (!(bufev->enabled & EV_WRITE) || bufev_p->write_suspended) {
                event_del(&bufev->ev_write);
                goto done;
            }
        }
    }

    atmost = _bufferevent_get_write_max(bufev_p);

    if (bufev_p->write_suspended)
        goto done;

    if (evbuffer_get_length(bufev->output)) {
        evbuffer_unfreeze(bufev->output, 1);
        res = evbuffer_write_atmost(bufev->output, fd, atmost);
        evbuffer_freeze(bufev->output, 1);
        if (res == -1) {
            int err = evutil_socket_geterror(fd);

            if (EVUTIL_ERR_RW_RETRIABLE(err))
                goto reschedule;
            what |= BEV_EVENT_ERROR;
        }
        else if (res == 0) {
            /* eof case XXXX Actually, a 0 on write doesn't indicate an EOF. An ECONNRESET might be more typical. */
            what |= BEV_EVENT_EOF;
        }
        if (res <= 0)
            goto error;

        _bufferevent_decrement_write_buckets(bufev_p, res);
    }

    if (evbuffer_get_length(bufev->output) == 0) {
        event_del(&bufev->ev_write);
    }

    /*
     * Invoke the user callback if our buffer is drained or below the
     * low watermark.
     */
    if ((res || !connected) && evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
        _bufferevent_run_writecb(bufev);
    }

    goto done;

  reschedule:
    if (evbuffer_get_length(bufev->output) == 0) {
        event_del(&bufev->ev_write);
    }
    goto done;

  error:
    bufferevent_disable(bufev, EV_WRITE);
    _bufferevent_run_eventcb(bufev, what);

  done:
    _bufferevent_decref_and_unlock(bufev);
}