Concurrency: Implement resume(returning:) and resume(throwing:) for Unsafe*Continuation.

Author: jckarter

include/swift/Runtime/Concurrency.h

@@ -22,6 +22,8 @@
 namespace swift {
 class DefaultActor;
 
+struct SwiftError;
+
 struct AsyncTaskAndContext {
   AsyncTask *Task;
   AsyncContext *InitialContext;
@@ -271,6 +273,24 @@ void swift_defaultActor_destroy(DefaultActor *actor);
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 void swift_defaultActor_enqueue(Job *job, DefaultActor *actor);
 
+/// Resume a task from its continuation, given a normal result value.
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+void swift_continuation_resume(/* +1 */ OpaqueValue *result,
+                               void *continuation,
+                               const Metadata *resumeType);
+
+/// Resume a task from its throwing continuation, given a normal result value.
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+void swift_continuation_throwingResume(/* +1 */ OpaqueValue *result,
+                                       void *continuation,
+                                       const Metadata *resumeType);
+
+/// Resume a task from its throwing continuation by throwing an error.
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+void swift_continuation_throwingResumeWithError(/* +1 */ SwiftError *error,
+                                                void *continuation,
+                                                const Metadata *resumeType);
+
 }
 
 #endif

stdlib/public/Concurrency/PartialAsyncTask.swift

@@ -24,25 +24,28 @@ public struct PartialAsyncTask {
 public struct UnsafeContinuation<T> {
   @usableFromInline internal var context: Builtin.RawUnsafeContinuation
 
-  public func resume(returning: __owned T) { }
-
   @_alwaysEmitIntoClient
   internal init(_ context: Builtin.RawUnsafeContinuation) {
     self.context = context
   }
+
+  @_silgen_name("swift_continuation_resume")
+  public func resume(returning value: __owned T)
 }
 
 @frozen
 public struct UnsafeThrowingContinuation<T> {
   @usableFromInline internal var context: Builtin.RawUnsafeContinuation
 
-  public func resume(returning: __owned T) { }
-  public func resume(throwing: __owned Error) { }
-
   @_alwaysEmitIntoClient
   internal init(_ context: Builtin.RawUnsafeContinuation) {
     self.context = context
   }
+
+  @_silgen_name("swift_continuation_throwingResume")
+  public func resume(returning: __owned T)
+  @_silgen_name("swift_continuation_throwingResumeWithError")
+  public func resume(throwing: __owned Error)
 }
 
 #if _runtime(_ObjC)
@@ -92,4 +95,4 @@ public func withUnsafeThrowingContinuation<T>(
   return try await Builtin.withUnsafeThrowingContinuation {
     fn(UnsafeThrowingContinuation<T>($0))
   }
-}
+}

stdlib/public/Concurrency/Task.cpp

@@ -20,6 +20,9 @@
 #include "swift/Runtime/HeapObject.h"
 #include "TaskPrivate.h"
 
+// TODO: We shouldn't need this
+#include <dispatch/dispatch.h>
+
 using namespace swift;
 using FutureFragment = AsyncTask::FutureFragment;
 
@@ -355,3 +358,70 @@ void swift::swift_task_run(AsyncTask *taskToRun) {
 JobFlags swift::swift_task_getJobFlags(AsyncTask *task) {
   return task->Flags;
 }
+
+namespace {
+  
+/// Structure that gets filled in when a task is suspended by `withUnsafeContinuation`.
+struct AsyncContinuationContext {
+  // These fields are unnecessary for resuming a continuation.
+  void *Unused1;
+  void *Unused2;
+  // Storage slot for the error result, if any.
+  SwiftError *ErrorResult;
+  // Pointer to where to store a normal result.
+  OpaqueValue *NormalResult;
+  
+  // Executor on which to resume execution.
+  ExecutorRef ResumeExecutor;
+};
+
+static void resumeTaskAfterContinuation(AsyncTask *task,
+                                        AsyncContinuationContext *context) {
+#if __APPLE__
+  // TODO: Enqueue the task on the specific executor in the continuation
+  // context.
+  //
+  // For now, just enqueue the task resumption on the global concurrent queue
+  // so that we're able to return back to the caller of resume.
+  
+  dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0),
+                 ^{
+                  task->run(context->ResumeExecutor);
+                 });
+#else
+  swift_unreachable("not implemented");
+#endif
+}
+
+
+}
+
+SWIFT_CC(swift)
+void swift::swift_continuation_resume(/* +1 */ OpaqueValue *result,
+                                      void *continuation,
+                                      const Metadata *resumeType) {
+  auto task = reinterpret_cast<AsyncTask*>(continuation);
+  auto context = reinterpret_cast<AsyncContinuationContext*>(task->ResumeContext);
+  resumeType->vw_initializeWithTake(context->NormalResult, result);
+  
+  resumeTaskAfterContinuation(task, context);
+}
+
+SWIFT_CC(swift)
+void swift::swift_continuation_throwingResume(/* +1 */ OpaqueValue *result,
+                                              void *continuation,
+                                              const Metadata *resumeType) {
+  return swift_continuation_resume(result, continuation, resumeType);
+}
+
+
+SWIFT_CC(swift)
+void swift::swift_continuation_throwingResumeWithError(/* +1 */ SwiftError *error,
+                                                       void *continuation,
+                                                       const Metadata *resumeType) {
+  auto task = reinterpret_cast<AsyncTask*>(continuation);
+  auto context = reinterpret_cast<AsyncContinuationContext*>(task->ResumeContext);
+  context->ErrorResult = error;
+  
+  resumeTaskAfterContinuation(task, context);
+}