Rebased to use new global executor

Author: ktoso

include/swift/ABI/Task.h

@@ -43,11 +43,6 @@ class alignas(2 * alignof(void*)) Job {
   enum {
     /// The next waiting task link, an AsyncTask that is waiting on a future.
     NextWaitingTaskIndex = 0,
-//    /// The next completed task link, an AsyncTask that is completed however
-//    /// has not been polled yet (by `group.next()`), so the channel task keeps
-//    /// the list in completion order, such that they can be polled out one by
-//    /// one.
-//    NextChannelCompletedTaskIndex = 1,
   };
 
 public:
@@ -218,7 +213,7 @@ class AsyncTask : public HeapObject, public Job {
     return reinterpret_cast<ChildFragment*>(this + 1);
   }
 
-  // ==== TaskGroup Channel ----------------------------------------------------
+  // ==== TaskGroup ------------------------------------------------------------
 
   class GroupFragment {
   public:
@@ -450,11 +445,6 @@ class AsyncTask : public HeapObject, public Job {
     NaiveQueue<ReadyQueueItem> readyQueue;
 //     mpsc_queue_t<ReadyQueueItem> readyQueue; // TODO: can we get away with an MPSC queue here once actor executors land?
 
-// NOTE: this style of "queue" is not very nice for the group,
-//        because it acts more like a stack, and we really want completion order
-//        for the task group, thus not using this style (which the wait queue does)
-//    std::atomic<ReadyQueueItem> readyQueue;
-
     /// Queue containing all of the tasks that are waiting in `get()`.
     ///
     /// A task group is also a future, and awaits on the group's result *itself*
@@ -481,13 +471,6 @@ class AsyncTask : public HeapObject, public Job {
       return oldStatus.pendingTasks() == 0;
     }
 
-    GroupStatus statusLoad() {
-      return GroupStatus {
-          // status.load(std::memory_order_acquire)
-           status.load(std::memory_order_seq_cst) // TODO: acquire instead
-      };
-    }
-
     /// Returns *assumed* new status, including the just performed +1.
     GroupStatus statusAddReadyTaskAcquire() {
       auto old = status.fetch_add(GroupStatus::oneReadyTask, std::memory_order_acquire);
@@ -706,12 +689,6 @@ class AsyncTask : public HeapObject, public Job {
         SchedulerPrivate[NextWaitingTaskIndex]);
   }
 
-//  /// Access the next completed task, which establishes a singly linked list of
-//  /// tasks that are waiting to be polled from a task group channel.
-//  AsyncTask *&getNextChannelReadyTask() {
-//    return reinterpret_cast<AsyncTask *&>(
-//        SchedulerPrivate[NextChannelCompletedTaskIndex]);
-//  }
 };
 
 // The compiler will eventually assume these.

include/swift/Runtime/Concurrency.h

@@ -133,6 +133,9 @@ struct TaskFutureWaitResult {
   OpaqueValue *storage;
 };
 
+using TaskFutureWaitSignature =
+  AsyncSignature<TaskFutureWaitResult(AsyncTask *), /*throws*/ false>;
+
 /// Wait for a future task to complete.
 ///
 /// This can be called from any thread. Its Swift signature is
@@ -151,10 +154,10 @@ swift_task_future_wait;
 ///
 /// \code
 /// func swift_task_group_wait_next(on groupTask: Builtin.NativeObject) async
-///     -> RawGroupPollResult?
+///     -> (hadErrorResult: Bool, storage: UnsafeRawPointer?)
 /// \endcode
-SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
-AsyncFunctionType<TaskFutureWaitResult(AsyncTask *task)>
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swiftasync)
+TaskFutureWaitSignature::FunctionType
 swift_task_group_wait_next;
 
 /// This can be called from any thread. Its Swift signature is

stdlib/public/Concurrency/CMakeLists.txt

@@ -44,9 +44,8 @@ add_swift_target_library(swift_Concurrency ${SWIFT_STDLIB_LIBRARY_BUILD_TYPES} I
   _TimeTypes.swift
   TaskAlloc.cpp
   TaskStatus.cpp
-  TaskGroup.swift
-  TaskQueues.h
   TaskGroup.cpp
+  TaskGroup.swift
   Mutex.cpp
   ${swift_concurrency_objc_sources}
 

stdlib/public/Concurrency/Task.cpp

@@ -182,7 +182,6 @@ static void completeTask(AsyncTask *task, ExecutorRef executor,
 
   // Release the task, balancing the retain that a running task has on itself.
   // If it was a group child task, it will remain until the group returns it.
-  assert(task && "task to be released MUST NOT BE NULL");
   swift_release(task);
 }
 

stdlib/public/Concurrency/Task.swift

@@ -353,6 +353,28 @@ public func _taskFutureGetThrowing<T>(
 }
 
 public func _runChildTask<T>(
+  operation: @escaping () async throws -> T
+) async -> Builtin.NativeObject {
+  let currentTask = Builtin.getCurrentAsyncTask()
+
+  // Set up the job flags for a new task.
+  var flags = Task.JobFlags()
+  flags.kind = .task
+  flags.priority = getJobFlags(currentTask).priority
+  flags.isFuture = true
+  flags.isChildTask = true
+
+  // Create the asynchronous task future.
+  let (task, _) = Builtin.createAsyncTaskFuture(
+      flags.bits, currentTask, operation)
+
+  // Enqueue the resulting job.
+  _enqueueJobGlobal(Builtin.convertTaskToJob(task))
+
+  return task
+}
+
+public func _runGroupChildTask<T>(
   overridingPriority priorityOverride: Task.Priority? = nil,
   operation: @escaping () async throws -> T
 ) async -> Builtin.NativeObject {

stdlib/public/Concurrency/TaskGroup.cpp

@@ -19,9 +19,6 @@
 #include "swift/ABI/Metadata.h"
 #include "swift/Runtime/HeapObject.h"
 #include "TaskPrivate.h"
-#include <iostream>
-#include <pthread.h>
-#include <stdio.h>
 
 using namespace swift;
 using GroupFragment = AsyncTask::GroupFragment;
@@ -34,52 +31,24 @@ using GroupPollResult = GroupFragment::GroupPollResult;
 // =============================================================================
 // ==== destroy ----------------------------------------------------------------
 
-// FIXME: implement this properly, unsure if this is correct
 void GroupFragment::destroy() {
   // TODO: need to release all waiters as well
 //  auto waitHead = waitQueue.load(std::memory_order_acquire);
 //  switch (waitHead.getStatus()) {
 //    case GroupFragment::WaitStatus::Waiting:
-//      assert(false && "destroying a task that never completed");
-//
-//    case GroupFragment::WaitStatus::Success:
-//      resultType->vw_destroy(getStoragePtr());
-//      break;
-//
-//    case GroupFragment::WaitStatus::Error:
-//      swift_unknownObjectRelease(reinterpret_cast<OpaqueValue *>(getError()));
-//      break;
-//  }
-//
-//  // FIXME: all tasks still in the readyQueue must be: swift_release(...)
-//  auto readyHead = readyQueue.load(std::memory_order_acquire);
-//  while (auto readyTask = readyHead.getTask()) {
-//    // Find the next waiting task.
-//    auto newReadyTask = readyTask->getNextChannelReadyTask();
-//    auto nextReadyQueueItem = GroupFragment::ReadyQueueItem::get(
-//        GroupFragment::ReadyStatus::Executing,
-//        newReadyTask
-//    );
-//
-//    // Attempt to claim it, we are the future that is going to complete it.
-//    // TODO: there may be other futures trying to do the same right now? FIXME: not really because the status right?
-//    if (fragment->waitQueue.compare_exchange_weak(readyHead, nextReadyQueueItem,
-//        /*success*/ std::memory_order_release,
-//        /*failure*/ std::memory_order_acquire)) {
-//      switch (readyHead.getStatus()) {
-//        case GroupFragment::ReadyStatus::Empty:
-//          return;
-//
-//        case GroupFragment::WaitStatus::Success:
-//          swift_unknownObjectRelease(reinterpret_cast<OpaqueValue *>(getError()));
-//          break;
-//
-//        case GroupFragment::WaitStatus::Error:
-//          swift_unknownObjectRelease(reinterpret_cast<OpaqueValue *>(getError()));
-//          break;
-//      }
-//    }
+//      assert(false && "destroying a task group that still has waiting tasks");
 //  }
+
+  mutex.lock(); // TODO: remove fragment lock, and use status for synchronization
+  // Release all ready tasks which are kept retained, the group destroyed,
+  // so no other task will ever await on them anymore;
+  ReadyQueueItem item;
+  bool taskDequeued = readyQueue.dequeue(item);
+  while (taskDequeued) {
+    swift_release(item.getTask());
+    bool taskDequeued = readyQueue.dequeue(item);
+  }
+  mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
 }
 
 // =============================================================================
@@ -209,7 +178,7 @@ void swift::swift_task_group_wait_next(
 GroupFragment::GroupPollResult AsyncTask::groupPoll(AsyncTask *waitingTask) {
   assert(isTaskGroup());
   auto fragment = groupFragment();
-  fragment->mutex.lock();
+  fragment->mutex.lock(); // TODO: remove fragment lock, and use status for synchronization
 
   // immediately update the status counter
   auto assumed = fragment->statusAddWaitingTaskAcquire();

stdlib/public/Concurrency/TaskGroup.swift

@@ -11,10 +11,7 @@
 ////===----------------------------------------------------------------------===//
 
 import Swift
-import Dispatch
 @_implementationOnly import _SwiftConcurrencyShims
-import Darwin
-import Foundation
 
 // ==== Task Group -------------------------------------------------------------
 
@@ -78,7 +75,6 @@ extension Task {
     groupFlags.isTaskGroup = true
     groupFlags.isFuture = true
 
-    // 1. Prepare the Group task
     let (groupTask, _) =
       Builtin.createAsyncTaskFuture(groupFlags.bits, parent) { () async throws -> BodyResult in
         let task = Builtin.getCurrentAsyncTask()
@@ -98,15 +94,11 @@ extension Task {
 
         return result
       }
-    let groupHandle = Handle<BodyResult>(task: groupTask)
 
-    // 2.0) Run the task!
-    DispatchQueue.global(priority: .default).async { // FIXME: use executors when they land
-      groupHandle.run()
-    }
+    // Enqueue the resulting job.
+    _enqueueJobGlobal(Builtin.convertTaskToJob(groupTask))
 
-    // 2.2) Await the group completing
-    return await try groupHandle.get()
+    return await try Handle<BodyResult>(task: groupTask).get()
   }
 
   /// A task group serves as storage for dynamically started tasks.
@@ -121,6 +113,7 @@ extension Task {
     /// No public initializers
     init(task: Builtin.NativeObject) {
       self.task = task
+      _swiftRetain(task)
     }
 
     // Swift will statically prevent this type from being copied or moved.
@@ -143,25 +136,15 @@ extension Task {
       overridingPriority priorityOverride: Priority? = nil,
       operation: @escaping () async throws -> TaskResult
     ) async -> Task.Handle<TaskResult> {
-      // 1) Increment the number of pending tasks immediately;
+      // Increment the number of pending tasks immediately;
       // We don't need to know which specific task is pending, just that pending
       // ones exist, so that next() can know if to wait or return nil.
-      let groupTask = Builtin.getCurrentAsyncTask()
-      _taskGroupAddPendingTask(groupTask)
+      _taskGroupAddPendingTask(self.task)
 
-      // 2) Create and run the child task
       let childTask =
-        await _runChildTask(overridingPriority: priorityOverride, operation: operation)
-
-      let handle = Handle<TaskResult>(task: childTask)
+        await _runGroupChildTask(overridingPriority: priorityOverride, operation: operation)
 
-      // FIXME: use executors or something else to launch the task
-      DispatchQueue.global(priority: .default).async {
-        // print(">>> run (task added at \(file):\(line))")
-        handle.run()
-      }
-
-      return handle
+      return Handle<TaskResult>(task: childTask)
     }
 
     /// Wait for the a child task that was added to the group to complete,
@@ -208,21 +191,18 @@ extension Task {
       if rawResult.hadErrorResult {
         // Throw the result on error.
         let error = unsafeBitCast(rawResult.storage, to: Error.self)
-        fputs("error: next[\(#file):\(#line)]: after await, error: \(rawResult) -> \(error)\n", stderr)
         throw error
       }
 
       guard let storage = rawResult.storage else {
         // The group was empty, return nil
-        fputs("error: next[\(#file):\(#line)]: after await, result: \(rawResult) -> nil\n", stderr)
         return nil
       }
 
       // Take the value on success.
       let storagePtr =
         storage.bindMemory(to: TaskResult.self, capacity: 1)
       let value = UnsafeMutablePointer<TaskResult>(mutating: storagePtr).pointee
-      fputs("error: next[\(pthread_self()) \(#file):\(#line)]: after await, result: \(rawResult) -> \(value)\n", stderr)
       return value
     }
 
@@ -273,7 +253,6 @@ extension Task.Group {
     //
     // Failures of tasks are ignored.
     while !self.isEmpty {
-      print("[\(#function)] self.isEmpty == \(self.isEmpty)")
       _ = await try? self.next()
       // TODO: Should a failure cause a cancellation of the task group?
       //       This looks very much like supervision trees,
@@ -285,6 +264,16 @@ extension Task.Group {
 
 /// ==== -----------------------------------------------------------------------
 
+@_silgen_name("swift_retain")
+func _swiftRetain(
+  _ task: Builtin.NativeObject
+)
+
+@_silgen_name("swift_task_group_add_pending")
+func _taskGroupAddPendingTask(
+  _ groupTask: Builtin.NativeObject
+)
+
 @_silgen_name("swift_task_group_offer")
 func taskGroupOffer(
   group: Builtin.NativeObject,
@@ -296,12 +285,6 @@ func _taskGroupWaitNext(
   on groupTask: Builtin.NativeObject
 ) async -> (hadErrorResult: Bool, storage: UnsafeRawPointer?)
 
-/// SeeAlso: GroupPollResult
-struct RawGroupPollResult {
-  let status: GroupPollStatus
-  let storage: UnsafeRawPointer
-}
-
 enum GroupPollStatus: Int {
   case empty   = 0
   case waiting = 1
@@ -313,8 +296,3 @@ enum GroupPollStatus: Int {
 func _taskGroupIsEmpty(
   _ groupTask: Builtin.NativeObject
 ) -> Bool
-
-@_silgen_name("swift_task_group_add_pending")
-func _taskGroupAddPendingTask(
-  _ groupTask: Builtin.NativeObject
-)

stdlib/public/Concurrency/TaskPrivate.h

@@ -21,8 +21,6 @@
 #include "swift/ABI/Task.h"
 #include "swift/ABI/Metadata.h"
 #include "swift/Runtime/HeapObject.h"
-#include <pthread.h>
-#include <stdio.h>
 
 namespace swift {
 
@@ -134,10 +132,11 @@ static void runTaskWithGroupPollResult(
   // TODO: schedule this task on the executor rather than running it directly.
   waitingTask->run(executor);
 
+  // TODO: Not entirely sure when to release; we synchronously run the code above so we can't before
   // if we need to, release the now completed task so it can be destroyed
-  if (result.retainedTask) {
-    swift_release(result.retainedTask);
-  }
+//  if (result.retainedTask) {
+//    swift_release(result.retainedTask);
+//  }
 }
 
 } // end namespace swift