<repo_name>pponnuvel/liburing4cpp<file_sep>/CMakeLists.txt
cmake_minimum_required(VERSION 2.8)

project(liburing-http-demo)

find_package(Boost OPTIONAL_COMPONENTS context)

set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_CXX_EXTENSIONS OFF)

option(USE_LIBAIO "Use libaio instead of liburing for old kernel support")

add_executable(${PROJECT_NAME} "main.cpp" "global.cpp")
target_link_libraries(${PROJECT_NAME} LINK_PRIVATE fmt)

if(USE_LIBAIO)
    add_compile_definitions(USE_LIBAIO=1)
    target_link_libraries(${PROJECT_NAME} LINK_PRIVATE aio)
else()
    target_link_libraries(${PROJECT_NAME} LINK_PRIVATE uring)
endif()
if(Boost_FOUND)
    include_directories(${Boost_INCLUDE_DIR})
    target_link_libraries(${PROJECT_NAME} LINK_PRIVATE ${Boost_LIBRARIES})
endif()

string(APPEND CMAKE_CXX_FLAGS_DEBUG " -fno-omit-frame-pointer -fsanitize=address -D_GLIBCXX_DEBUG")
string(APPEND CMAKE_LINKER_FLAGS_DEBUG " -fno-omit-frame-pointer -fsanitize=address")
<file_sep>/global.hpp
#pragma once
#include <unordered_map>
#include <string_view>
#include <vector>
#if !USE_LIBAIO
#   include <liburing.h>   // http://git.kernel.dk/liburing
#else
#   include <libaio.h>     // http://git.infradead.org/users/hch/libaio.git
#endif

extern const std::unordered_map<std::string_view, std::string_view> MimeDicts;

#if !USE_LIBAIO
extern io_uring ring;
#else
extern io_context_t context;
#endif
<file_sep>/coroutine.hpp
#pragma once
#include <functional>
#include <sys/timerfd.h>
#if !USE_LIBAIO
#   include <liburing.h>   // http://git.kernel.dk/liburing
#else
#   include <libaio.h>     // http://git.infradead.org/users/hch/libaio.git
#endif
#include <sys/poll.h>
#include <fmt/format.h> // https://github.com/fmtlib/fmt

#include "yield.hpp"
#include "global.hpp"

// 填充 iovec 结构体
constexpr inline iovec to_iov(void *buf, size_t size) {
    return { buf, size };
}
constexpr inline iovec to_iov(std::string_view sv) {
    return to_iov(const_cast<char *>(sv.data()), sv.size());
}
template <size_t N>
constexpr inline iovec to_iov(std::array<char, N>& array) {
    return to_iov(array.data(), array.size());
}

template <typename Fn>
struct on_scope_exit {
    on_scope_exit(Fn &&fn): _fn(std::move(fn)) {}
    ~on_scope_exit() { this->_fn(); }

private:
    Fn _fn;
};

[[noreturn]]
void panic(std::string_view sv, int err = 0) { // 简单起见，把错误直接转化成异常抛出，终止程序
    if (err == 0) err = errno;
    fmt::print(stderr, "errno: {}\n", err);
    if (err == EPIPE) {
        throw std::runtime_error("Broken pipe: client socket is closed");
    }
    throw std::system_error(err, std::generic_category(), sv.data());
}

class Coroutine: public FiberSpace::Fiber<int, std::function<void ()>> {
public:
    using BaseType = FiberSpace::Fiber<int, std::function<void ()>>;
    using FuncType = BaseType::FuncType;

public:
    static int runningCoroutines;

    explicit Coroutine(FuncType f): BaseType(std::move(f)) {
        ++Coroutine::runningCoroutines;
        this->next(); // We have to make sure that this->yield() is executed at least once.
    }

    explicit Coroutine(FuncType f, std::function<void ()> cleanup): BaseType(std::move(f)) {
        this->localData = std::move(cleanup);
        ++Coroutine::runningCoroutines;
        this->next();
    }

