Roll back semaphore change (#59)

* Revert "Final removal of actors (#57)"

This reverts commit 2f8d44c.

* Update dock, add types from Atomics, fix Counter bug

Signed-off-by: EandJsFilmCrew <789213+rvsrvs@users.noreply.github.com>

Signed-off-by: EandJsFilmCrew <789213+rvsrvs@users.noreply.github.com>

Sources/FreeCombine/Decombinator/DistributorRepeaterState.swift

@@ -68,7 +68,7 @@ public struct DistributorRepeaterState<ID: Hashable & Sendable, Output: Sendable
                     if case .completion = output { mostRecentDemand = .done }
                 }
                 catch { mostRecentDemand = .done }
-                semaphore.decrement(with: .repeated(id, mostRecentDemand))
+                await semaphore.decrement(with: .repeated(id, mostRecentDemand))
                 return mostRecentDemand == .done ? .completion(.exit) : .none
         }
     }

Sources/FreeCombine/Decombinator/DistributorState.swift

@@ -148,12 +148,11 @@ public struct DistributorState<Output: Sendable> {
         with result: AsyncStream<Output>.Result
     ) async throws -> Void {
         guard currentRepeaters.count > 0 else { return }
-        var semaphore: Semaphore<[Int], RepeatedAction<Int>>!
         try await withResumption { (completedResumption: Resumption<[Int]>) in
             // Note that the semaphore's reducer constructs a list of repeaters
             // which have responded with .done and that the elements of that list
             // are removed at completion of the sends
-            semaphore = Semaphore<[Int], RepeatedAction<Int>>(
+            let semaphore = Semaphore<[Int], RepeatedAction<Int>>(
                 resumption: completedResumption,
                 reducer: { completedIds, action in
                     guard case let .repeated(id, .done) = action else { return }
@@ -169,7 +168,7 @@ public struct DistributorState<Output: Sendable> {
                     case .enqueued:
                         ()
                     case .terminated:
-                        semaphore.decrement(with: .repeated(key, .done))
+                        Task { await semaphore.decrement(with: .repeated(key, .done)) }
                     case .dropped:
                         fatalError("Should never drop")
                     @unknown default:
@@ -180,7 +179,6 @@ public struct DistributorState<Output: Sendable> {
         .forEach { key in
             repeaters.removeValue(forKey: key)
         }
-        assert(semaphore.count == 0, "Exiting with semaphore unresolved")
     }
 
     mutating func process(

Sources/FreeCombine/Decombinator/PromiseRepeaterState.swift

@@ -57,7 +57,7 @@ public struct PromiseRepeaterState<ID: Hashable & Sendable, Output: Sendable>: I
         switch action {
             case let .complete(output, semaphore):
                     try? await downstream(output)
-                semaphore.decrement(with: .repeated(id, .done))
+                await semaphore.decrement(with: .repeated(id, .done))
                 return .completion(.exit)
         }
     }

Sources/FreeCombine/Decombinator/PromiseState.swift

@@ -176,7 +176,7 @@ public struct PromiseState<Output: Sendable> {
                     case .enqueued:
                         ()
                     case .terminated:
-                        semaphore.decrement(with: .repeated(key, .done))
+                        Task { await semaphore.decrement(with: .repeated(key, .done)) }
                     case .dropped:
                         fatalError("Should never drop")
                     @unknown default:

Sources/FreeCombine/Synchronization/Semaphore.swift

@@ -18,18 +18,15 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 //
-public class Semaphore<State, Action> {
+public actor Semaphore<State, Action> {
     public enum Error: Swift.Error {
         case complete
     }
     private let resumption: Resumption<State>
     private let reducer: (inout State, Action) -> Void
 
     private var state: State
-    private var counter: Counter
-    private var actions: SingleConsumerStack<Action> = .init()
-
-    var count: Int { counter.count }
+    private var count: Int
 
     public init(
         resumption: Resumption<State>,
@@ -40,27 +37,26 @@ public class Semaphore<State, Action> {
         self.resumption = resumption
         self.reducer = reducer
         self.state = initialState
-        self.counter = .init(count: count)
+        self.count = count
         if count == 0 { resumption.resume(returning: initialState) }
     }
 
-    public func decrement(
-        function: String = #function,
-        file: String = #file,
-        line: Int = #line,
-        with action: Action
-    ) -> Void {
-        actions.push(action)
-        switch counter.decrement() {
-            case Int.min ..< 0:
-                fatalError("Semaphore decremented after complete in \(function) @\(file):\(line)")
-            case 1 ... Int.max:
-                return
-            default:
-                while let prevAction = actions.pop() {
-                    reducer(&state, prevAction)
-                }
+    public func decrement(with action: Action, function: String = #function, file: String = #file, line: Int = #line) -> Void {
+        guard count > 0 else {
+            fatalError("Semaphore decremented after complete in \(function) @\(file):\(line)")
+        }
+        count -= 1
+        reducer(&state, action)
+        if count == 0 {
+            resumption.resume(returning: state)
+        }
+    }
+
+    public func increment(with action: Action, function: String = #function, file: String = #file, line: Int = #line) -> Void {
+        guard count > 0 else {
+            fatalError("Semaphore incremented after complete in \(function) @\(file):\(line)")
         }
-        resumption.resume(returning: state)
+        count += 1
+        reducer(&state, action)
     }
 }

Tests/FreeCombineTests/PromiseTests.swift

@@ -160,8 +160,8 @@ final class PromiseTests: XCTestCase {
             )
             for _ in 0 ..< maxAttempts {
                 Task {
-                    do { try await promise.succeed(13); succeedCounter.increment(); semaphore.decrement(with: ()) }
-                    catch { failureCounter.increment(); semaphore.decrement(with: ()) }
+                    do { try await promise.succeed(13); succeedCounter.increment(); await semaphore.decrement(with: ()) }
+                    catch { failureCounter.increment(); await semaphore.decrement(with: ()) }
                 }
             }
         }