TCP echo server

A TCP echo server and its clients in one program. The server listens on 127.0.0.1 on a port the system picks, accepts four connections and runs one task per connection that sends back whatever it receives. Four client tasks connect, send three messages each, check the echo and close their connection gracefully. Every socket operation suspends only the task that calls it, so each task reads like plain blocking code.

Starting up

// Winsock and its extension functions (AcceptEx, ConnectEx, ...), then the
// core's completion port.
if (WinsockError::Failed(TC::Windows::Socket::Startup()))
    return 1;
if (!TC::AsyncIo::EnableAsyncIo())
    return 1;

Socket::Startup initializes Winsock and loads the extension functions (AcceptEx, ConnectEx, …) the asynchronous sockets use; EnableAsyncIo starts the completion port their completions arrive on. Both must have succeeded before the first socket is created.

Listening

// Port 0: the system picks a free port; GetSockName tells which.
AsyncTcpSocket4 listener;
sockaddr_in     address{};
address.sin_family      = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
const bool listening    = WinsockError::Succeeded(listener.Create())
    && WinsockError::Succeeded(listener.Bind(address))
    && WinsockError::Succeeded(listener.Listen())
    && WinsockError::Succeeded(listener.GetSockName(address));

Port 0 lets the system choose a free port; GetSockName tells which one.

One task per connection

// One connection: receive, send back, until the client shuts down its side (a
// receive of 0 bytes).
class EchoTask : public TC::Tasks::Task
{
public:
    AsyncTcpSocket4
                socket; // moved in by the accept task before the task is started
    std::size_t echoed = 0;

    void Run() noexcept override
    {
        std::byte buffer[4096];
        for (;;) {
            DWORD received = 0;
            if (WinsockError::Failed(socket.Recv(buffer, sizeof(buffer), &received))
                || received == 0)
                break;
            if (WinsockError::Failed(SendAll(socket, buffer, received)))
                break;
            echoed += received;
        }
        socket.Close(); // nothing is pending: Recv and Send have returned
    }
};

Recv returns 0 bytes when the peer has shut down its sending side – the signal to stop. A send may transfer less than asked, so SendAll loops.

Accepting

// Accepts a known number of connections and starts one EchoTask for each. The echo
// tasks are owned by main and counted into the group before they are started.
class AcceptTask : public TC::Tasks::Task
{
public:
    AcceptTask(AsyncTcpSocket4& listener, EchoTask* echoTasks,
               TC::Synchronization::TaskGroup_s& group) noexcept
        : listener_(listener),
          echoTasks_(echoTasks),
          group_(group)
    {}

    WinsockErrorCode error = WinsockError::Success;

    void Run() noexcept override
    {
        for (std::size_t i = 0; i < Clients; ++i) {
            AsyncTcpSocket4 connection; // Accept creates it
            error = listener_.Accept(connection);
            if (WinsockError::Failed(error))
                return;
            echoTasks_[i].socket = std::move(connection);
            group_.AddTask(&echoTasks_[i]);
            echoTasks_[i].Schedule(); // on the scheduler of the calling task
        }
    }

private:
    AsyncTcpSocket4&                  listener_;
    EchoTask*                         echoTasks_;
    TC::Synchronization::TaskGroup_s& group_;
};

Accept creates the connection's socket itself and suspends the accept task until a client connects. The accepted socket is moved into an echo task, which is counted into the group and started with Schedule on the accept task's own scheduler.

The clients

// Connects, sends each message and reads the echo back, then shuts down its sending
// side and waits until the server has closed the connection.
class ClientTask : public TC::Tasks::Task
{
public:
    std::size_t index = 0;
    sockaddr_in server{};
    bool        ok = false;

