Bygg en chattserver med Boost.Asio, del 2 - Asynkront med callbacks

Tvåhundra klienter gav 202 trådar, och en enda klient som slutade läsa fick hela chattservern att stanna. Så slutade förra delen, och det var inte ett fel i koden utan en egenskap hos synkron I/O: varje blockerande anrop binder en tråd, och trådar är en knapp resurs. I den här delen skriver vi om samma server med Boost.Asios asynkrona operationer. Resultatet är en server som hanterar alla klienter i en tråd, inte behöver en enda mutex, och där en långsam klient bara sinkar sig själv. Det kostar något annat i stället, nämligen att koden byter form. Varje anrop returnerar direkt, och det som ska hända efteråt ligger i en callback.

Jag förutsätter att du läst förra delen, eller i alla fall känner till io_context, I/O-objekt och error_code. Trådar och shared_ptr dyker upp igen, men den här gången i en skepnad du kanske inte sett förut.

Artikeln är den andra av fem. Vi bygger vidare på samma chattserver genom hela serien, och den här delen är den där servern byter från synkront till asynkront. Det bytet är det största steget i hela serien, för allt som kommer sedan, coroutines, TLS och WebSocket, bygger på den asynkrona modellen vi lägger fast här. Alla program finns i ett eget GitHub-repo, se länkarna sist i artikeln. Artikeln visar de rader som bär berättelsen, och i repot kan du se all programkod, plus att det finns instruktioner om hur du bygger och kör samtliga demoprogram.

Den asynkrona modellen

I förra delen anropade vi aldrig io_context::run(). Synkrona operationer blockerar tråden själva och behöver ingen eventloop. Nu blir run() det viktigaste anropet i programmet, så låt oss börja med det minsta program som visar vad det gör. Inga sockets, bara en timer:

int main() {
    auto start = clk::now();
    auto ioc = asio::io_context{};
    auto timer = asio::steady_timer{ioc, 2s}; // initiera med 2 sekunders fördröjning

    timer.async_wait([start](boost::system::error_code ec) {
        std::println("[{:5} ms] handlern körs i tråd {}: {}",
                     ms_since(start), std::this_thread::get_id(), ec ? ec.message() : "timern gick av");
    });
    std::println("[{:5} ms] async_wait returnerade direkt, huvudtråden är {}",
                 ms_since(start), std::this_thread::get_id());

    ioc.run();
    std::println("[{:5} ms] run() returnerade: inget mer arbete", ms_since(start));
}

$ ./build/release/timer/timer
[    0 ms] async_wait returnerade direkt, huvudtråden är 133791323453312
[ 2000 ms] handlern körs i tråd 133791323453312: timern gick av
[ 2000 ms] run() returnerade: inget mer arbete

Tre saker att ta med sig. async_wait() startar operationen och returnerar omedelbart, vid 0 ms. Handlern, lambdan, körs två sekunder senare, och den körs inne i run(), i den tråd som anropade run(). Det är samma tråd-id på båda raderna. Och run() returnerar när det inte finns mer arbete, det vill säga när inga asynkrona operationer längre är utestående. Det sista är viktigt för en server: så länge en async_accept() väntar på nästa klient har run() något att göra, och programmet lever vidare. Det fungerar på ett liknande sätt i ett Node.js-program, med skillnaden att man där inte startar eventloopen själv.

Modellen är alltså: initiera, returnera, och få resultatet levererat senare. Den tråd som kör run() är den som levererar. Allt annat i den här artikeln är variationer på detta.

Completion tokens

Sista argumentet till varje async_*-funktion kallas completion token. Det är denna som avgör hur resultatet levereras. I den här artikeln är token alltid en callback-funktion (eller lambda), det vill säga något anropbart (callable) med rätt signatur, och det är den enklaste formen. Men samma funktion tar också andra typer av tokens: use_future ger en std::future, och use_awaitable gör att anropet kan co_await:as. Det är hela poängen med nästa artikel, så jag nämner det bara.

