[Concurrency] Initial implementation of TaskGroups #34999

ktoso · 2020-12-08T16:57:33Z

I need to do some more cleanups and remove all the println debugging... but still fixing some final things.

I fully expect this to still have some bugs.

rdar://70141994

ktoso · 2020-12-08T16:58:24Z

@swift-ci please smoke test

ktoso · 2020-12-10T00:33:36Z

@swift-ci Please smoke test OS X platform

ktoso · 2020-12-10T04:21:50Z

@swift-ci Please smoke test OS X platform

ktoso · 2020-12-10T04:37:15Z

Ok rebased again and given up un lockless version for now; at least this way we can have things working first.

Will remove debugging and some of the noise and ask for review then...

@swift-ci Please smoke test OS X platform

ktoso · 2020-12-10T11:29:10Z

Actually... not fully sure about the command, is it macOS or OS X? the readme has both hm...

@swift-ci please smoke test

ktoso · 2020-12-10T01:26:30Z

stdlib/public/Concurrency/TaskGroup.swift

+
+        /// Drain all remaining tasks by awaiting on them;
+        /// Their failures are ignored;
+        await group._tearDown()


So the semantics are:

run body; run all tasks in it

if body throws, cancel all, just return, without awaiting on the remaining ones

if body returns and there are remaining pending tasks, await on them

if one of those "ignored in body" throws, this is ignored and we keep waiting <- not sure that's right, perhaps we should bail out whenever at least one of them throws? It wasn't clear what to do in this case in our writeups. Other impls do bail out early I believe then

The logic is based on my reading of Kotlin's coroutineScopes, Trio's nurseries and also similar to what Loom does: https://github.com/openjdk/loom/blob/fibers/src/java.base/share/classes/java/util/concurrent/ExecutorService.java#L329

stdlib/public/Concurrency/TaskGroup.swift

stdlib/public/Concurrency/TaskGroup.cpp

ktoso · 2020-12-10T15:35:35Z

stdlib/public/Concurrency/TaskGroup.cpp

+
+  assert(isTaskGroup());
+  auto fragment = groupFragment();
+  fragment->mutex.lock(); // TODO: remove fragment lock, and use status for synchronization


I've been spending a lot of time on the lockless dance and perhaps we should first get things going as I try to get the lockless exchanges correct.

One of the things I was unsure about is the queue style to use here -- I'm not completely sure if we perhaps could get away with an MPSC queue for the readyQueue? Initially I assumed yes, but then I thought no, and now I'm actually not sure... Can the group assume it will be run in an actor exclusive executor? I was kind of assuming "no", but even if, would it give enough guarantees for the consuming side of the ready queue... Have not figured that all out, and since we don't have the actor runtime yet I was not sure really.

(So all of the dance with atomics inside here is being developed with the assumption we'll get rid of this lock, but I wanted to offer the PR as some incremental progress for the time being rather than block progress on this)

ktoso · 2020-12-10T15:59:25Z

include/swift/ABI/Task.h

+    mutable std::mutex mutex;
+
+    /// Used for queue management, counting number of waiting and ready tasks
+    std::atomic<unsigned long> status;


The status is of utmost importance; it allows us to implement: isEmpty and properly implement scheduling including as well as decide if next() should return nil or suspend.

All offer and poll initially are to acquire the status and release it when they're done processing -- poll processes it by "if there is e.g. 2 ready tasks and 5 pending tasks" we know we can -1 both -- we then perform a CAS, to know if that polling task "won" and shall pull. This way even if we get multiple polls, (because they may be done in async lets -- ???), polls can be correct and we won't pull/decrement more than we indeed have ready.

I am not clear about the concurrent poller case -- I was assuming that because async lets, there may be concurrent polls issued... but perhaps I'm assuming too much? Are async lets all confined to the same exclusive actor executor and thus will not be able to cause concurrent polls? Clarification here would be helpful - thanks! If we can assume no concurrent polls, we will be able to get away with a simple queue for readyQueue as well, which would be excellent.

@DougGregor so all of this dance around the status and the two queues is remnants of attempts of the lockless impl;

I'm planning to get back to it and making it happen now that John's runtime landed -- I did not know if you'd rather we remove all of it and add proper thing later; or simply keep status quo + the lock because it works just fine enough; Honestly I'd prefer keeping as is as then I can spare a cycle of restoring similar infra...

Let me know if you'd rather only merge a perfectly clean locking impl for now. I did remove all the other not currently used MPSC queues and stuff so at least we do not pollute with not used datastructures.

include/swift/ABI/Task.h

ktoso · 2020-12-10T16:00:37Z

include/swift/ABI/Task.h

+    void destroy();
+
+    bool isEmpty() {
+      auto oldStatus = GroupStatus { status.load(std::memory_order_relaxed) };


This may be my lack of C++ experience showing...

Or could I model this as the atomic being of type GroupStatus?

ktoso · 2020-12-10T16:01:31Z

include/swift/ABI/Task.h

+      offset += sizeof(GroupFragment);
+    }
+
+    return reinterpret_cast<FutureFragment *>(offset);


@DougGregor I think I got this right; last time we talked this was an intense if/else dance if you remember.

ktoso · 2020-12-10T16:02:06Z

include/swift/ABI/Task.h

@@ -293,12 +663,19 @@ class AsyncTask : public HeapObject, public Job {

  FutureFragment *futureFragment() {
    assert(isFuture());
+
+    auto offset = reinterpret_cast<uintptr_t>(this); // TODO: char* instead?


I'm not sure what style to follow -- other files do char* but this file seems to do uintptr_t for similar things.

WDYT / does it matter?

ktoso · 2020-12-11T02:10:36Z

Okey I realized we can get back to lockless and vastly simplify the internal queues by just using one linked one like Futures do and CAS over it, since we don't need to guarantee ordering as strongly as I had thought initially I suppose.

ktoso · 2020-12-14T10:35:35Z

@swift-ci please smoke test

ktoso · 2020-12-14T12:36:12Z

stdlib/public/Concurrency/TaskGroup.cpp

+    bool taskDequeued = readyQueue.dequeue(item);
+  }
+  mutex.unlock(); // TODO: remove fragment lock, and use status for synchronization
+}


Unsure if anything else to cleanup here; the wait queue never has values in it, just waiting tasks; so nothing to cleanup there I think

DougGregor · 2020-12-14T15:47:51Z

@swift-ci please smoke test

ktoso · 2020-12-14T15:58:20Z

stdlib/public/Concurrency/Task.cpp

-  // directly.
-  waitingTask->run(executor);
-}
-


