- improved performance

- fixed bugs and crashes

Author: Alex Belozierov

Sources/SwiftCoroutine/Coroutine/CoroutineDispatcher/CoroutineDispatcher.swift

@@ -6,7 +6,7 @@
 //  Copyright © 2020 Alex Belozierov. All rights reserved.
 //
 
-@usableFromInline protocol _CoroutineTaskExecutor: class {
+@usableFromInline protocol CoroutineTaskExecutor: class {
 
     func execute(on scheduler: CoroutineScheduler, task: @escaping () -> Void)
 
@@ -22,7 +22,7 @@
         return CoroutineDispatcher(executor: executor)
     }
 
-    @usableFromInline let executor: _CoroutineTaskExecutor
+    @usableFromInline let executor: CoroutineTaskExecutor
 
     @inlinable internal func execute(on scheduler: CoroutineScheduler, task: @escaping () -> Void) {
         executor.execute(on: scheduler, task: task)

Sources/SwiftCoroutine/Coroutine/Shared/SharedCoroutineQueue.swift

@@ -2,71 +2,90 @@
 //  SharedCoroutine.swift
 //  SwiftCoroutine
 //
-//  Created by Alex Belozierov on 09.03.2020.
+//  Created by Alex Belozierov on 03.04.2020.
 //  Copyright © 2020 Alex Belozierov. All rights reserved.
 //
 
-#if SWIFT_PACKAGE
-import CCoroutine
-#endif
-
-internal final class SharedCoroutine: CoroutineProtocol {
+internal final class SharedCoroutine {
+    
+    internal typealias CompletionState = SharedCoroutineQueue.CompletionState
 
     private enum State: Int {
-        case running, suspending, suspended, restarting
+        case running, suspending, suspended, restarting, finished
     }
 
     private struct StackBuffer {
         let stack: UnsafeMutableRawPointer, size: Int
     }
 
-    private let dispatcher: SharedCoroutineDispatcher
+    internal let dispatcher: SharedCoroutineDispatcher
     internal let queue: SharedCoroutineQueue
     private(set) var scheduler: CoroutineScheduler
-    private(set) var environment: UnsafeMutablePointer<CoroutineContext.SuspendData>!
-    private var stackBuffer: StackBuffer!
     private var state = AtomicEnum(value: State.running)
 
+    private var environment: UnsafeMutablePointer<CoroutineContext.SuspendData>!
+    private var stackBuffer: StackBuffer!
+    
     internal init(dispatcher: SharedCoroutineDispatcher, queue: SharedCoroutineQueue, scheduler: CoroutineScheduler) {
         self.dispatcher = dispatcher
         self.queue = queue
         self.scheduler = scheduler
     }
 
-    // MARK: - actions
+    // MARK: - Actions
+    
+    internal func start() -> CompletionState {
+        performAsCurrent { perform(queue.context.start) }
+    }
+    
+    internal func resume() -> CompletionState {
+        performAsCurrent(resumeContext)
+    }
 
-    internal func start(_ task: @escaping () -> Void) {
-        performAsCurrent { perform { queue.start(task) } }
+    private func resumeContext() -> CompletionState {
+        perform { queue.context.resume(from: environment.pointee.env) }
     }
 
-    internal func resume() {
-        performAsCurrent { perform { queue.resume(self) } }
+    private func perform(_ block: () -> Bool) -> CompletionState {
+        if block() { return .finished }
+        switch state.update(.suspended) {
+        case .running: return resumeContext()
+        case .restarting: return .restarting
+        default: return .suspended
+        }
     }
 
     private func suspend() {
         if environment == nil {
             environment = .allocate(capacity: 1)
             environment.initialize(to: .init())
         }
-        queue.suspend(self)
+        queue.context.suspend(to: environment)
     }
 
-    private func perform(_ block: () -> Bool) {
-        queue.isFree = true
-        if block() { return }
-        switch state.update(.suspended) {
-        case .running:
-            state.value = .running
-            perform { queue.resume(self) }
-        case .restarting:
-            state.value = .running
-            queue.isFree = false
-            dispatcher.restart(self)
-        default: break
-        }
+    // MARK: - Stack
+    
+    internal func saveStack() {
+        let size = environment.pointee.sp.distance(to: queue.context.stackTop)
+        let stack = UnsafeMutableRawPointer.allocate(byteCount: size, alignment: 16)
+        stack.copyMemory(from: environment.pointee.sp, byteCount: size)
+        stackBuffer = .init(stack: stack, size: size)
     }
 
-    // MARK: - await
+    internal func restoreStack() {
+        environment.pointee.sp.copyMemory(from: stackBuffer.stack, byteCount: stackBuffer.size)
+        stackBuffer.stack.deallocate()
+        stackBuffer = nil
+    }
+    
+    deinit {
+        environment?.pointee.env.deallocate()
+        environment?.deallocate()
+    }
+    
+}
+
+extension SharedCoroutine: CoroutineProtocol {
 
     internal func await<T>(_ callback: (@escaping (T) -> Void) -> Void) -> T {
         state.value = .suspending
@@ -95,26 +114,4 @@ internal final class SharedCoroutine: CoroutineProtocol {
         suspend()
     }
 
-    deinit {
-        environment?.pointee.env.deallocate()
-        environment?.deallocate()
-    }
-    
-}
-
-extension SharedCoroutine {
-    
-    internal func saveStack() {
-        let size = environment.pointee.sp.distance(to: queue.context.stackTop)
-        let stack = UnsafeMutableRawPointer.allocate(byteCount: size, alignment: 16)
-        stack.copyMemory(from: environment.pointee.sp, byteCount: size)
-        stackBuffer = .init(stack: stack, size: size)
-    }
-    
-    internal func restoreStack() {
-        environment.pointee.sp.copyMemory(from: stackBuffer.stack, byteCount: stackBuffer.size)
-        stackBuffer.stack.deallocate()
-        stackBuffer = nil
-    }
-    
 }

Sources/SwiftCoroutine/Coroutine/Shared/SharedCoroutine.swift renamed to Sources/SwiftCoroutine/Coroutine/SharedCoroutineDispatcher/SharedCoroutine.swift

@@ -2,24 +2,25 @@
 //  SharedCoroutineDispatcher.swift
 //  SwiftCoroutine
 //
-//  Created by Alex Belozierov on 09.03.2020.
+//  Created by Alex Belozierov on 03.04.2020.
 //  Copyright © 2020 Alex Belozierov. All rights reserved.
 //
 
 import Dispatch
 
-internal final class SharedCoroutineDispatcher: _CoroutineTaskExecutor {
+internal final class SharedCoroutineDispatcher: CoroutineTaskExecutor {
 
-    private struct Task {
+    internal struct Task {
         let scheduler: CoroutineScheduler, task: () -> Void
     }
 
     private let mutex = PsxLock()
     private let stackSize: Int
+    private var tasks = FifoQueue<Task>()
+    
     private var contextsCount: Int
     private var freeQueues = [SharedCoroutineQueue]()
     private var suspendedQueues = Set<SharedCoroutineQueue>()
-    private var tasks = FifoQueue<Task>()
     private var freeCount: AtomicInt
 
     internal init(contextsCount: Int, stackSize: Int) {
@@ -31,14 +32,15 @@ internal final class SharedCoroutineDispatcher: _CoroutineTaskExecutor {
         startDispatchSource()
     }
 
+    // MARK: - Start
+    
     internal func execute(on scheduler: CoroutineScheduler, task: @escaping () -> Void) {
         func perform() {
             freeCount.update { max(0, $0 - 1) }
             mutex.lock()
             if let queue = freeQueue {
                 mutex.unlock()
-                SharedCoroutine(dispatcher: self, queue: queue, scheduler: scheduler).start(task)
-                performNext(for: queue)
+                queue.start(dispatcher: self, task: .init(scheduler: scheduler, task: task))
             } else {
                 tasks.push(.init(scheduler: scheduler, task: task))
                 mutex.unlock()
@@ -58,8 +60,7 @@ internal final class SharedCoroutineDispatcher: _CoroutineTaskExecutor {
         if let queue = freeQueues.popLast() { return queue }
         if contextsCount > 0 {
             contextsCount -= 1
-            let context = CoroutineContext(stackSize: stackSize)
-            return SharedCoroutineQueue(context: context)
+            return SharedCoroutineQueue(stackSize: stackSize)
         } else if suspendedQueues.count < 2 {
             return suspendedQueues.popFirst()
         }
@@ -74,28 +75,9 @@ internal final class SharedCoroutineDispatcher: _CoroutineTaskExecutor {
         return suspendedQueues.remove(min)
     }
 
+    // MARK: - Resume
+    
     internal func resume(_ coroutine: SharedCoroutine) {
-//        func perform() {
-//            mutex.lock()
-//            if suspendedQueues.remove(coroutine.queue) == nil {
-//                coroutine.queue.push(coroutine)
-//                mutex.unlock()
-//            } else {
-//                freeCount.decrease()
-//                mutex.unlock()
-//                coroutine.resume()
-//                performNext(for: coroutine.queue)
-//            }
-//        }
-//        mutex.lock()
-//        if suspendedQueues.contains(coroutine.queue) {
-//            mutex.unlock()
-//            coroutine.scheduler.scheduleTask(perform)
-//        } else {
-//            coroutine.queue.push(coroutine)
-//            mutex.unlock()
-//        }
-        
         mutex.lock()
         if suspendedQueues.remove(coroutine.queue) == nil {
             coroutine.queue.push(coroutine)
@@ -104,54 +86,33 @@ internal final class SharedCoroutineDispatcher: _CoroutineTaskExecutor {
             mutex.unlock()
             freeCount.decrease()
             coroutine.scheduler.scheduleTask {
-                coroutine.resume()
-                self.performNext(for: coroutine.queue)
+                coroutine.queue.resume(coroutine: coroutine)
             }
         }
     }
 
-    internal func restart(_ coroutine: SharedCoroutine) {
-        coroutine.scheduler.scheduleTask {
-            coroutine.resume()
-            self.performNext(for: coroutine.queue)
-        }
-    }
+    // MARK: - Next
 
-    private enum NextState: Int {
-        case running, none
-    }
-    
-    private func performNext(for queue: SharedCoroutineQueue) {
-        var state = AtomicEnum(value: NextState.none)
-        while queue.isFree {
-            mutex.lock()
-            if let coroutine = queue.pop() {
-                mutex.unlock()
-                state.value = .running
-                coroutine.scheduler.scheduleTask {
-                    coroutine.resume()
-                    if state.update(.none) == .running { return }
-                    self.performNext(for: queue)
-                }
-            } else if let task = tasks.pop() {
-                mutex.unlock()
-                state.value = .running
-                task.scheduler.scheduleTask {
-                    SharedCoroutine(dispatcher: self, queue: queue, scheduler: task.scheduler)
-                        .start(task.task)
-                    if state.update(.none) == .running { return }
-                    self.performNext(for: queue)
-                }
-            } else if queue.started == 0 {
+    internal func performNext(for queue: SharedCoroutineQueue) {
+        mutex.lock()
+        if let coroutine = queue.pop() {
+            mutex.unlock()
+            coroutine.scheduler.scheduleTask {
+                queue.resume(coroutine: coroutine)
+            }
+        } else if let task = tasks.pop() {
+            mutex.unlock()
+            task.scheduler.scheduleTask {
+                queue.start(dispatcher: self, task: task)
+            }
+        } else {
+            if queue.started == 0 {
                 freeQueues.append(queue)
-                freeCount.increase()
-                return mutex.unlock()
             } else {
                 suspendedQueues.insert(queue)
-                freeCount.increase()
-                return mutex.unlock()
             }
-            if state.update(.none) == .running { return }
+            freeCount.increase()
+            mutex.unlock()
         }
     }