Signaturen för handlern beror på operationen. En timer ger (error_code), en läsning eller skrivning ger (error_code, std::size_t) med antalet överförda byte, och async_accept ger (error_code, tcp::socket). Den första parametern är alltid en error_code, och det är en konsekvens av modellen. Det finns ingen att kasta ett exception till: när handlern körs är det anropande kodstycket sedan länge borta. Så tumregeln från förra delen, exceptions som default och error_code där fel är normalt flöde, byter till error_code överallt. Exceptions finns kvar bara i main, för fel vid uppstart som "porten är upptagen".

För dig som kan Node.js. Det här är Nodes error-first callbacks: (err, data) i Node motsvarar (error_code, n) i Asio. Skillnaden är eventloopen. I Node finns den, och den startar så fort ditt skript nått slutet. I Asio heter den io_context, och du startar den själv med run(). Det låter som en detalj, men det är det som gör att du kan välja hur många trådar som kör loopen, vilket vi gör i steg 2b. En annan skillnad är push mot pull: en Node-ström pushar 'data'-events till dig så länge du lyssnar, medan Asio bara läser när du ber om det. Varje async_read_some() ger exakt en leverans, och vill du ha nästa ber du om den igen.

Echo-server med callbacks

Nu skriver vi om echo-servern från förra delen. Den synkrona versionen var en loop: läs, skriv, läs, skriv. Den asynkrona är samma loop, men utan loop-sats. Varje steg startar nästa:

class Session : public std::enable_shared_from_this<Session> {
    using super = std::enable_shared_from_this<Session>;

    tcp::socket socket;
    std::string peer;
    std::array<char, 1024> buf{};  // medlem, inte lokal: måste leva tills handlern körs

public:
    explicit Session(tcp::socket sock)
        : socket{std::move(sock)}, peer{to_string(socket.remote_endpoint())} {}

    ~Session() { std::println("nedkopplad: {}", peer); }

    void start() {
        std::println("ansluten: {}", peer);
        do_read();
    }

private:
    void do_read() {
        auto self = super::shared_from_this();
        socket.async_read_some(asio::buffer(buf), [self](boost::system::error_code ec, std::size_t n) {
            if (ec) {
                if (ec != asio::error::eof) std::println("fel mot {}: {}", self->peer, ec.message());
                return;  // ingen ny operation: sessionen dör när self går ur scope
            }
            self->do_write(n);
        });
    }

    void do_write(std::size_t n) {
        auto self = super::shared_from_this();
        asio::async_write(socket, asio::buffer(buf, n), [self](boost::system::error_code ec, std::size_t) {
            if (ec) {
                std::println("fel mot {}: {}", self->peer, ec.message());
                return;
            }
            self->do_read();
        });
    }
};

do_read() startar en läsning, och när den är klar startar handlern en skrivning. När skrivningen är klar startar dess handler en ny läsning. Kedjan do_read → do_write → do_read är den synkrona loopen, utspridd över två funktioner och två lambdas. Basklassen och raden auto self = ... förklarar jag i nästa avsnitt. Först accept-loopen, som följer samma mönster:

void do_accept(tcp::acceptor& acceptor) {
    acceptor.async_accept([&acceptor](boost::system::error_code ec, tcp::socket sock) {
        if (ec) {
            std::println("accept: {}", ec.message());
            return;
        }
        std::make_shared<Session>(std::move(sock))->start(); // hantera ny klient
        do_accept(acceptor); // vänta på nästa klient
    });
}

int main(int argc, char** argv) try {
    auto port = static_cast<unsigned short>(argc > 1 ? std::stoi(argv[1]) : 7000);

    auto ioc = asio::io_context{};
    auto acceptor = tcp::acceptor{ioc, tcp::endpoint{tcp::v4(), port}};
    std::println("echo-server (callbacks) lyssnar på port {}", port);

    do_accept(acceptor);
    ioc.run();
} catch (std::exception const& e) {
    std::println(stderr, "fel: {}", e.what());
    return 1;
}