    void Run() noexcept override
    {
        AsyncTcpSocket4 socket;
        // ConnectEx needs a bound socket: any port on the loopback address.
        sockaddr_in local{};
        local.sin_family      = AF_INET;
        local.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        if (WinsockError::Failed(socket.Create())
            || WinsockError::Failed(socket.Bind(local))
            || WinsockError::Failed(socket.Connect(server)))
            return;

        std::vector<std::byte> message;
        std::vector<std::byte> echo;
        for (std::size_t m = 0; m < MessagesPerClient; ++m) {
            message.resize(MessageSize(index, m));
            for (std::size_t i = 0; i < message.size(); ++i)
                message[i] = MessageByte(index, m, i);
            if (WinsockError::Failed(
                    SendAll(socket, message.data(),
                            static_cast<std::uint32_t>(message.size()))))
                return;

            // TCP is a byte stream: the echo may arrive in several pieces.
            echo.resize(message.size());
            std::size_t received = 0;
            while (received < echo.size()) {
                DWORD bytes = 0;
                if (WinsockError::Failed(socket.Recv(
                        echo.data() + received,
                        static_cast<std::uint32_t>(echo.size() - received), &bytes))
                    || bytes == 0)
                    return;
                received += bytes;
            }
            if (echo != message)
                return;
        }

        // Graceful end: no more data from this side; the server's receive returns 0
        // bytes, it closes, and this side's receive returns 0 bytes too.
        DWORD bytes = 0;
        ok          = WinsockError::Succeeded(socket.Shutdown(SD_SEND))
            && WinsockError::Succeeded(socket.Recv(echo.data(), 1, &bytes))
            && bytes == 0;
    }
};

Connect uses ConnectEx, which needs a bound socket – hence the Bind to port 0 first. TCP is a byte stream, so a client collects the echo until it has as many bytes as it sent. To close, it shuts down its sending side; the server's receive returns 0, the server closes, and the client's last receive returns 0 too.

Running it

// All task objects and the group outlive the tasks: they are destroyed
// after Stop.
EchoTask   echoTasks[Clients];
ClientTask clientTasks[Clients];
TC::Synchronization::TaskGroup_s
           group; // main is the only waiter; the accept task adds to it
AcceptTask acceptTask(listener, echoTasks, group);

group.AddTask(&acceptTask);
scheduler->AddTask(&acceptTask);
for (std::size_t i = 0; i < Clients; ++i) {
    clientTasks[i].index  = i;
    clientTasks[i].server = address;
    group.AddTask(&clientTasks[i]);
    scheduler->AddTask(&clientTasks[i]);
}
group.Wait();     // accept task, clients and every echo task the accept
                  // task started
listener.Close(); // no Accept is pending any more

TC::AsyncIo::DisableAsyncIo(); // every socket is closed, no operation is
                               // pending
TC::Windows::Socket::Cleanup();

Full program

// TCP echo server: a listening socket on 127.0.0.1, one task per accepted
// connection that echoes what it receives, and client tasks in the same program
// that check the echo.
#define NOMINMAX
#include <WinSock2.h>
#include <Windows.h>

#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <vector>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncTcpSocket;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::AsyncTcpSocket4;
using TC::Windows::WinsockError;
using TC::Windows::WinsockErrorCode;

constexpr std::size_t Clients           = 4;
constexpr std::size_t MessagesPerClient = 3;

std::uint32_t MessageSize(std::size_t client, std::size_t message) noexcept
{
    return static_cast<std::uint32_t>(1 + (client * 977 + message * 3001) % 8000);
}

std::byte MessageByte(std::size_t client, std::size_t message,
                      std::size_t i) noexcept
{
    return static_cast<std::byte>(client * 31 + message * 7 + i);
}

// Sends the whole buffer; a send may transfer less than asked.
WinsockErrorCode SendAll(AsyncTcpSocket4& socket, const std::byte* data,
                         std::uint32_t size) noexcept
{
    while (size > 0) {
        DWORD                  sent  = 0;
        const WinsockErrorCode error = socket.Send(data, size, &sent);
        if (WinsockError::Failed(error))
            return error;
        data += sent;
        size -= sent;
    }
    return WinsockError::Success;
}

