[Concurrency] Task: isCancelled,checkCancelled implementation

move comments to the wired up continuations

remove duplicated continuations; leep the wired up ones

before moving to C++ for queue impl

trying to next wait via channel_poll

submitting works; need to impl next()

Author: ktoso

docs/SIL.rst

@@ -656,7 +656,7 @@ the coroutine until the continuation is invoked to resume it.  A use of
 
   func waitForCallback() async -> Int {
     return await withUnsafeContinuation { cc in
-      registerCallback { cc.resume($0) }
+      registerCallback { cc.resume(returning: $0) }
     }
   }
 

include/swift/ABI/MetadataValues.h

@@ -1925,8 +1925,15 @@ class JobFlags : public FlagSet<size_t> {
 
     // Kind-specific flags.
 
-    Task_IsChildTask  = 24,
-    Task_IsFuture     = 25
+    Task_IsChildTask      = 24,
+    Task_IsFuture         = 25,
+    // A TaskGroup is a Channel Task which allows children to offer values into
+    // it as they complete
+    Task_IsChannel        = 26,
+    // A child task started in a TaskGroup.
+    // Its parent MUST be the group's channel task,
+    // and the child task MUST offer its completion value into it when it completes.
+    Task_IsGroupChild     = 27,
   };
 
   explicit JobFlags(size_t bits) : FlagSet(bits) {}
@@ -1954,6 +1961,13 @@ class JobFlags : public FlagSet<size_t> {
                                 task_isFuture,
                                 task_setIsFuture)
 
+  FLAGSET_DEFINE_FLAG_ACCESSORS(Task_IsChannel,
+                                task_isChannel,
+                                task_setIsChannel)
+  FLAGSET_DEFINE_FLAG_ACCESSORS(Task_IsGroupChild,
+                                task_isGroupChild,
+                                task_setGroupChild)
+
 };
 
 /// Kinds of task status record.

include/swift/ABI/Task.h

@@ -161,8 +161,8 @@ class AsyncTask : public HeapObject, public Job {
     ResumeTask(this, currentExecutor, ResumeContext);
   }
 
-  /// Check whether this task has been cancelled.  Checking this is,
-  /// of course, inherently race-prone on its own.
+  /// Check whether this task has been cancelled.
+  /// Checking this is, of course, inherently race-prone on its own.
   bool isCancelled() const {
     return Status.load(std::memory_order_relaxed).isCancelled();
   }
@@ -190,12 +190,16 @@ class AsyncTask : public HeapObject, public Job {
     }
   };
 
+  // TODO: rename? all other functions are is... rather than has...Fragment
   bool hasChildFragment() const { return Flags.task_isChildTask(); }
+
   ChildFragment *childFragment() {
     assert(hasChildFragment());
     return reinterpret_cast<ChildFragment*>(this + 1);
   }
 