Varje accepterad klient blir en Session, som startar sin läskedja, och sedan startas nästa accept. Notera att main inte gör annat än att starta den första accepten och sedan anropa run(). Allt arbete sker i handlers, och alla handlers körs i run().

Starta servern och anslut med två nc i varsin terminal:

$ nc localhost 7000
hej
hej
hallå igen
hallå igen

$ nc localhost 7000
jag också
jag också

Båda får sitt eko direkt. I förra delen hängde den andra klienten i accept-kön tills den första kopplade ned, och nu betjänas de samtidigt, i en enda tråd. Det är den första konkreta vinsten med den asynkrona modellen. Observera att det fortfarande är en echo-server: var och en får tillbaka sin egen text, och ingen ser den andras. Broadcast mellan klienter är chattserverns jobb, och den kommer strax.

echo-server (callbacks) lyssnar på port 7000
ansluten: 127.0.0.1:54794
ansluten: 127.0.0.1:54800
nedkopplad: 127.0.0.1:54800
nedkopplad: 127.0.0.1:54794

Livslängd och ägarskap

Här kommer den asynkrona modellens egentliga svårighet. Ett async_read_some() returnerar direkt, och handlern körs någon gång senare. Under tiden måste socketen och bufferten fortfarande vara aktiva. I den synkrona versionen var bufferten en lokal variabel i echo(), och det fungerade, eftersom funktionen var kvar i read_some() tills datan kommit. Nu returnerar do_read() omedelbart, och en lokal buffert hade dött (gått ur scope) långt innan handlern anropats. Därför är buf en medlemsvariabel i Session, så vi har en garanti att denna är aktiv så länge dess omslutande Session-objekt är aktivt.

Men vem äger Session? Titta på do_accept(): den skapar sessionen med make_shared, anropar start() och kastar bort pekaren. Ingen lista, ingen tråd, ingen variabel någonstans håller i den. Ändå lever den tills klienten kopplar ned. Det är basklassen std::enable_shared_from_this som gör det möjligt, och den förtjänar en ordentlig förklaring, för de flesta har aldrig sett den här basklassen.

Problemet den löser är att du inne i en medlemsfunktion bara har this, en rå pekare, men behöver en shared_ptr som delar ägarskap med den shared_ptr som redan äger objektet. Frestelsen är att skriva std::shared_ptr<Session>{this}. Det kompilerar, men skapar en ny, oberoende ägarkedja med ett eget kontrollblock och räknaren 1, för ett objekt som redan ägs av någon annan. När båda räknarna når noll får du två delete på samma objekt. En shared_ptr har ingen aning om att det finns en annan, för räknaren sitter i kontrollblocket, inte i objektet.

enable_shared_from_this<Session> innehåller en enda medlem: en std::weak_ptr<Session>. När du skapar objektet med std::make_shared<Session>(...) kontrollerar make_shared om typen ärver från just den basklassen, och i så fall fyller den i weak_ptr:n så att den pekar på det kontrollblock som precis skapades. Därefter kan du när som helst anropa shared_from_this(), som gör weak_ptr.lock() och ger dig en shared_ptr till samma kontrollblock, med räknaren uppräknad. Mallargumentet <Session> behövs för att basklassen ska veta vilken typ weak_ptr:n ska ha, och arvet måste vara public för att make_shared ska hitta den. I koden deklarerar jag using super = std::enable_shared_from_this<Session>; och skriver super::shared_from_this(), bara för att det ska synas varifrån metoden kommer.

Villkoret är att objektet redan ägs av en shared_ptr när shared_from_this() anropas. I konstruktorn finns ingen sådan ännu, och på ett stackobjekt finns den aldrig. Då kastas std::bad_weak_ptr. Det är därför Session har en separat start(), och do_accept() skriver make_shared<Session>(...)->start(): först ägarskap, sedan första asynkrona anropet.

