C++ for Networking

// Standalone asio (no Boost dependency)
#include <asio.hpp>

int main() {
    asio::io_context io;

    // Single-threaded: run all handlers on this thread
    // Multi-threaded: pool of threads calling io.run()

    io.run();
}

#include <asio.hpp>
#include <asio/co_spawn.hpp>
#include <asio/detached.hpp>
#include <print>

using asio::ip::tcp;

asio::awaitable<void> handle_client(tcp::socket socket) {
    char buf[1024];
    for (;;) {
        // Read data — suspends coroutine until data arrives
        std::size_t n = co_await socket.async_read_some(
            asio::buffer(buf), asio::use_awaitable);

        // Echo it back
        co_await asio::async_write(
            socket, asio::buffer(buf, n), asio::use_awaitable);
    }
}

asio::awaitable<void> accept_loop(tcp::acceptor& acceptor) {
    for (;;) {
        tcp::socket socket = co_await acceptor.async_accept(asio::use_awaitable);
        // Spawn each client as an independent coroutine
        co_spawn(acceptor.get_executor(),
            handle_client(std::move(socket)), asio::detached);
    }
}

int main() {
    asio::io_context io;
    tcp::acceptor acceptor(io, {tcp::v4(), 8080});

    co_spawn(io, accept_loop(acceptor), asio::detached);

    io.run();  // blocks until all work is done
}

asio::awaitable<void> connect_and_send(asio::io_context& io) {
    tcp::resolver resolver(io.get_executor());
    tcp::socket socket(io.get_executor());

    auto endpoints = co_await resolver.async_resolve(
        "example.com", "80", asio::use_awaitable);

    co_await asio::async_connect(socket, endpoints, asio::use_awaitable);

    std::string request = "GET / HTTP/1.0\r\nHost: example.com\r\n\r\n";
    co_await asio::async_write(socket, asio::buffer(request), asio::use_awaitable);

    std::string response;
    co_await asio::async_read_until(socket, asio::dynamic_buffer(response),
        "\r\n\r\n", asio::use_awaitable);

    std::println("{}", response);
}

asio::io_context io;
auto work = asio::make_work_guard(io);  // keep io_context alive

// Thread pool — N threads sharing the same io_context
std::vector<std::thread> threads;
int N = std::thread::hardware_concurrency();
for (int i = 0; i < N; ++i) {
    threads.emplace_back([&io] { io.run(); });
}

// Handlers are dispatched to any free thread — must be thread-safe!
// Use strand to serialize handlers for a given connection:
asio::strand<asio::io_context::executor_type> strand =
    asio::make_strand(io);

// Post work on the strand
asio::post(strand, []{
    // Serialized with other work on the same strand
});

// Join
work.reset();
for (auto& t : threads) t.join();

#include <crow.h>

int main() {
    crow::SimpleApp app;

    CROW_ROUTE(app, "/")
    ([] {
        return "Hello, World!";
    });

    CROW_ROUTE(app, "/api/user/<int>")
    ([](int user_id) {
        crow::json::wvalue res;
        res["id"] = user_id;
        res["name"] = "Alice";
        return crow::response{res};
    });

    CROW_ROUTE(app, "/api/data").methods(crow::HTTPMethod::POST)
    ([](const crow::request& req) {
        auto body = crow::json::load(req.body);
        if (!body) return crow::response(400, "Invalid JSON");
        auto name = body["name"].s();
        return crow::response{200, std::string("Hello ") + name};
    });

    app.port(8080).multithreaded().run();
}

#include <boost/beast/core.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/asio/co_spawn.hpp>

namespace beast     = boost::beast;
namespace websocket = beast::websocket;
namespace net       = boost::asio;
using tcp           = net::ip::tcp;

net::awaitable<void> ws_session(tcp::socket socket) {
    websocket::stream<tcp::socket> ws{std::move(socket)};

    // Perform WebSocket handshake
    co_await ws.async_accept(net::use_awaitable);

    beast::flat_buffer buf;
    for (;;) {
        co_await ws.async_read(buf, net::use_awaitable);

        ws.text(ws.got_text());
        co_await ws.async_write(buf.data(), net::use_awaitable);
        buf.consume(buf.size());
    }
}

net::awaitable<std::string> http_get(std::string host, std::string target) {
    auto executor = co_await net::this_coro::executor;

    tcp::resolver resolver{executor};
    beast::tcp_stream stream{executor};

    auto results = co_await resolver.async_resolve(host, "80", net::use_awaitable);
    co_await stream.async_connect(results, net::use_awaitable);

    beast::http::request<beast::http::empty_body> req{
        beast::http::verb::get, target, 11};
    req.set(beast::http::field::host, host);
    req.set(beast::http::field::user_agent, "beast/1.0");

    co_await beast::http::async_write(stream, req, net::use_awaitable);

    beast::flat_buffer buf;
    beast::http::response<beast::http::dynamic_body> res;
    co_await beast::http::async_read(stream, buf, res, net::use_awaitable);

    co_return beast::buffers_to_string(res.body().data());
}