// One connection: receive, send back, until the client shuts down its side (a
// receive of 0 bytes).
class EchoTask : public TC::Tasks::Task
{
public:
    AsyncTcpSocket4
                socket; // moved in by the accept task before the task is started
    std::size_t echoed = 0;

    void Run() noexcept override
    {
        std::byte buffer[4096];
        for (;;) {
            DWORD received = 0;
            if (WinsockError::Failed(socket.Recv(buffer, sizeof(buffer), &received))
                || received == 0)
                break;
            if (WinsockError::Failed(SendAll(socket, buffer, received)))
                break;
            echoed += received;
        }
        socket.Close(); // nothing is pending: Recv and Send have returned
    }
};

// Accepts a known number of connections and starts one EchoTask for each. The echo
// tasks are owned by main and counted into the group before they are started.
class AcceptTask : public TC::Tasks::Task
{
public:
    AcceptTask(AsyncTcpSocket4& listener, EchoTask* echoTasks,
               TC::Synchronization::TaskGroup_s& group) noexcept
        : listener_(listener),
          echoTasks_(echoTasks),
          group_(group)
    {}

    WinsockErrorCode error = WinsockError::Success;

    void Run() noexcept override
    {
        for (std::size_t i = 0; i < Clients; ++i) {
            AsyncTcpSocket4 connection; // Accept creates it
            error = listener_.Accept(connection);
            if (WinsockError::Failed(error))
                return;
            echoTasks_[i].socket = std::move(connection);
            group_.AddTask(&echoTasks_[i]);
            echoTasks_[i].Schedule(); // on the scheduler of the calling task
        }
    }

private:
    AsyncTcpSocket4&                  listener_;
    EchoTask*                         echoTasks_;
    TC::Synchronization::TaskGroup_s& group_;
};

// Connects, sends each message and reads the echo back, then shuts down its sending
// side and waits until the server has closed the connection.
class ClientTask : public TC::Tasks::Task
{
public:
    std::size_t index = 0;
    sockaddr_in server{};
    bool        ok = false;

    void Run() noexcept override
    {
        AsyncTcpSocket4 socket;
        // ConnectEx needs a bound socket: any port on the loopback address.
        sockaddr_in local{};
        local.sin_family      = AF_INET;
        local.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        if (WinsockError::Failed(socket.Create())
            || WinsockError::Failed(socket.Bind(local))
            || WinsockError::Failed(socket.Connect(server)))
            return;

        std::vector<std::byte> message;
        std::vector<std::byte> echo;
        for (std::size_t m = 0; m < MessagesPerClient; ++m) {
            message.resize(MessageSize(index, m));
            for (std::size_t i = 0; i < message.size(); ++i)
                message[i] = MessageByte(index, m, i);
            if (WinsockError::Failed(
                    SendAll(socket, message.data(),
                            static_cast<std::uint32_t>(message.size()))))
                return;

            // TCP is a byte stream: the echo may arrive in several pieces.
            echo.resize(message.size());
            std::size_t received = 0;
            while (received < echo.size()) {
                DWORD bytes = 0;
                if (WinsockError::Failed(socket.Recv(
                        echo.data() + received,
                        static_cast<std::uint32_t>(echo.size() - received), &bytes))
                    || bytes == 0)
                    return;
                received += bytes;
            }
            if (echo != message)
                return;
        }

        // Graceful end: no more data from this side; the server's receive returns 0
        // bytes, it closes, and this side's receive returns 0 bytes too.
        DWORD bytes = 0;
        ok          = WinsockError::Succeeded(socket.Shutdown(SD_SEND))
            && WinsockError::Succeeded(socket.Recv(echo.data(), 1, &bytes))
            && bytes == 0;
    }
};