Och nu svaret på frågan om vem som äger sessionen: den utestående operationen. Lambdan i do_read() fångar self som värde, och Asio lagrar lambdan tills operationen är klar. Så länge kedjan do_read → do_write → do_read fortsätter finns alltid minst en handler som håller en shared_ptr. När klienten kopplar ned ser handlern eof, startar ingen ny operation och returnerar. Asio förstör lambdan, räknaren går till noll, destruktorn körs, och socketen stängs. Objektet äger sig självt genom sina egna pågående operationer, och städar efter sig när de tar slut. Raden "nedkopplad" i serverloggen skrivs av destruktorn.

Hade lambdan fångat [this] i stället, så hade objektet dött direkt efter start(), och handlern hade skrivit i frigjort minne. Det är det klassiska felet i callback-baserad Asio-kod, och anledningen till att nästan all sådan kod har raden auto self = shared_from_this(); överst i varje funktion som startar en operation.

I förra delen ägdes Session av läs-tråden, klientlistan och broadcast-trådens kopia. Här finns inga trådar och ingen lista, så ägarna är i stället de handlers som väntar på att få köras. Samma shared_ptr-mekanik, men ägarna är operationer i stället för trådar.

Chattserver, steg 2a: en tråd

Nu till chattservern. Protokollet är detsamma som förut: varje rad från en klient skickas till alla andra. Men designen är ny. I stället för en läs-tråd per klient och en broadcast-tråd har vi tre klasser som alla kör i samma tråd:

Server:   accept-loop, skapar en Session per klient
Session:  async_read_until en rad  -->  room.broadcast()
          deliver() lägger meddelandet i sessionens skrivkö, async_write skriver ett i taget
Room:     mängden av sessioner, broadcast till alla utom avsändaren

Rummet först, för det är kortast. Jämför med ClientList från förra delen: ingen mutex, ingen kopia.

class Room {
    std::set<SessionPtr> sessions;

public:
    void join(SessionPtr session) { sessions.insert(std::move(session)); }
    void leave(SessionPtr const& session) { sessions.erase(session); }
    void broadcast(std::string const& msg, SessionPtr const& from) {
        for (auto const& session : sessions) {
            if (session != from) session->deliver(msg);
        }
    }
    auto size() const -> std::size_t { return sessions.size(); }
};

Alla anrop till Room sker från handlers, och alla handlers körs i den enda tråd som anropat run(). Två handlers kan aldrig köras samtidigt, så det finns inget att skydda. Det är samma garanti som Node.js ger, och den är anledningen till att Node-kod nästan aldrig behöver tänka på lås.

Sessionen är större, så jag visar den i delar. Läsningen känner du igen från echo-servern, fast med async_read_until och en dynamic_buffer, precis som i förra delens read_until, och med samma fallgrop om att bara radera de n byte som utgör raden:

void do_read() {
    auto self = super::shared_from_this();
    asio::async_read_until(socket, asio::dynamic_buffer(read_buf), '\n',
        [self](boost::system::error_code ec, std::size_t n) {
            if (ec) {
                self->close();
                return;
            }
            auto line = strip_eol(std::string_view{self->read_buf}.substr(0, n));
            self->room.broadcast(std::format("[{}] {}\n", self->name, line), self);
            self->read_buf.erase(0, n);
            self->do_read();
        });
}

Det nya är skrivsidan, och det är här förra delens head-of-line blocking löses. Room::broadcast() anropar deliver() på varje session, och deliver() får inte blockera, för då står hela servern stilla. Den får inte heller anropa async_write() rakt av, för async_write är en composed operation som gör flera write_some() tills allt är skickat, och startar man en ny innan den förra är klar flätas de två meddelandena ihop. Lösningen är en kö per session:

constexpr auto max_queue = std::size_t{1000};
//...
std::deque<std::string> queue;        // väntar på att skrivas
std::optional<std::string> in_flight; // skrivs just nu, får inte röras
std::size_t dropped = 0;

bool is_full() const { return queue.size() >= max_queue; }

void deliver(std::string msg) {
    if (is_full()) {
        queue.pop_front();  // äldsta ej påbörjade. in_flight rörs inte.
        ++dropped;
    }
    queue.push_back(std::move(msg));
    if (!in_flight) do_write();
}

