{"id":357885,"date":"2024-05-21T00:00:04","date_gmt":"2024-05-21T00:00:04","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=357885"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-05-21T00:00:04","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=357885","title":{"rendered":"High-performance network library using C++20 coroutines<\/span>"},"content":{"rendered":"
<\/div>\n
\n
\n
\n
\n
<\/figure>\n

Asynchronous programming is commonly employed for efficient implementation of network interactions in C++. The essence of this approach lies in the fact that the results of socket read\/write functions are not immediately available but become accessible after some time. This approach allows for loading the processor with useful work during the wait for data. Various implementations of this approach exist, such as callbacks<\/a>, actors<\/a>, future\/promise<\/a>, coroutines<\/a>. In C++, these implementations are available as libraries from third-party developers or can be implemented independently.<\/p>\n

Coroutines are the most challenging to implement as they require writing platform-dependent code. However, the recent version of the C++ 20 language standard introduces support for coroutines at the compiler and standard library levels. Coroutines are functions that can suspend their execution, preserving their state, and later return to that state to resume the function’s work. The compiler automatically creates a checkpoint with the coroutine’s state.<\/p>\n

For a comprehensive understanding of C++ 20 coroutines, refer to this article. Below, we examine a code example using coroutines and describe important points applied during implementation.<\/p>\n

For a complete understanding and assimilation of the material presented in this article, it is necessary for the reader to possess basic knowledge in the field of sockets and the TCP\/IP protocol. Fundamental concepts, such as mechanisms of interaction through sockets, principles of data transmission via the TCP\/IP protocol, as well as an understanding of key terms in this domain, will be crucial for the successful comprehension of the provided information. GeeksforGeeks offers useful tutorials on TCP Server-Client Implementation in C++<\/a> and Socket Programming in C\/C++<\/a>, which can serve as valuable resources for enhancing your understanding of these foundational concepts.<\/p>\n

The code of the library discussed below, along with usage examples, can be downloaded from the following link: https:\/\/github.com\/resetius\/coroio<\/a>.<\/p>\n

Contents<\/h3>\n
    \n
  1. \n

    Network Interaction: Echo Client and Server Example<\/a><\/p>\n<\/li>\n

  2. \n

    Echo Client and Server Code Overview<\/a><\/p>\n<\/li>\n

  3. \n

    promise and coroutine_handle<\/a><\/p>\n<\/li>\n

  4. \n

    Using final_suspend in Unit Tests<\/a><\/p>\n<\/li>\n

  5. \n

    Event Handling Using select<\/a><\/p>\n<\/li>\n

  6. \n

    Performance and benchmarks<\/a><\/p>\n<\/li>\n<\/ol>\n\n

    <\/a><\/p>\n

    Network Interaction: Echo Client and Server Example<\/h3>\n

    Let’s examine a classic example of network interaction\u2014the implementation of an echo client and server. This example illustrates the task of implementing network programs and effectively demonstrates the details of a specific approach used for asynchronous programming. The echo client\/server example is often included in the documentation of third-party libraries.<\/p>\n

    The echo server receives a message from the client and immediately sends it back unchanged as a response. The echo client sends a message to the server and waits for a response. Here’s an example of the process.<\/p>\n

    This is a client session:<\/p>\n

    $ .\/echoclient test Received: test  message Received: message  yet another message Received: yet another message <\/code><\/pre>\n
    This is a server session:<\/p>\n
    $ .\/echoserver Received: test  Received: message  Received: yet another message <\/code><\/pre>\n
    <\/a><\/p>\n
    Echo Client and Server Code Overview<\/h3>\n
    Let’s examine the Echo client and server code from top to bottom. We’ll start
     by looking at the overall code structure, then move on to exploring the C++
     coroutine API, and finally, we’ll descend to the POSIX API level for socket
     operations. It’s worth noting that no third-party libraries are used in this
     implementation. Only the standard C++ library and the POSIX API are employed.<\/p>\n
    The Echo client reads from standard input, sends the input to the socket,
     and waits for a response:<\/p>\n
    TSimpleTask client(TLoop* loop) {     char out[128] = {0};     char in[128] = {0};     ssize_t size = 1;      try {         TSocket input{TAddress{}, 0 \/* stdin *\/, loop->Poller()};         TSocket socket{TAddress{\"127.0.0.1\", 8888}, loop->Poller()};          co_await socket.Connect();         while (size && (size = co_await input.ReadSome(out, sizeof(out)))) {             co_await socket.WriteSome(out, size);             size = co_await socket.ReadSome(in, sizeof(in));             std::cout << \"Received: \" << std::string_view(in, size) << \"\\n\";         }     } catch (const std::exception& ex) {         std::cout << \"Exception: \" << ex.what() << \"\\n\";     }     loop->Stop();     co_return; }  int main(int argc, char** argv) {     TLoop loop;     client(&loop);     loop.Loop();     return 0; } <\/code><\/pre>\n
    The Echo server listens on a port, accepts incoming connections (calls
     accept<\/code>), and for each connection, a separate coroutine is created. This
     coroutine reads data from the socket and immediately sends the read data back
     to the socket:<\/p>\n
    TSimpleTask client_handler(TSocket socket, TLoop* loop) {     char buffer[128] = {0}; ssize_t size = 0;      try {         while ((size = co_await socket.ReadSome(buffer, sizeof(buffer))) > 0) {             std::cerr << \"Received: \" << std::string_view(buffer, size) << \"\\n\";             co_await socket.WriteSome(buffer, size);         }     } catch (const std::exception& ex) {         std::cerr << \"Exception: \" << ex.what() << \"\\n\";     }     co_return; }  TSimpleTask server(TLoop* loop) {     TSocket socket(TAddress{\"0.0.0.0\", 8888}, loop->Poller());     socket.Bind();     socket.Listen();      while (true) {         auto client = co_await socket.Accept();         client_handler(std::move(client), loop);     }     co_return; }  int main(int argc, char** argv) {     TLoop loop;     server(&loop);     loop.Loop();     return 0; } <\/code><\/pre>\n
    The client and server share a similar structure: the main<\/code> function starts an
     event handling loop (TLoop<\/code>) and launches a coroutine. The event handling loop
     interacts with an object (TPoller<\/code>), which performs input-output multiplexing.
     In this implementation, input-output multiplexing is achieved using the select<\/code>
     mechanism, and changing it to poll<\/code>, epoll<\/code>, kqueue<\/code>, etc., should not pose a
     problem. TPoller<\/code> waits for events on the socket or a timeout, passes these
     events to the TLoop<\/code> loop, which, in turn, wakes up the coroutines subscribed to
     these events.<\/p>\n
    The following diagram illustrates how the program switches the execution flow between the
     client<\/code> coroutine and the Loop.loop()<\/code> function when the coroutine sleeps on
     each co_await<\/code> operator call (in reality, it may not sleep every time;
     details below).<\/p>\n
    \"coroutine<\/p>\n
    coroutine suspend\/resume<\/figcaption><\/div>\n<\/figure>\n
    Below is a diagram of the main classes implemented in our library. TPoller<\/code> has several implementations (select<\/code>, poll<\/code>, epoll<\/code>, kqueue<\/code>, uring<\/code>). TLoop<\/code> contains TPoller<\/code> — its responsibility is to initiate the Poll<\/code>. TSocket<\/code> contains a reference to TPoller<\/code> and adds events to it when ReadSome<\/code>, WriteSome<\/code>, Connect<\/code>, etc., are called.<\/p>\n
    \"class<\/p>\n
    class diagram<\/figcaption><\/div>\n<\/figure>\n
    Before moving on to the description of the main components, let’s take a look at the coroutine API in C++20.<\/p>\n
    <\/a><\/p>\n
    Objects promise and coroutine_handle<\/h3>\n
    Each coroutine is associated with two entities: promise<\/code> and coroutine_handle<\/code>.
     coroutine_handle<\/code> is necessary for managing the coroutine externally; for
     example, it can be used to resume (resume<\/code>) the coroutine, destroy (destroy<\/code>)
     it, or check if the coroutine is done (done<\/code>). With the promise<\/code> object, you
     can define the behavior of the coroutine at the start and end of its
     execution and manage the coroutine from within. It’s important to note that the
     methods of the promise<\/code> object should not be called directly; instead, the
     co_return<\/code> and co_yield<\/code> operators are used, and the compiler substitutes the
     appropriate methods.<\/p>\n
    In the example, the following pair of promise<\/code>\/coroutine_handle<\/code> is used
     (TVoidPromise<\/code>\/TSimpleTask<\/code>):<\/p>\n
    struct TVoidPromise;  struct TSimpleTask : std::coroutine_handle<TVoidPromise> {     using promise_type = TVoidPromise; };  struct TVoidPromise {     TSimpleTask get_return_object() {        return { TSimpleTask::from_promise(*this) };      }     std::suspend_never initial_suspend() { return {}; }     std::suspend_never final_suspend() noexcept { return {}; }     void return_void() {}     void unhandled_exception() {} }; <\/code><\/pre>\n
    Here, the coroutine doesn’t return anything (return_void<\/code>), starts
     immediately (std::suspend_never initial_suspend<\/code>), and does not sleep upon
     completion (std::suspend_never final_suspend<\/code>).<\/p>\n
    In some cases, such as when implementing unit tests, it is convenient to sleep
     the coroutine upon completion:<\/p>\n
    struct TTestPromise;  struct TTestTask : std::coroutine_handle<TTestPromise> {     using promise_type = TTestPromise; };  struct TTestPromise {     TTestTask get_return_object() {        return { TTestTask::from_promise(*this) };      }     std::suspend_never initial_suspend() { return {}; }     std::suspend_always \/* the difference here *\/ final_suspend() noexcept      { return {}; }     void return_void() {}     void unhandled_exception() {} }; <\/code><\/pre>\n
    Please note that in this case, the promise<\/code> differs only in the signature of
     the final_suspend<\/code> method.<\/p>\n
    <\/a><\/p>\n
    Using final_suspend in Unit Tests<\/h3>\n
    Let’s consider a unit test that launches client and server coroutines and then
     checks the correctness of the message exchange:<\/p>\n
    void test_write_after_connect(void**) {     TLoop loop;     TSocket socket(TAddress{\"127.0.0.1\", 8898}, loop.Poller());     socket.Bind();     socket.Listen();     char send_buf[128] = \"Hello\";     char rcv_buf[128] = {0};      TTestTask h1 = [](TLoop* loop, char* buf, int size) -> TTestTask     {         TSocket client(TAddress{\"127.0.0.1\", 8898}, loop->Poller());         co_await client.Connect();         co_await client.WriteSome(buf, size);         co_return;     }(&loop, send_buf, sizeof(send_buf));      TTestTask h2 = [](TSocket* socket, char* buf, int size) -> TTestTask     {         TSocket clientSocket = std::move(co_await socket->Accept());         co_await clientSocket.ReadSome(buf, size);         co_return;     }(&socket, rcv_buf, sizeof(rcv_buf));      while (!(h1.done() && h2.done())) {         loop.Step();     }     h1.destroy(); h2.destroy();      assert_true(memcmp(&send_buf, &rcv_buf, sizeof(send_buf))==0); } <\/code><\/pre>\n
    In this case, the event handling loop runs until both coroutines have
     completed. The done<\/code> method checks whether the coroutine has finished.
     However, this method can only be called if the coroutine is «asleep,» which is
     why the promise<\/code> with the std::suspend_always final_suspend<\/code> signature is used
     here.<\/p>\n
    Before moving on to the implementation of the multiplexer and asynchronous
     socket operations, let’s describe the main classes, structures, and functions
     used in the program.<\/p>\n
    Basic Structures, Classes, and Functions<\/h3>\n
    For convenience, the following using<\/code> statements are employed for std::chrono<\/code>
     (clock-related operations) and std::coroutine_handle<><\/code> objects:<\/p>\n
    using TClock = std::chrono::steady_clock; using TTime = TClock::time_point; using THandle = std::coroutine_handle<>; <\/code><\/pre>\n
    The low-level structure for working with the network address sockaddr_in<\/code> is encapsulated within the following class:<\/p>\n
    class TAddress { public:     TAddress(const std::string& addr, int port);     TAddress(sockaddr_in addr); }; <\/code><\/pre>\n
    The POSIX API uses the timeval<\/code> structure to specify a timeout. We need a
     function that converts a time interval specified by two time values into the
     timeval<\/code> structure:<\/p>\n
    timeval GetTimeval(TTime now, TTime deadline); <\/code><\/pre>\n
    Implementations of TAddress<\/code> and GetTimeval<\/code> are straightforward, and I do
     not provide them here. The code can be viewed     here<\/a>.<\/p>\n
    To work with timers and socket events, the TTimer<\/code> and TEvent<\/code> structures are used:<\/p>\n
    struct TTimer {     TTime Deadline;     int Fd;     THandle Handle;     bool operator<(const TTimer& e) const {         return std::tuple(Deadline, Fd, !static_cast<bool>(Handle))              < std::tuple(e.Deadline, e.Fd, !static_cast<bool>(e.Handle));     } };  struct TEvent {     int Fd;     enum {         READ = 1,         WRITE = 2     };     int Type;     THandle Handle;      bool Match(const TEvent& other) const {         return Fd == other.Fd && (Type & other.Type);     } }; <\/code><\/pre>\n
    TTimer<\/code> will be used to implement timeouts on the socket (e.g., connection
     timeout) and for the implementation of the asynchronous Sleep<\/code> function.
     Objects of type TTimer<\/code> will be stored in a priority queue ordered by
     Deadline<\/code>. When the Deadline<\/code> arrives, the Handle<\/code> coroutine is
     awakened. Also, note the comparison function\u2014timers with the same
     Deadline<\/code> and Fd<\/code>, but with an empty Handle<\/code> field, come before
     timers with a filled Handle<\/code> field. This is needed for timer removal.
     Instead of deleting, we will insert a timer with an empty Handle<\/code> into the
     priority queue. During the priority queue processing, we will use this to skip
     removed timers.<\/p>\n
    TEvent<\/code> describes an event on the socket. The event type is specified by
     the Type<\/code> field. An event can be either a read event, a write event, or
     both. When the event occurs, the Handle<\/code> coroutine is awakened.<\/p>\n
    Using these structures, we will implement an event multiplexer based on select<\/code>.<\/p>\n
    <\/a><\/p>\n
    Event Handling Using select<\/h3>\n
    First, let’s describe the class fields. For each socket, we need to store
     THandle<\/code>, which is awakened on read\/write. We will store them in the
     THandlePair<\/code> structure, which consists of the pair THandle Read<\/code> \/
     THandle Write<\/code>. We will also store a priority queue for handling timeouts.
     This queue stores structures of type TTimer<\/code>. After select<\/code> wakes up,
     some coroutines need to be awakened, and for some, we need to send them back to
     select<\/code>. For these purposes, we will maintain two vectors: Changes_<\/code>
     and ReadyEvents_<\/code>. The class also stores a pair of utility fields used in
     the select<\/code> call.<\/p>\n
    class TSelect { private:     int MaxFd_ = 0; \/\/ The maximum file descriptor number passed to select     \/\/ Events that have changed (coroutine awakened for them)      \/\/  and need to be sent back to select     std::vector<TEvent> Changes_;      \/\/ Events for which a coroutine needs to be awakened     std::vector<TEvent> ReadyEvents_;      std::priority_queue<TTimer> Timers_;     TTime LastTimersProcessTime_; \/\/ The time when timers were last processed      std::vector<THandlePair> InEvents_; \/\/ Read\/write events sent to select     std::vector<fd_mask> ReadFds_; \/\/ Auxiliary array for select     std::vector<fd_mask> WriteFds_; \/\/ Auxiliary array for select <\/code><\/pre>\n
    Now let’s look at the method that prepares arguments for the select<\/code> call,
     invokes select<\/code>, and populates the vector with coroutines to be awakened.<\/p>\n
        fd_set* ReadFds() {         return reinterpret_cast<fd_set*>(&ReadFds_[0]);     }     fd_set* WriteFds() {         return reinterpret_cast<fd_set*>(&WriteFds_[0]);     }  public:     void Poll() {         auto deadline = Timers_.empty() ? TTime::max() : Timers_.top().Deadline;         auto tv = GetTimeval(TClock::now(), deadline);          constexpr int bits = sizeof(fd_mask)*8;          if (InEvents_.size() <= MaxFd_) {             InEvents_.resize(MaxFd_+1);         }         if (MaxFd_ >= ReadFds_.size()*bits) {             ReadFds_.resize((MaxFd_+bits)\/bits);             WriteFds_.resize((MaxFd_+bits)\/bits);         }          for (const auto& ch : Changes_) {             int fd = ch.Fd;             auto& ev = InEvents_[fd];             if (ch.Handle) {                 if (ch.Type & TEvent::READ) {                     FD_SET(fd, ReadFds()); ev.Read = ch.Handle;                 }                 if (ch.Type & TEvent::WRITE) {                     FD_SET(fd, WriteFds()); ev.Write = ch.Handle;                 }             } else {                 if (ch.Type & TEvent::READ) {                     FD_CLR(fd, ReadFds()); ev.Read = {};                 }                 if (ch.Type & TEvent::WRITE) {                     FD_CLR(fd, WriteFds()); ev.Write = {};                 }             }         }          ReadyEvents_.clear(); Changes_.clear(); MaxFd_ = 0;          if (select(InEvents_.size(), ReadFds(), WriteFds(), nullptr, &tv) < 0) {             throw std::system_error(errno, std::generic_category(), \"select\");         }  <\/code><\/pre>\n
    Reading\/writing events on the socket are indicated in select<\/code> by setting bits
     in the bit arrays ReadFds_<\/code> and WriteFds_<\/code>. Therefore, we must iterate
     through all the changed events and set or clear bits for them in ReadFds_<\/code> and
     WriteFds_<\/code>. The occurrence of events like Timeout<\/code> is determined by setting
     the last argument of select<\/code>. The tv<\/code> argument is set so that select<\/code> wakes
     up when the Deadline<\/code> of the first TTimer<\/code> in the priority queue is reached.<\/p>\n
    After waking up from select<\/code>, we iterate through the array of events and
     check the read\/write bit in the bit arrays ReadFds_<\/code> and WriteFds_<\/code>. If the
     corresponding bit is set, we place the event in the ReadyEvents_<\/code> array. We
     will traverse this array later and awaken all the coroutines in it.<\/p>\n
            for (int k=0; k < static_cast<int>(InEvents_.size()); ++k) {             auto ev = InEvents_[k];              if (FD_ISSET(k, WriteFds())) {                 assert(ev.Write);                 ReadyEvents_.emplace_back(TEvent{k, TEvent::WRITE, ev.Write});             } else if (ev.Write) {                 \/\/ fd was cleared by select, set it                 FD_SET(k, WriteFds());             }             if (FD_ISSET(k, ReadFds())) {                 assert(ev.Read);                 ReadyEvents_.emplace_back(TEvent{k, TEvent::READ, ev.Read});             } else if (ev.Read) {                 \/\/ fd was cleared by select, set it                 FD_SET(k, ReadFds());             }         }          ProcessTimers();     } <\/code><\/pre>\n
    Now let’s consider the processing of timers:<\/p>\n
        void ProcessTimers() {         auto now = TClock::now();         int prevFd = -1;         while (!Timers_.empty()&&Timers_.top().Deadline <= now) {             TTimer timer = std::move(Timers_.top());              if ((prevFd == -1 || prevFd != timer.Fd) && timer.Handle) {                 ReadyEvents_.emplace_back(TEvent{-1, 0, timer.Handle});             }              prevFd = timer.Fd;             Timers_.pop();         }         LastTimersProcessTime_ = now;     } <\/code><\/pre>\n
    We iterate through the priority queue, and if the Deadline<\/code> has arrived, we
     place the event in ReadyEvents_<\/code>. It’s important to consider that the timer
     might have been removed; in this case, we need to skip it and not place it in
     ReadyEvents_<\/code>.<\/p>\n
    Adding read\/write events and timers is done trivially:<\/p>\n
        void AddTimer(int fd, TTime deadline, THandle h) {         Timers_.emplace(TTimer{deadline, fd, h});     }      void AddRead(int fd, THandle h) {         MaxFd_ = std::max(MaxFd_, fd);         Changes_.emplace_back(TEvent{fd, TEvent::READ, h});     }      void AddWrite(int fd, THandle h) {         MaxFd_ = std::max(MaxFd_, fd);         Changes_.emplace_back(TEvent{fd, TEvent::WRITE, h});     } <\/code><\/pre>\n
    Removing an event is accomplished by inserting an empty event into Changes_<\/code>:<\/p>\n
        void RemoveEvent(int fd) {         MaxFd_ = std::max(MaxFd_, fd);         Changes_.emplace_back(TEvent{fd, TEvent::READ|TEvent::WRITE, {}});     } <\/code><\/pre>\n
    Removing a timer is done by inserting an empty timer into Timers_<\/code>:<\/p>\n
        bool RemoveTimer(int fd, TTime deadline) {         bool fired = deadline < LastTimersProcessTime_;         if (!fired) {             Timers_.emplace(TTimer{deadline, fd, {}});         }         return fired;     } <\/code><\/pre>\n
    Now let’s consider the awakening of coroutines:<\/p>\n
        void Wakeup(TEvent&& change) {         change.Handle.resume();         if (Changes_.empty() || !Changes_.back().Match(change)) {             if (change.Fd >= 0) {                 change.Handle = {};                 Changes_.emplace_back(std::move(change));             }         }     }      void WakeupReadyHandles() {         int i = 0;         for (auto&& ev : ReadyEvents_) {             Wakeup(std::move(ev));         }     } }; <\/code><\/pre>\n
    The Wakeup<\/code> method awakens a coroutine. After awakening and execution, the
     coroutine might fall asleep again, waiting for the same descriptor on which we
     previously woke up, or it might fall asleep on another descriptor. If the
     coroutine falls asleep on a different descriptor, the one that triggered the
     awakening must be removed from select<\/code>. This check and removal are performed
     in the condition if (Changes_.empty() ...)<\/code>. The WakeupReadyHandles<\/code> method
     traverses all events in ReadyEvents_<\/code> and awakens coroutines.<\/p>\n
    That concludes the description of the TSelect<\/code> class. The Poll<\/code> and
     WakeupReadyHandles<\/code> methods will be used in the event handling loop, and the
     other methods will be used by the socket class in read\/write\/connection
     methods.<\/p>\n
    The event handling loop is implemented trivially:<\/p>\n
    template<typename TPoller> class TLoop { public:     void Loop() {         while (Running_) {             Step();         }     }      void Stop() {         Running_ = false;     }      void Step() {         Poller_.Poll();         Poller_.WakeupReadyHandles();     }      TPoller& Poller() {         return Poller_;     }  private:     TPoller Poller_;     bool Running_ = true; }; <\/code><\/pre>\n
    This is a template class parameterized by the type TPoller<\/code>, which can be
     TSelect<\/code>. This is done to allow for swapping out the implementation of the
     multiplexer (using epoll<\/code> instead of select<\/code>, etc.). Also, for convenience
     and future code extension, let’s place all methods and fields needed by the
     socket for operation in the base class TPollerBase<\/code> and inherit TSelect<\/code>
     from TPollerBase<\/code>. Thus, TPollerBase<\/code> will contain methods AddTimer<\/code>,
     RemoveTimer<\/code>, AddRead<\/code>, AddWrite<\/code>, RemoveEvent<\/code>. The code for this
     enhancement is trivial. Later, the socket class will work with the base class.<\/p>\n
    Socket<\/h3>\n
    As fields, the socket has a descriptor Fd_<\/code>, a pointer to the base class
     poller Poller_<\/code>, and an address Addr_<\/code>. Constructors and destructors are
     trivial. In the constructor, a socket is created and set to non-blocking mode
     using fcntl<\/code>. In the destructor, the descriptor is closed with close<\/code>, and
     events associated with the socket are removed from Poller_<\/code>.<\/p>\n
    class TSocket {     int Fd_ = -1;     TPollerBase* Poller_ = nullptr;     TAddress Addr_;  public:     \/\/ Obvious things: Constructors, assignment operators, destructors <\/code><\/pre>\n
    Let’s present the code for the main socket methods — establishing a
     connection, accepting a connection, reading, and writing. All these methods
     will return a structure called TAwaitable<\/code>, containing the methods
     await_ready<\/code>, await_suspend<\/code>, await_resume<\/code>. The compiler will insert
     calls to these methods into the code. First, the await_ready<\/code> method is
     called; if it returns true<\/code>, the coroutine will not go into a sleeping
     state, and instead, the await_resume<\/code> method will be immediately called to
     get the result. If the await_ready<\/code> method returns false<\/code>, the
     await_suspend<\/code> method is called, and a reference to the coroutine is passed
     as an argument. This reference can be saved and later used to awaken the
     coroutine. Upon awakening, the await_resume<\/code> method is always called,
     returning the result.<\/p>\n
    Let’s examine the Connect<\/code> method, which establishes a connection. Here,
     in await_ready<\/code>, we attempt to establish a connection in non-blocking mode,
     and if successful, we return true<\/code>. If it’s not possible to do it
     immediately, we return false<\/code> and go to sleep. We wake up when select<\/code>
     indicates that the connection is established and operations can be performed
     with the socket (in this case, write operations). Note that here, with the
     help of a timer (AddTimer<\/code>) and the deadline<\/code> parameter, you can set the
     maximum connection waiting time.<\/p>\n
    auto Connect(TTime deadline = TTime::max()) {     struct TAwaitable {         bool await_ready() {             int ret = connect(fd, (struct sockaddr*) &addr, sizeof(addr));             if (ret < 0                  && !(errno == EINTR||errno==EAGAIN||errno==EINPROGRESS)) {                 throw std::system_error(errno,                                          std::generic_category(), \"connect\");             }             return ret >= 0;         }          void await_suspend(std::coroutine_handle<> h) {             poller->AddWrite(fd, h);             if (deadline != TTime::max()) {                 poller->AddTimer(fd, deadline, h);             }         }          void await_resume() {             if (deadline != TTime::max()                  && poller->RemoveTimer(fd, deadline)) {                 throw std::system_error(                   std::make_error_code(std::errc::timed_out));             }         }          TPollerBase* poller;         int fd;         sockaddr_in addr;         TTime deadline;     };     return TAwaitable{Poller_, Fd_, Addr_.Addr(), deadline}; } <\/code><\/pre>\n
    The read method works similarly, but in this case, we try to read data using
     read<\/code>. If it fails, we put the coroutine to sleep, and the descriptor is
     sent to wait in select<\/code>.<\/p>\n
    auto ReadSome(char* buf, size_t size) {     struct TAwaitable  {         bool await_ready() {             Run();             return ready = (ret >= 0);         }          int await_resume() {             if (!ready) { Run(); }             return ret;         }          void Run() {             ret = read(fd, b, s);             if (ret < 0 && !(err == EINTR||err==EAGAIN||err==EINPROGRESS)) {                 throw std::system_error(errno,                                          std::generic_category(), \"read\");             }         }          void await_suspend(std::coroutine_handle<> h) {             poller->AddRead(fd, h);         }          TSelect* poller;         int fd;         char* b; size_t s;         int ret;         bool ready;     };     return TAwaitable{Poller_,Fd_,buf,size}; } <\/code><\/pre>\n
    Let’s add another method, ReadSomeYield<\/code>, which will always put the coroutine
     to sleep immediately and send the descriptor to wait in select<\/code>. This method
     is useful for tests and benchmarks.<\/p>\n
        auto ReadSomeYield(char* buf, size_t size) {         struct TAwaitable  {             bool await_ready() {                 return (ready = false);             }             ...             same code here             ...         };         return TAwaitable{Poller_,Fd_,buf,size};     } <\/code><\/pre>\n
    The write methods WriteSome<\/code> and WriteSomeYield<\/code> are implemented similarly,
     but instead of calling read<\/code>, write<\/code> is used everywhere. In case of
     sleeping, an event of type Write is added to the poller using AddWrite<\/code>.<\/p>\n
        auto WriteSome(char* buf, size_t size) {         \/\/ Same as ReadSome     }      auto WriteSomeYield(char* buf, size_t size) {         \/\/ Same as ReadSomeYield     } <\/code><\/pre>\n
    The method for accepting connections, Accept<\/code>, used by the server, is
     implemented as follows.<\/p>\n
    auto Accept() {     struct TAwaitable {         bool await_ready() const { return false; }         void await_suspend(std::coroutine_handle<> h) {             poller->AddRead(fd, h);         }         TSocket await_resume() {             sockaddr_in clientaddr;             socklen_t len = sizeof(clientaddr);              int clientfd = accept(fd, (sockaddr*)&clientaddr, &len);             if (clientfd < 0) {                 throw std::system_error(errno,                                          std::generic_category(), \"accept\");             }              return TSocket{clientaddr, clientfd, *poller};         }          TSelect* poller;         int fd;     };      return TAwaitable{Poller_, Fd_}; } <\/code><\/pre>\n
    In await_ready<\/code>, we go to sleep immediately and wait for select<\/code> to respond
     that the server socket is readable (AddRead<\/code>). If this happens, it means a
     new client has arrived. In this case, we obtain the client socket using
     the accept<\/code> call, wrap it in TSocket<\/code>, and return the result.<\/p>\n
    There are also the Bind<\/code> and Listen<\/code> methods, which are trivial wrappers
     around the bind<\/code> and listen<\/code> calls. The first method binds the socket to an
     address, and the second method puts the socket in passive mode, allowing the
     accept<\/code> method to be called to accept connections.<\/p>\n
        \/* assign the socket the         required address to which clients will send requests *\/     void Bind() { \/* trivial *\/ }      void Listen() { \/* trivial *\/ } }; <\/code><\/pre>\n\n
    <\/a><\/p>\n
    Performance<\/h3>\n
    The full code of the examples discussed above can be downloaded using the link<\/a>.
     The complete version includes support not only for select<\/code> but also for poll<\/code>, epoll<\/code>, kqueue<\/code>, and even uring<\/code>.<\/p>\n
    Our solution has proven its effectiveness in terms of usability, as seen in
     the client and server code. Now let’s look at the performance of the
     written library and compare it with a similar solution that uses callbacks
     for code implementation.<\/p>\n
    We’ll apply the methodology from libevent<\/a>. Create N pipes that pass
     data to each other as follows:<\/p>\n
    struct Stat {     int writes = 0;     int fired = 0;     int count = 0;     int failures = 0;     int out = 0; };  template<typename TSocket> TTestTask pipe_reader(TSocket& r, TSocket& w, Stat& s) {     ssize_t size;     char buf[1] = {0};      try {         while ((size = co_await r.ReadSomeYield(buf, 1)) != 0) {             s.count += size;             if (s.writes) {                 s.writes--;                 if (co_await w.WriteSome(buf, 1) != 1) {                     s.failures ++;                 }                 s.fired++;             }         }     } catch (const std::exception& ex) {         s.failures ++;     }     s.out ++;     co_return; } <\/code><\/pre>\n
    We read data from one pipe and write it to the next one in the chain. We
     collect statistics about the number of bytes read, the number of write<\/code> calls,
     the number of remaining write<\/code> calls, and so on in the Stat<\/code> structure.
     A similar benchmark included in libevent also collects similar statistics. To
     ensure that the coroutine benchmark operates exactly like the callback-based
     libevent benchmark, I compared the counter values of the statistics after
     running the benchmark. The counter values precisely matched the values
     provided by the libevent benchmark.<\/p>\n
    We write data into M pipes as follows:<\/p>\n
    template<typename TSocket> TTestTask write_one(TSocket& w, Stat& s) {     char buf[1] = {'e'};     co_await w.WriteSome(buf, 1);     s.fired ++;     co_return; } <\/code><\/pre>\n
    Let’s run two benchmarks. In the first one, we initially write data into one
     pipe with M=1, meaning one pipe is active while the others are passive. We
     increase the total number of pipes and observe how the processing time for all
     events changes as we increase the number of pipes. In the second benchmark, we
     set M=100. The results for various operating systems and CPUs are provided
     below.
     As you can see, the developed solution performs as well as libevent, and for
     backend select<\/code> and poll<\/code>, it even performs better.<\/p>\n
    Below are the results of two benchmarks for the i7-12800H and i5-11400F processors, as well as the Apple M1. Results for the libevent library are also provided for comparison.    <\/p>\n
    \"CPU<\/p>\n
    CPU i7-12800H, Ubuntu 23.04, clang 16<\/figcaption><\/div>\n<\/figure>\n
    \"CPU<\/p>\n
    CPU i7-12800H, Ubuntu 23.04, clang 16<\/figcaption><\/div>\n<\/figure>\n
    \"CPU<\/p>\n
    CPU i5-11400F, Ubuntu 23.04, WSL2, kernel 6.1.21.1-microsoft-standard-WSL2+<\/figcaption><\/div>\n<\/figure>\n
    \"CPU<\/p>\n
    CPU i5-11400F, Ubuntu 23.04, WSL2, kernel 6.1.21.1-microsoft-standard-WSL2+<\/figcaption><\/div>\n<\/figure>\n
    \"CPU<\/p>\n
    CPU Apple M1, MacBook Air M1 16G, MacOS 12.6.3<\/figcaption><\/div>\n<\/figure>\n
    \"CPU<\/p>\n
    CPU Apple M1, MacBook Air M1 16G, MacOS 12.6.3<\/figcaption><\/div>\n<\/figure>\n
    Conclusion<\/h3>\n
    We have successfully developed a high-performance network library that not only excels in efficiency but also offers ease of use and extendability. The library’s unique design supports asynchronous I\/O for both sockets and files by leveraging the capabilities of the uring backend. The code snippets provided are straightforward, mirroring the structure of typical synchronous code, thereby simplifying the learning curve for new users. This combination of performance, simplicity, and extendability makes our library a robust tool for building efficient, scalable network applications. It stands as a testament to the power and flexibility of modern C++ programming using coroutines.<\/p>\n
    Links<\/h3>\n
      \n
    1. \n
      Callbacks<\/a><\/p>\n<\/li>\n
    2. \n
      Actors<\/a><\/p>\n<\/li>\n
    3. \n
      Future\/Promise<\/a><\/p>\n<\/li>\n
    4. \n
      Coroutines<\/a><\/p>\n<\/li>\n
    5. \n
      Coroutines in C++<\/a><\/p>\n<\/li>\n
    6. \n
      Libevent<\/a><\/p>\n<\/li>\n
    7. \n
      resetius\/coroio<\/a><\/p>\n<\/li>\n<\/ol>\n<\/div>\n<\/div>\n<\/div>\n
      <\/div>\n

       \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438        https:\/\/habr.com\/ru\/articles\/768418\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"
      <\/div>\n
      \n
      \n
      \n
      \n
      <\/figure>\n
      Asynchronous programming is commonly employed for efficient implementation of network interactions in C++. The essence of this approach lies in the fact that the results of socket read\/write functions are not immediately available but become accessible after some time. This approach allows for loading the processor with useful work during the wait for data. Various implementations of this approach exist, such as callbacks<\/a>, actors<\/a>, future\/promise<\/a>, coroutines<\/a>. In C++, these implementations are available as libraries from third-party developers or can be implemented independently.<\/p>\n
      Coroutines are the most challenging to implement as they require writing platform-dependent code. However, the recent version of the C++ 20 language standard introduces support for coroutines at the compiler and standard library levels. Coroutines are functions that can suspend their execution, preserving their state, and later return to that state to resume the function’s work. The compiler automatically creates a checkpoint with the coroutine’s state.<\/p>\n
      For a comprehensive understanding of C++ 20 coroutines, refer to this article. Below, we examine a code example using coroutines and describe important points applied during implementation.<\/p>\n
      For a complete understanding and assimilation of the material presented in this article, it is necessary for the reader to possess basic knowledge in the field of sockets and the TCP\/IP protocol. Fundamental concepts, such as mechanisms of interaction through sockets, principles of data transmission via the TCP\/IP protocol, as well as an understanding of key terms in this domain, will be crucial for the successful comprehension of the provided information. GeeksforGeeks offers useful tutorials on TCP Server-Client Implementation in C++<\/a> and Socket Programming in C\/C++<\/a>, which can serve as valuable resources for enhancing your understanding of these foundational concepts.<\/p>\n
      The code of the library discussed below, along with usage examples, can be downloaded from the following link: https:\/\/github.com\/resetius\/coroio<\/a>.<\/p>\n
      Contents<\/h3>\n
        \n
      1. \n
        Network Interaction: Echo Client and Server Example<\/a><\/p>\n<\/li>\n
      2. \n
        Echo Client and Server Code Overview<\/a><\/p>\n<\/li>\n
      3. \n
        promise and coroutine_handle<\/a><\/p>\n<\/li>\n
      4. \n
        Using final_suspend in Unit Tests<\/a><\/p>\n<\/li>\n
      5. \n
        Event Handling Using select<\/a><\/p>\n<\/li>\n
      6. \n
        Performance and benchmarks<\/a><\/p>\n<\/li>\n<\/ol>\n\n
        <\/a><\/p>\n
        Network Interaction: Echo Client and Server Example<\/h3>\n
        Let’s examine a classic example of network interaction\u2014the implementation of an echo client and server. This example illustrates the task of implementing network programs and effectively demonstrates the details of a specific approach used for asynchronous programming. The echo client\/server example is often included in the documentation of third-party libraries.<\/p>\n
        The echo server receives a message from the client and immediately sends it back unchanged as a response. The echo client sends a message to the server and waits for a response. Here’s an example of the process.<\/p>\n
        This is a client session:<\/p>\n
        $ .\/echoclient test Received: test  message Received: message  yet another message Received: yet another message <\/code><\/pre>\n
        This is a server session:<\/p>\n
        $ .\/echoserver Received: test  Received: message  Received: yet another message <\/code><\/pre>\n
        <\/a><\/p>\n
        Echo Client and Server Code Overview<\/h3>\n
        Let’s examine the Echo client and server code from top to bottom. We’ll start
         by looking at the overall code structure, then move on to exploring the C++
         coroutine API, and finally, we’ll descend to the POSIX API level for socket
         operations. It’s worth noting that no third-party libraries are used in this
         implementation. Only the standard C++ library and the POSIX API are employed.<\/p>\n
        The Echo client reads from standard input, sends the input to the socket,
         and waits for a response:<\/p>\n
        TSimpleTask client(TLoop* loop) {     char out[128] = {0};     char in[128] = {0};     ssize_t size = 1;      try {         TSocket input{TAddress{}, 0 \/* stdin *\/, loop->Poller()};         TSocket socket{TAddress{\"127.0.0.1\", 8888}, loop->Poller()};          co_await socket.Connect();         while (size && (size = co_await input.ReadSome(out, sizeof(out)))) {             co_await socket.WriteSome(out, size);             size = co_await socket.ReadSome(in, sizeof(in));             std::cout << \"Received: \" << std::string_view(in, size) << \"\\n\";         }     } catch (const std::exception& ex) {         std::cout << \"Exception: \" << ex.what() << \"\\n\";     }     loop->Stop();     co_return; }  int main(int argc, char** argv) {     TLoop loop;     client(&loop);     loop.Loop();     return 0; } <\/code><\/pre>\n
        The Echo server listens on a port, accepts incoming connections (calls
         accept<\/code>), and for each connection, a separate coroutine is created. This
         coroutine reads data from the socket and immediately sends the read data back
         to the socket:<\/p>\n
        TSimpleTask client_handler(TSocket socket, TLoop* loop) {     char buffer[128] = {0}; ssize_t size = 0;      try {         while ((size = co_await socket.ReadSome(buffer, sizeof(buffer))) > 0) {             std::cerr << \"Received: \" << std::string_view(buffer, size) << \"\\n\";             co_await socket.WriteSome(buffer, size);         }     } catch (const std::exception& ex) {         std::cerr << \"Exception: \" << ex.what() << \"\\n\";     }     co_return; }  TSimpleTask server(TLoop* loop) {     TSocket socket(TAddress{\"0.0.0.0\", 8888}, loop->Poller());     socket.Bind();     socket.Listen();      while (true) {         auto client = co_await socket.Accept();         client_handler(std::move(client), loop);     }     co_return; }  int main(int argc, char** argv) {     TLoop loop;     server(&loop);     loop.Loop();     return 0; } <\/code><\/pre>\n
        The client and server share a similar structure: the main<\/code> function starts an
         event handling loop (TLoop<\/code>) and launches a coroutine. The event handling loop
         interacts with an object (TPoller<\/code>), which performs input-output multiplexing.
         In this implementation, input-output multiplexing is achieved using the select<\/code>
         mechanism, and changing it to poll<\/code>, epoll<\/code>, kqueue<\/code>, etc., should not pose a
         problem. TPoller<\/code> waits for events on the socket or a timeout, passes these
         events to the TLoop<\/code> loop, which, in turn, wakes up the coroutines subscribed to
         these events.<\/p>\n
        The following diagram illustrates how the program switches the execution flow between the
         client<\/code> coroutine and the Loop.loop()<\/code> function when the coroutine sleeps on
         each co_await<\/code> operator call (in reality, it may not sleep every time;
         details below).<\/p>\n
        \n
        coroutine suspend\/resume<\/figcaption><\/div>\n<\/figure>\n
        Below is a diagram of the main classes implemented in our library. TPoller<\/code> has several implementations (select<\/code>, poll<\/code>, epoll<\/code>, kqueue<\/code>, uring<\/code>). TLoop<\/code> contains TPoller<\/code> — its responsibility is to initiate the Poll<\/code>. TSocket<\/code> contains a reference to TPoller<\/code> and adds events to it when ReadSome<\/code>, WriteSome<\/code>, Connect<\/code>, etc., are called.<\/p>\n
        \n
        class diagram<\/figcaption><\/div>\n<\/figure>\n
        Before moving on to the description of the main components, let’s take a look at the coroutine API in C++20.<\/p>\n
        <\/a><\/p>\n
        Objects promise and coroutine_handle<\/h3>\n
        Each coroutine is associated with two entities: promise<\/code> and coroutine_handle<\/code>.
         coroutine_handle<\/code> is necessary for managing the coroutine externally; for
         example, it can be used to resume (resume<\/code>) the coroutine, destroy (destroy<\/code>)
         it, or check if the coroutine is done (done<\/code>). With the promise<\/code> object, you
         can define the behavior of the coroutine at the start and end of its
         execution and manage the coroutine from within. It’s important to note that the
         methods of the promise<\/code> object should not be called directly; instead, the
         co_return<\/code> and co_yield<\/code> operators are used, and the compiler substitutes the
         appropriate methods.<\/p>\n
        In the example, the following pair of promise<\/code>\/coroutine_handle<\/code> is used
         (TVoidPromise<\/code>\/TSimpleTask<\/code>):<\/p>\n
        struct TVoidPromise;  struct TSimpleTask : std::coroutine_handle<TVoidPromise> {     using promise_type = TVoidPromise; };  struct TVoidPromise {     TSimpleTask get_return_object() {        return { TSimpleTask::from_promise(*this) };      }     std::suspend_never initial_suspend() { return {}; }     std::suspend_never final_suspend() noexcept { return {}; }     void return_void() {}     void unhandled_exception() {} }; <\/code><\/pre>\n
        Here, the coroutine doesn’t return anything (return_void<\/code>), starts
         immediately (std::suspend_never initial_suspend<\/code>), and does not sleep upon
         completion (std::suspend_never final_suspend<\/code>).<\/p>\n
        In some cases, such as when implementing unit tests, it is convenient to sleep
         the coroutine upon completion:<\/p>\n
        struct TTestPromise;  struct TTestTask : std::coroutine_handle<TTestPromise> {     using promise_type = TTestPromise; };  struct TTestPromise {     TTestTask get_return_object() {        return { TTestTask::from_promise(*this) };      }     std::suspend_never initial_suspend() { return {}; }     std::suspend_always \/* the difference here *\/ final_suspend() noexcept      { return {}; }     void return_void() {}     void unhandled_exception() {} }; <\/code><\/pre>\n
        Please note that in this case, the promise<\/code> differs only in the signature of
         the final_suspend<\/code> method.<\/p>\n
        <\/a><\/p>\n
        Using final_suspend in Unit Tests<\/h3>\n
        Let’s consider a unit test that launches client and server coroutines and then
         checks the correctness of the message exchange:<\/p>\n
        void test_write_after_connect(void**) {     TLoop loop;     TSocket socket(TAddress{\"127.0.0.1\", 8898}, loop.Poller());     socket.Bind();     socket.Listen();     char send_buf[128] = \"Hello\";     char rcv_buf[128] = {0};      TTestTask h1 = [](TLoop* loop, char* buf, int size) -> TTestTask     {         TSocket client(TAddress{\"127.0.0.1\", 8898}, loop->Poller());         co_await client.Connect();         co_await client.WriteSome(buf, size);         co_return;     }(&loop, send_buf, sizeof(send_buf));      TTestTask h2 = [](TSocket* socket, char* buf, int size) -> TTestTask     {         TSocket clientSocket = std::move(co_await socket->Accept());         co_await clientSocket.ReadSome(buf, size);         co_return;     }(&socket, rcv_buf, sizeof(rcv_buf));      while (!(h1.done() && h2.done())) {         loop.Step();     }     h1.destroy(); h2.destroy();      assert_true(memcmp(&send_buf, &rcv_buf, sizeof(send_buf))==0); } <\/code><\/pre>\n
        In this case, the event handling loop runs until<\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-357885","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/357885","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=357885"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/357885\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=357885"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=357885"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=357885"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}