Add task execution package

Add task execution package that includes ConcurrentSequenceExecutor that supports executing sequences of Task in a highly concurrent fashion.

Author: neakor

Concurrency.xcodeproj/project.pbxproj

@@ -21,6 +21,11 @@
 /* End PBXAggregateTarget section */
 
 /* Begin PBXBuildFile section */
+		41B94843210A4744007E59C8 /* SerialSequenceExecutor.swift in Sources */ = {isa = PBXBuildFile; fileRef = 41B9483F210A4744007E59C8 /* SerialSequenceExecutor.swift */; };
+		41B94844210A4744007E59C8 /* SequenceExecutor.swift in Sources */ = {isa = PBXBuildFile; fileRef = 41B94840210A4744007E59C8 /* SequenceExecutor.swift */; };
+		41B94845210A4744007E59C8 /* ConcurrentSequenceExecutor.swift in Sources */ = {isa = PBXBuildFile; fileRef = 41B94841210A4744007E59C8 /* ConcurrentSequenceExecutor.swift */; };
+		41B94846210A4744007E59C8 /* Task.swift in Sources */ = {isa = PBXBuildFile; fileRef = 41B94842210A4744007E59C8 /* Task.swift */; };
+		41B94849210A4756007E59C8 /* ConcurrentSequenceExecutorTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 41B94848210A4756007E59C8 /* ConcurrentSequenceExecutorTests.swift */; };
 		OBJ_27 /* AtomicBool.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_9 /* AtomicBool.swift */; };
 		OBJ_28 /* AtomicInt.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_10 /* AtomicInt.swift */; };
 		OBJ_29 /* AtomicReference.swift in Sources */ = {isa = PBXBuildFile; fileRef = OBJ_11 /* AtomicReference.swift */; };
@@ -51,6 +56,11 @@
 /* End PBXContainerItemProxy section */
 
 /* Begin PBXFileReference section */
+		41B9483F210A4744007E59C8 /* SerialSequenceExecutor.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = SerialSequenceExecutor.swift; sourceTree = "<group>"; };
+		41B94840210A4744007E59C8 /* SequenceExecutor.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = SequenceExecutor.swift; sourceTree = "<group>"; };
+		41B94841210A4744007E59C8 /* ConcurrentSequenceExecutor.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConcurrentSequenceExecutor.swift; sourceTree = "<group>"; };
+		41B94842210A4744007E59C8 /* Task.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = Task.swift; sourceTree = "<group>"; };
+		41B94848210A4756007E59C8 /* ConcurrentSequenceExecutorTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ConcurrentSequenceExecutorTests.swift; sourceTree = "<group>"; };
 		"Concurrency::Concurrency::Product" /* Concurrency.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; path = Concurrency.framework; sourceTree = BUILT_PRODUCTS_DIR; };
 		"Concurrency::ConcurrencyTests::Product" /* ConcurrencyTests.xctest */ = {isa = PBXFileReference; lastKnownFileType = file; path = ConcurrencyTests.xctest; sourceTree = BUILT_PRODUCTS_DIR; };
 		OBJ_10 /* AtomicInt.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AtomicInt.swift; sourceTree = "<group>"; };
@@ -83,6 +93,25 @@
 /* End PBXFrameworksBuildPhase section */
 
 /* Begin PBXGroup section */