void do_write() {
    in_flight = std::move(queue.front());
    queue.pop_front();

    auto self = super::shared_from_this();
    asio::async_write(socket, asio::buffer(*in_flight),
        [self](boost::system::error_code ec, std::size_t) {
            if (ec) {
                self->close();
                return;
            }
            self->in_flight.reset();
            if (!self->queue.empty()) self->do_write();
        });
}

deliver() lägger meddelandet sist i kön, och om ingen skrivning pågår startar den en. Skriv-handlern tar nästa ur kön, tills den är tom. Det garanterar att det bara finns en async_write i taget per socket, och att meddelandena kommer i rätt ordning. Och det garanterar att deliver() returnerar på mikrosekunder oavsett hur klienten mår. Om klienten slutat läsa fylls kärnans buffertar, async_write blir inte klar, och kön växer. Men det är den klientens kö. Alla andra fortsätter.

Kön får inte växa obegränsat, så den har ett tak. När den är full kastas det äldsta meddelandet som ännu inte börjat skrivas. För en chatt är det rimligt: ett meddelande som inte levererats på länge är värt mindre än ett nytt, och en klient som pausat en stund får det senaste när den kommer tillbaka, inte en flodvåg av gammalt. Lägg märke till att meddelandet som just nu skrivs ligger utanför kön, i in_flight. async_write refererar sin buffert tills handlern körts, och hade meddelandet legat kvar i kön hade pop_front() kunnat frigöra minne som en pågående skrivning läser från. Det är "bufferten måste leva tills handlern körs" i praktiken.

Hur stort ska taket vara? Jag satte det först till 100, och då kastades meddelanden hos alla sessioner när simulatorn körde 200 klienter, trots att de läste flitigt. Orsaken är skurar. När N klienter skickar en rad i samma ögonblick körs N läs-handlers i följd innan någon skriv-handler hinner köras, och varje kö växer med N. Jag verifierade det med 200 klienter som alla skickar efter exakt två sekunder. Taket är nu 1000, och kön måste rymma den största skur du väntar dig.

Till sist close(), som anropas från både läs- och skriv-handlern, och Server:

void close() {
    if (!socket.is_open()) return;
    auto ec = boost::system::error_code{};
    socket.close(ec);

    auto self = super::shared_from_this();
    room.leave(self);
    std::println("nedkopplad: {} ({} kvar{})", name, room.size(),
                 dropped ? std::format(", kastade {} meddelanden", dropped) : "");
    room.broadcast(std::format("*** {} kopplade ned\n", name), self);
}

class Server {
    tcp::acceptor acceptor;
    Room room;

public:
    Server(asio::io_context& ioc, unsigned short port)
        : acceptor{ioc, tcp::endpoint{tcp::v4(), port}} {
        do_accept();
    }

private:
    void do_accept() {
        acceptor.async_accept([this](boost::system::error_code ec, tcp::socket sock) {
            if (ec) return;
            std::make_shared<Session>(std::move(sock), room)->start();
            do_accept();
        });
    }
};

close() är idempotent av en anledning. Stänger läs-handlern socketen medan en skrivning pågår, så avbryts skrivningen med operation_aborted, och dess handler hamnar i close() igen. Då är socketen redan stängd. Och Server fångar [this], inte self, för den är en lokal variabel i main och lever tills run() returnerar. Skillnaden mot Session är just att någon äger Server.

Nu chattar det, i en tråd

Två nc ger samma utskrift som i förra delen, med avsändaren som prefix och "anslöt" och "kopplade ned" som systemmeddelanden. Det intressanta är kostnaden. Simulatorn från förra delen följer med i repot, och medan den kör 200 klienter räknar vi trådar:

$ client-simulator/chat-sim.py --clients 200 --messages 5
$ ps -T -p $(pgrep -x chatserver-one-) | tail -n +2 | wc -l
1