    ~Coroutine() {
        --Coroutine::runningCoroutines;
        if (this->localData) this->localData();
    }

public:

// 异步读操作，不使用缓冲区
#if !USE_LIBAIO
#define DEFINE_AWAIT_OP(operation) \
template <unsigned int N> \
int await_##operation (int fd, iovec (&&ioves) [N], off_t offset = 0) { \
    auto* sqe = io_uring_get_sqe(&ring); \
    assert(sqe && "sqe should not be NULL"); \
    io_uring_prep_##operation (sqe, fd, ioves, N, offset); \
    io_uring_sqe_set_data(sqe, this); \
    io_uring_submit(&ring); \
    this->yield(); \
    int res = this->current().value(); \
    if (res < 0) panic(#operation, -res); \
    return res; \
}
#else
#define DEFINE_AWAIT_OP(operation) \
template <unsigned int N> \
int await_##operation (int fd, iovec (&&ioves) [N], off_t offset = 0) { \
    iocb ioq, *pioq = &ioq; \
    io_prep_p##operation(&ioq, fd, ioves, N, offset); \
    ioq.data = this; \
    io_submit(context, 1, &pioq); \
    this->yield(); \
    int res = this->current().value(); \
    if (res < 0) panic(#operation, -res); \
    return res; \
}
#endif
    DEFINE_AWAIT_OP(readv)
    DEFINE_AWAIT_OP(writev)
#undef DEFINE_AWAIT_OP

#if !USE_LIBAIO
#define DEFINE_AWAIT_OP(operation) \
template <unsigned int N> \
int await_##operation(int sockfd, iovec (&&ioves) [N], uint32_t flags) { \
    msghdr msg = { \
        .msg_iov = ioves, \
        .msg_iovlen = N, \
    }; \
    auto* sqe = io_uring_get_sqe(&ring); \
    io_uring_prep_##operation(sqe, sockfd, &msg, flags); \
    io_uring_sqe_set_data(sqe, this); \
    io_uring_submit(&ring); \
    this->yield(); \
    int res = this->current().value(); \
    if (res < 0) panic(#operation, -res); \
    return res; \
}

    DEFINE_AWAIT_OP(recvmsg)
    DEFINE_AWAIT_OP(sendmsg)
#else
template <unsigned int N>
int await_recvmsg(int sockfd, iovec (&&ioves) [N], uint32_t flags) {
    return this->await_readv(sockfd, std::move(ioves));
}

template <unsigned int N>
int await_sendmsg(int sockfd, iovec (&&ioves) [N], uint32_t flags) {
    return this->await_writev(sockfd, std::move(ioves));
}
#endif

    int await_poll(int fd) {
#if !USE_LIBAIO
        auto* sqe = io_uring_get_sqe(&ring);
        assert(sqe && "sqe should not be NULL");
        io_uring_prep_poll_add(sqe, fd, POLLIN);
        io_uring_sqe_set_data(sqe, this);
        io_uring_submit(&ring);
#else
        iocb ioq, *pioq = &ioq;
        io_prep_poll(&ioq, fd, POLLIN);
        ioq.data = this;
        io_submit(context, 1, &pioq);
#endif
        this->yield();
        int res = this->current().value();
        if (res < 0) panic("poll", -res);
        return res;
    }

    // 把控制权交给其他协程
    int yield_execution() {
#if !USE_LIBAIO
        auto* sqe = io_uring_get_sqe(&ring);
        assert(sqe && "sqe should not be NULL");
        io_uring_prep_nop(sqe);
        io_uring_sqe_set_data(sqe, this);
        io_uring_submit(&ring);
#else
        iocb ioq, *pioq = &ioq;
        memset(&ioq, 0, sizeof(ioq));
        ioq.aio_lio_opcode = IO_CMD_NOOP;
        ioq.data = this;
        io_submit(context, 1, &pioq);
#endif
        this->yield();
        return this->current().value();
    }

    int delay(int second) {
        itimerspec exp = { {}, { second, 0 } };
        auto tfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
        if (timerfd_settime(tfd, 0, &exp, nullptr)) panic("timerfd");
        on_scope_exit closefd([=]() { close(tfd); });
        return this->await_poll(tfd);
    }

public:
    static std::pair<Coroutine *, int> wait_for_event() {
        // 获取已完成的IO事件
#if !USE_LIBAIO
        io_uring_cqe* cqe;
        Coroutine *coro;

        do {
            if (io_uring_wait_cqe(&ring, &cqe)) panic("wait_cqe");
            io_uring_cqe_seen(&ring, cqe);
        } while ((coro = static_cast<Coroutine *>(io_uring_cqe_get_data(cqe))));

        auto res = cqe->res;
#else
        io_event event;
        io_getevents(context, 1, 1, &event, nullptr);

        auto* coro = static_cast<Coroutine *>(event.data);
        auto res = event.res;
#endif
        return { coro, res };
    }

    static std::optional<std::pair<Coroutine *, int>> peek_a_event() {
#if !USE_LIBAIO
        io_uring_cqe* cqe;
        while (io_uring_peek_cqe(&ring, &cqe) >= 0 && cqe) {
            io_uring_cqe_seen(&ring, cqe);

            if (auto* coro = static_cast<Coroutine *>(io_uring_cqe_get_data(cqe))) {
                return std::make_pair(coro, cqe->res);
            }
        }
        return std::nullopt;
#else
        return timedwait_for_event({0, 0});
#endif
    }

