Repository: apple/swift-nio
File: Sources/NIOPosix/NIOThreadPool.swift (Swift)

Code before:
/// treated as a last resort.
///
/// - note: The prime example for missing non-blocking APIs is file IO on UNIX.
///   The OS does not provide a usable and truly non-blocking API but with
///   `NonBlockingFileIO` NIO provides a high-level API for file IO that should
///   be preferred to running blocking file IO system calls directly on
///   `NIOThreadPool`. Under the covers `NonBlockingFileIO` will use
///   `NIOThreadPool` on all currently supported platforms though.
public final class NIOThreadPool {

    /// The state of the `WorkItem`.
    public enum WorkItemState: Sendable {
        /// The `WorkItem` is active now and in process by the `NIOThreadPool`.
        case active
        /// The `WorkItem` was cancelled and will not be processed by the `NIOThreadPool`.
        case cancelled
    }

    /// The work that should be done by the `NIOThreadPool`.
    public typealias WorkItem = @Sendable (WorkItemState) -> Void

    /// `WorkItem`-s encapsulated in a reference type in order to avoid CoW when mutating the queue.
    /// Since `WorkItems` are embedded in `State` enum, it's hard to reliably avoid CoW otherwise.
    ///
    /// Also the closures are wrapped in a struct, to avoid the cost of allocation as discussed on https://bugs.swift.org/browse/SR-15872
    final class WorkItems {
        struct WorkItemBox {
            let item: WorkItem
        }

        private var items = CircularBuffer<WorkItemBox>(initialCapacity: 128)

        // Implement minimal, non-generic APIs required by `NIOThreadPool`
        func forEach(_ fn: (WorkItem) throws -> Void) rethrows {
            try self.items.forEach {
                try fn($0.item)
            }
        }

        func append(_ item: @escaping WorkItem) {
            self.items.append(WorkItemBox(item: item))
        }

        func removeFirst() -> WorkItem {
            return self.items.removeFirst().item
        }
    }

    private enum State {
        /// The `NIOThreadPool` is already stopped.
        case stopped

Review comment (suggestion):
Instead of boxing the buffer here with a class we should be able to avoid the CoW by introducing a new state called `modifying` into the `State` enum and before mutating the queue we transition to that state and then back to the running state. We have used that approach in most other places recently and it avoids this class entirely. 

The other upside of this pattern is that once we language gets `mutating/inout` switches in the future we can easily transition over to that and remove the `modifying` case again.

Code after:
/// APIs exist. In those cases `NIOThreadPool` can be used but should be
/// treated as a last resort.
///
/// - note: The prime example for missing non-blocking APIs is file IO on UNIX.
///   The OS does not provide a usable and truly non-blocking API but with
///   `NonBlockingFileIO` NIO provides a high-level API for file IO that should
///   be preferred to running blocking file IO system calls directly on
///   `NIOThreadPool`. Under the covers `NonBlockingFileIO` will use
///   `NIOThreadPool` on all currently supported platforms though.
public final class NIOThreadPool {

    /// The state of the `WorkItem`.
    public enum WorkItemState: Sendable {
        /// The `WorkItem` is active now and in process by the `NIOThreadPool`.
        case active
        /// The `WorkItem` was cancelled and will not be processed by the `NIOThreadPool`.
        case cancelled
    }

    /// The work that should be done by the `NIOThreadPool`.
    public typealias WorkItem = @Sendable (WorkItemState) -> Void
    private enum State {
        /// The `NIOThreadPool` is already stopped.
        case stopped
        /// The `NIOThreadPool` is shutting down, the array has one boolean entry for each thread indicating if it has shut down already.
        case shuttingDown([Bool])
        /// The `NIOThreadPool` is up and running, the `CircularBuffer` containing the yet unprocessed `WorkItems`.
        case running(Deque<WorkItem>)
        /// Temporary state used when mutating the .running(items). Used to avoid CoW copies.
        /// It should never be "leaked" outside of the lock block.
        case modifying
    }
    private let semaphore = DispatchSemaphore(value: 0)
    private let lock = NIOLock()
    private var threads: [NIOThread]? = nil  // protected by `lock`
    private var state: State = .stopped
    private let numberOfThreads: Int
    private let canBeStopped: Bool

    /// Gracefully shutdown this `NIOThreadPool`. All tasks will be run before shutdown will take place.
    ///
    /// - parameters:
    ///     - queue: The `DispatchQueue` used to executed the callback
    ///     - callback: The function to be executed once the shutdown is complete.
    @preconcurrency
    public func shutdownGracefully(queue: DispatchQueue, _ callback: @escaping @Sendable (Error?) -> Void) {
        self._shutdownGracefully(queue: queue, callback)
    }

    private func _shutdownGracefully(queue: DispatchQueue, _ callback: @escaping (Error?) -> Void) {
        guard self.canBeStopped else {