En tråd. Förra gången var det 202. Och head-of-line-experimentet, där en nc pausas med Ctrl-Z, en annan nc lyssnar, och en tredje klient skickar 40 000 rader på 200 tecken? Förra gången stannade allt efter 3,4 MB. Nu får den vakna klienten alla 40 000 rader, och den pausade fyller sin egen kö. Tar man tillbaka den med fg får den de senaste tusen raderna, och när den avslutas loggar servern:

nedkopplad: 127.0.0.1:44734 (1 kvar, kastade 24208 meddelanden)

Båda problemen från förra delen är lösta, med färre rader kod än den trådade versionen, och utan en enda mutex.

Chattserver, steg 2b: N trådar

En tråd räcker förvånansvärt långt. Nästan allt en chattserver gör är att vänta, och en väntande tråd behöver inte sällskap. Men förr eller senare vill man använda fler kärnor, och med Asio är det enkelt: anropa run() från flera trådar på samma io_context. Det är allt. Handlers plockas ur samma kö av alla trådarna, och kan nu köras parallellt.

auto num_threads = argc > 2 ? std::stoul(argv[2]) : std::thread::hardware_concurrency();

auto ioc = asio::io_context{};
auto server = Server{ioc, port};

auto threads = std::vector<std::jthread>{};
for (auto i = 0u; i < num_threads; ++i) {
    threads.emplace_back([&ioc] { ioc.run(); });
}

Det är den enda ändringen i första försöket. Och nu har vi förstört garantin som hela steg 2a vilade på. Två läs-handlers för två olika klienter körs samtidigt i två trådar, och båda anropar room.broadcast(), som itererar över samma std::set och anropar deliver() på samma sessioner, som skriver till samma std::deque. Det är en data race, och data races är odefinierat beteende. Ibland kraschar det, ibland tappas ett meddelande, och ibland händer ingenting på flera dagar.

Så i stället för att vänta på kraschen frågar vi ThreadSanitizer. Repot har ett CMake-preset tsan som bygger med -fsanitize=thread, och programmet chatserver-race är steg 2a med trådloopen ovan. Kör det med fyra trådar under simulatorn:

$ ./build/tsan/chatserver-race/chatserver-race 4711 4
...
WARNING: ThreadSanitizer: data race (pid=753483)
  Read of size 8 at 0x724400014230 by thread T3:
    #2 Session::is_full() const chatserver.cxx:99
    #3 Session::deliver(std::string) chatserver.cxx:90
  Previous write of size 8 at 0x724400014230 by thread T1:
    #2 std::deque::push_back(std::string&&)
    #3 Session::deliver(std::string) chatserver.cxx:94
SUMMARY: ThreadSanitizer: data race chatserver.cxx:99 in Session::is_full() const
...
ERROR: ThreadSanitizer: SEGV on unknown address 0x000000000000
SUMMARY: ThreadSanitizer: SEGV chatserver.cxx:94 in Session::deliver(std::string)

Med 50 klienter gav det 102 rapporter, och till slut en krasch i deliver(), för en std::deque som två trådar skriver till samtidigt är inte en deque länge. Tråd T3 läser köns storlek medan tråd T1 lägger till. Det är exakt det vi förutsåg, och sanitizern pekar ut raderna.

Den reflexmässiga lösningen är en mutex i Session och en i Room. Det fungerar, men det är fel verktyg. En mutex får en tråd att vänta, och hela poängen med den asynkrona modellen är att trådarna aldrig väntar. Asio har ett bättre svar.

Strand:ar

Ordet strand är engelska för en kardel, alltså en av de parter som tvinnas till ett rep, och har inget med sand och sol att göra. Bilden är ett rep där flera kardeler löper parallellt, men var och en är en enda sammanhängande sträng. Det är precis vad en strand i Asio är: en executor med en enda garanti. Handlers som skickas genom samma strand körs aldrig samtidigt, och i den ordning de skickades.

Inget mer än så. En strand binder inte handlers till en viss tråd. Vilken som helst av de N trådarna kan köra nästa handler på strand:en, men bara en i taget. Internt är den en liten kö plus en flagga som säger "någon kör just nu". Vill en tråd köra en handler på en strand som är upptagen, så lägger den handlern i strand:ens kö och går vidare med annat arbete. Ingen tråd blockerar någonsin. Det är skillnaden mot en mutex: en mutex får tråden att vänta, en strand får arbetet att vänta.