    static std::optional<std::pair<Coroutine *, int>> timedwait_for_event(timespec timeout) {
#if !USE_LIBAIO
        // io_uring doesn't support timed wait (yet), simulate it by polling a timer

        // First clear all cqes, to make sure there's no timer-poll completion event in queue
        if (auto result = Coroutine::peek_a_event()) {
            return result;
        }

        itimerspec exp = { {}, timeout }, old;
        static const int tfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);

        if (timerfd_settime(tfd, 0, &exp, &old)) panic("timerfd");

        // See if we can reuse the existing timer-poll request
        if (old.it_value.tv_sec == 0 && old.it_value.tv_nsec == 0) {
            // The old timer has expired, restart the poll request
            auto* sqe = io_uring_get_sqe(&ring);
            io_uring_prep_poll_add(sqe, tfd, POLLIN);
            // io_uring_sqe_set_data(sqe, nullptr);
            if (io_uring_submit_and_wait(&ring, 1) < 0) panic("io_uring_submit_and_wait");
        }

        io_uring_cqe* cqe;
        if (io_uring_wait_cqe(&ring, &cqe) < 0) panic("io_uring_wait_cqe");
        io_uring_cqe_seen(&ring, cqe);

        if (auto* coro = static_cast<Coroutine *>(io_uring_cqe_get_data(cqe))) {
            // We don't cancel the running time-poll request, it will be handled in next call
            return std::make_pair(coro, cqe->res);
        } else {
            return std::nullopt;
        }
#else
        io_event event;
        if (io_getevents(context, 1, 1, &event, &timeout)) {
            return std::make_pair(static_cast<Coroutine *>(event.data), event.res);
        } else {
            return std::nullopt;
        }
#endif
    }
};
int Coroutine::runningCoroutines = 0;
<file_sep>/README.md
# liburing-http-demo