#include <liburing.h>
#include <fcntl.h>

// io_uring: submits I/O requests to the kernel; kernel completes them async
// Up to 2× throughput vs epoll for mixed read/write workloads

struct io_uring ring;
io_uring_queue_init(256, &ring, 0);  // 256 submission queue entries

// --- Submit a read ---
int fd = open("data.bin", O_RDONLY | O_DIRECT);
char* buf;
posix_memalign((void**)&buf, 4096, 4096);  // aligned for O_DIRECT

struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);
io_uring_prep_read(sqe, fd, buf, 4096, 0);
sqe->user_data = (uint64_t)buf;  // tag for result matching

io_uring_submit(&ring);

// --- Collect completion ---
struct io_uring_cqe* cqe;
io_uring_wait_cqe(&ring, &cqe);

if (cqe->res < 0)
    fprintf(stderr, "Read error: %s\n", strerror(-cqe->res));
else
    printf("Read %d bytes\n", cqe->res);

io_uring_cqe_seen(&ring, cqe);
io_uring_queue_exit(&ring);

// Asio has native io_uring backend on Linux (much simpler than raw liburing)
asio::io_context io(1);  // io_uring backend selected automatically on Linux

// All async_read, async_write operations use io_uring under the hood
// No code change needed — just link with -luring

// readv/writev — read/write multiple buffers in one syscall
#include <sys/uio.h>

struct iovec iov[3];
char header[16], body[1024], footer[4];
iov[0] = { header, sizeof(header) };
iov[1] = { body,   sizeof(body) };
iov[2] = { footer, sizeof(footer) };

ssize_t n = readv(fd, iov, 3);  // one syscall fills all three buffers

// sendfile — zero-copy file to socket (no user-space buffer)
#include <sys/sendfile.h>
off_t offset = 0;
sendfile(socket_fd, file_fd, &offset, file_size);

using udp = asio::ip::udp;

// UDP sender
udp::socket sender(io, udp::endpoint(udp::v4(), 0));
udp::endpoint target(asio::ip::address::from_string("239.0.0.1"), 9999);
sender.send_to(asio::buffer("hello"), target);

// UDP receiver — multicast join
udp::socket recv_sock(io, udp::endpoint(udp::v4(), 9999));
recv_sock.set_option(asio::ip::multicast::join_group(
    asio::ip::address::from_string("239.0.0.1")));

char buf[512];
udp::endpoint sender_ep;
recv_sock.receive_from(asio::buffer(buf), sender_ep);

// Common socket options via Asio
socket.set_option(asio::ip::tcp::no_delay(true));       // TCP_NODELAY
socket.set_option(asio::socket_base::reuse_address(true));
socket.set_option(asio::socket_base::receive_buffer_size(1 << 20));  // 1 MB
socket.set_option(asio::socket_base::send_buffer_size(1 << 20));

// libcurl with HTTP/2 multiplexing
#include <curl/curl.h>

size_t write_cb(char* ptr, size_t size, size_t nmemb, std::string* out) {
    out->append(ptr, size * nmemb);
    return size * nmemb;
}

std::string http2_get(const std::string& url) {
    CURL* curl = curl_easy_init();
    std::string response;

    curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
    curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_cb);
    curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response);
    curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);

    CURLcode res = curl_easy_perform(curl);
    if (res != CURLE_OK)
        throw std::runtime_error(curl_easy_strerror(res));

    curl_easy_cleanup(curl);
    return response;
}