Regeln för att använda dem är samma som för lås: varje bit delad data har exakt en strand, och all kod som rör datan går genom den. I chattservern finns två sorters delad data, sessionen (socket, läsbuffert, skrivkö) och rummet (mängden av sessioner), och alltså två sorters strand:ar. Sessionens strand får vi nästan gratis, genom att ge en executor som första argument till async_accept:

void do_accept() {
    acceptor.async_accept(asio::make_strand(ioc), [this](boost::system::error_code ec, tcp::socket sock) {
        if (ec) return;
        std::make_shared<Session>(std::move(sock), room)->start();
        do_accept();
    });
}

Socketen skapas då med en ny strand som sin executor, och alla handlers för operationer på den socketen körs automatiskt på den strand:en. do_read() och do_write() är oförändrade från steg 2a, och close() har bara tappat sin loggrad. Ändå är de nu trådsäkra, för två av sessionens handlers kan aldrig köras samtidigt.

Rummet har ingen socket, så det får en strand explicit, och dess publika metoder postar sitt arbete dit:

class Room {
    asio::strand<asio::io_context::executor_type> strand;
    std::set<SessionPtr> sessions;

public:
    explicit Room(asio::io_context& ioc) : strand{asio::make_strand(ioc)} {}

    void broadcast(std::string msg, SessionPtr from) {
        asio::post(strand, [this, msg = std::move(msg), from = std::move(from)] {
            for (auto const& session : sessions) {
                if (session != from) session->deliver(msg);
            }
        });
    }
    ...
};

asio::post(strand, f) lägger f i strand:ens kö och returnerar direkt. Någon av trådarna kör f senare, när strand:en är ledig. Så broadcast() anropas från sessionens strand, men själva loopen över sessions körs på rummets. Och eftersom loopen anropar deliver() på sessioner som tillhör andra strand:ar, får deliver() inte röra kön direkt utan måste posta i sin tur:

void deliver(std::string msg) {
    auto self = super::shared_from_this();
    asio::post(socket.get_executor(), [self, msg = std::move(msg)]() mutable {
        self->enqueue(std::move(msg));  // kön från steg 2a, nu på sessionens strand
    });
}

Flödet för ett meddelande blir alltså: A:s läs-handler på A:s strand postar till rummets strand, rummets handler postar till B:s strand, och B:s handler lägger i kön och startar skrivningen. Tre hopp, inga lås. Varje bit data rörs bara från sin egen strand.

Det finns också asio::dispatch(strand, f), som kör f direkt om anroparen redan står på rätt strand, och köar annars. Det sparar ett hopp när man ändå är på plats, men post är lättare att resonera om: den köar alltid, så handlern körs aldrig inne i anroparen.

En sak till ändrades. I steg 2a loggade Session antalet anslutna med room.size(). Nu vore det en läsning av sessions från sessionens strand, alltså en race, så Room::join() och Room::leave() loggar själva, på rummets strand. Det är så strand:ar påverkar designen: data och kod som rör den flyttar ihop.

Samma körning som förut, nu med chatserver-strands:

$ ./build/tsan/chatserver-strands/chatserver-strands 4711 4
chattserver (strand:ar) lyssnar på port 4711 med 4 trådar
ansluten: 127.0.0.1:37728 (1 totalt)
...
nedkopplad: 127.0.0.1:38062 (0 kvar)

Inga varningar. Och i release-bygget med standardantalet trådar och 200 simulatorklienter visar ps -T nio trådar på min maskin, åtta som kör run() plus main som väntar, och inga kastade meddelanden.

