פתרון הפרויקט: מנוע טבעי עם השלמה מנוהלת¶
נממש תחילה מנוע x64 מלא שמשתמש ב־thread לכל פעולה כדי להפוך את חיי הפעולה לקלים לצפייה. זהו בסיס לימודי מוגבל ל־64 MiB לבקשה, ולא מנוע שמיועד לאלפי threads במקביל. ההגבלה של שתי פעולות חיות תופיע בצד המנוהל, ואפשר להחליף בהמשך את התזמון הטבעי בלי לשנות את חוזה ההשלמה.
הקובץ sum_native.cpp הוא DLL מלא. החוזה מתיר input שהוא null רק כאשר length הוא אפס, ומבטיח callback אחד בדיוק אחרי start מוצלח. במקרה start שנכשל אין callback ואין operation לבעלות הקורא.
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <atomic>
#include <cstdint>
#include <memory>
#include <new>
#include <vector>
struct options_v1 { std::uint32_t size; std::uint32_t version; };
using completion_fn = void(__cdecl*)(void*, std::int32_t, std::uint64_t);
struct operation {
std::vector<std::uint8_t> input;
std::atomic<bool> canceled{false};
completion_fn callback;
void* context;
HANDLE thread{};
};
static DWORD WINAPI work(void* raw) {
auto* op = static_cast<operation*>(raw);
std::uint64_t total = 0;
std::int32_t status = 0;
for (std::size_t i = 0; i < op->input.size(); ++i) {
if ((i & 4095u) == 0 && op->canceled.load(std::memory_order_relaxed)) {
status = 2;
break;
}
total += op->input[i];
}
op->callback(op->context, status, status == 0 ? total : 0);
return 0;
}
extern "C" __declspec(dllexport) std::int32_t __cdecl sum_start(
const options_v1* options, const std::uint8_t* input, std::uint32_t length,
completion_fn callback, void* context, operation** output) noexcept {
if (!output) return 1;
*output = nullptr;
if (!options || options->size != sizeof(options_v1) || options->version != 1 ||
!callback || (!input && length != 0) || length > 64u * 1024u * 1024u)
return 1;
try {
auto op = std::make_unique<operation>();
if (length != 0) op->input.assign(input, input + length);
op->callback = callback;
op->context = context;
op->thread = CreateThread(nullptr, 0, work, op.get(), 0, nullptr);
if (!op->thread) return 3;
*output = op.release();
return 0;
} catch (const std::bad_alloc&) { return 4; }
catch (...) { return 3; }
}
extern "C" __declspec(dllexport) void __cdecl sum_cancel(operation* op) noexcept {
if (op) op->canceled.store(true, std::memory_order_relaxed);
}
extern "C" __declspec(dllexport) std::int32_t __cdecl sum_join(operation* op) noexcept {
if (!op) return 1;
if (WaitForSingleObject(op->thread, INFINITE) != WAIT_OBJECT_0) return 3;
if (!CloseHandle(op->thread)) return 3;
delete op;
return 0;
}
ה־callback יכול לרוץ לפני ש־sum_start חזר, אבל הוא משתמש רק בשדות שכבר אותחלו. הוא אינו קורא את שדה thread, שנכתב לאחר CreateThread. לאחר יצירת ה־thread אין במסלול start פעולה רגילה שזורקת exception לפני העברת הבעלות. ה־relaxed של canceled מספיק כאן משום שזהו דגל עצמאי שאינו מפרסם נתונים אחרים. ה־join מסתמך על השלמת thread כדי לסיים את כל הגישות ל־operation.
נבנה עם toolchain של Windows x64. הפקודה הראשונה מיועדת ל־Developer PowerShell של MSVC, והשנייה היא חלופה עם Zig שמספק GNU ABI ל־C++ אבל מייצאת את אותו C ABI שהלקוח צורך. אין צורך לערבב ביניהן באותו build.
הפרויקט המנוהל הוא console של net10.0 עם AllowUnsafeBlocks=true ו־PlatformTarget=x64. הוסיפו את ה־DLL לתיקיית הפלט באמצעות פריט Content עם CopyToOutputDirectory=PreserveNewest, או העתיקו אותו לשם כחלק מסקריפט בנייה מבוקר. הקובץ Program.cs הבא מלא ומשתמש ב־resolver שמעדיף את עותק הספרייה שבתיקיית האפליקציה.
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
NativeLibrary.SetDllImportResolver(Assembly.GetExecutingAssembly(),
(name, _, _) => name == "sum_native"
? NativeLibrary.Load(Path.Combine(AppContext.BaseDirectory, "sum_native.dll")) : 0);
byte[] input = Enumerable.Range(0, 100_000).Select(i => (byte)i).ToArray();
ulong expected = 0;
foreach (byte value in input) expected += value;
using var capacity = new SemaphoreSlim(2);
using var stop = new CancellationTokenSource();
Task<ulong>[] jobs = Enumerable.Range(0, 4).Select(_ => RunBounded()).ToArray();
foreach (ulong result in await Task.WhenAll(jobs))
if (result != expected) throw new InvalidOperationException("incorrect sum");
Console.WriteLine($"four results={expected}");
async Task<ulong> RunBounded()
{
await capacity.WaitAsync(stop.Token);
try { return await Bridge.SumAsync(input, stop.Token); }
finally { capacity.Release(); }
}
sealed class State
{
public readonly TaskCompletionSource<(int Status, ulong Total)> Source =
new(TaskCreationOptions.RunContinuationsAsynchronously);
}
static class Bridge
{
public static async Task<ulong> SumAsync(byte[] input, CancellationToken token)
{
ArgumentNullException.ThrowIfNull(input);
token.ThrowIfCancellationRequested();
var state = new State();
GCHandle root = GCHandle.Alloc(state);
nint operation;
try { operation = Start(input, GCHandle.ToIntPtr(root)); }
catch { root.Free(); throw; }
CancellationTokenRegistration cancellation = default;
try
{
cancellation = token.Register(() => Native.Cancel(operation));
var result = await state.Source.Task.ConfigureAwait(false);
return result.Status switch
{
0 => result.Total,
2 => throw new OperationCanceledException(token),
_ => throw new InvalidOperationException($"native status {result.Status}")
};
}
finally
{
cancellation.Dispose();
int status = await Task.Run(() => Native.Join(operation)).ConfigureAwait(false);
if (status != 0) Environment.FailFast("native join failed; context retained");
root.Free();
}
}
private static unsafe nint Start(byte[] input, nint context)
{
var options = new Options { Size = (uint)sizeof(Options), Version = 1 };
fixed (byte* pointer = input)
{
int status = Native.Start(in options, pointer, checked((uint)input.Length),
&Completed, context, out nint operation);
if (status != 0) throw new InvalidOperationException($"start status {status}");
return operation;
}
}
[UnmanagedCallersOnly(CallConvs = new[] { typeof(CallConvCdecl) })]
private static void Completed(nint context, int status, ulong total)
{
try
{
var state = (State)GCHandle.FromIntPtr(context).Target!;
if (!state.Source.TrySetResult((status, total)))
Environment.FailFast("duplicate native completion");
}
catch { Environment.FailFast("invalid native callback context"); }
}
}
[StructLayout(LayoutKind.Sequential)]
struct Options { public uint Size; public uint Version; }
internal static unsafe partial class Native
{
[LibraryImport("sum_native", EntryPoint = "sum_start")]
[UnmanagedCallConv(CallConvs = new[] { typeof(CallConvCdecl) })]
internal static partial int Start(in Options options, byte* input, uint length,
delegate* unmanaged[Cdecl]<nint, int, ulong, void> callback,
nint context, out nint operation);
[LibraryImport("sum_native", EntryPoint = "sum_cancel")]
[UnmanagedCallConv(CallConvs = new[] { typeof(CallConvCdecl) })]
internal static partial void Cancel(nint operation);
[LibraryImport("sum_native", EntryPoint = "sum_join")]
[UnmanagedCallConv(CallConvs = new[] { typeof(CallConvCdecl) })]
internal static partial int Join(nint operation);
}
ה־finally האסינכרוני הוא חלק מן ה־Task הציבורי, ולכן הצרכן אינו מקבל השלמה סופית לפני join ו־Free. הניתוק של רישום הביטול קודם ל־join וממתין ל־callbacks של הביטול לפי חוזה CancellationTokenRegistration.Dispose. כך לא יכולה להתחיל קריאת Cancel חדשה על operation שכבר נמחק. ה־join אינו רץ מתוך callback, ולכן אינו מחכה לעצמו.
ההעתקה הטבעית מנתקת את חיי קלט המערך מחיי הפעולה. אפשר להפוך את התכנון לאפס העתקות, אבל אז נצטרך לשמור קיבוע ובעלות בלעדית או חוזה קריאה יציב עד join, ולא רק עד סוף fixed. מספר הפעולות החיות מוגבל כאן באמצעות SemaphoreSlim; כדי להוסיף תור של בקשות ממתינות נשתמש ב־Channel bounded מן ההרצאה ונשמור גם מגבלת byte-ים, מפני שקלטים ממתינים צורכים זיכרון לפני start.
ביטול שמגיע אחרי החישוב יכול להפסיד להצלחה, וזה חלק מן החוזה. כדי לבדוק את מסלול הביטול באופן נשלט נוסיף בגרסת בדיקה barrier טבעי לפני עיבוד, נמתין לאות כניסה, נקרא Cancel ואז נשחרר אותו. בדיקה שמסתמכת על "בטח מיליון byte-ים לוקחים מספיק זמן" אינה דטרמיניסטית. את ממשק המשתמש נחבר ל־RunBounded עם דפוס Closing שממתין למשימת ה־cleanup; אין להחזיר buffer או לסגור חלון שמחזיק native state לפני שהחוזה הזה הושלם.
בגרסת ABI שנייה נוסיף struct חדש עם size/version/algorithm, נשמור את exports הישנים או נוסיף export חדש ונבדוק כל שילוב לקוח־ספרייה. שינוי packing או משמעות status בלי שינוי חוזה הוא שבירה גם כאשר DLL נטען. בתחקיר הסופי נצרף trace עם מזהה בקשה ואת סדר CallbackExited, Joined ו־Released כדי להוכיח את החזרת הבעלות, ולא להסתפק בתוצאה חשבונית נכונה.