[Concurrency] Implement TaskGroup.isEmpty via readyQueue

before reversing order of fragments; future must be last since dynamic
size

offer fixed

before implementing poll

Author: ktoso

include/swift/ABI/Task.h

@@ -133,29 +133,6 @@ struct TaskFutureWaitResult {
   OpaqueValue *storage;
 };
 
-/// The result of waiting on a Channel (TaskGroup).
-struct TaskChannelPollResult {
-  /// Whether the storage represents an existing value, or "known no value,"
-  /// instructing the `group.next()` call to return `nil` immediately.
-  bool hadAnyResult;
-
-  /// Whether the storage represents the error result vs. the successful
-  /// result.
-  bool hadErrorResult;
-
-  /// Storage for the result of the future.
-  ///
-  /// When the future completed normally, this is a pointer to the storage
-  /// of the result value, which lives inside the future task itself.
-  ///
-  /// When the future completed by throwing an error, this is the error
-  /// object itself.
-  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
@@ -174,12 +151,23 @@ swift_task_future_wait;
 ///
 /// \code
 /// func swift_task_channel_poll(on channelTask: Builtin.NativeObject) async
-///     -> RawTaskChannelPollResult?
+///     -> RawChannelPollResult?
 /// \endcode
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
-AsyncFunctionType<TaskChannelPollResult(AsyncTask *task)>
+AsyncFunctionType<AsyncTask::ChannelFragment::ChannelPollResult(AsyncTask *task)>
 swift_task_channel_poll;
 
+/// Check the readyQueue of a Channel, return true if it has no pending tasks.
+///
+/// This can be called from any thread. Its Swift signature is
+///
+/// \code
+/// func swift_task_channel_is_empty(on channelTask: Builtin.NativeObject) -> Bool
+/// \endcode
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+bool
+swift_task_channel_is_empty(AsyncTask *task);
+
 /// Add a status record to a task.  The record should not be
 /// modified while it is registered with a task.
 ///

include/swift/Runtime/Concurrency.h

@@ -2278,8 +2278,13 @@ void IRGenSILFunction::visitFunctionRefBaseInst(FunctionRefBaseInst *i) {
       fn, NotForDefinition, false /*isDynamicallyReplaceableImplementation*/,
       isa<PreviousDynamicFunctionRefInst>(i));
   llvm::Value *value;
+//  auto isSpecialAsyncWithoutCtxtSize =
+//      fn->isAsync() && fn->getName().equals("swift_task_future_wait");
   auto isSpecialAsyncWithoutCtxtSize =
-      fn->isAsync() && fn->getName().equals("swift_task_future_wait");
+      fn->isAsync() && (
+          fn->getName().equals("swift_task_future_wait") ||
+          fn->getName().equals("swift_task_channel_poll") // TODO: do we need this?
+      );
   if (fn->isAsync() && !isSpecialAsyncWithoutCtxtSize) {
     value = IGM.getAddrOfAsyncFunctionPointer(fn);
     value = Builder.CreateBitCast(value, fnPtr->getType());

lib/IRGen/IRGenSIL.cpp

@@ -24,6 +24,7 @@
 
 using namespace swift;
 using FutureFragment = AsyncTask::FutureFragment;
+using ChannelFragment = AsyncTask::ChannelFragment;
 
 void FutureFragment::destroy() {
   auto queueHead = waitQueue.load(std::memory_order_acquire);
@@ -41,10 +42,15 @@ void FutureFragment::destroy() {
   }
 }
 
-FutureFragment::Status AsyncTask::waitFuture(AsyncTask *waitingTask) {
+FutureFragment::Status
+AsyncTask::waitFuture(AsyncTask *waitingTask) {
   using Status = FutureFragment::Status;
   using WaitQueueItem = FutureFragment::WaitQueueItem;
 
+//  if (isChannel()) {
+//    assert(false && "waiting on channel!");
+//  }
+
   assert(isFuture());
   auto fragment = futureFragment();
 
@@ -263,15 +269,20 @@ AsyncTaskAndContext swift::swift_task_create_future_f(
          initialContextSize >= sizeof(FutureAsyncContext));
   assert((parent != nullptr) == flags.task_isChildTask());
 
-  // TODO: cleanup
-  if (flags.task_isGroupChild()) {
-    assert(flags.task_isGroupChild() && flags.task_isChildTask());
-    assert(flags.task_isGroupChild() && parent->Flags.task_isChannel());
+  if (flags.task_isGroupChild()) { // TODO: express without the `if`?
+    assert(flags.task_isChildTask());
+    assert(parent->Flags.task_isChannel());
   }
 
   // Figure out the size of the header.
   size_t headerSize = sizeof(AsyncTask);
-  if (parent) headerSize += sizeof(AsyncTask::ChildFragment);
+  if (parent) {
+    headerSize += sizeof(AsyncTask::ChildFragment);
+  }
+
+  if (flags.task_isChannel()) {
+    headerSize += ChannelFragment::fragmentSize();
+  }
 
   if (futureResultType) {
     headerSize += FutureFragment::fragmentSize(futureResultType);
@@ -304,6 +315,12 @@ AsyncTaskAndContext swift::swift_task_create_future_f(
     new (childFragment) AsyncTask::ChildFragment(parent);
   }
 
+  // Initialize the channel fragment if applicable.
+  if (flags.task_isChannel()) {
+    auto channelFragment = task->channelFragment();
+    new (channelFragment) ChannelFragment(futureResultType);
+  }
+
   // Initialize the future fragment if applicable.
   if (futureResultType) {
     auto futureFragment = task->futureFragment();
@@ -314,18 +331,6 @@ AsyncTaskAndContext swift::swift_task_create_future_f(
     auto futureContext = static_cast<FutureAsyncContext *>(initialContext);
     futureContext->errorResult = nullptr;
     futureContext->indirectResult = futureFragment->getStoragePtr();
-
-    if (flags.task_isChannel()) {
-      auto parentChannelFragment = parent->channelFragment();
-      // new(channelFragment) ChannelFragment(futureResultType); // TODO make sure parent does init
-      assert(parentChannelFragment && "group child task could not find parent channel fragment!");
-
-//
-//      // Set up the context for the future such that results ar *offered*
-//      // into the channel fragment's message queue.
-//      futureContext->indirectResult = parentChannelFragment;
-    }
-
   }
 
   // Configure the initial context.
@@ -355,9 +360,16 @@ void swift::swift_task_future_wait(
   waitingTask->ResumeTask = rawContext->ResumeParent;
   waitingTask->ResumeContext = rawContext;
 
-  // Wait on the future.
   auto context = static_cast<TaskFutureWaitAsyncContext *>(rawContext);
   auto task = context->task;
+
+  if (task->isChannel()) {
+    // assert(false && "WAITING ON GROUP CHILD TASK");
+    swift::swift_task_channel_poll(waitingTask, executor, rawContext);
+    return;
+  }
+
+  // Wait on the future.
   assert(task->isFuture());
   switch (task->waitFuture(waitingTask)) {
   case FutureFragment::Status::Executing: