Add Linux support using Synchronization.Mutex

.github/workflows/ci.yml

@@ -1,6 +1,6 @@
 name: Build and test
 
-on: [push]
+on: [push, pull_request]
 
 jobs:
   Build:
@@ -9,7 +9,7 @@ jobs:
 
     steps:
       - name: Checkout branch
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
       - name: Build
         run: swift build -Xswiftc -suppress-warnings
 
@@ -19,7 +19,7 @@ jobs:
 
     steps:
       - name: Checkout branch
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
       - name: Test
         run: swift test --enable-code-coverage -Xswiftc -suppress-warnings
       - name: Generate coverage
@@ -29,3 +29,21 @@ jobs:
         uses: codecov/codecov-action@v3
         with:
           files: ${{join(fromJSON(steps.coverage-files.outputs.files), ',')}}
+
+  Linux:
+    runs-on: ubuntu-22.04
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        swift: ["6.1.3", "6.4.0"]
+    container: swift:${{ matrix.swift }}-jammy
+    steps:
+      - name: Checkout branch
+        uses: actions/checkout@v4
+      - name: Show Swift version
+        run: swift --version
+      - name: Build release
+        run: swift build -c release
+      - name: Test
+        run: swift test

CHANGELOG.md

@@ -6,7 +6,8 @@
 
 - Breaking: use Swift Async Algorithms for `Sequence.async` and two- or three-input `zip`/`merge`, removing import ambiguities. Add the `AsyncAlgorithms` product dependency and import when migrating these APIs. Preserve variadic `zip`/`merge` and the explicit `AsyncLazySequence` constructor.
 - Breaking: rename the buffered Date timer from `AsyncTimerSequence` to `AsyncBufferedTimerSequence` to avoid ambiguity with Apple's clock-based timer.
