diff --git a/crates/csharp/src/AsyncSupport.cs b/crates/csharp/src/AsyncSupport.cs index 37c617511..5b31f4522 100644 --- a/crates/csharp/src/AsyncSupport.cs +++ b/crates/csharp/src/AsyncSupport.cs @@ -35,6 +35,7 @@ public struct ContextTask { public int WaitableSetHandle; public int FutureHandle; + public int CallbackDepth; } public static class AsyncSupport @@ -94,6 +95,39 @@ internal static void Join(int readerWriterHandle, int waitableHandle, WaitableIn Interop.WaitableJoin(readerWriterHandle, waitableHandle); } + internal static unsafe void Unregister(int handle) + { + ContextTask* contextTaskPtr = ContextGet(); + if (contextTaskPtr == null) + { + return; + } + + var waitables = pendingTasks[contextTaskPtr->WaitableSetHandle]; + waitables.Remove(handle, out _); + Interop.WaitableJoin(handle, 0); + + if (contextTaskPtr->CallbackDepth == 0) + { + TryCleanupContext(contextTaskPtr, waitables); + } + } + + private static unsafe bool TryCleanupContext(ContextTask* contextTaskPtr, ConcurrentDictionary waitables) + { + if (waitables.Count != 0) + { + return false; + } + + var waitableSetHandle = contextTaskPtr->WaitableSetHandle; + pendingTasks.Remove(waitableSetHandle, out _); + WaitableSetDrop(waitableSetHandle); + ContextSet(null); + Marshal.FreeHGlobal((IntPtr)contextTaskPtr); + return true; + } + // TODO: Revisit this to see if we can remove it. // Only allow joining to a handle directly when there is no waitable. public static void Join(int handle) @@ -140,51 +174,58 @@ public static unsafe int Callback(EventWaitable e) var waitables = pendingTasks[contextTaskPtr->WaitableSetHandle]; var waitableInfoState = waitables[e.Waitable]; - if (e.IsDropped) - { - waitableInfoState.FutureStream!.OtherSideDropped(); - } - if (e.IsCompleted || e.IsDropped) { - // The operation is complete so we can free the buffer and remove the waitable from our dicitonary - waitables.Remove(e.Waitable, out _); - if (e.IsSubtask) + contextTaskPtr->CallbackDepth++; + try { - switch (e.SubtaskStatus) + if (e.IsDropped) { - case { IsStarting: true }: - throw new Exception("unexpected subtask status Starting " + e.Code); + waitableInfoState.FutureStream!.OtherSideDropped(); + } - case { IsStarted: true }: - break; + // The operation is complete so we can free the buffer and remove the waitable from our dicitonary + waitables.Remove(e.Waitable, out _); + Interop.WaitableJoin(e.Waitable, 0); + if (e.IsSubtask) + { + switch (e.SubtaskStatus) + { + case { IsStarting: true }: + throw new Exception("unexpected subtask status Starting " + e.Code); - case { IsReturned: true }: - waitableInfoState.SetResult(e.WaitableCount); - Interop.SubtaskDrop(e.Waitable); - break; + case { IsStarted: true }: + break; - default: - throw new Exception("TODO: subtask status " + e.Code); - } - } - else - { - if (e.IsDropped) - { - waitableInfoState.SetException(new StreamDroppedException()); + case { IsReturned: true }: + waitableInfoState.SetResult(e.WaitableCount); + Interop.SubtaskDrop(e.Waitable); + break; + + default: + throw new Exception("TODO: subtask status " + e.Code); + } } else { - // This may add a new waitable to the set. - waitableInfoState.SetResult(e.WaitableCount); + if (e.IsDropped) + { + waitableInfoState.SetException(new StreamDroppedException()); + } + else + { + // This may add a new waitable to the set. + waitableInfoState.SetResult(e.WaitableCount); + } } } + finally + { + contextTaskPtr->CallbackDepth--; + } - if (waitables.Count == 0) + if (contextTaskPtr->CallbackDepth == 0 && TryCleanupContext(contextTaskPtr, waitables)) { - ContextSet(null); - Marshal.FreeHGlobal((IntPtr)contextTaskPtr); return (int)CallbackCode.Exit; } @@ -289,6 +330,7 @@ internal LiftingTaskCompletionSource(TaskCompletionSource innerTaskCompleti { var contextTaskPtr = (ContextTask *)Marshal.AllocHGlobal(Marshal.SizeOf()); contextTaskPtr->WaitableSetHandle = WaitableSetNew(); + contextTaskPtr->CallbackDepth = 0; ContextSet(contextTaskPtr); return contextTaskPtr; } @@ -320,7 +362,9 @@ public class CancelableRead(IVTable cancelableVTable, int handle) : ICancelable { public uint Cancel() { - return cancelableVTable.CancelRead(handle); + var status = cancelableVTable.CancelRead(handle); + AsyncSupport.Unregister(handle); + return status; } } @@ -328,7 +372,9 @@ public class CancelableWrite(IVTable cancelableVTable, int handle) : ICancelable { public uint Cancel() { - return cancelableVTable.CancelWrite(handle); + var status = cancelableVTable.CancelWrite(handle); + AsyncSupport.Unregister(handle); + return status; } } @@ -1122,4 +1168,4 @@ public override void SetException(Exception e) } public T Result => tcs.Task.Result; -} \ No newline at end of file +} diff --git a/tests/runtime/future-cancel-read/disabled/runner.cs b/tests/runtime/future-cancel-read/runner.cs similarity index 79% rename from tests/runtime/future-cancel-read/disabled/runner.cs rename to tests/runtime/future-cancel-read/runner.cs index b14b778b8..83bd03ca8 100644 --- a/tests/runtime/future-cancel-read/disabled/runner.cs +++ b/tests/runtime/future-cancel-read/runner.cs @@ -25,11 +25,16 @@ public static async Task Run() var testTask = IIImports.StartReadThenCancel(dataReader, signalReader); async Task WriterAsync() { - await signalWriter.Write(); + // Make the data read ready first so that completing the signal + // synchronously cancels the last other operation. await dataWriter.Write(4); + await signalWriter.Write(); } await WriterAsync(); + await testTask; + dataWriter.Dispose(); + signalWriter.Dispose(); } } } diff --git a/tests/runtime/future-cancel-read/disabled/test.cs b/tests/runtime/future-cancel-read/test.cs similarity index 92% rename from tests/runtime/future-cancel-read/disabled/test.cs rename to tests/runtime/future-cancel-read/test.cs index 60485cc40..78c0f20fc 100644 --- a/tests/runtime/future-cancel-read/disabled/test.cs +++ b/tests/runtime/future-cancel-read/test.cs @@ -20,6 +20,7 @@ public static Task CancelAfterRead(FutureReader future) // If the cancel occurs before the read is complete (or the writer ignores the cancel) we return Cancelled. Debug.Assert(task.Cancel() == CancelCode.Cancelled); + future.Dispose(); return Task.CompletedTask; } @@ -31,6 +32,8 @@ public static async Task StartReadThenCancel(FutureReader future, FutureRe await signal.Read(); Debug.Assert(task.Cancel() == CancelCode.Completed); + future.Dispose(); + signal.Dispose(); } } -} \ No newline at end of file +} diff --git a/tests/runtime/ping-pong/disabled/runner.cs b/tests/runtime/ping-pong/disabled/runner.cs deleted file mode 100644 index f5621eb1b..000000000 --- a/tests/runtime/ping-pong/disabled/runner.cs +++ /dev/null @@ -1,36 +0,0 @@ -using System.Diagnostics; -using RunnerWorld.wit.Imports.my.test; -using RunnerWorld; - -public class RunnerWorldExportsImpl -{ - public static async Task Run() - { - try - { - string pingResult; - { - var (reader, writer) = IIImports.FutureNewString(); - var pingTask = IIImports.Ping(reader, "world"); - await writer.Write("hello"); - var pingFutureResult = await pingTask; - var result = await pingFutureResult.Read(); - Debug.Assert(result == "helloworld"); - - pingResult = result; - } - - { - var (reader, writer) = IIImports.FutureNewString(); - var pongTask = IIImports.Pong(reader); - await writer.Write(pingResult); - var pongResult = await pongTask; - Debug.Assert(pongResult == "helloworld"); - } - } - catch(Exception e) - { - Console.WriteLine(e); - } - } -} diff --git a/tests/runtime/ping-pong/runner.cs b/tests/runtime/ping-pong/runner.cs new file mode 100644 index 000000000..85e36171b --- /dev/null +++ b/tests/runtime/ping-pong/runner.cs @@ -0,0 +1,32 @@ +using System.Diagnostics; +using RunnerWorld.wit.Imports.my.test; +using RunnerWorld; + +public class RunnerWorldExportsImpl +{ + public static async Task Run() + { + string pingResult; + { + var (reader, writer) = IIImports.FutureNewString(); + var pingTask = IIImports.Ping(reader, "world"); + await writer.Write("hello"); + writer.Dispose(); + var pingFutureResult = await pingTask; + var result = await pingFutureResult.Read(); + pingFutureResult.Dispose(); + Debug.Assert(result == "helloworld"); + + pingResult = result; + } + + { + var (reader, writer) = IIImports.FutureNewString(); + var pongTask = IIImports.Pong(reader); + await writer.Write(pingResult); + writer.Dispose(); + var pongResult = await pongTask; + Debug.Assert(pongResult == "helloworld"); + } + } +} diff --git a/tests/runtime/ping-pong/disabled/test.cs b/tests/runtime/ping-pong/test.cs similarity index 83% rename from tests/runtime/ping-pong/disabled/test.cs rename to tests/runtime/ping-pong/test.cs index fbfc7b4b2..797af5322 100644 --- a/tests/runtime/ping-pong/disabled/test.cs +++ b/tests/runtime/ping-pong/test.cs @@ -9,10 +9,12 @@ public class IExportsImpl : IIExports public static async Task> Ping(FutureReader future, string s) { var msg = (await future.Read()) + s; + future.Dispose(); var (newFutureReader, newFutureWriter) = IIExports.FutureNewString(); var writeTask = newFutureWriter.Write(msg); writeTask.ContinueWith(t => { + newFutureWriter.Dispose(); if(t.Exception != null) { Debug.Fail("Exception in returned future write." + t.Exception); @@ -23,7 +25,9 @@ public static async Task> Ping(FutureReader future, public static async Task Pong(FutureReader future) { - return await future.Read(); + var result = await future.Read(); + future.Dispose(); + return result; } } }