A coroutine nyelvi mechanizmus, nem I/O-runtime: a
co_await felfüggeszti a coroutine-t, de a folytatásról, az
eseményekről és az erőforrások élettartamáról a program gondoskodik. Ez
a labor egy kis poll-alapú eseményloophoz készít read/write
awaitable-t, hogy látható legyen a teljes szerződés: ki birtokolja a
buffert, meddig él az operation, hogyan lehet cancelálni, és hogyan
biztosítható a pontosan egyszeri completion. A co_await
önmagában nem indít szálat és nem végez hálózati I/O-t; a scheduler, az
I/O és a cancellation policy külön implementációs szerződés. A rész a generátorok felfüggesztési mintájára, a poll/epoll eseményállapotgépre és a jthread cancellation fogalmaira épül, de azokat nem
ismétli meg.
Az aszinkron I/O adapter feladata, hogy egy operációs rendszerbeli eseményt coroutine-felfüggesztéssé és folytatássá alakítson. Ehhez négy fogalmat kell pontosan rögzíteni.
Operation (művelet): egy regisztrált I/O-kérés, amelyhez fájlleíró, várt esemény, completion-művelet és buffer tartozik. Az operation addig él, amíg a completion vagy a cancellation le nem zárja.
Awaitable: a co_await operandusa. Három
tagfüggvénye van: await_ready() eldöntheti, hogy nincs
szükség felfüggesztésre; await_suspend(handle) regisztrálja
a műveletet és felfüggeszti a coroutine-t; await_resume()
átadja az eredményt. Az awaitable élettartama a teljes
co_await kifejezés, beleértve a felfüggesztést is.
Completion: az operation eredményének átadása a coroutine-nak. Pontosan egyszer történhet meg. Utána az operation nem hivatkozható, a hozzá tartozó awaitable pedig hamarosan megsemmisül.
Cancellation: az operation eredmény nélküli
lezárása. A coroutine vagy hibát kap az await_resume()-ban,
vagy a coroutine frame megsemmisül, és az awaitable destruktora
deregisztrálja a műveletet.
A szerződés feltételei a következők. Az awaitable élettartamának fednie kell a felfüggesztést; ha a coroutine frame megsemmisül felfüggesztett állapotban, az awaitable destruktorának el kell távolítania az operationt az eseménylooptól. A buffer tulajdonosa a hívó marad; az adapter nem másol, csak pointert tárol, ezért a buffer élettartama legalább a completionig tart. A terminális események — completion, cancel, timeout — közül pontosan egy nyerhet. A versenyt az eseményloop szálán, az operation eltávolításával kell eldönteni: amelyik ág előbb eltávolítja az operationt, az nyer, a másik ág már nem hivatkozhat rá. Végül a shutdown alatt új operation nem regisztrálható, a meglévőket pedig drainelni vagy cancelálni kell; az outstanding operation számlálója ezt teszi ellenőrizhetővé.
A callback, az explicit állapotgép és a coroutine ugyanazt a
folyamatot írja le, de más szerződéssel. A callbacknél az állapot egy
structban él, és a struct élettartama a cancellation határa. Az explicit
állapotgépnél az állapot egy enum és egy struct, a vezérlést switch
adja. A coroutine-nál az állapot a coroutine frame-ben van, a vezérlés
lineáris, a cancellation a frame és az awaitable élettartamához kötött.
A coroutine frame allokálhat, és a
promise_type::operator new testreszabható, de a coroutine
nem feltétlenül gyorsabb a callbacknél.
| Szempont | Callback | Explicit állapotgép | Coroutine |
|---|---|---|---|
| Állapot helye | struct | enum + struct | coroutine frame |
| Vezérlés | visszahívási lánc | switch | lineáris kód |
| Cancel | a struct élettartama | state flag | frame + awaitable |
| Allokáció | stack vagy heap | stack vagy heap | frame, testreszabható |
| Debug | callback-lánc | állapotváltozó | frame és handle |
A Loop poll-t használ, és Op
rekordokat tart nyilván. Az Op tartalmazza a fájlleírót, a
várt eseményt, a felfüggesztett coroutine handle-jét, a
completion-függvényt és az awaitable-re mutató owner
pointert. Az awaitable az Op-ot saját tagjaként tárolja,
tehát az Op a coroutine frame-ben él. A Loop
csak pointert tárol rá, nem birtokolja. Ez teszi lehetővé, hogy a
completion után az awaitable destruktora remove-olja
magát.
A run() ciklus minden iterációban pollfd
tömböt épít és poll-ol. Esemény esetén először a completion
fut, amely a read/write rendszerhívást végzi;
a döntést ennek eredménye hozza, nem önmagában a
poll-esemény. Átmeneti hibánál (EAGAIN,
EINTR) a completion hamisat ad, és a művelet regisztrálva
marad a következő körre; egyébként a Loop eltávolítja a
műveletet, és felébreszti a coroutine-t. Az eredményt és a végleges
hibát az awaitable result és error tagja
hordozza.
Az alábbi példa egyetlen fájlban, oktatási célra készült. Nem éles
runtime: egy szálon fut, és a Loop nem szálbiztos. A
descriptorprofil rögzített: az adapter csak
O_NONBLOCK módú pipe- vagy socket-fd-vel használható, mert
csak így garantált, hogy egy poll utáni
read/write nem blokkolja az egész loopot, és
egy lassú kimenet (EAGAIN) nem állítja meg a többi
műveletet. Az awaitable az fd-t kölcsönzi; a tulajdonos a hívó
unique_fd-je (RAII wrapper file
descriptorhoz és lockhoz (21.2)). A Loop háromállapotú:
nyitott (fogad), leálló (új regisztrációt elutasít, a függőket
ECANCELED-del lemondja) és leállt. A laborban a fordítás és
futtatás a tanuló feladata; a kód a megadott profillal
ellenőrizendő.
// echo_coro.cpp
// Descriptorprofil: az adapter csak O_NONBLOCK módú pipe/socket fd-vel
// használható. Az awaitable az fd-t kölcsönzi, nem birtokolja.
#include <algorithm>
#include <cerrno>
#include <coroutine>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <exception>
#include <fcntl.h>
#include <poll.h>
#include <stdexcept>
#include <unistd.h>
#include <utility>
#include <vector>
// A [RAII wrapper file descriptorhoz] rész unique_fd-je, rövidítve.
class unique_fd {
public:
unique_fd() noexcept = default;
explicit unique_fd(int fd) noexcept : fd_(fd) {}
unique_fd(const unique_fd&) = delete;
unique_fd& operator=(const unique_fd&) = delete;
unique_fd(unique_fd&& o) noexcept : fd_(std::exchange(o.fd_, -1)) {}
~unique_fd() { reset(); }
[[nodiscard]] int get() const noexcept { return fd_; }
void reset(int fd = -1) noexcept {
if (fd_ >= 0) (void)::close(fd_);
fd_ = fd;
}
private:
int fd_ = -1;
};
#ifdef INJECT_TRANSIENT
// Tesztbuild: az első két read és az első két write EAGAIN, majd EINTR hibát ad.
static int injected = 0;
static ssize_t sys_read(int fd, void* b, std::size_t n) {
if (injected < 2) { errno = injected++ ? EINTR : EAGAIN; return -1; }
return ::read(fd, b, n);
}
static int injected_w = 0;
static ssize_t sys_write(int fd, const void* b, std::size_t n) {
if (injected_w < 2) { errno = injected_w++ ? EINTR : EAGAIN; return -1; }
return ::write(fd, b, n);
}
#else
static ssize_t sys_read(int fd, void* b, std::size_t n) { return ::read(fd, b, n); }
static ssize_t sys_write(int fd, const void* b, std::size_t n) { return ::write(fd, b, n); }
#endif
struct Op {
std::uint64_t id = 0; // a Loop adja; a pillanatkép ezt tárolja
int fd = -1;
short events = 0;
std::coroutine_handle<> waiter{};
// A rendszerhívást végzi; hamis, ha átmeneti hiba miatt újra kell próbálni.
bool (*complete)(Op*) = nullptr;
void* owner = nullptr;
int cancel_error = 0; // ECANCELED, ha a Loop lemondta a műveletet
};
class Loop {
public:
// Nyitott: fogad műveletet. Leálló: nem fogad újat, a függőket lemondja.
// Leállt: a run() visszatért, nincs függő művelet.
enum class State { open, stopping, stopped };
// Hamis, ha a Loop már nem fogad műveletet: ilyenkor a hívó nem függeszt fel.
[[nodiscard]] bool add(Op* op) {
if (state_ != State::open) return false;
op->id = next_id_++;
ops_.push_back(op);
return true;
}
// Csak a még regisztrált műveletet adja vissza. A megszűnt awaitable
// destruktora remove-olja az Op-ot, így annak azonosítója nem található.
Op* lookup(std::uint64_t id) const {
for (auto* op : ops_)
if (op->id == id) return op;
return nullptr;
}
void remove(Op* op) {
auto it = std::find(ops_.begin(), ops_.end(), op);
if (it != ops_.end()) ops_.erase(it);
}
void stop() noexcept {
if (state_ == State::open) state_ = State::stopping;
}
std::size_t outstanding() const noexcept { return ops_.size(); }
State state() const noexcept { return state_; }
void run() {
while (!ops_.empty()) {
if (state_ == State::stopping) { cancel_all(); continue; }
std::vector<pollfd> pfds;
std::vector<std::uint64_t> ids; // nem pointer: az Op közben megszűnhet
pfds.reserve(ops_.size());
ids.reserve(ops_.size());
for (auto* op : ops_) {
pfds.push_back(pollfd{op->fd, op->events, 0});
ids.push_back(op->id);
}
int n = ::poll(pfds.data(), pfds.size(), 10);
if (n < 0) {
if (errno == EINTR) continue;
throw std::runtime_error("poll failed");
}
for (std::size_t i = 0; i < pfds.size(); ++i) {
if (pfds[i].revents == 0) continue;
// Egy korábbi resume ugyanebben a kötegben megszüntethette ezt a
// műveletet; ekkor nincs completion és nincs resume.
Op* op = lookup(ids[i]);
if (op == nullptr) continue;
// A rendszerhívás eredménye dönt, nem a poll-esemény: átmeneti
// hibánál (EAGAIN, EINTR) a művelet regisztrálva marad.
if (!op->complete(op)) continue;
// Előbb eltávolítjuk, csak utána folytatjuk a coroutine-t.
remove(op);
op->waiter.resume();
}
}
state_ = State::stopped;
}
private:
// Minden függő műveletet ECANCELED-del zár le: remove, majd resume.
void cancel_all() {
while (!ops_.empty()) {
Op* op = ops_.front();
remove(op);
op->cancel_error = ECANCELED;
op->waiter.resume();
}
}
std::vector<Op*> ops_;
std::uint64_t next_id_ = 1;
State state_ = State::open;
};
struct ReadAwaitable {
Loop& loop;
int fd; // kölcsönzött, nem birtokolt
void* buf;
std::size_t len;
ssize_t result = 0;
int error = 0;
Op op;
ReadAwaitable(Loop& l, int f, void* b, std::size_t n)
: loop(l), fd(f), buf(b), len(n) {}
~ReadAwaitable() { loop.remove(&op); }
bool await_ready() const noexcept { return false; }
bool await_suspend(std::coroutine_handle<> h) {
op.fd = fd;
op.events = POLLIN;
op.waiter = h;
op.complete = &on_complete;
op.owner = this;
if (loop.add(&op)) return true; // felfüggesztés
error = ECANCELED; // a Loop már nem fogad: azonnal folytat
return false;
}
ssize_t await_resume() {
if (op.cancel_error) error = op.cancel_error;
if (error) {
errno = error;
return -1;
}
return result;
}
static bool on_complete(Op* o) {
auto* self = static_cast<ReadAwaitable*>(o->owner);
self->result = sys_read(o->fd, self->buf, self->len);
if (self->result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return false;
self->error = errno;
}
return true;
}
};
struct WriteAwaitable {
Loop& loop;
int fd; // kölcsönzött, nem birtokolt
const void* buf;
std::size_t len;
ssize_t result = 0;
int error = 0;
Op op;
WriteAwaitable(Loop& l, int f, const void* b, std::size_t n)
: loop(l), fd(f), buf(b), len(n) {}
~WriteAwaitable() { loop.remove(&op); }
bool await_ready() const noexcept { return false; }
bool await_suspend(std::coroutine_handle<> h) {
op.fd = fd;
op.events = POLLOUT;
op.waiter = h;
op.complete = &on_complete;
op.owner = this;
if (loop.add(&op)) return true;
error = ECANCELED;
return false;
}
ssize_t await_resume() {
if (op.cancel_error) error = op.cancel_error;
if (error) {
errno = error;
return -1;
}
return result;
}
static bool on_complete(Op* o) {
auto* self = static_cast<WriteAwaitable*>(o->owner);
self->result = sys_write(o->fd, self->buf, self->len);
if (self->result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return false;
self->error = errno;
}
return true;
}
};
struct Task {
struct promise_type {
Task get_return_object() {
return Task{std::coroutine_handle<promise_type>::from_promise(*this)};
}
std::suspend_always initial_suspend() noexcept { return {}; }
std::suspend_always final_suspend() noexcept { return {}; }
void return_value(int err) noexcept { error = err; }
void unhandled_exception() noexcept { exception = std::current_exception(); }
int error = 0; // 0: siker, egyébként errno
std::exception_ptr exception; // a coroutine-ból kiszökő kivétel
};
std::coroutine_handle<promise_type> h{};
explicit Task(std::coroutine_handle<promise_type> handle) : h(handle) {}
Task(Task&& other) noexcept : h(other.h) { other.h = {}; }
Task& operator=(Task&& other) noexcept {
if (this != &other) {
if (h) h.destroy();
h = other.h;
other.h = {};
}
return *this;
}
Task(const Task&) = delete;
Task& operator=(const Task&) = delete;
~Task() {
if (h) h.destroy();
}
void resume() {
if (h && !h.done()) h.resume();
}
bool done() const noexcept { return !h || h.done(); }
// Csak befejezett coroutine-ra hívható; kivételt továbbdob.
int result() const {
if (h.promise().exception) std::rethrow_exception(h.promise().exception);
return h.promise().error;
}
};
// Eredmény: 0, ha a bemenet EOF-ig átment; különben a végleges hiba errno-ja
// (ECANCELED, ha a Loop leállítása mondta le). Az átmeneti hibákat az adapter
// kezeli, ide nem jutnak el.
Task echo(Loop& loop, int in_fd, int out_fd) {
char buf[4096];
for (;;) {
ssize_t n = co_await ReadAwaitable{loop, in_fd, buf, sizeof buf};
if (n == 0) co_return 0; // EOF: siker
if (n < 0) co_return errno; // végleges hiba
std::size_t off = 0; // n > 0: a konverzió értéktartó
while (off < static_cast<std::size_t>(n)) {
ssize_t w = co_await WriteAwaitable{
loop, out_fd, buf + off, static_cast<std::size_t>(n) - off};
if (w < 0) co_return errno;
off += static_cast<std::size_t>(w);
}
}
}
static bool set_nonblock(int fd) {
int fl = ::fcntl(fd, F_GETFL);
return fl >= 0 && ::fcntl(fd, F_SETFL, fl | O_NONBLOCK) == 0;
}
// Egy echo-futás saját pipe-pal; a kimenet egy kölcsönzött, már nem
// blokkoló fd. Minden pipe-vég a unique_fd-k révén pontosan egyszer záródik,
// a korai returnokon is.
[[maybe_unused]] static int run_echo(int out_fd) {
int fds[2];
if (::pipe(fds) != 0) return 1;
unique_fd rd{fds[0]}, wr{fds[1]};
if (!set_nonblock(rd.get())) return 1;
Loop loop;
Task task = echo(loop, rd.get(), out_fd);
task.resume();
const char msg[] = "hello from coroutine\n";
if (::write(wr.get(), msg, sizeof msg - 1) < 0) return 1;
wr.reset(); // EOF jelzése az olvasónak
loop.run();
if (!task.done()) return 1;
if (int err = task.result(); err != 0) {
std::fprintf(stderr, "echo: %s\n", std::strerror(err));
return 1;
}
return 0;
}
#if !defined(LOOP_CANCEL_TEST) && !defined(LOOP_STOP_TEST)
int main() {
// A stdout fájlleírását a szülővel osztjuk meg: a nem blokkoló módot csak
// a futás idejére állítjuk be, utána visszaállítjuk.
int old = ::fcntl(STDOUT_FILENO, F_GETFL);
if (old < 0 || !set_nonblock(STDOUT_FILENO)) return 1;
int rc = run_echo(STDOUT_FILENO);
#ifdef FD_LEAK_TEST
auto count_fds = [] { int c = 0; for (int fd = 0; fd < 1024; ++fd) if (::fcntl(fd, F_GETFD) >= 0) ++c; return c; };
int before = count_fds();
for (int i = 0; i < 100 && rc == 0; ++i) rc = run_echo(STDOUT_FILENO);
std::fprintf(stderr, "fd before=%d after=%d\n", before, count_fds());
#endif
(void)::fcntl(STDOUT_FILENO, F_SETFL, old);
return rc;
}
#elif defined(LOOP_STOP_TEST)
// Leállítás függő olvasással: a függő művelet ECANCELED-et kap, a run()
// véges időben visszatér, és a stop után új regisztráció elutasított.
int main() {
int fds[2];
if (::pipe(fds) != 0) return 1;
unique_fd rd{fds[0]}, wr{fds[1]};
if (!set_nonblock(rd.get())) return 1;
Loop loop;
Task pending = echo(loop, rd.get(), STDOUT_FILENO);
pending.resume(); // olvasásra vár, adat nem jön
loop.stop();
loop.run(); // lemondja, majd visszatér
int e1 = pending.done() ? pending.result() : -1;
Task late = echo(loop, rd.get(), STDOUT_FILENO);
late.resume(); // a stop után: azonnal elutasítva
int e2 = late.done() ? late.result() : -1;
std::printf("pending=%s late=%s outstanding=%zu\n", std::strerror(e1),
std::strerror(e2), loop.outstanding());
return (e1 == ECANCELED && e2 == ECANCELED && loop.outstanding() == 0) ? 0 : 1;
}
#else
// Két egyszerre kész művelet: az első folytatása megszünteti a másodikat.
#include <optional>
std::optional<Task> victim;
int completions = 0;
Task reader(Loop& loop, int fd, bool kill_victim) {
char c;
ssize_t n = co_await ReadAwaitable{loop, fd, &c, 1};
++completions;
std::printf("reader fd=%d n=%zd\n", fd, n);
if (kill_victim) victim.reset();
co_return 0;
}
int main() {
int a[2], b[2];
if (::pipe(a) != 0) return 1;
unique_fd ar{a[0]}, aw{a[1]};
if (::pipe(b) != 0) return 1;
unique_fd br{b[0]}, bw{b[1]};
if (!set_nonblock(ar.get()) || !set_nonblock(br.get())) return 1;
Loop loop;
Task first = reader(loop, ar.get(), true);
victim.emplace(reader(loop, br.get(), false));
first.resume();
victim->resume();
if (::write(aw.get(), "x", 1) != 1 || ::write(bw.get(), "y", 1) != 1) return 1;
loop.run();
std::printf("completions=%d\n", completions);
return completions == 1 ? 0 : 1;
}
#endifA GCC-sor a lab-gcc, a Clang-sor a
lab-clang profilhoz tartozik:
g++ -std=c++20 -Wall -Wextra -Wpedantic -O2 -g -o echo_coro echo_coro.cpp
clang++ -std=c++20 -Wall -Wextra -Wpedantic -O2 -g -o echo_coro echo_coro.cpp
./echo_coro | cat # normál pipe: a teljes üzenet, 0-s kód
./echo_coro > /dev/full # írási hiba: diagnosztika, nem nulla kód
# tesztbuildek ASan/UBSan alatt:
for t in LOOP_CANCEL_TEST LOOP_STOP_TEST INJECT_TRANSIENT FD_LEAK_TEST; do
g++ -std=c++20 -O1 -g -fsanitize=address,undefined -D$t echo_coro.cpp -o t_$t &&
./t_$t > /dev/null; echo "$t: $?"
doneKét szerződést érdemes kiemelni. Az első a diszpécser
pillanatképe. A run() egy kötegben több kész
eseményt is feldolgozhat. Egy korábbi elem resume()-ja
közben megszüntethet egy olyan coroutine-t, amelynek a művelete szintén
kész: a frame-mel együtt az awaitable és benne az Op is
megszűnik. Ha a pillanatkép nyers Op* pointereket tárolna,
a következő elem feldolgozása már felszabadított memóriát érne el. Ezért
a pillanatkép az Op azonosítóját tárolja, és minden elem
előtt a lookup ellenőrzi, hogy a művelet még regisztrálva
van-e. A megszűnt awaitable destruktora remove-olja az
Op-ot, így annak sem completionje, sem resume-ja nem
fut.
A második az eredménycsatorna. Az echo
különválasztja az EOF-ot (siker), az átmeneti hibát (EINTR,
EAGAIN: újrapróbálás) és a végleges hibát (az
errno értéke). A Task promise-a eltárolja a
visszatérési értéket, illetve a coroutine-ból kiszökő kivételt
std::exception_ptr-ként. A main ebből képez
kilépési kódot és diagnosztikát.
Buildjegyzőkönyv (lab-gcc profil). A
program mind az öt változatban figyelmeztetés nélkül fordult. Normál
pipe-on a kimenet hello from coroutine, a kód 0;
/dev/full-ra echo: No space left on device, a
kód 1. A lemondási próba reader fd=3 n=1,
completions=1, 0-s kód. A leállítási próba
(LOOP_STOP_TEST)
pending=Operation canceled late=Operation canceled outstanding=0
kimenettel, 0 kóddal, 5 másodperces időkorláton belül futott: a függő
olvasás ECANCELED-et kapott, a run()
visszatért, és a stop() utáni regisztrációt az adapter
azonnal elutasította. Az INJECT_TRANSIENT build az első két
read és az első két write hívásra
EAGAIN, majd EINTR hibát injektál; a kimenet
ekkor is a teljes hello from coroutine, 0-s kóddal — a
művelet nem veszett el, és nem lett hamis siker. Az
FD_LEAK_TEST 101-szer futtatja az echót: az élő fd-k száma
előtte és utána is 3 (a javítás előtti, pipe olvasóvégét
nyitva hagyó változatnál 3-ról 103-ra nőtt). Az ASan/UBSan egyik
próbában sem jelzett. A Clang-sort Clang 22-vel (ziglang) próbáltuk:
figyelmeztetés nélkül fordult, 0-s kóddal futott.
A példában a Task initial_suspend-je
suspend_always, ezért a task.resume() indítja
el a coroutine-t. A final_suspend szintén felfüggeszt, így
a Task destruktora tudja destroy-olni a frame-et. A
ReadAwaitable és a WriteAwaitable temporális
objektum a co_await kifejezésben jön létre, és a
felfüggesztés alatt is él. Az Op az awaitable tagja, tehát
a frame-ben van; a Loop csak pointert tárol. Ha a coroutine
felfüggesztett állapotban megsemmisül, az awaitable destruktora
remove-olja az Op-ot, így a Loop
nem tart dangling pointert. A completion után az Op már
nincs a Loop vektorában, ezért a destruktor
remove-ja nem talál semmit.
A callbackes változat ugyanezt a folyamatot explicit állapotgéppel
írja le. Az Echo struct tartalmazza a buffert, az offsetet
és a fájlleírókat; az on_read és on_write
függvények lépkednek. A coroutine ugyanezt a lineáris kódban tartja, az
állapot a frame-ben van. A callbacknél a cancellation a struct
élettartamához kötött, a coroutine-nál a frame és az awaitable
élettartamához. A callbackes változatban a buffer lehet a struct tagja,
a coroutine-ban a frame lokális változója; mindkettő addig él, amíg a
művelet hivatkozik rá.
// Callbackes vázlat, nem teljes program.
struct Echo {
Loop& loop;
int in_fd;
int out_fd;
char buf[4096];
std::size_t off = 0;
ssize_t n = 0;
void start_read();
void on_read();
void on_write();
};A start_read regisztrál egy olvasási operationt,
amelynek completionje on_read; az on_read
elindítja a start_write-ot, a on_write pedig
vagy befejez, vagy újabb olvasást indít. Az állapotátmeneteket a
n és az off hordozza. A coroutine-ban ugyanez
a for és while ciklus, az állapot a frame-ben.
A coroutine kódja rövidebb és lineárisabb, de a frame allokálhat, és a
debuggerben a frame-et kell érteni. A callbackes változat stacken is
elférhet, de a hívási lánc és az állapot kézi követése nehezebb.
Az outstanding operation számlálója a
Loop::outstanding(). A leállítási policy a példában
cancel: a stop() a nyitott állapotból
leálló állapotba lép; a run() ekkor nem pollol tovább,
hanem a cancel_all()-lal minden függő műveletet
ECANCELED-del zár le, majd — ha nincs több függő művelet —
leállt állapotban visszatér. A leállás így véges, akkor is, ha egy
olvasásra soha nem érkezne adat. A leálló vagy leállt Loop
az add-ot elutasítja (false); az awaitable
ilyenkor nem függeszt fel, hanem azonnal ECANCELED-del
folytat. (Egy drain policy ehelyett megvárná a függő műveletek
befejeződését, de akkor a véges leállást egy időkorlátnak kell
biztosítania.) A Task destruktora csak akkor biztonságos,
ha a Loop már nem hivatkozik az Op-ra. Ezt a
laborban úgy ellenőrizhetjük, hogy a run() végén
ops_.empty() igaz, és a Task csak ezután
semmisül meg.
A timeout, a cancel és a completion kizáró szabálya a következő. Az
operation pontosan egy terminális eseményt kaphat. A completion a
poll által jelzett esemény után, a timeout a lejárat után,
a cancel a stop() vagy egy külső kérés után következik be.
A versenyt az eseményloop szálán, az Op eltávolításával
kell eldönteni: amelyik ág előbb remove-ol, az nyer. A
másik ág a remove után nem hivatkozhat az
Op-ra, és nem hívhatja meg a completiont. Ezért a
completion és a cancel is csak a remove után, a resume
előtt futhat. Ha a timeout és a poll esemény ugyanabban az
iterációban érkezik, a remove dönti el a sorrendet; a másik
ág már nem találja az Op-ot.
A cancel_all() minden Op-ra beállítja a
lemondás hibakódját, remove-ol, majd resume-ol. Az
awaitable await_resume()-a ilyenkor -1-et ad
ECANCELED errno-val. A coroutine ilyenkor
hibát lát, és co_return-nal lezárhat. A timeout hasonló, de
csak a lejárt operationre vonatkozik; a poll timeoutja
önmagában nem elég, mert a lejáratot operation szinten kell
nyilvántartani.
A debuggerben a coroutine frame a Task::h handle-ből
érhető el. A h.done() a befejezést, a
h.address() a frame címét mutatja. A co_await
ponton a frame a stacken vagy a heapen van; a debugger a
promise és a lokális változók alapján azonosítja. Az
AddressSanitizer a frame-allokációt és a buffer túlcsordulását, a
ThreadSanitizer a több szálú loopot ellenőrzi. A ssize_t →
size_t konverzió nem UB: negatív értékre is definiált
(modulo) eredményt ad, ezért az UBSan -fsanitize=undefined
nem jelez rá. A védelem a kódban van: az echo a konverzió
előtt kezeli a negatív eredményt, így a
static_cast<std::size_t>(n) már csak pozitív értéket
kap. Ha az implicit előjelváltó konverziókat külön is figyelni akarjuk,
az fordítófüggő kiegészítés: a Clang
-fsanitize=implicit-integer-sign-change kapcsolója (GCC-ben
nincs megfelelője). A laborban érdemes külön futtatni
-fsanitize=address,undefined és
-fsanitize=thread kapcsolókkal, dokumentált compiler- és
library-verzió mellett.
Gyakori hibák: a buffer a co_await után megszűnik, de az
adapter még használja; a Loop a resume után hivatkozik az
Op-ra; a poll kötegének pillanatképe nyers
Op*-t tárol, és egy korábbi resume által megszüntetett
műveletet is feldolgoz; a completion kétszer hívódik, mert a timeout és
a poll ág is lefut; a cancel után a coroutine még
resume-ol; a Task destruktora felfüggesztett coroutine-t
destroy-ol, miközben az Op a Loop-ban van. A
javítás mindegyiknél ugyanaz a szerződés betartása: az Op
az awaitable élettartamához kötött, a Loop csak pointert
tárol, a terminális eseményt a remove dönti el, és a resume
után az Op nem használható.
Egészítsd ki a ReadAwaitable-t timeouttal. A
Loop kapjon egy now() függvényt és egy
timeout-listát, amely a lejáratokat tartja nyilván. A timeout cancelálja
az operationt, ha még nincs completion. Írd le, hogyan biztosítod, hogy
a timeout és a completion közül csak az egyik fusson le.
A példa cancel_all()-ja a leállításkor minden függő
műveletet lemond. Valósíts meg egy
Loop::cancel(std::uint64_t id) függvényt, amely
egyetlen függő műveletet zár le ECANCELED-del, a
többit érintetlenül hagyja. Ellenőrizd, hogy a lemondott coroutine
await_resume()-ban kapja meg a hibát, hogy a
Task destruktora nem hagy dangling Op-ot, és
hogy a run() végén az outstanding számláló nulla.
Írd meg ugyanazt az echo-folyamatot callbackes formában, majd hasonlítsd össze a coroutine verzióval. Készíts táblázatot a kódméretről, az állapot láthatóságáról, a cancellation szerződéséről, a hibakezelésről és az allokációról. Mérést csak dokumentált laborprofillal és legalább három ismétléssel végezz; a mérés nélküli állítást ne tekintsd bizonyítottnak.
A Loop minden Op-hoz eltárolja a lejárat
idejét. A run() minden iterációban a poll
timeoutját a legközelebbi lejáratig csökkenti. A poll
visszatérése után először a lejárt operationöket kell feldolgozni:
mindegyikre remove, hiba beállítása
ETIMEDOUT-ra, majd resume. Ezután jönnek a
poll eseményei. Ha egy Op már nincs a
vektorban, mert a timeout ág eltávolította, a poll ág nem
találja, így nem hívja meg a completiont. A kizáró szabályt a
remove és az ops_ vektor konzisztens
használata biztosítja; a két ág ugyanazon a szálon fut, ezért nincs
adatverseny. Ellenőrzési feltétel: egy Op-ra sem fut le
kétszer a completion, és a timeout után a coroutine
ETIMEDOUT-ot kap.
A cancel(id) a lookup(id)-vel megkeresi a
műveletet; ha már nincs regisztrálva (közben befejeződött vagy
megszűnt), nincs teendő — ez a kizáró szabály: az nyer, aki előbb
remove-ol. Ha megvan, ugyanazt teszi, mint a
cancel_all() egy elemre: remove,
cancel_error = ECANCELED, majd resume. A resume hatására az
awaitable await_resume()-a -1-et ad, az
errno ECANCELED, a coroutine
co_return-nel lezár. A run() végén az
outstanding() nulla, ha minden más művelet is lezárult.
Ellenőrzési feltétel: a lemondott művelet completionje nem fut le, a
többi művelet eredménye változatlan.
A callbackes változatban az Echo struct tartalmazza a
buffert és az állapotot; a start_read és
start_write függvények regisztrálják az operationöket, a
completionök pedig a struct metódusai. A coroutine változatban ugyanez a
folyamat a for és while ciklusban van, az
állapot a frame-ben. A táblázatban a callbackes változat kisebb frame-et
használhat, de az állapotátmeneteket kézzel kell követni; a coroutine
kódja lineárisabb, de a frame allokálhat. A cancellation a callbacknél a
struct élettartamához, a coroutine-nál a frame és az awaitable
élettartamához kötött. A mérésnél rögzíteni kell a compiler, a library,
a target és az optimalizálási szint pontos verzióját; a mérés nélküli
összehasonlítás csak szerződés szintű.
A coroutine csak felfüggesztés; az aszinkron I/O adapter adja a
completiont, a schedulert és a cancellationt. A laborban a
poll eseményei és a coroutine frame találkoznak. A
szerződés négy pontja: a buffer tulajdonosa a hívó, az awaitable
élettartama fedi a felfüggesztést, a terminális esemény pontosan egyszer
fut le, és a shutdown alatt az outstanding számláló nullázódik. A
debugger a frame-et, a sanitizer a buffer és az Op
élettartamát ellenőrzi. A callback, az állapotgép és a coroutine
ugyanazt a folyamatot írja le, de eltérő élettartam- és
költségszerződéssel.