cleanup after rebase; this is likely incorrect in some way

Author: ktoso

include/swift/ABI/TaskGroup.h

@@ -1,4 +1,4 @@
-//===--- Task.h - ABI structures for asynchronous tasks ---------*- C++ -*-===//
+//===--- TaskGroup.h - ABI structures for task groups -00--------*- C++ -*-===//
 //
 // This source file is part of the Swift.org open source project
 //
@@ -53,7 +53,7 @@ namespace swift {
         Error = 0b11,
     };
 
-    enum class GroupPollStatus : uintptr_t {
+    enum class PollStatus : uintptr_t {
         /// The channel is known to be empty and we can immediately return nil.
         Empty = 0,
 
@@ -70,7 +70,7 @@ namespace swift {
 
     /// The result of waiting on a Channel (TaskGroup).
     struct PollResult {
-        GroupPollStatus status; // TODO: pack it into storage pointer or not worth it?
+        PollStatus status; // TODO: pack it into storage pointer or not worth it?
 
         /// Storage for the result of the future.
         ///
@@ -91,18 +91,18 @@ namespace swift {
         AsyncTask *retainedTask;
 
         bool isStorageAccessible() {
-          return status == GroupPollStatus::Success ||
-              status == GroupPollStatus::Error ||
-              status == GroupPollStatus::Empty;
+          return status == PollStatus::Success ||
+              status == PollStatus::Error ||
+              status == PollStatus::Empty;
         }
 
         static PollResult get(AsyncTask *asyncTask, bool hadErrorResult,
                                    bool needsSwiftRelease) {
           auto fragment = asyncTask->futureFragment();
           return PollResult{
               /*status*/ hadErrorResult ?
-                  TaskGroup::GroupPollStatus::Error :
-                  TaskGroup::GroupPollStatus::Success,
+                  TaskGroup::PollStatus::Error :
+                  TaskGroup::PollStatus::Success,
               /*storage*/ hadErrorResult ?
                   reinterpret_cast<OpaqueValue *>(fragment->getError()) :
                   fragment->getStoragePtr(),
@@ -380,9 +380,9 @@ namespace swift {
     /// Attempt to dequeue ready tasks and complete the waitingTask.
     ///
     /// If unable to complete the waiting task immediately (with an readily
-    /// available completed task), either returns an `GroupPollStatus::Empty`
+    /// available completed task), either returns an `PollStatus::Empty`
     /// result if it is known that no pending tasks in the group,
-    /// or a `GroupPollStatus::MustWait` result if there are tasks in flight
+    /// or a `PollStatus::MustWait` result if there are tasks in flight
     /// and the waitingTask eventually be woken up by a completion.
     TaskGroup::PollResult poll(AsyncTask *waitingTask);
 

include/swift/Runtime/Concurrency.h

@@ -161,19 +161,25 @@ SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swiftasync)
 TaskFutureWaitThrowingSignature::FunctionType
 swift_task_future_wait_throwing;
 
+//using TaskFutureWaitSignature =
+//  AsyncSignature<void(AsyncTask *, OpaqueValue *), /*throws*/ false>;
+
+using TaskGroupFutureWaitSignature =
+  AsyncSignature<void(AsyncTask *, TaskGroup *, OpaqueValue *), /*throws*/ true>; // TODO: is this correct?
+
 /// Wait for a readyQueue of a Channel to become non empty.
 ///
 /// This can be called from any thread. Its Swift signature is
 ///
 /// \code
-/// func swift_task_group_wait_next(
+/// func swift_task_group_wait_next_throwing(
 ///     waitingTask: Builtin.NativeObject, // current task
 ///     group: Builtin.NativeObject,
-/// ) async -> (hadErrorResult: Bool, storage: UnsafeRawPointer?)
+/// ) async -> T
 /// \endcode
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swiftasync)
-TaskFutureWaitSignature::FunctionType
-swift_task_group_wait_next;
+TaskGroupFutureWaitSignature::FunctionType
+swift_task_group_wait_next_throwing;
 
 /// Create a new `TaskGroup` using the task's allocator.
 /// The caller is responsible for retaining and managing the group's lifecycle.

lib/IRGen/IRGenSIL.cpp

@@ -2290,7 +2290,7 @@ static FunctionPointer::Kind classifyFunctionPointerKind(SILFunction *fn) {
       return SpecialKind::TaskFutureWait;
     if (name.equals("swift_task_future_wait_throwing"))
       return SpecialKind::TaskFutureWaitThrowing;
-    if (name.equals("swift_task_group_wait_next"))
+    if (name.equals("swift_task_group_wait_next_throwing"))
       return SpecialKind::TaskGroupWaitNext;
   }
 

stdlib/public/Concurrency/TaskGroup.cpp

@@ -93,7 +93,7 @@ void TaskGroup::destroy(AsyncTask *task) {
   bool taskDequeued = readyQueue.dequeue(item);
   while (taskDequeued) {
     swift_release(item.getTask());
-    bool taskDequeued = readyQueue.dequeue(item);
+    taskDequeued = readyQueue.dequeue(item);
   }
   mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
 
@@ -105,17 +105,18 @@ void TaskGroup::destroy(AsyncTask *task) {
 // ==== offer ------------------------------------------------------------------
 
 static void fillGroupNextResult(TaskFutureWaitAsyncContext *context,
-                     AsyncTask::GroupFragment::GroupPollResult result) {
+                                TaskGroup::PollResult result) {
+  /// Fill in the result value
   switch (result.status) {
-  case GroupFragment::GroupPollStatus::Waiting:
+  case TaskGroup::PollStatus::MustWait:
     assert(false && "filling a waiting status?");
     return;
 
-  case GroupFragment::GroupPollStatus::Error:
+  case TaskGroup::PollStatus::Error:
     context->fillWithError(reinterpret_cast<SwiftError*>(result.storage));
     return;
 
-  case GroupFragment::GroupPollStatus::Success: {
+  case TaskGroup::PollStatus::Success: {
     // Initialize the result as an Optional<Success>.
     const Metadata *successType = context->successType;
     OpaqueValue *destPtr = context->successResultPointer;
@@ -127,7 +128,7 @@ static void fillGroupNextResult(TaskFutureWaitAsyncContext *context,
     return;
   }
 
-  case GroupFragment::GroupPollStatus::Empty: {
+  case TaskGroup::PollStatus::Empty: {
     // Initialize the result as a nil Optional<Success>.
     const Metadata *successType = context->successType;
     OpaqueValue *destPtr = context->successResultPointer;
@@ -182,7 +183,14 @@ void TaskGroup::offer(AsyncTask *completedTask, AsyncContext *context,
             completedTask, hadErrorResult, /*needsRelease*/ false);
 
         mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
-        swift::runTaskWithPollResult(waitingTask, completingExecutor, result);
+        // swift::runTaskWithPollResult(waitingTask, completingExecutor, result);
+        auto waitingContext =
+          static_cast<TaskFutureWaitAsyncContext *>(
+            waitingTask->ResumeContext);
+        fillGroupNextResult(waitingContext, result);
+
+        // TODO: allow the caller to suggest an executor
+        swift_task_enqueueGlobal(waitingTask);
         return;
       } // else, try again
 
@@ -216,31 +224,33 @@ void TaskGroup::offer(AsyncTask *completedTask, AsyncContext *context,
 // ==== group.next() implementation (wait_next and groupPoll) ------------------
 
 SWIFT_CC(swiftasync)
-void swift::swift_task_group_wait_next(
+void swift::swift_task_group_wait_next_throwing(
     AsyncTask *waitingTask,
     ExecutorRef executor,
     SWIFT_ASYNC_CONTEXT AsyncContext *rawContext) {
   waitingTask->ResumeTask = rawContext->ResumeParent;
   waitingTask->ResumeContext = rawContext;
 
-  auto context = static_cast<TaskGroupNextWaitAsyncContext *>(rawContext);
+  auto context = static_cast<TaskFutureWaitAsyncContext *>(rawContext);
   auto task = context->task;
   auto group = context->group;
-  TaskGroup::PollResult polled = group->poll(waitingTask);
+fprintf(stderr, "[%s:%d](%s) group: %d\n", __FILE_NAME__, __LINE__, __FUNCTION__, group);
+
+assert(group && "swift_task_group_wait_next_throwing was passed context without group!");
 
-  if (polled.status == TaskGroup::GroupPollStatus::MustWait) {
+  TaskGroup::PollResult polled = group->poll(waitingTask);
+  switch (polled.status) {
+  case TaskGroup::PollStatus::MustWait:
     // The waiting task has been queued on the channel,
     // there were pending tasks so it will be woken up eventually.
     return;
 
-  case GroupFragment::GroupPollStatus::Empty:
-  case GroupFragment::GroupPollStatus::Error:
-  case GroupFragment::GroupPollStatus::Success:
+  case TaskGroup::PollStatus::Empty:
+  case TaskGroup::PollStatus::Error:
+  case TaskGroup::PollStatus::Success:
     fillGroupNextResult(context, polled);
     return waitingTask->runInFullyEstablishedContext(executor);
   }
-
-  runTaskWithPollResult(waitingTask, executor, polled);
 }
 
 TaskGroup::PollResult TaskGroup::poll(AsyncTask *waitingTask) {
@@ -257,7 +267,7 @@ TaskGroup::PollResult TaskGroup::poll(AsyncTask *waitingTask) {
     // was issued, and if we parked here we'd potentially never be woken up.
     // Bail out and return `nil` from `group.next()`.
     statusRemoveWaiting();
-    result.status = TaskGroup::GroupPollStatus::Empty;
+    result.status = TaskGroup::PollStatus::Empty;
     mutex.unlock(); // TODO: remove group lock, and use status for synchronization
     return result;
   }
@@ -277,7 +287,7 @@ TaskGroup::PollResult TaskGroup::poll(AsyncTask *waitingTask) {
       ReadyQueueItem item;
       bool taskDequeued = readyQueue.dequeue(item);
       if (!taskDequeued) {
-        result.status = TaskGroup::GroupPollStatus::MustWait;
+        result.status = TaskGroup::PollStatus::MustWait;
         result.storage = nullptr;
         result.retainedTask = nullptr;
         mutex.unlock(); // TODO: remove group lock, and use status for synchronization
@@ -294,23 +304,23 @@ TaskGroup::PollResult TaskGroup::poll(AsyncTask *waitingTask) {
       switch (item.getStatus()) {
         case ReadyStatus::Success:
           // Immediately return the polled value
-          result.status = TaskGroup::GroupPollStatus::Success;
+          result.status = TaskGroup::PollStatus::Success;
           result.storage = futureFragment->getStoragePtr();
           assert(result.retainedTask && "polled a task, it must be not null");
           mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
           return result;
 
         case ReadyStatus::Error:
           // Immediately return the polled value
-          result.status = TaskGroup::GroupPollStatus::Error;
+          result.status = TaskGroup::PollStatus::Error;
           result.storage =
               reinterpret_cast<OpaqueValue *>(futureFragment->getError());
           assert(result.retainedTask && "polled a task, it must be not null");
           mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
           return result;
 
         case ReadyStatus::Empty:
-          result.status = TaskGroup::GroupPollStatus::Empty;
+          result.status = TaskGroup::PollStatus::Empty;
           result.storage = nullptr;
           result.retainedTask = nullptr;
           mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
@@ -330,7 +340,7 @@ TaskGroup::PollResult TaskGroup::poll(AsyncTask *waitingTask) {
         /*failure*/ std::memory_order_acquire)) {
       mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
       // no ready tasks, so we must wait.
-      result.status = TaskGroup::GroupPollStatus::MustWait;
+      result.status = TaskGroup::PollStatus::MustWait;
       return result;
     } // else, try again
   }

stdlib/public/Concurrency/TaskGroup.swift

@@ -219,29 +219,30 @@ extension Task {
         """)
       #endif
 
-      let rawResult = await _taskGroupWaitNext(waitingTask: _task, group: _group)
-
-      // FIXME: we are now in the parent task again, and can safely modify the
-      //        records again; we should remove this specific child from the
-      //        group->record
-
-      if rawResult.hadErrorResult {
-        // Throw the result on error.
-        let error = unsafeBitCast(rawResult.storage, to: Error.self)
-        throw error
-      }
-
-      guard let storage = rawResult.storage else {
-        // The group was empty, return nil
-        return nil
-      }
-
-      // Take the value on success.
-      let storagePtr =
-        storage.bindMemory(to: TaskResult.self, capacity: 1)
-      let value = UnsafeMutablePointer<TaskResult>(mutating: storagePtr).pointee
-
-      return value
+      return try await _taskGroupWaitNext(waitingTask: _task, group: _group)
+//      let rawResult = await _taskGroupWaitNext(waitingTask: _task, group: _group)
+//
+//      // FIXME: we are now in the parent task again, and can safely modify the
+//      //        records again; we should remove this specific child from the
+//      //        group->record
+//
+//      if rawResult.hadErrorResult {
+//        // Throw the result on error.
+//        let error = unsafeBitCast(rawResult.storage, to: Error.self)
+//        throw error
+//      }
+//
+//      guard let storage = rawResult.storage else {
+//        // The group was empty, return nil
+//        return nil
+//      }
+//
+//      // Take the value on success.
+//      let storagePtr =
+//        storage.bindMemory(to: TaskResult.self, capacity: 1)
+//      let value = UnsafeMutablePointer<TaskResult>(mutating: storagePtr).pointee
+//
+//      return value
     }
 
     /// Query whether the group has any remaining tasks.
@@ -363,13 +364,6 @@ func _taskGroupCreate(
   task: Builtin.NativeObject
 ) -> Builtin.NativeObject
 
-///// Attach the group to the task.
-//@_silgen_name("swift_task_group_attach")
-//func _taskGroupAttach(
-//  group: Builtin.NativeObject,
-//  to task: Builtin.NativeObject
-//) -> UnsafeRawPointer /*TaskGroupTaskStatusRecord*/
-
 /// Attach task group child to the group group to the task.
 @_silgen_name("swift_task_group_attachChild")
 func _taskGroupAttachChild(
@@ -378,16 +372,6 @@ func _taskGroupAttachChild(
   child: Builtin.NativeObject
 ) -> UnsafeRawPointer /*ChildTaskStatusRecord*/
 
-///// Detach the group to the task.
-/////
-///// - Parameters:
-/////   - record: the TaskGroupTaskStatusRecord returned by _taskGroupAttach
-//@_silgen_name("swift_task_group_detach")
-//func _taskGroupDetach(
-//  record: UnsafeRawPointer,
-//  from task: Builtin.NativeObject
-//)
-
 @_silgen_name("swift_task_group_destroy")
 func _taskGroupDestroy(
   task: Builtin.NativeObject,
@@ -413,13 +397,13 @@ func _taskGroupIsCancelled(
   group: Builtin.NativeObject
 ) -> Bool
 
-@_silgen_name("swift_task_group_wait_next")
+@_silgen_name("swift_task_group_wait_next_throwing")
 func _taskGroupWaitNext<T>(
   waitingTask: Builtin.NativeObject,
   group: Builtin.NativeObject
-) async -> (hadErrorResult: Bool, storage: UnsafeRawPointer?)
+) async throws -> T?
 
-enum GroupPollStatus: Int {
+enum PollStatus: Int {
   case empty   = 0
   case waiting = 1
   case success = 2

stdlib/public/Concurrency/TaskPrivate.h

@@ -82,6 +82,7 @@ class TaskFutureWaitAsyncContext : public AsyncContext {
   AsyncTask *task;
 
   // Only in swift_task_group_next
+  TaskGroup *group;
   const Metadata *successType;
 
   using AsyncContext::AsyncContext;

test/Concurrency/Runtime/async_taskgroup_next_on_completed.swift

@@ -33,7 +33,6 @@ func test_sum_nextOnCompleted() async {
     var sum = 0
     do {
       while let r = try await group.next() {
-        fputs("error: \(#function)[\(#file):\(#line)]: next: \(r)\n", stderr)
         print("next: \(r)")
         sum += r
         print("sum: \(sum)")
@@ -44,7 +43,6 @@ func test_sum_nextOnCompleted() async {
 
     assert(group.isEmpty, "Group must be empty after we consumed all tasks")
 
-    fputs("error: \(#function)[\(#file):\(#line)]: group returning: \(sum)", stderr)
     print("task group returning: \(sum)")
     return sum
   }