libevhtp初探

libevent的evhttp不适合多线程，libevhtp重新设计了libevent的http API，采用了和memcached类似的多线程模型。

worker线程的管道读事件的回调函数为htp__run_in_thread_:

#ifndef EVHTP_DISABLE_EVTHR
static void
htp__run_in_thread_(evthr_t * thr, void * arg, void * shared)
{
    evhtp_t            * htp        = shared;
    evhtp_connection_t * connection = arg;

    connection->evbase = evthr_get_base(thr);
    connection->thread = thr;

    if (htp__connection_accept_(connection->evbase, connection) < 0)
    {
        evhtp_connection_free(connection);

        return;
    }

    if (htp__run_post_accept_(htp, connection) < 0)
    {
        evhtp_connection_free(connection);

        return;
    }
}

#endif

htp__connection_accept_函数如下:

static int
htp__connection_accept_(struct event_base * evbase, evhtp_connection_t * connection)
{
    struct timeval * c_recv_timeo;
    struct timeval * c_send_timeo;

    if (htp__run_pre_accept_(connection->htp, connection) < 0)
    {
        evutil_closesocket(connection->sock);

        return -1;
    }

#ifndef EVHTP_DISABLE_SSL
    if (connection->htp->ssl_ctx != NULL)
    {
        connection->ssl = SSL_new(connection->htp->ssl_ctx);
        connection->bev = bufferevent_openssl_socket_new(evbase,
                                                         connection->sock,
                                                         connection->ssl,
                                                         BUFFEREVENT_SSL_ACCEPTING,
                                                         connection->htp->bev_flags);
        SSL_set_app_data(connection->ssl, connection);
        goto end;
    }
#endif

    connection->bev = bufferevent_socket_new(evbase,
                                             connection->sock,
                                             connection->htp->bev_flags);

    htp_log_debug("enter sock=%d\n", connection->sock);

#ifndef EVHTP_DISABLE_SSL
end:
#endif

    if (connection->recv_timeo.tv_sec || connection->recv_timeo.tv_usec)
    {
        c_recv_timeo = &connection->recv_timeo;
    } else if (connection->htp->recv_timeo.tv_sec ||
               connection->htp->recv_timeo.tv_usec)
    {
        c_recv_timeo = &connection->htp->recv_timeo;
    } else {
        c_recv_timeo = NULL;
    }

    if (connection->send_timeo.tv_sec || connection->send_timeo.tv_usec)
    {
        c_send_timeo = &connection->send_timeo;
    } else if (connection->htp->send_timeo.tv_sec ||
               connection->htp->send_timeo.tv_usec)
    {
        c_send_timeo = &connection->htp->send_timeo;
    } else {
        c_send_timeo = NULL;
    }

    evhtp_connection_set_timeouts(connection, c_recv_timeo, c_send_timeo);

    connection->resume_ev = event_new(evbase, -1, EV_READ | EV_PERSIST,
                                      htp__connection_resumecb_, connection);
    event_add(connection->resume_ev, NULL);

    bufferevent_enable(connection->bev, EV_READ);
    bufferevent_setcb(connection->bev,
                      htp__connection_readcb_,
                      htp__connection_writecb_,
                      htp__connection_eventcb_, connection);

    return 0;
} 

此时，conn_fd的读事件已经添加到worker线程的event_base中。

 

当读事件发生后，最终会运行htp__connection_readcb_，然后调用htparser_run，再调用htp__request_parse_path_:

static int
htp__request_parse_path_(htparser * p, const char * data, size_t len)
{
    evhtp_connection_t * c = htparser_get_userdata(p);
    evhtp_path_t       * path;

    if (htp__require_uri_(c) != 0)
    {
        return -1;
    }

    if (evhtp_unlikely(!(path = htp__path_new_(data, len))))
    {
        c->request->status = EVHTP_RES_FATAL;

        return -1;
    }

    c->request->uri->path   = path;
    c->request->uri->scheme = htparser_get_scheme(p);
    c->request->method      = htparser_get_method(p);

    htp__lock_(c->htp);
    {
        htp__request_set_callbacks_(c->request);
    }
    htp__unlock_(c->htp);

    if ((c->request->status = htp__hook_path_(c->request, path)) != EVHTP_RES_OK)
    {
        return -1;
    }

    return 0;
}

通过htp__request_set_callbacks_，我们可以设置每个请求的回调函数为evhtp_s结构中的回调函数。

htparser_run后续会调用htp__request_parse_fini_:

static int
htp__request_parse_fini_(htparser * p)
{
    evhtp_connection_t * c = htparser_get_userdata(p);

    if (c->flags & EVHTP_CONN_FLAG_PAUSED)
    {
        return -1;
    }

    /* check to see if we should use the body of the request as the query
     * arguments.
     */
    if (htp__should_parse_query_body_(c->request) == 1)
    {
        const char      * body;
        size_t            body_len;
        evhtp_uri_t     * uri;
        struct evbuffer * buf_in;

        uri            = c->request->uri;
        buf_in         = c->request->buffer_in;

        body_len       = evbuffer_get_length(buf_in);
        body           = (const char *)evbuffer_pullup(buf_in, body_len);

        uri->query_raw = calloc(body_len + 1, 1);
        evhtp_alloc_assert(uri->query_raw);

        memcpy(uri->query_raw, body, body_len);

        uri->query     = evhtp_parse_query(body, body_len);
    }


    /*
     * XXX c->request should never be NULL, but we have found some path of
     * execution where this actually happens. We will check for now, but the bug
     * path needs to be tracked down.
     *
     */
    if (c->request && c->request->cb)
    {
        (c->request->cb)(c->request, c->request->cbarg);
    }

    if (c->flags & EVHTP_CONN_FLAG_PAUSED)
    {
        return -1;
    }

    return 0;
}

可以看到，最终request的回调函数被执行。