Simple http server demo using [liburing](http://kernel.dk/io_uring.pdf) and [Cxx-yield](https://github.com/CarterLi/Cxx-yield/)

Tested on `Linux Ubuntu 5.3.0-10-generic #11-Ubuntu SMP Mon Sep 9 15:12:17 UTC 2019 x86_64 x86_64 x86_64 GNU/Linux`

## Dependencies

* [liburing](http://git.kernel.dk/liburing) ( requires Linux 5.1 or later )
* [libaio](http://git.infradead.org/users/hch/libaio.git) ( fallback, requires Linux 4.18 or later )
* [fmt](https://github.com/fmtlib/fmt)
* [boost::context](https://boost.org) ( optional )
* CMake ( build )
* Compiler that supports C++17

## Introduction ( Chinese )

* https://segmentfault.com/a/1190000019300089
* https://segmentfault.com/a/1190000019361819

## More info

https://github.com/libuv/libuv/issues/1947

## License

Public domain
<file_sep>/main.cpp
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <string>
#include <string_view>
#include <chrono>

#include "coroutine.hpp"

enum {
    SERVER_PORT = 8080,
    BUF_SIZE = 1024,
};

using namespace std::literals;

// 一些预定义的错误返回体
static constexpr const auto http_404_hdr = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"sv;
static constexpr const auto http_400_hdr = "HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n"sv;

// 解析到HTTP请求的文件后，发送本地文件系统中的文件
void http_send_file(Coroutine& coro, std::string filename, int clientfd, int dirfd) {
    if (filename == "./") filename = "./index.html";

    // 尝试打开待发送文件
    const auto infd = openat(dirfd, filename.c_str(), O_RDONLY);
    on_scope_exit closefd([=]() { close(infd); });

    if (struct stat st; infd < 0 || fstat(infd, &st) || !S_ISREG(st.st_mode)) {
        // 文件未找到情况下发送404 error响应
        fmt::print("{}: file not found!\n", filename);
        coro.await_sendmsg(clientfd, { to_iov(http_404_hdr) }, MSG_NOSIGNAL);
    } else {
        auto contentType = [filename_view = std::string_view(filename)]() {
            auto extension = filename_view.substr(filename_view.find_last_of('.') + 1);
            auto iter = MimeDicts.find(extension);
            if (iter == MimeDicts.end()) return "application/octet-stream"sv;
            return iter->second;
        }();

        // 发送响应头
        coro.await_sendmsg(clientfd, {
            to_iov(fmt::format("HTTP/1.1 200 OK\r\nContent-type: {}\r\nContent-Length: {}\r\n\r\n", contentType, st.st_size)),
        }, MSG_NOSIGNAL | MSG_MORE);

        off_t offset = 0;
        std::array<char, BUF_SIZE> filebuf;
        auto iov = to_iov(filebuf);
        for (; st.st_size - offset > BUF_SIZE; offset += BUF_SIZE) {
            coro.await_readv(infd, { iov }, offset);
            coro.await_sendmsg(clientfd, { iov }, MSG_NOSIGNAL | MSG_MORE);
//            coro.delay(1); // For debugging
        }
        if (st.st_size > offset) {
            iov.iov_len = size_t(st.st_size - offset);
            coro.await_readv(infd, { iov }, offset);
            coro.await_sendmsg(clientfd, { iov }, MSG_NOSIGNAL);
        }
    }
}

// HTTP请求解析
void serve(Coroutine& coro, int clientfd, int dirfd) {
    fmt::print("Serving connection, sockfd {}; number of running coroutines: {}\n",
         clientfd, Coroutine::runningCoroutines);

    std::string_view buf_view;
    std::array<char, BUF_SIZE> buffer;
    int res = coro.await_recvmsg(clientfd, { to_iov(buffer) }, MSG_NOSIGNAL);
    buf_view = std::string_view(buffer.data(), size_t(res));

    // 这里我们只处理GET请求
    if (buf_view.compare(0, 3, "GET") == 0) {
        // 获取请求的path
        auto file = "."s += buf_view.substr(4, buf_view.find(' ', 4) - 4);
        fmt::print("received request {} with sockfd {}\n", file, clientfd);
        http_send_file(coro, file, clientfd, dirfd);
    } else {
        // 其他HTTP请求处理，如POST，HEAD等，返回400错误
        fmt::print("unsupported request: {}\n", buf_view);
        coro.await_sendmsg(clientfd, { to_iov(http_400_hdr) }, MSG_NOSIGNAL);
    }
}

void accept_connection(Coroutine& coro, int serverfd, int dirfd) {
    while (coro.await_poll(serverfd)) {
        int clientfd = accept(serverfd, nullptr, nullptr);

        // 新建新协程处理请求
        new Coroutine(
            [=](auto& coro) {
                serve(static_cast<Coroutine &>(coro), clientfd, dirfd);
            },
            [=, start = std::chrono::high_resolution_clock::now()] () {
                // 请求结束时清理资源
                close(clientfd);
                fmt::print("sockfd {} is closed, time used {}\n",
                    clientfd,
                    (std::chrono::high_resolution_clock::now() - start).count());
            }
        );
    }
}

int main(int argc, char* argv[]) {
    if (argc != 2) {
        fmt::print("Usage: {} <ROOT_DIR>\n", argv[0]);
        return 1;
    }

    int dirfd = open(argv[1], O_DIRECTORY);
    if (dirfd < 0) panic("open dir");
    on_scope_exit closedir([=]() { close(dirfd); });

    // 初始化内核支持的原生异步IO操作实现
#if !USE_LIBAIO
    if (io_uring_queue_init(32, &ring, 0)) panic("queue_init");
    on_scope_exit closerg([&]() { io_uring_queue_exit(&ring); });
#else
    if (io_queue_init(32, &context)) panic("queue_init");
    on_scope_exit closerg([&]() { io_queue_release(context); });
#endif

    // 建立TCP套接字
    int sockfd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
    if (sockfd < 0) panic("socket creation");
    on_scope_exit closesock([=]() { close(sockfd); });

    // 设置允许端口重用
    if (int on = 1; setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on))) panic("SO_REUSEADDR");
    if (int on = 1; setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(on))) panic("SO_REUSEPORT");

    // 绑定端口
    if (sockaddr_in addr = {
        .sin_family = AF_INET,
        // 这里要注意，端口号一定要使用htons先转化为网络字节序，否则绑定的实际端口可能和你需要的不同
        .sin_port = htons(SERVER_PORT),
        .sin_addr = { INADDR_ANY },
        .sin_zero = {}, // 消除编译器警告
    }; bind(sockfd, reinterpret_cast<sockaddr *>(&addr), sizeof (sockaddr_in))) panic("socket binding");

    // 监听端口
    if (listen(sockfd, 128)) panic("listen");
    fmt::print("Listening: {}\n", SERVER_PORT);

    new Coroutine(
        [=](auto& coro) { accept_connection(static_cast<Coroutine &>(coro), sockfd, dirfd); }
    );

    // 事件循环
    while (Coroutine::runningCoroutines) {
        // 获取已完成的IO事件
        if (auto some = Coroutine::timedwait_for_event({1, 0})) {
            auto [coro, res] = some.value();

            // 有已完成的事件，回到协程继续
            try {
                if (!coro->next(res)) delete coro;
            } catch (std::runtime_error& e) {
                fmt::print("{}\n", e.what());
                delete coro;
            }
        } else {
            fmt::print("PING!\n");
        }
    }
}