+  // ==== Future ---------------------------------------------------------------
+
   class FutureFragment {
   public:
     /// Describes the status of the future.
@@ -248,7 +252,7 @@ class AsyncTask : public HeapObject, public Job {
     const Metadata *resultType;
 
     // Trailing storage for the result itself. The storage will be uninitialized,
-    // contain an instance of \c resultType, or contaon an an \c Error.
+    // contain an instance of \c resultType, or contain an an \c Error.
 
     friend class AsyncTask;
 
@@ -315,6 +319,197 @@ class AsyncTask : public HeapObject, public Job {
   /// executor.
   void completeFuture(AsyncContext *context, ExecutorRef executor);
 
+  // ==== Channel --------------------------------------------------------------
+
+  class ChannelFragment {
+  public:
+      /// Describes the status of the channel.
+      enum class ReadyQueueStatus : uintptr_t {
+          /// The channel is empty, no tasks are pending.
+          /// Return immediately, there is no point in suspending.
+          ///
+          /// The storage is not accessible.
+          Empty = 0,
+
+          /// The channel has pending tasks
+          ///
+          /// The storage is not accessible.
+          Pending = 1,
+
+          /// The future has completed with result (of type \c resultType).
+          Success = 2,
+
+          /// The future has completed by throwing an error (an \c Error
+          /// existential).
+          Error = 3,
+
+//          /// No tasks are ready, yet there exist pending tasks (they were
+//          /// `add()`-ed and not completed yet). Suspend and await a wake-up
+//          /// when any task completes.
+//          Pending = 1,
+//
+//          /// At least one task is completed and waiting in the queue.
+//          /// Pull immediately without suspending to obtain it.
+//          Ready = 2,
+      };
+
+      /// Describes the status of the future.
+      ///
+      /// Futures always begin in the "Executing" state, and will always
+      /// make a single state change to either Success or Error.
+      enum class Status : uintptr_t {
+          /// No-one is waiting.
+          /// The storage is not accessible.
+          Pending = 0,
+
+          /// The future has completed with result (of type \c resultType).
+          Success = 1,
+
+          /// The future has completed by throwing an error (an \c Error
+          /// existential).
+          Error = 2,
+      };
+
+      /// An item within the message queue of a channel.
+      struct ReadyQueueItem {
+          /// Mask used for the low status bits in a message queue item.
+          static const uintptr_t statusMask = 0x03;
+
+          uintptr_t storage;
+
+          ReadyQueueStatus getStatus() const {
+            return static_cast<ReadyQueueStatus>(storage & statusMask);
+          }
+
+          AsyncTask *getTask() const {
+            return reinterpret_cast<AsyncTask *>(storage & ~statusMask);
+          }
+
+          static ReadyQueueItem get(ReadyQueueStatus status, AsyncTask *task) {
+            return ReadyQueueItem{
+                reinterpret_cast<uintptr_t>(task) | static_cast<uintptr_t>(status)};
+          }
+      };
+
+      /// An item within the wait queue, which includes the status and the
+      /// head of the list of tasks.
+      struct WaitQueueItem {
+          /// Mask used for the low status bits in a wait queue item.
+          static const uintptr_t statusMask = 0x03;
+
+          uintptr_t storage;
+
+          Status getStatus() const {
+            return static_cast<Status>(storage & statusMask);
+          }
+
+          AsyncTask *getTask() const {
+            return reinterpret_cast<AsyncTask *>(storage & ~statusMask);
+          }
+
+          static WaitQueueItem get(Status status, AsyncTask *task) {
+            return WaitQueueItem{
+                reinterpret_cast<uintptr_t>(task) | static_cast<uintptr_t>(status)};
+          }
+      };
+
+  private:
+      // TODO we likely can collapse these into one queue if we try hard enough
+
+      /// Queue containing completed tasks offered into this channel.
+      ///
+      /// The low bits contain the status, the rest of the pointer is the
+      /// AsyncTask.
+      std::atomic<ReadyQueueItem> readyQueue;
+
+      /// Queue containing all of the tasks that are waiting in `get()`.
+      ///
+      /// The low bits contain the status, the rest of the pointer is the
+      /// AsyncTask.
+      std::atomic<WaitQueueItem> waitQueue;
+
+      /// The type of the result that will be produced by the channel.
+      const Metadata *resultType;
+
+      // FIXME: seems shady...?
+      // Trailing storage for the result itself. The storage will be uninitialized.
+      // Use the `readyQueue` to poll for values from the channel instead.
+      friend class AsyncTask; // TODO: remove this?
+
+  public:
+      explicit ChannelFragment(const Metadata *resultType)
+          : readyQueue(ReadyQueueItem::get(ReadyQueueStatus::Empty, nullptr)),
+            waitQueue(WaitQueueItem::get(Status::Pending, nullptr)),
+            resultType(resultType) { }
+
+      /// Destroy the storage associated with the channel.
+      void destroy();
+
+//      /// Offer a completed task to the channel to enable polling for it.
+//      void offer(AsyncTask *completed, AsyncContext *context, ExecutorRef executor);
+//
+//      /// Poll the channel for any
+//      ChannelFragment::ReadyQueueStatus channelPoll(AsyncTask *waitingTask);
+
+//      /// Retrieve a pointer to the storage of result.
+//      OpaqueValue *getStoragePtr() {
+//        return reinterpret_cast<OpaqueValue *>(
+//            reinterpret_cast<char *>(this) + storageOffset(resultType));
+//      }
+//
+//      /// Retrieve the error.
+//      SwiftError *&getError() {
+//        return *reinterpret_cast<SwiftError **>(
+//            reinterpret_cast<char *>(this) + storageOffset(resultType));
+//      }
+
+//      /// Compute the offset of the storage from the base of the channel
+//      /// fragment.
+//      static size_t storageOffset(const Metadata *resultType)  {
+//        size_t offset = sizeof(ChannelFragment);
+//        size_t alignment =
+//            std::max(resultType->vw_alignment(), alignof(SwiftError *));
+//        return (offset + alignment - 1) & ~(alignment - 1);
+//      }
+//
+//      /// Determine the size of the channel fragment given a particular channel
+//      /// result type.
+//      static size_t fragmentSize(const Metadata *resultType) {
+//        return storageOffset(resultType) +
+//               std::max(resultType->vw_size(), sizeof(SwiftError *));
+//      }
+  };
+
+  bool isChannel() const { return Flags.task_isChannel(); }
+
+  ChannelFragment *channelFragment() {
+    assert(isChannel());
+    // FIXME: isn't it also a future at the same time?
+    if (hasChildFragment()) { // TODO: make sure those all are correct; we can have many fragments
+      return reinterpret_cast<ChannelFragment *>(
+          reinterpret_cast<ChildFragment*>(this + 1) + 1);
+    }
+
+    return reinterpret_cast<ChannelFragment *>(this + 1);
+  }
+
+  /// Offer result of a task into this channel.
+  /// The value is enqueued at the end of the channel.
+  ///
+  /// Upon enqueue, any waiting tasks will be scheduled on the given executor. // TODO: not precisely right
+  void channelOffer(AsyncTask *completed, AsyncContext *context, ExecutorRef executor);
+
+  /// Wait for for channel to become non-empty.
+  ///
+  /// \returns the status of the queue.  TODO more docs
+  ChannelFragment::ReadyQueueStatus channelPoll(AsyncTask *waitingTask);
+
+  // ==== ----------------------------------------------------------------------
+
+  bool isGroupChild() const { return Flags.task_isGroupChild(); }
+
+  // ==== ----------------------------------------------------------------------
+
   static bool classof(const Job *job) {
     return job->isAsyncTask();
   }
@@ -431,6 +626,10 @@ class FutureAsyncContext : public AsyncContext {
   SwiftError *errorResult = nullptr;
   OpaqueValue *indirectResult;
 
+
+  // TODO: this is to support "offer into queue on complete"
+  AsyncContext *parentChannel = nullptr; // TODO: no idea if we need this or not
+
   using AsyncContext::AsyncContext;
 };
 

include/swift/Runtime/Concurrency.h

@@ -133,6 +133,26 @@ 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>;
 
@@ -148,6 +168,18 @@ SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swiftasync)
 TaskFutureWaitSignature::FunctionType
 swift_task_future_wait;
 
+/// 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_channel_poll(on channelTask: Builtin.NativeObject) async
+///     -> RawTaskChannelPollResult?
+/// \endcode
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+AsyncFunctionType<TaskChannelPollResult(AsyncTask *task)>
+swift_task_channel_poll;
+
 /// Add a status record to a task.  The record should not be
 /// modified while it is registered with a task.
 ///
@@ -188,6 +220,9 @@ bool swift_task_removeStatusRecord(AsyncTask *task,
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 size_t swift_task_getJobFlags(AsyncTask* task);
 
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+bool swift_task_isCancelled(AsyncTask* task);
+
 /// This should have the same representation as an enum like this:
 ///    enum NearestTaskDeadline {
 ///      case none

stdlib/public/Concurrency/CMakeLists.txt

@@ -42,9 +42,11 @@ add_swift_target_library(swift_Concurrency ${SWIFT_STDLIB_LIBRARY_BUILD_TYPES} I
   Task.swift
   TaskCancellation.swift
   _TimeTypes.swift
+  __LocksTemporaryWorkaround.swift
   TaskAlloc.cpp
   TaskStatus.cpp
   TaskGroup.swift
+  TaskChannel.cpp
   Mutex.cpp
   ${swift_concurrency_objc_sources}