// Multiplexed HTTP/2 — multiple requests over one connection
void http2_multi_example() {
    CURLM* multi = curl_multi_init();
    curl_multi_setopt(multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);

    // Add multiple easy handles to the multi handle
    std::vector<CURL*> handles;
    std::vector<std::string> responses(3);

    for (int i = 0; i < 3; ++i) {
        CURL* h = curl_easy_init();
        curl_easy_setopt(h, CURLOPT_URL, "https://example.com/api");
        curl_easy_setopt(h, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
        curl_easy_setopt(h, CURLOPT_WRITEFUNCTION, write_cb);
        curl_easy_setopt(h, CURLOPT_WRITEDATA, &responses[i]);
        curl_multi_add_handle(multi, h);
        handles.push_back(h);
    }

    // Event loop
    int still_running = 1;
    while (still_running) {
        curl_multi_perform(multi, &still_running);
        curl_multi_wait(multi, nullptr, 0, 100, nullptr);
    }

    for (auto h : handles) { curl_multi_remove_handle(multi, h); curl_easy_cleanup(h); }
    curl_multi_cleanup(multi);
}

# CMake: link libcurl
find_package(CURL REQUIRED)
target_link_libraries(myapp PRIVATE CURL::libcurl)

#include <uwebsockets/App.h>

struct PerSocketData {
    std::string user_id;
};

int main() {
    uWS::App()
        .ws<PerSocketData>("/*", {
            // Configuration
            .compression     = uWS::SHARED_COMPRESSOR,
            .maxPayloadLength = 16 * 1024,
            .idleTimeout     = 60,

            // Lifecycle callbacks
            .open = [](auto* ws) {
                ws->getUserData()->user_id = "anon";
                ws->subscribe("broadcast");
                std::println("Client connected");
            },
            .message = [](auto* ws, std::string_view msg, uWS::OpCode op) {
                // Echo — op is TEXT or BINARY
                ws->send(msg, op, /*compress=*/true);
            },
            .close = [](auto* ws, int code, std::string_view reason) {
                std::println("Closed: {} {}", code, reason);
            }
        })
        .listen(9001, [](auto* token) {
            if (token) std::println("Listening on port 9001");
        })
        .run();
}

// Beast: upgrade an HTTP connection to WebSocket manually
net::awaitable<void> beast_ws_server(tcp::socket raw_socket) {
    beast::tcp_stream stream(std::move(raw_socket));
    beast::flat_buffer buf;

    // Read the HTTP upgrade request first
    beast::http::request<beast::http::empty_body> req;
    co_await beast::http::async_read(stream, buf, req, net::use_awaitable);

    // Validate it is a WebSocket upgrade
    if (!websocket::is_upgrade(req))
        co_return;  // not a WS request — send 400 or handle as HTTP

    websocket::stream<beast::tcp_stream> ws(std::move(stream));

    // Accept with a custom decorator (add response headers)
    ws.set_option(websocket::stream_base::decorator(
        [](websocket::response_type& res) {
            res.set(beast::http::field::server, "MyServer/1.0");
        }));

    co_await ws.async_accept(req, net::use_awaitable);

    // Read frames
    beast::flat_buffer frame_buf;
    for (;;) {
        co_await ws.async_read(frame_buf, net::use_awaitable);
        if (ws.got_text())
            std::println("Text frame: {}", beast::buffers_to_string(frame_buf.data()));
        frame_buf.consume(frame_buf.size());
    }
}

#include <google/protobuf/message.h>
#include <arpa/inet.h>  // htonl / ntohl

// --- Send ---
asio::awaitable<void> send_proto(
    asio::ip::tcp::socket& sock,
    const google::protobuf::Message& msg)
{
    std::string payload;
    msg.SerializeToString(&payload);

    // 4-byte big-endian length prefix
    uint32_t net_len = htonl(static_cast<uint32_t>(payload.size()));

    std::array<asio::const_buffer, 2> bufs = {
        asio::buffer(&net_len, 4),
        asio::buffer(payload)
    };
    co_await asio::async_write(sock, bufs, asio::use_awaitable);
}

// --- Receive ---
asio::awaitable<std::string> recv_proto_bytes(asio::ip::tcp::socket& sock) {
    uint32_t net_len = 0;
    co_await asio::async_read(sock,
        asio::buffer(&net_len, 4), asio::use_awaitable);

    uint32_t len = ntohl(net_len);
    if (len > 64 * 1024 * 1024)  // sanity cap: 64 MB
        throw std::runtime_error("message too large");

    std::string payload(len, '\0');
    co_await asio::async_read(sock,
        asio::buffer(payload), asio::use_awaitable);

    co_return payload;
}

// Loopback server + client in the same process (integration test)
asio::io_context io;

tcp::acceptor acceptor(io, tcp::endpoint(tcp::v4(), 0));  // port 0 → OS picks
auto local_port = acceptor.local_endpoint().port();

// Spawn server coroutine
co_spawn(io, [&]() -> asio::awaitable<void> {
    tcp::socket server_sock = co_await acceptor.async_accept(asio::use_awaitable);
    char buf[4];
    co_await asio::async_read(server_sock, asio::buffer(buf), asio::use_awaitable);
    co_await asio::async_write(server_sock, asio::buffer(buf), asio::use_awaitable);
}, asio::detached);

// Spawn client coroutine
co_spawn(io, [&, local_port]() -> asio::awaitable<void> {
    tcp::socket client_sock(io);
    co_await client_sock.async_connect(
        {asio::ip::address_v4::loopback(), local_port}, asio::use_awaitable);
    co_await asio::async_write(client_sock, asio::buffer("ping"), asio::use_awaitable);
    char resp[4];
    co_await asio::async_read(client_sock, asio::buffer(resp), asio::use_awaitable);
    assert(std::string_view(resp, 4) == "ping");
}, asio::detached);

io.run();

#include <pcap.h>

// Open a saved capture file for replay
char errbuf[PCAP_ERRBUF_SIZE];
pcap_t* handle = pcap_open_offline("capture.pcap", errbuf);
if (!handle) throw std::runtime_error(errbuf);

pcap_pkthdr* header;
const u_char* packet;
while (pcap_next_ex(handle, &header, &packet) == 1) {
    // header->len  = original packet length
    // header->caplen = captured bytes (may be truncated)
    // packet points to raw Ethernet frame
    process_packet(packet, header->caplen);
}
pcap_close(handle);

# Capture 100 packets on loopback to a file
tcpdump -i lo -c 100 -w capture.pcap port 8080

# Replay at 2× speed
tcpreplay --multiplier=2 --intf1=lo capture.pcap

// Bounded async queue — producer blocks when full
template<typename T>
class BoundedQueue {
    std::deque<T>       queue_;
    std::mutex          mu_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
    const size_t        cap_;
    bool                closed_ = false;
public:
    explicit BoundedQueue(size_t cap) : cap_(cap) {}

    // Producer: blocks until space available
    bool push(T item) {
        std::unique_lock lk(mu_);
        not_full_.wait(lk, [&]{ return queue_.size() < cap_ || closed_; });
        if (closed_) return false;
        queue_.push_back(std::move(item));
        not_empty_.notify_one();
        return true;
    }

    // Consumer: blocks until item available
    std::optional<T> pop() {
        std::unique_lock lk(mu_);
        not_empty_.wait(lk, [&]{ return !queue_.empty() || closed_; });
        if (queue_.empty()) return std::nullopt;
        T item = std::move(queue_.front());
        queue_.pop_front();
        not_full_.notify_one();
        return item;
    }

    void close() { std::lock_guard lk(mu_); closed_ = true; not_full_.notify_all(); not_empty_.notify_all(); }
};

// Asio: pause reading when internal buffer exceeds threshold
class FlowControlledSession {
    tcp::socket    socket_;
    BoundedQueue<std::string> write_queue_{64};  // max 64 pending writes

    asio::awaitable<void> write_loop() {
        while (auto msg = write_queue_.pop()) {
            co_await asio::async_write(socket_, asio::buffer(*msg), asio::use_awaitable);
        }
    }
};

// Token bucket — allow burst_size tokens, refill at rate tokens/sec
class TokenBucket {
    double   tokens_;
    double   rate_;       // tokens per second
    double   burst_;
    std::chrono::steady_clock::time_point last_;
public:
    TokenBucket(double rate, double burst)
        : tokens_(burst), rate_(rate), burst_(burst),
          last_(std::chrono::steady_clock::now()) {}

    bool consume(double n = 1.0) {
        auto now = std::chrono::steady_clock::now();
        double elapsed = std::chrono::duration<double>(now - last_).count();
        tokens_ = std::min(burst_, tokens_ + elapsed * rate_);
        last_ = now;
        if (tokens_ < n) return false;
        tokens_ -= n;
        return true;
    }
};

// Server: bind to IPv6 any-address, accept IPv4-mapped addresses too
asio::ip::tcp::endpoint ep(asio::ip::tcp::v6(), 8080);
asio::ip::tcp::acceptor acceptor(io, ep);

// Disable IPV6_V6ONLY so IPv4 clients appear as ::ffff:a.b.c.d
acceptor.set_option(asio::ip::v6_only(false));

// Client: AF_UNSPEC — let the OS pick IPv4 or IPv6
asio::awaitable<void> dual_stack_connect(std::string host, std::string port) {
    auto executor = co_await asio::this_coro::executor;

    tcp::resolver resolver(executor);
    // async_resolve with no hints → returns both A and AAAA records
    auto results = co_await resolver.async_resolve(host, port, asio::use_awaitable);

    tcp::socket socket(executor);
    // async_connect tries each endpoint in order (Happy Eyeballs-like)
    co_await asio::async_connect(socket, results, asio::use_awaitable);

    auto remote = socket.remote_endpoint();
    std::println("Connected to {} ({})",
        remote.address().to_string(),
        remote.address().is_v4() ? "IPv4" : "IPv6");
}

// Raw POSIX: getaddrinfo with AF_UNSPEC
addrinfo hints{};
hints.ai_family   = AF_UNSPEC;    // accept IPv4 or IPv6
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags    = AI_ADDRCONFIG; // only return families with a configured interface

addrinfo* res = nullptr;
if (getaddrinfo("example.com", "443", &hints, &res) != 0)
    throw std::runtime_error("getaddrinfo failed");

// Iterate results; first successful connect wins
for (addrinfo* p = res; p; p = p->ai_next) {
    int fd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
    if (fd == -1) continue;
    if (connect(fd, p->ai_addr, p->ai_addrlen) == 0) {
        // success
        freeaddrinfo(res);
        break;
    }
    close(fd);
}