-- SwiftPM: require Swift 5.8 or later for the Swift Async Algorithms test dependency.
+- Breaking: require a Swift 6.1 or later compiler and iOS/tvOS 18, macOS 15, or watchOS 11. Preserve Swift 5 language mode. Swift 6.0.3 on Linux crashes when compiling generic `Mutex` storage in debug builds.
+- Linux: use `Synchronization.Mutex` for shared state, remove the unused Combine test import, and add Ubuntu CI with Swift 6.1.3 and 6.4.0.
 
 - SwitchToLatest: finish cancelled collection while the latest channel or outer sequence remains open, and discard late producer results (https://github.com/sideeffect-io/AsyncExtensions/issues/53).
 - Subjects: fix a deadlock when sending values or termination concurrently with consumer cancellation (https://github.com/sideeffect-io/AsyncExtensions/issues/52).

Package.swift

@@ -1,15 +1,15 @@
-// swift-tools-version:5.8
+// swift-tools-version:6.1
 // The swift-tools-version declares the minimum version of Swift required to build this package.
 
 import PackageDescription
 
 let package = Package(
     name: "AsyncExtensions",
     platforms: [
-            .iOS(.v13),
-            .macOS(.v10_15),
-            .tvOS(.v13),
-            .watchOS(.v6)
+            .iOS(.v18),
+            .macOS(.v15),
+            .tvOS(.v18),
+            .watchOS(.v11)
         ],
     products: [
         .library(
@@ -40,5 +40,6 @@ let package = Package(
                 .product(name: "AsyncAlgorithms", package: "swift-async-algorithms")
             ],
             path: "Tests"),
-    ]
+    ],
+    swiftLanguageModes: [.v5]
 )

README.md

@@ -19,7 +19,7 @@ An async sequence produces values over time. You read those values with `for awa
 
 ## Installation
 
-The package manifest requires Swift 5.8 or later and declares support for iOS 13, macOS 10.15, tvOS 13, and watchOS 6 or later. Resolved dependencies may require a newer Swift toolchain.
+The package requires a Swift 6.1 or later compiler and uses Swift 5 language mode. It supports Linux and iOS 18, macOS 15, tvOS 18, and watchOS 11 or later. Its shared-state synchronization uses `Synchronization.Mutex` on every platform.
 
 ### In Xcode
 
@@ -35,12 +35,12 @@ Add `import AsyncExtensions` to the Swift files that use the library. See Apple'
 Add the package to `dependencies`, then add its library product to the target that uses it. For example, a command-line app can use this `Package.swift`:
 
 ```swift
-// swift-tools-version: 5.8
+// swift-tools-version: 6.1
 import PackageDescription
 
 let package = Package(
     name: "MyApp",
-    platforms: [.macOS(.v10_15)],
+    platforms: [.macOS(.v15)],
     dependencies: [
         .package(
             url: "https://github.com/sideeffect-io/AsyncExtensions.git",
@@ -60,6 +60,17 @@ let package = Package(
 
 Put your app's code in `Sources/MyApp/main.swift`, import `AsyncExtensions`, and run it with `swift run`. For a library target, add the same product dependency to your existing `.target(...)`. SwiftPM's [package manifest reference](https://docs.swift.org/package-manager/PackageDescription/PackageDescription.html) describes these dependency declarations.
 
+### On Linux
+
+Install a [Swift toolchain for Linux](https://www.swift.org/install/linux/), then build and test this package with:
+
+```sh
+swift build -c release
+swift test
+```
+
+Linux CI runs the full test suite on Ubuntu 22.04 with Swift 6.1.3 and 6.4.0. The Apple deployment versions in `Package.swift` do not restrict Linux builds.
+
 ## A first example
 
 This sequence produces three numbers. `scan` keeps a running total, and the loop prints each total as it arrives:

Sources/Creators/AsyncBufferedTimerSequence.swift

@@ -6,6 +6,7 @@
 //
 
 @preconcurrency import Foundation
+import Dispatch
 
 private extension DispatchTimeInterval {
   var nanoseconds: UInt64 {

Sources/Supporting/ManagedCriticalState.swift

@@ -1,33 +1,19 @@
-import Darwin
+import Synchronization
 
-final class LockedBuffer<State>: ManagedBuffer<State, os_unfair_lock> {
-  deinit {
-    _ = self.withUnsafeMutablePointerToElements { lock in
-      lock.deinitialize(count: 1)
-    }
-  }
-}
-
-struct ManagedCriticalState<State> {
-  let buffer: ManagedBuffer<State, os_unfair_lock>
+// Copies of iterators and callbacks must keep sharing the same protected state.
+final class ManagedCriticalState<State> {
+  private let state: Mutex<State>
 
   init(_ initial: State) {
-    buffer = LockedBuffer.create(minimumCapacity: 1) { buffer in
-      buffer.withUnsafeMutablePointerToElements { lock in
-        lock.initialize(to: os_unfair_lock())
-      }
-      return initial
-    }
+    self.state = Mutex(initial)
   }
 
   @discardableResult
   func withCriticalRegion<R>(
     _ critical: (inout State) throws -> R
   ) rethrows -> R {
-    try buffer.withUnsafeMutablePointers { header, lock in
-      os_unfair_lock_lock(lock)
-      defer { os_unfair_lock_unlock(lock) }
-      return try critical(&header.pointee)
+    try self.state.withLock { state in
+      try critical(&state)
     }
   }
 
@@ -42,4 +28,4 @@ struct ManagedCriticalState<State> {
   }
 }
 
-extension ManagedCriticalState: @unchecked Sendable where State: Sendable { }
+extension ManagedCriticalState: Sendable where State: Sendable { }

Tests/AsyncSubjets/StreamedTests.swift

@@ -6,7 +6,6 @@
 //
 
 import AsyncExtensions
-import Combine
 import XCTest
 
 final class StreamedTests: XCTestCase {

Tests/Creators/AsyncLazySequenceTests.swift

@@ -23,9 +23,9 @@ final class AsyncLazySequenceTests: XCTestCase {
     XCTAssertEqual(receivedResult, sequence)
   }
 
-  func test_AsyncLazySequence_returns_an_asyncSequence_that_finishes_when_task_is_cancelled() {
+  func test_AsyncLazySequence_returns_an_asyncSequence_that_finishes_when_task_is_cancelled() async {
     let canCancelExpectation = expectation(description: "The first element has been emitted")
-    let hasCancelExceptation = expectation(description: "The task has been cancelled")
+    let hasCancelExpectation = expectation(description: "The task has been cancelled")
 
     let sequence = (0...1_000_000)
 
@@ -36,15 +36,16 @@ final class AsyncLazySequenceTests: XCTestCase {
       for await element in sut {
         firstElement = element
         canCancelExpectation.fulfill()
-        await fulfillment(of: [hasCancelExceptation], timeout: 5)
+        await fulfillment(of: [hasCancelExpectation], timeout: 5)
       }
-      XCTAssertEqual(firstElement!, 0) // the AsyncSequence is cancelled having only emitted the first element
+      XCTAssertEqual(firstElement, 0) // the AsyncSequence is cancelled having only emitted the first element
     }
 
-    wait(for: [canCancelExpectation], timeout: 5) // one element has been emitted, we can cancel the task
+    await fulfillment(of: [canCancelExpectation], timeout: 5) // one element has been emitted, we can cancel the task
 
     task.cancel()
 
-    hasCancelExceptation.fulfill() // we can release the lock in the for loop
+    hasCancelExpectation.fulfill() // we can release the lock in the for loop
+    await task.value // finish checking cancellation before the test returns
   }
 }

Tests/Supporting/ManagedCriticalStateTests.swift

@@ -0,0 +1,24 @@
+@testable import AsyncExtensions
+import Dispatch
+import XCTest
+
+final class ManagedCriticalStateTests: XCTestCase {
+  func test_throwing_critical_region_preserves_changes_and_releases_lock() async {
+    let state = ManagedCriticalState(0)
+    XCTAssertThrowsError(try state.withCriticalRegion { value in
+      value = 1
+      throw MockError(code: 1701)
+    }) { error in
+      XCTAssertEqual(error as? MockError, MockError(code: 1701))
+    }
+
+    let reacquired = expectation(description: "Another thread can acquire the lock after a throw")
+    DispatchQueue.global().async {
+      XCTAssertEqual(state.criticalState, 1)
+      state.apply(criticalState: 2)
+      XCTAssertEqual(state.criticalState, 2)
+      reacquired.fulfill()
+    }
+    await fulfillment(of: [reacquired], timeout: 2)
+  }
+}