Add recursive mutex type, remove protect

Author: digitaliz

Flow/Disposable.swift

@@ -84,7 +84,10 @@ public final class DisposeBag: Disposable {
 
     /// Returns true if there is currently no disposables to dispose.
     public var isEmpty: Bool {
-        return mutex.protect { disposables.isEmpty }
+        mutex.lock()
+        defer { mutex.unlock() }
+        
+        return disposables.isEmpty
     }
 
     public func dispose() {

Flow/Future.swift

@@ -329,7 +329,10 @@ func memPrint(_ str: String, _ count: Int32) {
 private extension Future {
 
     private var protectedState: State {
-        return mutex.protect { state }
+        mutex.lock()
+        defer { mutex.unlock() }
+        
+        return state
     }
 
     func lock() {

Flow/FutureQueue.swift

@@ -18,7 +18,7 @@ public final class FutureQueue<Resource> {
     private let queueScheduler: Scheduler
     private var _closedError: Error?
     private let isEmptyCallbacker = Callbacker<Bool>()
-    private var _mutex = pthread_mutex_t()
+    private var mutex = pthread_mutex_t()
 
     // enqueued items.
     private var items: [Executable] = [] {
@@ -41,9 +41,11 @@ public final class FutureQueue<Resource> {
         queueScheduler = executeOn
         OSAtomicIncrement32(&futureQueueUnitTestAliveCount)
         memPrint("Queue init", futureQueueUnitTestAliveCount)
+        mutex.initialize(as: .recursive)
     }
 
     deinit {
+        mutex.deinitialize()
         OSAtomicDecrement32(&futureQueueUnitTestAliveCount)
         memPrint("Queue deinit", futureQueueUnitTestAliveCount)
     }
@@ -61,9 +63,9 @@ public extension FutureQueue {
         return Future { completion in
             let item = QueueItem<Output>(operation: operation, completion: completion)
 
-            self.mutex.protect {
-                self.items.append(item)
-            }
+            self.mutex.lock()
+            self.items.append(item)
+            self.mutex.unlock()
 
             self.executeNextItem()
 
@@ -119,7 +121,10 @@ public extension FutureQueue {
 public extension FutureQueue {
     /// Do we have any enqueued operations?
     var isEmpty: Bool {
-        return mutex.protect { items.isEmpty }
+        mutex.lock()
+        defer { mutex.unlock() }
+
+        return items.isEmpty
     }
 
     /// Returns a signal that will signal when `isEmpty` is changed.
@@ -164,19 +169,21 @@ public extension FutureQueue {
 
     /// The error passed to `abortQueuedExecutionWithError()` if called with `shouldCloseQueue` as true.
     var closedError: Error? {
-        return mutex.protect { _closedError }
+        mutex.lock()
+        defer { mutex.unlock() }
+
+        return _closedError
     }
 }
 
 private extension FutureQueue {
-    var mutex: PThreadMutex { return PThreadMutex(&_mutex) }
     func lock() { mutex.lock() }
     func unlock() { mutex.unlock() }
 
     func removeItem(_ item: Executable) {
-        mutex.protect {
-            _ = items.firstIndex { $0 === item }.map { items.remove(at: $0) }
-        }
+        mutex.lock()
+        defer { mutex.unlock() }
+        _ = items.firstIndex { $0 === item }.map { items.remove(at: $0) }
     }
 
     func executeNextItem() {
@@ -188,9 +195,9 @@ private extension FutureQueue {
         unlock()
 
         item.execute(on: queueScheduler) {
-            self.mutex.protect {
-                self.concurrentCount -= 1
-            }
+            self.mutex.lock()
+            self.concurrentCount -= 1
+            self.mutex.unlock()
             self.removeItem(item)
             self.executeNextItem()
         }

Flow/Locking.swift

@@ -39,11 +39,24 @@ public final class Mutex {
     }
 }
 
+@usableFromInline
+enum MutexType {
+    case normal
+    case recursive
+
+    var attrType: Int32 {
+        switch self {
+        case .normal: return PTHREAD_MUTEX_NORMAL
+        case .recursive: return PTHREAD_MUTEX_RECURSIVE
+        }
+    }
+}
+
 internal extension pthread_mutex_t {
 
-    @inlinable mutating func initialize() {
+    @inlinable mutating func initialize(as type: MutexType = .normal) {
         withUnsafeMutablePointer(to: &self) {
-            $0.initialize()
+            $0.initialize(as: type)
         }
     }
 
@@ -65,12 +78,6 @@ internal extension pthread_mutex_t {
         }
     }
 
-    @inlinable mutating func protect<T>(_ block: () throws -> T) rethrows -> T {
-        try withUnsafeMutablePointer(to: &self) {
-            return try $0.protect(block)
-        }
-    }
-
 }
 
 typealias PThreadMutex = UnsafeMutablePointer<pthread_mutex_t>
@@ -79,14 +86,14 @@ typealias PThreadMutex = UnsafeMutablePointer<pthread_mutex_t>
 /// - Note: You have to explicity call `initialize()` before use (typically in a class init) and `deinitialize()` when done (typically in a class deinit)
 extension UnsafeMutablePointer where Pointee == pthread_mutex_t {
     @usableFromInline
-    func initialize() {
+    func initialize(as type: MutexType = .normal) {
         var attr = pthread_mutexattr_t()
         defer { pthread_mutexattr_destroy(&attr) }
         guard pthread_mutexattr_init(&attr) == 0 else {
             preconditionFailure()
         }
 
-        pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL)
+        pthread_mutexattr_settype(&attr, type.attrType)
 
         guard pthread_mutex_init(self, &attr) == 0 else {
             preconditionFailure()
@@ -110,13 +117,6 @@ extension UnsafeMutablePointer where Pointee == pthread_mutex_t {
         pthread_mutex_unlock(self)
     }
 
-    /// Will lock `self`, call `block`, then unlock `self`
-    @discardableResult @usableFromInline
-    func protect<T>(_ block: () throws -> T) rethrows -> T {
-        pthread_mutex_lock(self)
-        defer { pthread_mutex_unlock(self) }
-        return try block()
-    }
 }
 
 /// Internal helper to help manage state in stateful transforms.

Flow/OrderedCallbacker.swift

@@ -26,30 +26,40 @@ public final class OrderedCallbacker<OrderedValue, CallbackValue> {
 
     /// - Returns: True if no callbacks has been registered.
     public var isEmpty: Bool {
-        return mutex.protect { callbacks.isEmpty }
+        mutex.lock()
+        defer { mutex.unlock() }
+
+        return callbacks.isEmpty
     }
 
     /// Register a callback and orderedValue to be called when `callAll` is executed.
     /// - Parameter callback: The next callback won't be called until `callback` return `Future` completes
     /// - Parameter orderedValue: The value used to order this callback
     /// - Returns: A `Disposable` to be disposed to unregister the callback.
     public func addCallback(_ callback: @escaping (CallbackValue) -> Future<()>, orderedBy orderedValue: OrderedValue) -> Disposable {
-        return mutex.protect {
-            let key = generateKey()
-            callbacks[key] = (orderedValue, callback)
-            return Disposer {
-                self.mutex.protect { self.callbacks[key] = nil }
-            }
+        mutex.lock()
+        defer { mutex.unlock() }
+
+        let key = generateKey()
+        callbacks[key] = (orderedValue, callback)
+        return Disposer {
+            self.mutex.lock()
+            self.callbacks[key] = nil
+            self.mutex.unlock()
         }
     }
 
     /// Will call all registered callbacks with `value` in the order set by `isOrderedBefore`
     /// - Returns: A `Future` that will complete when all callbacks has been called.
     @discardableResult
     public func callAll(with value: CallbackValue, isOrderedBefore: (OrderedValue, OrderedValue) -> Bool) -> Future<()> {
-        return mutex.protect {
-            callbacks.values.sorted { isOrderedBefore($0.0, $1.0) }.map { $1 }
-        }.mapToFuture { $0(value) }.toVoid()
+
+        mutex.lock()
+        defer { mutex.unlock() }
+        return callbacks.values
+            .sorted { isOrderedBefore($0.0, $1.0) }
+            .map { $1 }
+            .mapToFuture { $0(value) }.toVoid()
     }
 }