int main()
{
    TC::ConvertThreadToTask();

    // Winsock and its extension functions (AcceptEx, ConnectEx, ...), then the
    // core's completion port.
    if (WinsockError::Failed(TC::Windows::Socket::Startup()))
        return 1;
    if (!TC::AsyncIo::EnableAsyncIo())
        return 1;

    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    bool ok = false;
    {
        // Port 0: the system picks a free port; GetSockName tells which.
        AsyncTcpSocket4 listener;
        sockaddr_in     address{};
        address.sin_family      = AF_INET;
        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        const bool listening    = WinsockError::Succeeded(listener.Create())
            && WinsockError::Succeeded(listener.Bind(address))
            && WinsockError::Succeeded(listener.Listen())
            && WinsockError::Succeeded(listener.GetSockName(address));

        // All task objects and the group outlive the tasks: they are destroyed
        // after Stop.
        EchoTask   echoTasks[Clients];
        ClientTask clientTasks[Clients];
        TC::Synchronization::TaskGroup_s
                   group; // main is the only waiter; the accept task adds to it
        AcceptTask acceptTask(listener, echoTasks, group);

        if (listening) {
            std::printf("Listening on 127.0.0.1:%u\n", ntohs(address.sin_port));
            group.AddTask(&acceptTask);
            scheduler->AddTask(&acceptTask);
            for (std::size_t i = 0; i < Clients; ++i) {
                clientTasks[i].index  = i;
                clientTasks[i].server = address;
                group.AddTask(&clientTasks[i]);
                scheduler->AddTask(&clientTasks[i]);
            }
            group.Wait();     // accept task, clients and every echo task the accept
                              // task started
            listener.Close(); // no Accept is pending any more

            ok                 = WinsockError::Succeeded(acceptTask.error);
            std::size_t echoed = 0;
            for (std::size_t i = 0; i < Clients; ++i) {
                std::printf("client %zu: %s\n", i,
                            clientTasks[i].ok ? "echo verified" : "FAILED");
                ok = ok && clientTasks[i].ok;
                echoed += echoTasks[i].echoed;
            }
            std::printf("server: %zu connections, %zu bytes echoed\n", Clients,
                        echoed);
        }
        scheduler->Stop();
    }
    scheduler->Destroy();

    TC::AsyncIo::DisableAsyncIo(); // every socket is closed, no operation is
                                   // pending
    TC::Windows::Socket::Cleanup();
    TC::ConvertTaskToThread();
    std::printf("%s\n", ok ? "ok" : "FAILED");
    return ok ? 0 : 1;
}

Source and project on GitHub

Build and run

A console program. Besides the core's props file and the library reference (see Set up a project) it links ws2_32.lib. It uses only the loopback address and a port chosen by the system, finishes in well under a second and prints (the port differs from run to run):

Listening on 127.0.0.1:57699
client 0: echo verified
client 1: echo verified
client 2: echo verified
client 3: echo verified
server: 4 connections, 45610 bytes echoed
ok

What to notice

Important

A socket must not be closed or destroyed while an operation on it is pending, and the buffers of a pending operation must stay alive. The waiting forms used here (Accept, Connect, Recv, Send) return only after the operation has completed, so closing a socket after they have returned is safe. See Lifetimes.

Tasks, not threads. The waiting forms suspend the calling task, so they must be called from a task (or a thread after ConvertThreadToTask). Four connections or four thousand – each is a task, not a thread.
Adding to a group while it is waited for. The accept task adds the echo tasks to the group while main already waits on it. That is safe because the accept task is itself in the group and has not exited yet, so the count cannot reach 0 in between. Each echo task is counted in before it is started.
Order of the shutdown. All tasks end, the listener is closed (no Accept pending), the scheduler stops, then DisableAsyncIo – no operation may be pending then – and Socket::Cleanup.
Timeouts. Every waiting form takes an optional Duration timeout (it needs EnableTimers). A timed-out operation is cancelled; the socket is then fit only for closing.