Improve select performance (#5)

* Actor

* refactor

* Fix close

* Update Benchmarks

Benchmarks/Readme.md

@@ -36,50 +36,54 @@ The below results are from running the whole benchmark suit which covers multipl
 
 Test | Type | Execution Time(ms)
 -----|------|---------------
-SPSC | `Int` | `1117`
-MPSC | `Int` | `737`
-SPMC | `Int` | `881`
-MPMC | `Int` | `959`
-MPSC Write Contention | `Int` | `723`
-SPSC Buffered(100) | `Int` | `513`
-MPSC Buffered(100) | `Int` | `637`
-SPMC Buffered(100) | `Int` | `646`
-MPMC Buffered(100) | `Int` | `956`
-SyncRW | `Int` | `1531`
-Channel multi-select | `Int` | `1547`
-SPSC | `String` | `1292`
-MPSC | `String` | `850`
-SPMC | `String` | `923`
-MPMC | `String` | `1004`
-MPSC Write Contention | `String` | `848`
-SPSC Buffered(100) | `String` | `760`
-MPSC Buffered(100) | `String` | `698`
-SPMC Buffered(100) | `String` | `709`
-MPMC Buffered(100) | `String` | `1011`
-SyncRW | `String` | `1707`
-Channel multi-select | `String` | `1590`
-SPSC | `ValueData` | `1210`
-MPSC | `ValueData` | `790`
-SPMC | `ValueData` | `902`
-MPMC | `ValueData` | `975`
-MPSC Write Contention | `ValueData` | `785`
-SPSC Buffered(100) | `ValueData` | `497`
-MPSC Buffered(100) | `ValueData` | `658`
-SPMC Buffered(100) | `ValueData` | `659`
-MPMC Buffered(100) | `ValueData` | `978`
-SyncRW | `ValueData` | `1526`
-Channel multi-select | `ValueData` | `1567`
-SPSC | `RefData` | `1320`
-MPSC | `RefData` | `934`
-SPMC | `RefData` | `986`
-MPMC | `RefData` | `1261`
-MPSC Write Contention | `RefData` | `1492`
-SPSC Buffered(100) | `RefData` | `698`
-MPSC Buffered(100) | `RefData` | `774`
-SPMC Buffered(100) | `RefData` | `807`
-MPMC Buffered(100) | `RefData` | `1224`
-SyncRW | `RefData` | `1725`
-Channel multi-select | `RefData` | `1567`
+SPSC | `Int` | `1090`
+MPSC | `Int` | `698`
+SPMC | `Int` | `887`
+MPMC | `Int` | `957`
+MPSC Write Contention | `Int` | `707`
+SPSC Buffered(100) | `Int` | `604`
+MPSC Buffered(100) | `Int` | `639`
+SPMC Buffered(100) | `Int` | `642`
+MPMC Buffered(100) | `Int` | `960`
+MPSC Write Contention Buffered(100) | `Int` | `1095`
+SyncRW | `Int` | `1505`
+Channel multi-select | `Int` | `1172`
+SPSC | `String` | `1299`
+MPSC | `String` | `830`
+SPMC | `String` | `911`
+MPMC | `String` | `1001`
+MPSC Write Contention | `String` | `828`
+SPSC Buffered(100) | `String` | `725`
+MPSC Buffered(100) | `String` | `703`
+SPMC Buffered(100) | `String` | `691`
+MPMC Buffered(100) | `String` | `1014`
+MPSC Write Contention Buffered(100) | `String` | `1152`
+SyncRW | `String` | `1703`
+Channel multi-select | `String` | `1190`
+SPSC | `ValueData` | `1167`
+MPSC | `ValueData` | `772`
+SPMC | `ValueData` | `904`
+MPMC | `ValueData` | `977`
+MPSC Write Contention | `ValueData` | `768`
+SPSC Buffered(100) | `ValueData` | `496`
+MPSC Buffered(100) | `ValueData` | `652`
+SPMC Buffered(100) | `ValueData` | `645`
+MPMC Buffered(100) | `ValueData` | `987`
+MPSC Write Contention Buffered(100) | `ValueData` | `1158`
+SyncRW | `ValueData` | `1485`
+Channel multi-select | `ValueData` | `1176`
+SPSC | `RefData` | `1306`
+MPSC | `RefData` | `933`
+SPMC | `RefData` | `969`
+MPMC | `RefData` | `1219`
+MPSC Write Contention | `RefData` | `946`
+SPSC Buffered(100) | `RefData` | `675`
+MPSC Buffered(100) | `RefData` | `729`
+SPMC Buffered(100) | `RefData` | `742`
+MPMC Buffered(100) | `RefData` | `1220`
+MPSC Write Contention Buffered(100) | `RefData` | `1221`
+SyncRW | `RefData` | `1685`
+Channel multi-select | `RefData` | `1198`
 SPSC Async alg | `Int` | `3000`
 MPSC Async alg | `Int` | `4030`
 SPMC Async alg | `Int` | `3951`

Sources/AsyncChannels/Channel.swift

@@ -41,7 +41,7 @@ public final class Channel<T: Sendable>: @unchecked Sendable {
         return buffer.count
     }
 
-    var selectWaiter: AsyncSemaphore?
+    var selectWaiter: SelectSignal?
 
     @usableFromInline
     var isClosed: Bool {
@@ -50,7 +50,7 @@ public final class Channel<T: Sendable>: @unchecked Sendable {
         return closed
     }
 
-    func receiveOrListen(_ sema: AsyncSemaphore) -> T? {
+    func receiveOrListen(_ sema: SelectSignal) -> T? {
         mutex.lock()
 
         if let val = nonBlockingReceive() {
@@ -67,7 +67,7 @@ public final class Channel<T: Sendable>: @unchecked Sendable {
         return nil
     }
 
-    func sendOrListen(_ sema: AsyncSemaphore, value: T) -> Bool {
+    func sendOrListen(_ sema: SelectSignal, value: T) -> Bool {
         mutex.lock()
 
         if nonBlockingSend(value) {
@@ -168,6 +168,8 @@ public final class Channel<T: Sendable>: @unchecked Sendable {
         mutex.lock()
         defer { mutex.unlock() }
         closed = true
+        selectWaiter?.signal()
+
 
         while let recvW = recvQueue.pop() {
             recvW.resume(returning: nil)

Sources/AsyncChannels/Select.swift

@@ -9,7 +9,7 @@ public struct SelectHandler {
 }
 
 protocol SelectProtocol {
-    func handle(_ sm: AsyncSemaphore) async -> Bool
+    func handle(_ sm: SelectSignal) async -> Bool
 }
 
 struct RxHandler<T>: SelectProtocol {
@@ -22,7 +22,7 @@ struct RxHandler<T>: SelectProtocol {
         self.outFunc = outFunc
     }
 
-    func handle(_ sm: AsyncSemaphore) async -> Bool {
+    func handle(_ sm: SelectSignal) async -> Bool {
         if let val = chan.receiveOrListen(sm) {
             await outFunc(val)
             return true
@@ -42,7 +42,7 @@ struct NoneHandler: SelectProtocol {
         self.handler = handler
     }
 
-    func handle(_ sm: AsyncSemaphore) async -> Bool {
+    func handle(_ sm: SelectSignal) async -> Bool {
         await handler()
         return true
     }
@@ -59,7 +59,7 @@ struct TxHandler<T>: SelectProtocol {
         self.onSend = onSend
     }
 
-    func handle(_ sm: AsyncSemaphore) async -> Bool {
+    func handle(_ sm: SelectSignal) async -> Bool {
         if chan.sendOrListen(sm, value: val) {
             await onSend()
             return true
@@ -76,7 +76,7 @@ public struct SelectCollector {
 }
 
 @usableFromInline
-func handle(_ sm: AsyncSemaphore, handlers: [SelectHandler]) async -> Bool {
+func handle(_ sm: SelectSignal, handlers: [SelectHandler]) async -> Bool {
     var defaultCase: NoneHandler?
 
     for handler in handlers.shuffled() {
@@ -94,7 +94,7 @@ func handle(_ sm: AsyncSemaphore, handlers: [SelectHandler]) async -> Bool {
 public func select(@SelectCollector cases: () -> ([SelectHandler])) async {
     let handlers = cases()
     while true {
-        let sm = AsyncSemaphore(value: 0)
+        let sm = SelectSignal()
         if await handle(sm, handlers: handlers) {
             return
         }

Sources/AsyncChannels/Synchronization.swift

@@ -1,39 +1,32 @@
 import Foundation
 
 @usableFromInline
-class AsyncSemaphore {
-    private var permits: Int
-    private var mutex = FastLock()
-    private var continuationQueue = LinkedList<UnsafeContinuation<Void, Never>>()
+actor SelectSignal {
+    private var signaled = false
+    private var continuation: UnsafeContinuation<Void, Never>?
 
     @usableFromInline
-    init(value: Int) {
-        self.permits = value
-    }
+    init() {}
 
     @usableFromInline
     func wait() async {
+        if signaled {
+            return
+        }
         await withUnsafeContinuation { continuation in
-            self.mutex.lock()
-            if self.permits > 0 {
-                self.permits -= 1
-                self.mutex.unlock()
-                continuation.resume()
-            } else {
-                self.continuationQueue.push(continuation)
-                self.mutex.unlock()
-            }
+            self.continuation = continuation
         }
     }
 
-    func signal() {
-        self.mutex.lock()
-        if let next = continuationQueue.pop() {
-            self.mutex.unlock()
-            next.resume()
-        } else {
-            permits += 1
-            self.mutex.unlock()
+    private func _signal() {
+        signaled = true
+        continuation?.resume()
+        continuation = nil
+    }
+
+    nonisolated func signal() {
+        Task {
+            await _signal()
         }
     }
 }