פתרון מנוע קריאה אסינכרוני¶
נשתמש ב־event loop יחיד ובטבלה קבועה של בקשות. ההגבלה מפשטת את מרוץ הרישום: חבילת completion יכולה להיכנס לתור בזמן ההגשה, אבל הקוד שלנו לא יצרוך אותה עד שנחזור ללולאה. כך אין חידוש מקביל מתוך await_suspend.
התוכנית הבאה מלאה עבור Windows ב־x64 ו־C++20. שמרו בשם async_scan.cpp ובנו עם תמיכת קורוטינות של המהדר:
cl /nologo /std:c++20 /EHsc /W4 async_scan.cpp /Fe:async_scan.exe
async_scan.exe sample.bin
async_scan.exe sample.bin --cancel
הקוד כולל גם מסלול ניקוי כאשר הקמת אחת המשימות נכשלת:
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <algorithm>
#include <array>
#include <coroutine>
#include <cstddef>
#include <cstdint>
#include <cwchar>
#include <exception>
#include <iostream>
#include <optional>
#include <stdexcept>
#include <system_error>
#include <utility>
[[noreturn]] void fail(DWORD error, const char* operation) {
throw std::system_error(
static_cast<int>(error), std::system_category(), operation);
}
struct Handle {
HANDLE value;
explicit Handle(HANDLE v) : value(v) {}
~Handle() {
if (value && value != INVALID_HANDLE_VALUE) CloseHandle(value);
}
Handle(const Handle&) = delete;
Handle& operator=(const Handle&) = delete;
};
struct Request {
OVERLAPPED operation{};
std::array<std::byte, 4096> buffer{};
std::coroutine_handle<> continuation{};
HANDLE file = nullptr;
DWORD bytes = 0;
DWORD error = ERROR_SUCCESS;
};
class Loop {
Handle port_;
std::array<Request*, 4> active_{};
bool stopping_ = false;
public:
Loop() : port_(CreateIoCompletionPort(
INVALID_HANDLE_VALUE, nullptr, 0, 1)) {
if (!port_.value) fail(GetLastError(), "create port");
}
~Loop() { if (count() != 0) std::terminate(); }
void attach(HANDLE file) {
if (!CreateIoCompletionPort(file, port_.value, 17, 0))
fail(GetLastError(), "attach port");
}
std::size_t count() const noexcept {
std::size_t result = 0;
for (auto* request : active_) if (request) ++result;
return result;
}
bool add(Request& request) noexcept {
if (stopping_) {
request.error = ERROR_OPERATION_ABORTED;
return false;
}
for (auto& slot : active_) {
if (!slot) {
slot = &request;
return true;
}
}
request.error = ERROR_BUSY;
return false;
}
void remove(Request& request) noexcept {
for (auto& slot : active_) {
if (slot == &request) {
slot = nullptr;
return;
}
}
std::terminate();
}
void request_stop() noexcept {
stopping_ = true;
for (auto* request : active_) {
if (request && !CancelIoEx(
request->file, &request->operation)) {
const DWORD error = GetLastError();
if (error != ERROR_NOT_FOUND)
std::cerr << "cancel: " << error << '\n';
}
}
}
void drain() {
while (count() != 0) {
DWORD bytes = 0;
ULONG_PTR key = 0;
OVERLAPPED* operation = nullptr;
const BOOL ok = GetQueuedCompletionStatus(
port_.value, &bytes, &key, &operation, INFINITE);
const DWORD error = ok ? ERROR_SUCCESS : GetLastError();
if (!operation || key != 17) std::terminate();
Request* found = nullptr;
for (auto* request : active_)
if (request && &request->operation == operation)
found = request;
if (!found) std::terminate();
remove(*found);
found->bytes = bytes;
found->error = error;
const auto continuation = found->continuation;
continuation.resume();
}
}
};
class Task {
public:
struct promise_type {
std::exception_ptr error;
Task get_return_object() {
return Task{
std::coroutine_handle<promise_type>::from_promise(*this)};
}
std::suspend_never initial_suspend() noexcept { return {}; }
std::suspend_always final_suspend() noexcept { return {}; }
void return_void() noexcept {}
void unhandled_exception() noexcept {
error = std::current_exception();
}
};
using Coroutine = std::coroutine_handle<promise_type>;
explicit Task(Coroutine coroutine) noexcept : coroutine_(coroutine) {}
Task(Task&& other) noexcept
: coroutine_(std::exchange(other.coroutine_, {})) {}
Task(const Task&) = delete;
Task& operator=(const Task&) = delete;
~Task() {
if (coroutine_) {
if (!coroutine_.done()) std::terminate();
coroutine_.destroy();
}
}
void result() const {
if (!coroutine_ || !coroutine_.done())
throw std::logic_error("task still active");
if (coroutine_.promise().error)
std::rethrow_exception(coroutine_.promise().error);
}
private:
Coroutine coroutine_;
};
struct Read {
Loop& loop;
Request request;
DWORD wanted;
Read(Loop& owner, HANDLE file, std::uint64_t offset, DWORD count)
: loop(owner), wanted(count) {
request.file = file;
request.operation.Offset = static_cast<DWORD>(offset);
request.operation.OffsetHigh = static_cast<DWORD>(offset >> 32);
}
Read(const Read&) = delete;
Read& operator=(const Read&) = delete;
bool await_ready() const noexcept { return false; }
bool await_suspend(std::coroutine_handle<> continuation) noexcept {
request.continuation = continuation;
if (!loop.add(request)) return false;
DWORD ignored = 0;
const BOOL ok = ReadFile(request.file, request.buffer.data(),
wanted, &ignored, &request.operation);
if (!ok) {
const DWORD error = GetLastError();
if (error != ERROR_IO_PENDING) {
loop.remove(request);
request.error = error;
return false;
}
}
return true;
}
DWORD await_resume() {
if (request.error != ERROR_SUCCESS)
fail(request.error, "read completion");
if (request.bytes > wanted)
throw std::runtime_error("invalid transfer count");
return request.bytes;
}
};
struct Stats {
std::uint64_t bytes = 0;
std::uint64_t sum = 0;
};
Task scan_lane(Loop& loop, HANDLE file, std::uint64_t size,
unsigned lane, Stats& stats) {
constexpr std::uint64_t block = 4096;
constexpr std::uint64_t stride = block * 2;
for (std::uint64_t offset = lane * block;
offset < size; offset += stride) {
const auto target = static_cast<DWORD>(std::min(block, size - offset));
DWORD done = 0;
while (done < target) {
Read read(loop, file, offset + done, target - done);
const DWORD count = co_await read;
if (count == 0) throw std::runtime_error("unexpected early EOF");
for (DWORD i = 0; i < count; ++i)
stats.sum += std::to_integer<unsigned char>(
read.request.buffer[i]);
stats.bytes += count;
done += count;
}
}
}
int wmain(int argc, wchar_t** argv) {
const bool cancel = argc == 3 && std::wcscmp(argv[2], L"--cancel") == 0;
if (argc != 2 && !cancel) return 2;
try {
Handle file(CreateFileW(argv[1], GENERIC_READ, FILE_SHARE_READ,
nullptr, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, nullptr));
if (file.value == INVALID_HANDLE_VALUE)
fail(GetLastError(), "open");
LARGE_INTEGER length{};
if (!GetFileSizeEx(file.value, &length))
fail(GetLastError(), "size");
if (length.QuadPart < 0) throw std::runtime_error("negative size");
Loop loop;
loop.attach(file.value);
std::array<Stats, 2> stats{};
std::array<std::optional<Task>, 2> tasks;
std::exception_ptr setup_error;
try {
for (unsigned lane = 0; lane < 2; ++lane)
tasks[lane].emplace(scan_lane(loop, file.value,
static_cast<std::uint64_t>(length.QuadPart),
lane, stats[lane]));
} catch (...) {
setup_error = std::current_exception();
loop.request_stop();
}
if (cancel) loop.request_stop();
loop.drain();
if (setup_error) std::rethrow_exception(setup_error);
bool failed = false;
for (auto& task : tasks) {
try {
task->result();
} catch (const std::exception& error) {
failed = true;
std::cerr << error.what() << '\n';
}
}
if (failed) return cancel ? 3 : 1;
const auto bytes = stats[0].bytes + stats[1].bytes;
const auto sum = stats[0].sum + stats[1].sum;
if (bytes != static_cast<std::uint64_t>(length.QuadPart))
throw std::runtime_error("coverage invariant failed");
std::cout << "bytes=" << bytes << " sum=" << sum << '\n';
return 0;
} catch (const std::exception& error) {
std::cerr << error.what() << '\n';
return 1;
}
}
חלוקת הקובץ והשלמות חלקיות¶
הקורוטינה הראשונה קוראת בלוקים שמספרם זוגי, והשנייה קוראת בלוקים שמספרם אי־זוגי. כל בלוק לוגי הוא עד ארבעת אלפים תשעים ושישה בתים. הלולאה הפנימית משלימה את אותו בלוק אם קריאה העבירה פחות מן המבוקש, ומעדכנת את offset לפי מספר הבתים שהתקבל בפועל.
ה־buffer שייך למשתנה Read בתוך מסגרת הקורוטינה. המשתנה נשאר חי עד סוף האיטרציה הפנימית, ולכן אפשר לסכם את תוכנו אחרי co_await. אין החזרת span אל buffer זמני שנהרס לפני שהקורא משתמש בו.
הסכום הוא חיבור unsigned בן שישים וארבעה ביט ולכן מוגדר מודולו הטווח של הטיפוס אם קובץ גדול מאוד גורם לגלישה. הסריקה הסינכרונית להשוואה צריכה להשתמש באותה הגדרה. מספר הבתים אינו אמור לגלוש, משום שגודל הקובץ נקרא לערך חתום חיובי באותו רוחב.
קיבולת וסיום¶
טבלת active מכילה ארבעה מקומות, בעוד התוכנית מפעילה שתי קורוטינות עם פעולה אחת פעילה לכל קורוטינה. לכן הקיבולת אינה אמורה להתמלא בהרצה הרגילה. אם מוסיפים משימות, בקשה מעבר לגבול מקבלת ERROR_BUSY לפני הגשה. זו דחייה מוגדרת, ולא תור בלתי מוגבל שמסתתר מאחורי הקורוטינות.
הפעולה request_stop מונעת רישום חדש לפני שהיא מבקשת ביטול של הפעולות הקיימות. פעולה יכולה להסתיים בהצלחה למרות הביטול; אם הקורוטינה מנסה להגיש בלוק נוסף, הרישום נדחה עם שגיאת ביטול. בכל המקרים drain ממשיכה עד שאין בקשות רשומות.
ה־destructor של Task מסיים את התהליך אם מנסים להרוס מסגרת שעדיין פעילה. זהו גילוי של הפרת חוזה פנימית, ולא תחליף לניקוי. המסלולים הרגילים מבטיחים קודם drain, ורק אחריו הרס. במקרה שהקמת המשימה השנייה זורקת, ה־catch הפנימי משאיר את הראשונה בחיים, מבקש עצירה ומרוקן לפני שמפרסם את השגיאה.
למה אין גישה אחרי resume¶
ה־dispatcher מסיר את הבקשה, כותב את התוצאה ומעתיק את continuation. הקריאה ל־resume היא הפעולה האחרונה שלו ביחס לאותה בקשה. הקורוטינה יכולה לסיים את Read, להתחיל Read חדש או להגיע לסיום; אין pointer ישן שה־dispatcher ממשיך לקרוא אחר כך.
ההוכחה נשענת על thread יחיד שמגיש וצורך. העברת drain ל־thread נוסף בלי שינוי הפרוטוקול אינה הרחבה בטוחה. גם request_stop כאן נקראת מאותו thread לפני drain, ולכן שדות המנוע אינם מוגנים לשימוש חיצוני מקביל. מערכת עם בקשת עצירה מתהליכון אחר צריכה לפרסם פקודת בקרה לתור או להוסיף סנכרון מתאים.
אימות ותיעוד גבולות¶
בדיקת בסיס סורקת את אותו קובץ באמצעות ifstream ומחשבת מספר בתים וסכום באותו טיפוס. עבור קובץ של תשעת אלפים בתים נבדוק גם את המעבר בין שתי הקורוטינות וגם בלוק אחרון חלקי. עבור קובץ ריק שתי הקורוטינות מסיימות בלי להגיש פעולה.
ב־backend בדיקה נחליף את פעולת ההגשה והאיסוף בממשק שמחזיר מספר בתים קטן מן המבוקש, שגיאת הגשה או completion לאחר בקשת ביטול. נבדוק שמספר ההשלמות הסופיות לכל בקשה הוא אחד ושהטבלה ריקה לפני destruction. ההחלפה חייבת לשמור על ההבחנה בין פעולה שלא התקבלה לבין פעולה שמחייבת חבילה.
הפתרון אינו טוען שנבדקה תגובת driver אמיתי לביטול או שהושג זמן סיום חסום. תיעוד IOCP ותיעוד ביטול פעולות מגדירים את גבולות ההבטחה. במעבדה יש לשמור פקודות בנייה, תוצאות אמיתיות וגרסאות כלים לצד ההוכחה.