c10k-server方面的坑

最近在写 https://github.com/htfy96/c10k-server一个玩具服务器。初步是打算采用一个主线程只Accept,然后把Accept的FD放到某个Worker的Eventloop里。每个WorkerThread跑一个Eventloop监听并处理事件。

基本流程

大致流程是这样:

// MainThread
for (;;)
{
    int accept_fd = accept(mainfd, nullptr, nullptr);
    workerPool.put(accept_fd);
}

// WorkerPool::put
cur_worker = (cur_worker + 1) % WORKER_NUM;
workers[cur_worker].add_new_connection(accept_fd);

// Worker::add_new_connection
auto handler = make_shared();
auto connection = make_shared(fd, loop_internal);
connection->register();
handler->handle_connected(connection);

// Connection::register
loop.add_event(fd, ERDHHUP, bind(shared_from_this(), &Connection::handler_events), _1));

// Worker::loop
for (;;)
{
    epoll_wait(event_set, &events, -1);
    for (auto e: events)
    {
        e.callback(e); // = connection::handle_events
    }
}

// user-defined: HandleT::handle_connected
void HandlerT::handle_connected(const shared_ptr &conn)
{
    auto self = static_pointer_cast(shared_from_this());
    static const char[] MSG = "hello world";
    conn->write_async_then(begin(MSG), end(MSG), [self](const shared_ptr &conn) {
        conn->close();
    });
}

// Connection::handle_events
void Connection::handle_events(const Event &e)
{
    // LOCK_GUARD
    if (!closed())
    {
        try {
            if (e.is(POLLIN))
                handle_in();
            if (e.is(POLLOUT))
                handle_out();
            if (e.is_err())
                close();
        } catch(exception &e) {
            LOG(e.what());
            close();
        }
    }
}

// Connection::handle_out
void Connection::handle_out()
{
    while (!write_req.empty())
    {
        write_first_request(); // may throw std::system_error
        if (write_req.front().finished())
        {
            write_req.front().callback(shared_from_this());
            write_req.pop(); 
        }
        else
            break; // wait for next time
    }
    if (write_req.empty())
        stop_listen_event(POLLOUT);
}

// Connection::write_async_then
void Connection::write_async_then(InputIt st, InputIt ed, CallbackT cb)
{
    auto new_req = WriteReq(vector(st, ed), cb);
    write_req.push(new_req);
    start_listen_event(POLLOUT);
}
        

TO BE CONTINUED…

发表评论