+		41B9483E210A4744007E59C8 /* Executor */ = {
+			isa = PBXGroup;
+			children = (
+				41B9483F210A4744007E59C8 /* SerialSequenceExecutor.swift */,
+				41B94840210A4744007E59C8 /* SequenceExecutor.swift */,
+				41B94841210A4744007E59C8 /* ConcurrentSequenceExecutor.swift */,
+				41B94842210A4744007E59C8 /* Task.swift */,
+			);
+			path = Executor;
+			sourceTree = "<group>";
+		};
+		41B94847210A4756007E59C8 /* Executor */ = {
+			isa = PBXGroup;
+			children = (
+				41B94848210A4756007E59C8 /* ConcurrentSequenceExecutorTests.swift */,
+			);
+			path = Executor;
+			sourceTree = "<group>";
+		};
 		OBJ_13 /* Tests */ = {
 			isa = PBXGroup;
 			children = (
@@ -94,6 +123,7 @@
 		OBJ_14 /* ConcurrencyTests */ = {
 			isa = PBXGroup;
 			children = (
+				41B94847210A4756007E59C8 /* Executor */,
 				OBJ_15 /* AtomicBoolTests.swift */,
 				OBJ_16 /* AtomicIntTests.swift */,
 				OBJ_17 /* AtomicReferenceTests.swift */,
@@ -133,6 +163,7 @@
 		OBJ_8 /* Concurrency */ = {
 			isa = PBXGroup;
 			children = (
+				41B9483E210A4744007E59C8 /* Executor */,
 				OBJ_9 /* AtomicBool.swift */,
 				OBJ_10 /* AtomicInt.swift */,
 				OBJ_11 /* AtomicReference.swift */,
@@ -225,9 +256,13 @@
 			isa = PBXSourcesBuildPhase;
 			buildActionMask = 0;
 			files = (
+				41B94846210A4744007E59C8 /* Task.swift in Sources */,
 				OBJ_27 /* AtomicBool.swift in Sources */,
+				41B94844210A4744007E59C8 /* SequenceExecutor.swift in Sources */,
+				41B94845210A4744007E59C8 /* ConcurrentSequenceExecutor.swift in Sources */,
 				OBJ_28 /* AtomicInt.swift in Sources */,
 				OBJ_29 /* AtomicReference.swift in Sources */,
+				41B94843210A4744007E59C8 /* SerialSequenceExecutor.swift in Sources */,
 				OBJ_30 /* CountDownLatch.swift in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
@@ -246,6 +281,7 @@
 			files = (
 				OBJ_48 /* AtomicBoolTests.swift in Sources */,
 				OBJ_49 /* AtomicIntTests.swift in Sources */,
+				41B94849210A4756007E59C8 /* ConcurrentSequenceExecutorTests.swift in Sources */,
 				OBJ_50 /* AtomicReferenceTests.swift in Sources */,
 				OBJ_51 /* CountDownLatchTests.swift in Sources */,
 			);

README.md

@@ -31,6 +31,12 @@ Provides locking-free synchronization of a mutable object reference. It provides
 ### `CountDownLatch`
 A utility class that allows coordination between threads. A count down latch starts with an initial count. Threads can then decrement the count until it reaches zero, at which point, the suspended waiting thread shall proceed. A `CountDownLatch` behaves differently from a `DispatchSemaphore` once the latch is open. Unlike a semaphore where subsequent waits would still block the caller thread, once a `CountDownLatch` is open, all subsequent waits can directly passthrough.
 
+### `ConcurrentSequenceExecutor`
+An execution utility that executes sequences of tasks and returns the final result in a highly concurrent environment.
+
+### `SerialSequenceExecutor`
+A debugging executor that executes sequences of tasks and returns the final result serially on the caller thread.
+
 ## Installation
 
 ### Carthage
@@ -88,4 +94,4 @@ Or you can follow the steps above to generate a Xcode project and run tests with
 
 
 ## License
-[![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fuber%2Fswift-concurrency.svg?type=large)](https://app.fossa.io/projects/git%2Bgithub.com%2Fuber%2Fswift-concurrency?ref=badge_large)
+[![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fuber%2Fswift-concurrency.svg?type=large)](https://app.fossa.io/projects/git%2Bgithub.com%2Fuber%2Fswift-concurrency?ref=badge_large)

Sources/Concurrency/Executor/ConcurrentSequenceExecutor.swift

@@ -0,0 +1,114 @@
+//
+//  Copyright (c) 2018. Uber Technologies
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+import Foundation
+
+/// An executor that executes sequences of tasks concurrently.
+///
+/// - seeAlso: `SequenceExecutor`.
+/// - seeAlso: `Task`.
+public class ConcurrentSequenceExecutor: SequenceExecutor {
+
+    /// Initializer.
+    ///
+    /// - parameter name: The name of the executor.
+    /// - parameter qos: The quality of service of this executor. This
+    /// defaults to `userInitiated`.
+    public init(name: String, qos: DispatchQoS = .userInitiated) {
+        taskQueue = DispatchQueue(label: "Executor.taskQueue-\(name)", qos: qos, attributes: .concurrent)
+    }
+
+    /// Execute a sequence of tasks concurrently from the given initial task.
+    ///
+    /// - parameter initialTask: The root task of the sequence of tasks
+    /// to be executed.
+    /// - parameter execution: The execution defining the sequence of tasks.
+    /// When a task completes its execution, this closure is invoked with
+    /// the task and its produced result. This closure is invoked from
+    /// multiple threads concurrently, therefore it must be thread-safe.
+    /// The tasks provided by this closure are executed concurrently.
+    /// - returns: The execution handle that allows control and monitoring
+    /// of the sequence of tasks being executed.
+    public func executeSequence<SequenceResultType>(from initialTask: Task, with execution: @escaping (Task, Any) -> SequenceExecution<SequenceResultType>) -> SequenceExecutionHandle<SequenceResultType> {
+        let handle: SynchronizedSequenceExecutionHandle<SequenceResultType> = SynchronizedSequenceExecutionHandle()
+        execute(initialTask, with: handle, execution)
+        return handle
+    }
+
+    // MARK: - Private
+
+    private let taskQueue: DispatchQueue
+
+    private func execute<SequenceResultType>(_ task: Task, with sequenceHandle: SynchronizedSequenceExecutionHandle<SequenceResultType>, _ execution: @escaping (Task, Any) -> SequenceExecution<SequenceResultType>) {
+        taskQueue.async {
+            guard !sequenceHandle.isCancelled else {
+                return
+            }
+
+            let result = task.typeErasedExecute()
+            let nextExecution = execution(task, result)
+            switch nextExecution {
+            case .continueSequence(let nextTask):
+                self.execute(nextTask, with: sequenceHandle, execution)
+            case .endOfSequence(let result):
+                sequenceHandle.sequenceDidComplete(with: result)
+            }
+        }
+    }
+}
+
+private class SynchronizedSequenceExecutionHandle<SequenceResultType>: SequenceExecutionHandle<SequenceResultType> {
+
+    private let latch = CountDownLatch(count: 1)
+    private let didCancel = AtomicBool(initialValue: false)
+
+    // Use a lock to ensure result is properly accessed, since the read
+    // `await` method may be invoked on a different thread than the write
+    // `sequenceDidComplete` method.
+    private let resultLock = NSRecursiveLock()
+    private var result: SequenceResultType?
+
+    fileprivate var isCancelled: Bool {
+        return didCancel.value
+    }
+
+    fileprivate override func await(withTimeout timeout: TimeInterval?) throws -> SequenceResultType {
+        let didComplete = latch.await(timeout: timeout)
+        if !didComplete {
+            throw SequenceExecutionError.awaitTimeout
+        }
+
+        resultLock.lock()
+        defer {
+            resultLock.unlock()
+        }
+        // If latch was counted down, the result must have been set. Therefore,
+        // this forced unwrap is safe.
+        return result!
+    }
+
+    fileprivate func sequenceDidComplete(with result: SequenceResultType) {
+        resultLock.lock()
+        self.result = result
+        resultLock.unlock()
+
+        latch.countDown()
+    }
+
+    fileprivate override func cancel() {
+        didCancel.compareAndSet(expect: false, newValue: true)
+    }
+}

Sources/Concurrency/Executor/SequenceExecutor.swift

@@ -0,0 +1,71 @@
+//
+//  Copyright (c) 2018. Uber Technologies
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+import Foundation
+
+/// Errors that can occur during a sequence execution.
+public enum SequenceExecutionError: Error {
+    /// The waiting on sequence completion timed out.
+    case awaitTimeout
+}
+
+/// The handle of the execution of a sequence of tasks, that allows control
+/// and monitoring of the said sequence of tasks.
+// This cannot be a protocol, since `SequenceExecutor` references this as a
+// type. Protocols with associatedType cannot be directly used as types.
+public class SequenceExecutionHandle<SequenceResultType> {
+
+    /// Block the caller thread until the sequence of tasks all finished
+    /// execution or the specified timeout period has elapsed.
+    ///
+    /// - parameter timeout: The duration to wait before the timeout error
+    /// is thrown. `nil` to wait indefinitely until the sequence execution
+    /// completes.
+    /// - throws: `SequenceExecutionError.awaitTimeout` if the given timeout
+    /// period elapsed before the sequence execution completed.
+    func await(withTimeout timeout: TimeInterval?) throws -> SequenceResultType {
+        fatalError("await not yet implemented.")
+    }
+
+    /// Cancel the sequence execution at the point this function is invoked.
+    func cancel() {}
+}
+
+/// The execution of a sequence.
+public enum SequenceExecution<ResultType> {
+    /// The execution of the sequence should continue with the associated
+    /// value task.
+    case continueSequence(Task)
+    /// The end of the entire task sequence with associated value result.
+    case endOfSequence(ResultType)
+}
+
+/// Executor of sequences of tasks.
+///
+/// - seeAlso: `Task`.
+public protocol SequenceExecutor {
+
+    /// Execute a sequence of tasks from the given initial task.
+    ///
+    /// - parameter initialTask: The root task of the sequence of tasks
+    /// to be executed.
+    /// - parameter execution: The execution defining the sequence of tasks.
+    /// When a task completes its execution, this closure is invoked with
+    /// the task and its produced result.
+    /// - returns: The execution handle that allows control and monitoring
+    /// of the sequence of tasks being executed.
+    func executeSequence<SequenceResultType>(from initialTask: Task, with execution: @escaping (Task, Any) -> SequenceExecution<SequenceResultType>) -> SequenceExecutionHandle<SequenceResultType>
+}

Sources/Concurrency/Executor/SerialSequenceExecutor.swift

@@ -0,0 +1,85 @@
+//
+//  Copyright (c) 2018. Uber Technologies
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//
+
+import Foundation
+
+/// An executor that executes sequences of tasks serially from the
+/// caller thread.
+///
+/// - note: Generally this implementation should only be used for debugging
+/// purposes, as debugging highly concurrent task executions can be very
+/// challenging. Production code should use `ConcurrentSequenceExecutor`.
+/// - seeAlso: `SequenceExecutor`.
+/// - seeAlso: `Task`.
+class SerialSequenceExecutor: SequenceExecutor {
+
+    /// Execute a sequence of tasks serially from the given initial task
+    /// on the caller thread.
+    ///
+    /// - parameter initialTask: The root task of the sequence of tasks
+    /// to be executed.
+    /// - parameter execution: The execution defining the sequence of tasks.
+    /// When a task completes its execution, this closure is invoked with
+    /// the task and its produced result. This closure is invoked from
+    /// the caller thread serially as each task completes. The tasks provided
+    /// by this closure are executed serially on the initial caller thread.
+    /// - returns: The execution handle that allows control and monitoring
+    /// of the sequence of tasks being executed.
+    func executeSequence<SequenceResultType>(from initialTask: Task, with execution: @escaping (Task, Any) -> SequenceExecution<SequenceResultType>) -> SequenceExecutionHandle<SequenceResultType> {
+        let handle: SequenceExecutionHandleImpl<SequenceResultType> = SequenceExecutionHandleImpl()
+        execute(initialTask, with: handle, execution)
+        return handle
+    }
+
+    // MARK: - Private
+
+    private func execute<SequenceResultType>(_ task: Task, with sequenceHandle: SequenceExecutionHandleImpl<SequenceResultType>, _ execution: @escaping (Task, Any) -> SequenceExecution<SequenceResultType>) {
+        guard !sequenceHandle.isCancelled else {
+            return
+        }
+
+        let result = task.typeErasedExecute()
+        let nextExecution = execution(task, result)
+        switch nextExecution {
+        case .continueSequence(let nextTask):
+            self.execute(nextTask, with: sequenceHandle, execution)
+        case .endOfSequence(let result):
+            sequenceHandle.sequenceDidComplete(with: result)
+        }
+    }
+}
+
+private class SequenceExecutionHandleImpl<SequenceResultType>: SequenceExecutionHandle<SequenceResultType> {
+
+    private var didCancel = false
+    private var result: SequenceResultType?
+
+    fileprivate var isCancelled: Bool {
+        return didCancel
+    }
+
+    fileprivate override func await(withTimeout timeout: TimeInterval?) throws -> SequenceResultType {
+        return result!
+    }
+
+    fileprivate func sequenceDidComplete(with result: SequenceResultType) {
+        self.result = result
+    }
+
+    fileprivate override func cancel() {
+        didCancel = true
+    }
+}