För dig som kan Node.js. Det här avsnittet har ingen motsvarighet i Node. En Node-process har en eventloop och en tråd, punkt. worker_threads ger fler trådar, men de delar inte sockets eller objekt, utan pratar med varandra via meddelanden. Asio låter dig köra samma eventloop i flera trådar, och strand:ar är priset för det. Den andra skillnaden är livslängd. I Node hålls en socket vid liv av att eventloopen refererar den, och GC tar resten. I C++ är shared_from_this() det som gör samma jobb: varje utestående operation håller en referens, och när sista handlern kört är objektet borta.

Callback-helvetet

Chattservern är ett tacksamt program för callbacks, för den är en loop: läs, broadcast, läs. Varje steg startar samma steg igen, och kedjan blir aldrig djupare än två. Men vad händer med ett program som är en sekvens? Daytime-klienten från förra delen gör tre saker i tur och ordning: slå upp namnet, anslut, läs svaret. Synkront var det tre rader. Asynkront blir det så här:

resolver.async_resolve(host, port, [&](boost::system::error_code ec, tcp::resolver::results_type endpoints) {
    if (ec) {
        std::println(stderr, "fel: resolve: {}", ec.message());
        status = 1;
        return;
    }

    asio::async_connect(sock, endpoints, [&](boost::system::error_code ec, tcp::endpoint ep) {
        if (ec) {
            std::println(stderr, "fel: connect: {}", ec.message());
            status = 1;
            return;
        }
        std::println("ansluten till {}", to_string(ep));

        asio::async_read(sock, asio::dynamic_buffer(reply), [&](boost::system::error_code ec, std::size_t) {
            if (ec && ec != asio::error::eof) {
                std::println(stderr, "fel: read: {}", ec.message());
                status = 1;
                return;
            }
            std::print("svar: {}", reply);
        });
    });
});

ioc.run();
return status;

Tre steg, tre nivåer. Varje handler har sin egen felkontroll, och koden för steg två ligger inne i handlern för steg ett. Lägg till ett fjärde steg, så blir det en fjärde nivå. Och lägg märke till status: en lambda kan inte returnera från main, så felkoden måste smugglas ut via en variabel. Vill man ha en loop, säg "läs rader tills servern stänger", går det inte att skriva en for-sats, utan loopen måste bli en funktion som anropar sig själv via sin egen handler, precis som do_read() i servern.

Det är det här som kallas callback-helvetet, och det är inte specifikt för C++. Node-världen gick igenom exakt samma sak, och svaret där blev Promises och sedan async/await. C++ har gått samma väg, och svaret heter coroutines. Nästa del skriver om daytime-klienten, echo-servern och chattservern med co_await, och daytime-klienten blir tre rader igen, fast fortfarande asynkron. Samma async_resolve, samma async_connect, samma async_read, med en annan completion token.

Sammanfattning

Asios asynkrona modell är tre steg: initiera, returnera direkt, och få resultatet levererat till en handler som körs inne i io_context::run(). Sista argumentet till varje async_*-funktion är en completion token, och en callback är den enklaste. Handlern får alltid en error_code, för det finns ingen att kasta till.

Priset är livslängd. Buffertar blir medlemmar i stället för lokala variabler, och en session måste äga sig själv genom sina utestående operationer. Det är vad enable_shared_from_this och auto self = shared_from_this() gör: varje handler håller en shared_ptr, och när sista handlern kört är objektet borta.

Chattservern i steg 2a löste båda problemen från förra delen. En tråd i stället för 202, och en skrivkö per session, med ett tak som kastar äldst, så att en klient som slutar läsa bara sinkar sig själv. Ingen mutex, eftersom alla handlers körs i samma tråd. Steg 2b körde run() i åtta trådar, lät ThreadSanitizer visa racen, och löste den med strand:ar: en per session, som socketen fick via async_accept, och en för rummet, som dess metoder postar till. En strand utför ett jobb liknande en mutex, men låter arbetsuppgiften vänta i stället för exekveringstråden.

Och så såg vi vad callbacks kostar när programmet är en sekvens i stället för en loop. Tre steg blev tre nivåer av nästlade lambdas med varsin felkontroll. Det är det nästa del löser med coroutines, och det bästa är att det är samma API: samma funktioner, samma io_context, samma strand:ar, bara en annan completion token.