פתרון מאגר מקומי עמיד¶
הפתרון משתמש בשלושה קובצי מקור מלאים. הם נמצאים גם בתיקיית הקוד של הפרויקט: קובץ העזר, השרת והלקוח. העתיקו אותם לאותה תיקיית בנייה ובנו ב־MSVC עבור x64. אין צורך להפעיל את השרת כמנהל.
cl /nologo /std:c++20 /EHsc /W4 wal_server.cpp /Fe:wal_server.exe /link Advapi32.lib
cl /nologo /std:c++20 /EHsc /W4 wal_client.cpp /Fe:wal_client.exe /link Advapi32.lib
wal_server.exe create lab.wal
בחלון שני מריצים בקשות. הפרמטרים הם סוג פעולה, מזהה, מפתח וערך; עבור get שדה הערך חייב להיות אפס:
wal_client.exe put 1 7 99
wal_client.exe put 2 7 100
wal_client.exe put 1 7 99
wal_client.exe get 3 7 0
wal_client.exe put 1 7 101
הבקשה השלישית מחזירה את התוצאה המקורית תשעים ותשע, אך הקריאה שאחריה מחזירה מאה. הבקשה האחרונה נדחית כקונפליקט מזהה. לאחר סיום השרת אפשר לפתוח שוב באמצעות open במקום create.
קובץ העזר¶
הקובץ מרכז קידוד bytes, משאבי Windows ופעולות pipe. מסלול הצלחה מיידית משתמש בספירת הבתים שחזרה מן ההגשה; רק pending מוביל להמתנה. אירוע של פעולה מיידית אינו חייב לעבור את אותו שינוי מצב כמו פעולה ממתינה.
#pragma once
#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <sddl.h>
#include <algorithm>
#include <array>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <cwchar>
#include <iostream>
#include <limits>
#include <map>
#include <memory>
#include <span>
#include <stdexcept>
#include <string>
#include <system_error>
#include <vector>
#include <utility>
#pragma comment(lib, "Advapi32.lib")
[[noreturn]] inline void fail(const char* operation) {
const DWORD error = GetLastError();
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 LocalDelete {
void operator()(void* p) const noexcept { if (p) LocalFree(p); }
};
using LocalMemory = std::unique_ptr<void, LocalDelete>;
inline LocalMemory current_user_security() {
HANDLE raw = nullptr;
if (!OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &raw))
fail("OpenProcessToken");
Handle token(raw);
DWORD required = 0;
GetTokenInformation(token.value, TokenUser, nullptr, 0, &required);
if (GetLastError() != ERROR_INSUFFICIENT_BUFFER) fail("token size");
std::vector<std::max_align_t> storage(
(required + sizeof(std::max_align_t) - 1) / sizeof(std::max_align_t));
if (!GetTokenInformation(token.value, TokenUser, storage.data(),
required, &required)) fail("TokenUser");
auto* user = reinterpret_cast<TOKEN_USER*>(storage.data());
wchar_t* raw_sid = nullptr;
if (!ConvertSidToStringSidW(user->User.Sid, &raw_sid)) fail("SID text");
LocalMemory sid(raw_sid);
const std::wstring text =
L"D:P(A;;GA;;;SY)(A;;GA;;;" + std::wstring(raw_sid) + L")";
PSECURITY_DESCRIPTOR raw_descriptor = nullptr;
if (!ConvertStringSecurityDescriptorToSecurityDescriptorW(
text.c_str(), SDDL_REVISION_1, &raw_descriptor, nullptr))
fail("security descriptor");
return LocalMemory(raw_descriptor);
}
inline std::uint64_t get(std::span<const std::byte> bytes,
std::size_t offset, unsigned width) {
if (width > 8 || offset > bytes.size() ||
width > bytes.size() - offset) throw std::runtime_error("field range");
std::uint64_t value = 0;
for (unsigned i = 0; i < width; ++i)
value |= std::uint64_t(std::to_integer<unsigned char>(
bytes[offset + i])) << (i * 8);
return value;
}
inline void put(std::span<std::byte> bytes, std::size_t offset,
std::uint64_t value, unsigned width) {
if (width > 8 || offset > bytes.size() ||
width > bytes.size() - offset) throw std::runtime_error("field range");
for (unsigned i = 0; i < width; ++i)
bytes[offset + i] = std::byte((value >> (i * 8)) & 0xffu);
}
inline std::uint32_t crc32(std::span<const std::byte> bytes) {
std::uint32_t crc = 0xffffffffu;
for (auto byte : bytes) {
crc ^= std::to_integer<unsigned char>(byte);
for (int bit = 0; bit < 8; ++bit)
crc = (crc >> 1) ^ (0xedb88320u & (0u - (crc & 1u)));
}
return ~crc;
}
using Record = std::array<std::byte, 32>;
inline Record record(unsigned kind, std::uint32_t id,
std::uint32_t key, std::uint32_t value) {
Record bytes{};
bytes[0] = std::byte{'A'}; bytes[1] = std::byte{'W'};
bytes[2] = std::byte{'L'}; bytes[3] = std::byte{'1'};
put(bytes, 4, kind, 1); put(bytes, 5, 1, 1); put(bytes, 6, 32, 2);
put(bytes, 8, id, 4); put(bytes, 16, key, 4); put(bytes, 20, value, 4);
put(bytes, 28, crc32(std::span(bytes).first(28)), 4);
return bytes;
}
inline bool valid_record(std::span<const std::byte> bytes, unsigned kind) {
if (bytes.size() != 32) return false;
if (bytes[0] != std::byte{'A'} || bytes[1] != std::byte{'W'} ||
bytes[2] != std::byte{'L'} || bytes[3] != std::byte{'1'} ||
get(bytes, 4, 1) != kind || get(bytes, 5, 1) != 1 ||
get(bytes, 6, 2) != 32 || get(bytes, 12, 4) != 0 ||
get(bytes, 24, 4) != 0) return false;
return get(bytes, 28, 4) == crc32(bytes.first(28));
}
inline void write_all(HANDLE file, std::span<const std::byte> bytes) {
while (!bytes.empty()) {
DWORD count = 0;
const auto request = static_cast<DWORD>(
std::min<std::size_t>(bytes.size(), 1024 * 1024));
if (!WriteFile(file, bytes.data(), request, &count, nullptr))
fail("journal write");
if (count == 0 || count > request)
throw std::runtime_error("invalid write progress");
bytes = bytes.subspan(count);
}
}
struct IoResult { DWORD error; DWORD bytes; };
inline IoResult finish_pending(HANDLE pipe, OVERLAPPED& operation,
DWORD timeout) {
if (WaitForSingleObject(operation.hEvent, timeout) != WAIT_OBJECT_0) {
if (!CancelIoEx(pipe, &operation) && GetLastError() != ERROR_NOT_FOUND)
std::cerr << "cancel request failed\n";
}
DWORD count = 0;
if (!GetOverlappedResult(pipe, &operation, &count, TRUE))
return {GetLastError(), count};
return {ERROR_SUCCESS, count};
}
inline IoResult pipe_io(HANDLE pipe, bool writing, void* bytes,
DWORD count, DWORD timeout = 5000) {
Handle event(CreateEventW(nullptr, TRUE, FALSE, nullptr));
if (!event.value) return {GetLastError(), 0};
OVERLAPPED operation{};
operation.hEvent = event.value;
DWORD immediate = 0;
const BOOL ok = writing
? WriteFile(pipe, bytes, count, &immediate, &operation)
: ReadFile(pipe, bytes, count, &immediate, &operation);
if (ok) return {ERROR_SUCCESS, immediate};
const DWORD error = GetLastError();
if (error != ERROR_IO_PENDING) return {error, immediate};
return finish_pending(pipe, operation, timeout);
}
inline DWORD connect_pipe(HANDLE pipe) {
Handle event(CreateEventW(nullptr, TRUE, FALSE, nullptr));
if (!event.value) return GetLastError();
OVERLAPPED operation{};
operation.hEvent = event.value;
if (ConnectNamedPipe(pipe, &operation)) return ERROR_SUCCESS;
const DWORD error = GetLastError();
if (error == ERROR_PIPE_CONNECTED) return ERROR_SUCCESS;
if (error != ERROR_IO_PENDING) return error;
return finish_pending(pipe, operation, 30000).error;
}
השרת והיומן¶
השרת פותח את היומן ללא שיתוף, כך שיש כותב מעבדה יחיד. הוא בונה מצב מועמד לפני append, כותב שתי רשומות, מבצע flush ורק אז מפרסם את המצב בזיכרון. כשל כתיבה או flush יוצא מגבול השירות במקום להמשיך append אחרי מצב אחסון לא ודאי.
#include "wal_lab.hpp"
struct Saved { std::uint32_t key, value; };
struct Reply { std::uint32_t status, value, flags; };
class Store {
HANDLE file_;
std::map<std::uint32_t, std::uint32_t> values_;
std::map<std::uint32_t, Saved> requests_;
std::wstring crash_;
void crash_at(const wchar_t* point) const {
if (crash_ == point) ExitProcess(70);
}
public:
Store(HANDLE file, std::wstring crash) : file_(file), crash_(std::move(crash)) {
LARGE_INTEGER size{};
if (!GetFileSizeEx(file_, &size)) fail("journal size");
if (size.QuadPart < 0 || size.QuadPart > 128 * 64)
throw std::runtime_error("journal exceeds bounded format");
std::vector<std::byte> bytes(static_cast<std::size_t>(size.QuadPart));
std::size_t read = 0;
while (read < bytes.size()) {
DWORD count = 0;
if (!ReadFile(file_, bytes.data() + read,
static_cast<DWORD>(bytes.size() - read), &count, nullptr))
fail("journal read");
if (count == 0) throw std::runtime_error("journal changed");
read += count;
}
std::size_t valid_end = 0;
while (bytes.size() - valid_end >= 64) {
const auto data = std::span(bytes).subspan(valid_end, 32);
const auto commit = std::span(bytes).subspan(valid_end + 32, 32);
if (!valid_record(data, 1) || !valid_record(commit, 2))
throw std::runtime_error("corrupt full transaction");
const auto id = static_cast<std::uint32_t>(get(data, 8, 4));
const auto key = static_cast<std::uint32_t>(get(data, 16, 4));
const auto value = static_cast<std::uint32_t>(get(data, 20, 4));
if (id == 0 || get(commit, 8, 4) != id ||
get(commit, 16, 4) != key || get(commit, 20, 4) != value ||
requests_.contains(id))
throw std::runtime_error("invalid transaction identity");
values_[key] = value;
requests_.emplace(id, Saved{key, value});
valid_end += 64;
}
if (valid_end != bytes.size()) {
const auto tail = std::span(bytes).subspan(valid_end);
if (tail.size() >= 32 && !valid_record(tail.first(32), 1))
throw std::runtime_error("corrupt tail data record");
std::cerr << "discarding incomplete unacknowledged tail: "
<< bytes.size() - valid_end << " bytes\n";
LARGE_INTEGER end{};
end.QuadPart = static_cast<LONGLONG>(valid_end);
if (!SetFilePointerEx(file_, end, nullptr, FILE_BEGIN) ||
!SetEndOfFile(file_) || !FlushFileBuffers(file_))
fail("truncate incomplete tail");
}
LARGE_INTEGER zero{};
if (!SetFilePointerEx(file_, zero, nullptr, FILE_END))
fail("seek append");
}
Reply get_value(std::uint32_t key) const {
const auto found = values_.find(key);
if (found == values_.end()) return {1, 0, 0};
return {0, found->second, 1};
}
Reply put_value(std::uint64_t wide_id, std::uint32_t key,
std::uint32_t value) {
if (wide_id == 0 || wide_id > std::numeric_limits<std::uint32_t>::max())
return {3, 0, 0};
const auto id = static_cast<std::uint32_t>(wide_id);
if (auto found = requests_.find(id); found != requests_.end()) {
if (found->second.key != key || found->second.value != value)
return {2, 0, 0};
return {0, found->second.value, 1};
}
if (requests_.size() >= 128) return {4, 0, 0};
auto next_values = values_;
auto next_requests = requests_;
next_values[key] = value;
next_requests.emplace(id, Saved{key, value});
const auto data = record(1, id, key, value);
const auto commit = record(2, id, key, value);
write_all(file_, data);
crash_at(L"--crash-after-data");
write_all(file_, commit);
crash_at(L"--crash-after-commit");
if (!FlushFileBuffers(file_)) fail("journal flush");
crash_at(L"--crash-after-flush");
values_.swap(next_values);
requests_.swap(next_requests);
return {0, value, 1};
}
};
int wmain(int argc, wchar_t** argv) {
if (argc < 3 || argc > 4) return 2;
const bool create = std::wcscmp(argv[1], L"create") == 0;
const bool open = std::wcscmp(argv[1], L"open") == 0;
if (!create && !open) return 2;
const std::wstring crash = argc == 4 ? argv[3] : L"";
if (!crash.empty() && crash != L"--crash-after-data" &&
crash != L"--crash-after-commit" && crash != L"--crash-after-flush")
return 2;
try {
Handle journal(CreateFileW(argv[2], GENERIC_READ | GENERIC_WRITE, 0,
nullptr, create ? CREATE_NEW : OPEN_EXISTING,
FILE_ATTRIBUTE_NORMAL, nullptr));
if (journal.value == INVALID_HANDLE_VALUE) fail("open journal");
Store store(journal.value, crash);
auto descriptor = current_user_security();
SECURITY_ATTRIBUTES attributes{sizeof(attributes), descriptor.get(), FALSE};
Handle pipe(CreateNamedPipeW(L"\\\\.\\pipe\\AmitTech.WalLab",
PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE,
PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT |
PIPE_REJECT_REMOTE_CLIENTS,
1, 256, 256, 5000, &attributes));
if (pipe.value == INVALID_HANDLE_VALUE) fail("create pipe");
for (unsigned connection = 0; connection < 8; ++connection) {
const DWORD connected = connect_pipe(pipe.value);
if (connected != ERROR_SUCCESS) {
std::cerr << "connect: " << connected << '\n';
break;
}
struct Disconnect {
HANDLE pipe;
~Disconnect() { DisconnectNamedPipe(pipe); }
} disconnect{pipe.value};
std::array<std::byte, 24> request{};
const auto received = pipe_io(pipe.value, false, request.data(), 24);
if (received.error != ERROR_SUCCESS || received.bytes != 24)
continue;
if (request[0] != std::byte{'K'} || request[1] != std::byte{'V'} ||
request[2] != std::byte{'P'} || request[3] != std::byte{'1'} ||
get(request, 4, 2) != 1) continue;
const auto operation = get(request, 6, 2);
const auto id = get(request, 8, 8);
const auto key = static_cast<std::uint32_t>(get(request, 16, 4));
const auto value = static_cast<std::uint32_t>(get(request, 20, 4));
Reply result{3, 0, 0};
if (operation == 1) result = store.put_value(id, key, value);
else if (operation == 2 && value == 0) result = store.get_value(key);
std::array<std::byte, 24> response{};
response[0] = std::byte{'K'}; response[1] = std::byte{'V'};
response[2] = std::byte{'R'}; response[3] = std::byte{'1'};
put(response, 4, result.status, 4);
put(response, 8, id, 8);
put(response, 16, result.value, 4);
put(response, 20, result.flags, 4);
const auto sent = pipe_io(pipe.value, true, response.data(), 24);
if (sent.error != ERROR_SUCCESS || sent.bytes != 24) continue;
std::byte acknowledgment{};
const auto ack = pipe_io(pipe.value, false, &acknowledgment, 1);
if (ack.error != ERROR_SUCCESS || ack.bytes != 1 ||
acknowledgment != std::byte{0xac})
std::cerr << "reply acknowledgment not received\n";
}
} catch (const std::exception& error) {
std::cerr << error.what() << '\n';
return 1;
}
}
הלקוח¶
הלקוח מאמת את מסגרת התשובה ואת מזהה הבקשה לפני שהוא מציג תוצאה. כישלון בקבלת תשובה משאיר את התוצאה לא ידועה; הוא אינו מייצר מזהה חדש אוטומטית. שליחת ACK נועדה לאפשר ניתוק מסודר, ואינה נקודת ה־commit של העסקה.
#include "wal_lab.hpp"
int wmain(int argc, wchar_t** argv) {
if (argc != 5) return 2;
try {
const std::wstring command = argv[1];
if (command != L"put" && command != L"get") return 2;
auto number = [](const wchar_t* text) {
const std::wstring input(text);
if (input.empty() ||
input.find_first_not_of(L"0123456789") != std::wstring::npos)
throw std::invalid_argument("decimal number required");
std::size_t used = 0;
const auto value = std::stoull(input, &used);
if (used != input.size()) throw std::invalid_argument("trailing input");
return static_cast<std::uint64_t>(value);
};
const auto id = number(argv[2]);
const auto key = number(argv[3]);
const auto value = number(argv[4]);
if (key > 0xffffffffULL || value > 0xffffffffULL) return 2;
constexpr auto name = L"\\\\.\\pipe\\AmitTech.WalLab";
if (!WaitNamedPipeW(name, 2000)) fail("wait pipe");
Handle pipe(CreateFileW(name, GENERIC_READ | GENERIC_WRITE, 0,
nullptr, OPEN_EXISTING, FILE_FLAG_OVERLAPPED |
SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION, nullptr));
if (pipe.value == INVALID_HANDLE_VALUE) fail("open pipe");
DWORD mode = PIPE_READMODE_MESSAGE;
if (!SetNamedPipeHandleState(pipe.value, &mode, nullptr, nullptr))
fail("pipe mode");
std::array<std::byte, 24> request{};
request[0] = std::byte{'K'}; request[1] = std::byte{'V'};
request[2] = std::byte{'P'}; request[3] = std::byte{'1'};
put(request, 4, 1, 2); put(request, 6, command == L"put" ? 1 : 2, 2);
put(request, 8, id, 8); put(request, 16, key, 4); put(request, 20, value, 4);
const auto sent = pipe_io(pipe.value, true, request.data(), 24);
if (sent.error || sent.bytes != 24) throw std::runtime_error("send failed");
std::array<std::byte, 24> response{};
const auto received = pipe_io(pipe.value, false, response.data(), 24);
if (received.error || received.bytes != 24)
throw std::runtime_error("result unknown: receive failed");
if (response[0] != std::byte{'K'} || response[1] != std::byte{'V'} ||
response[2] != std::byte{'R'} || response[3] != std::byte{'1'} ||
get(response, 8, 8) != id || get(response, 4, 4) > 4 ||
get(response, 20, 4) > 1)
throw std::runtime_error("invalid response");
const auto status = get(response, 4, 4);
const auto flags = get(response, 20, 4);
if (status == 0 && flags != 1)
throw std::runtime_error("success lacks durable state flag");
std::byte ack{0xac};
const auto acknowledged = pipe_io(pipe.value, true, &ack, 1);
if (acknowledged.error || acknowledged.bytes != 1)
std::cerr << "ack failed; received result remains known\n";
std::cout << "status=" << status << " value=" << get(response, 16, 4)
<< " durable=" << flags << '\n';
return status == 0 ? 0 : 3;
} catch (const std::exception& error) {
std::cerr << error.what() << '\n';
return 1;
}
}
הוכחת recovery¶
כל עסקה מאושרת מורכבת משישים וארבעה בתים: data ואחריו commit עם אותם מזהה, מפתח וערך. כל רשומה נבדקת בנפרד, וכל זוג נבדק גם כיחידה. רק לאחר התאמה מעדכנים את שתי המפות. מזהה כפול ביומן נחשב הפרת פורמט, משום ש־replay תקין אינו מוסיף זוג חדש.
סיומת קצרה מזוג היא תוצר אפשרי של append שלא אושר. הפתרון מדווח וחותך אותה לגבול העסקה האחרון, ואז מבצע flush לפני המשך. המדיניות הזו מניחה כשל append בסוף יומן המעבדה; היא אינה כלי תיקון כללי להשחתה שרירותית. זוג שלם עם checksum שגוי גורם לעצירה, כדי לא להסתיר אובדן מידע שהיה אולי מאושר.
הפורמט מוגבל מראש לשמונה קילובתים של עסקאות. ההגבלה מאפשרת לקרוא ולבדוק את כולו בזיכרון, ומונעת קלט שמבקש הקצאה בלתי מוגבלת. אין eviction של מזהי בקשות, ולכן ניסיון חוזר נשאר מוכר במשך חיי היומן הזה. הרחבה מעבר לקיבולת דורשת פרוטוקול retention או checkpoint מפורש.
שלוש נקודות הקריסה¶
כדי לבדוק יציאה אחרי data, יוצרים יומן חדש עם האפשרות המתאימה:
wal_server.exe create crash-data.wal --crash-after-data
wal_client.exe put 1 7 99
wal_server.exe open crash-data.wal
wal_client.exe put 1 7 99
הניסוי מפעיל ExitProcess רק בתוך תהליך השרת שנבנה למעבדה. הוא אינו מפסיק תהליכים אחרים. לאחר data בלבד, העסקה אינה שלמה והסיומת מוסרת. לאחר commit ולפני flush, ייתכן שהזוג יופיע בפתיחה מחדש, אך אין ללקוח אישור קודם. לאחר flush, הזוג אמור להיות ניתן לשחזור לפי חוזה האחסון, וה־retry מחזיר תוצאה שמורה.
קריסת תהליך אינה מדמה באופן מלא אובדן חשמל. ה־cache של המערכת עשוי להמשיך לכתוב גם אחרי יציאת התהליך. לכן הדוח צריך לומר איזה מסלול נבדק בפועל, ולא להציג את הניסוי כהוכחה לכל התקן ותצורה.
חיי הצינור והאמון¶
כל פעולה חופפת מחזיקה OVERLAPPED ואירוע מקומיים עד השלמה. מנגנון הזמן מבקש ביטול ואז אוסף תוצאה, ולכן אינו מבטיח גבול זמן קשיח מול driver שאינו מתקדם. הודעה גדולה או קצרה נדחית, והשרת אינו מקצה מתוך שדה אורך שנשלח על ידי הלקוח.
ה־DACL מאפשרת את המשתמש הנוכחי ואת SYSTEM. משתמשים אחרים ולקוחות מרוחקים אינם חלק מחוזה המעבדה. תהליך אחר באותו חשבון עדיין שייך לאותו גבול אמון, והשם אינו מאמת את זהות התוכנית. שירות אמיתי צריך להחליט מי הלקוח, אילו פעולות מותרות לו ואיך מזהים שרת מהימן.
לא קוראים FlushFileBuffers על ה־pipe כדי לחכות ללקוח, משום שהמתנה כזו יכולה לחסום. במקום זאת קוראים הודעת ACK עם אותה תשתית ביטול. אם התשובה כבר נכתבה אבל ה־ACK לא מגיע, העסקה נשארת מאושרת וה־retry פותר את אי־הוודאות.
בדיקות נוספות¶
לבדיקת corruption העתיקו יומן מעבדה תקין לקובץ חדש ושנו בית בתוך עסקה שלמה. פתיחת העותק צריכה להיכשל בלי לחתוך עסקאות קודמות. לבדיקת קיבולת שלחו מאה עשרים ושמונה מזהים שונים ובדקו שהבא נדחה ללא גידול ביומן.
לבדיקת מסגרת גדולה, שלחו הודעת pipe ארוכה מן הגודל הקבוע ובדקו שהחיבור נסגר בלי append. לבדיקת replay השוו את גודל היומן לפני ואחרי בקשה זהה. כל בדיקה צריכה לתעד תוצאה אמיתית; קוד הפתרון אינו מצרף פלט מומצא.
החוזים של שרת pipe עם פעולות חופפות ושל אבטחת pipe תומכים בצד התקשורת. חוזה FlushFileBuffers משמש את גבול האישור של היומן.