(this only moved)

ktoso · 2020-12-14T15:59:33Z

stdlib/public/Concurrency/TaskGroup.cpp

+//  switch (waitHead.getStatus()) {
+//    case GroupFragment::WaitStatus::Waiting:
+//      assert(false && "destroying a task group that still has waiting tasks");
+//  }


Not really I guess... since they never store any values, unlike in the Future case 🤔

ktoso · 2020-12-14T22:22:53Z

Fixed the SIL issue, introduced new _runGroupChildTask rather than change _runChildTask

@swift-ci please smoke test

ktoso · 2020-12-16T09:21:40Z

include/swift/ABI/Task.h

@@ -190,12 +205,350 @@ class AsyncTask : public HeapObject, public Job {
    }
  };

+  // TODO: rename? all other functions are `is...` rather than `has...Fragment`


actually, TaskLocals also read as hasLocalValues so I'll remove this TODO

cleanups

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()

before reversing order of fragments; future must be last since dynamic size offer fixed before implementing poll

ktoso · 2020-12-16T21:59:45Z

@swift-ci please smoke test and merge

ktoso mentioned this pull request Dec 8, 2020

[Concurrency] Task Groups implementation #34889

Closed

10 tasks

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from ce5a1c2 to c11e1b1 Compare December 10, 2020 04:20

ktoso changed the title ~~[Concurrency]Initial implementation of TaskGroups~~ [Concurrency] Initial implementation of TaskGroups Dec 10, 2020

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from c11e1b1 to b115f9c Compare December 10, 2020 04:35

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from 54e980b to 824bbe0 Compare December 10, 2020 11:16

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch 3 times, most recently from 082fe29 to ff52070 Compare December 10, 2020 16:02

ktoso commented Dec 10, 2020

View reviewed changes

ktoso requested review from DougGregor and rjmccall December 10, 2020 16:05

ktoso marked this pull request as ready for review December 11, 2020 00:19

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch 2 times, most recently from bf368d2 to 557969a Compare December 14, 2020 10:26

ktoso commented Dec 14, 2020

View reviewed changes

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from 557969a to 25ff6cc Compare December 14, 2020 13:11

ktoso commented Dec 14, 2020

View reviewed changes

stdlib/public/Concurrency/Task.cpp

// directly.

waitingTask->run(executor);

}

Copy link

Contributor Author

ktoso Dec 14, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(this only moved)

ktoso commented Dec 14, 2020

View reviewed changes

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from 25ff6cc to a0b9d73 Compare December 14, 2020 22:22

ktoso commented Dec 16, 2020

View reviewed changes

[Concurrency][TaskGroup] example "map async" using task groups

a346953

DougGregor and others added 7 commits December 17, 2020 06:05

+concurrency task groups initial work in progress

35acd40

+task Implement Task.currentPriority

9162d40

cleanups

[Concurrency] Task: isCancelled,checkCancelled implementation

520b513

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()

[Concurrency] Implement TaskGroup.isEmpty via readyQueue

e294c7c

before reversing order of fragments; future must be last since dynamic size offer fixed before implementing poll

[Concurrency] Initial TaskGroup implementation working

7b37554

[Concurrency] guard offer/poll with a lock for now; cleanups

9e1ecc5

Rebased to use new global executor

b267778

ktoso force-pushed the wip-task-groups-impl-channel-rebased branch from a0b9d73 to b267778 Compare December 16, 2020 21:36

swift-ci merged commit f23f5b8 into swiftlang:main Dec 17, 2020

ktoso deleted the wip-task-groups-impl-channel-rebased branch February 10, 2021 07:41

[Concurrency] Initial implementation of TaskGroups #34999

[Concurrency] Initial implementation of TaskGroups #34999

Uh oh!

Conversation

ktoso commented Dec 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ktoso commented Dec 8, 2020

Uh oh!

ktoso commented Dec 10, 2020

Uh oh!

ktoso commented Dec 10, 2020

Uh oh!

ktoso commented Dec 10, 2020

Uh oh!

ktoso commented Dec 10, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ktoso commented Dec 11, 2020

Uh oh!

ktoso commented Dec 14, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DougGregor commented Dec 14, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ktoso commented Dec 14, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ktoso commented Dec 16, 2020

Uh oh!

Uh oh!

ktoso commented Dec 8, 2020 •

edited

Loading