Revert "Adopt NIOThrowingAsyncSequenceProducer (#2879)" (#2892)

This reverts commit 282f593.

### Motivation:

Repeated testing in linux containers has surfaced issues with the
implementation which require some rethinking of the code, let's revert
it for now and fix it up.

### Modifications:

Revert the adoption commit.

### Result:

The change is backed out.

Author: rnro

Sources/NIOFileSystem/DirectoryEntries.swift

@@ -16,7 +16,6 @@
 import CNIODarwin
 import CNIOLinux
 import NIOConcurrencyHelpers
-import NIOCore
 import NIOPosix
 @preconcurrency import SystemPackage
 
@@ -90,17 +89,17 @@ extension DirectoryEntries {
         public typealias AsyncIterator = BatchedIterator
         public typealias Element = [DirectoryEntry]
 
-        private let stream: BufferedOrAnyStream<[DirectoryEntry], DirectoryEntryProducer>
+        private let stream: BufferedOrAnyStream<[DirectoryEntry]>
 
         /// Creates a ``DirectoryEntries/Batched`` sequence by wrapping an `AsyncSequence`
         /// of directory entry batches.
         public init<S: AsyncSequence>(wrapping sequence: S) where S.Element == Element {
-            self.stream = BufferedOrAnyStream<[DirectoryEntry], DirectoryEntryProducer>(wrapping: sequence)
+            self.stream = BufferedOrAnyStream(wrapping: sequence)
         }
 
         fileprivate init(handle: SystemFileHandle, recursive: Bool) {
             // Expanding the batches yields watermarks of 256 and 512 directory entries.
-            let stream = NIOThrowingAsyncSequenceProducer.makeBatchedDirectoryEntryStream(
+            let stream = BufferedStream.makeBatchedDirectoryEntryStream(
                 handle: handle,
                 recursive: recursive,
                 entriesPerBatch: 64,
@@ -117,11 +116,9 @@ extension DirectoryEntries {
 
         /// An `AsyncIteratorProtocol` of `Array<DirectoryEntry>`.
         public struct BatchedIterator: AsyncIteratorProtocol {
-            private var iterator: BufferedOrAnyStream<[DirectoryEntry], DirectoryEntryProducer>.AsyncIterator
+            private var iterator: BufferedOrAnyStream<[DirectoryEntry]>.AsyncIterator
 
-            fileprivate init(
-                wrapping iterator: BufferedOrAnyStream<[DirectoryEntry], DirectoryEntryProducer>.AsyncIterator
-            ) {
+            init(wrapping iterator: BufferedOrAnyStream<[DirectoryEntry]>.AsyncIterator) {
                 self.iterator = iterator
             }
 
@@ -138,95 +135,52 @@ extension DirectoryEntries.Batched.AsyncIterator: Sendable {}
 // MARK: - Internal
 
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
-extension NIOThrowingAsyncSequenceProducer
-where
-    Element == [DirectoryEntry],
-    Failure == (any Error),
-    Strategy == NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark,
-    Delegate == DirectoryEntryProducer
-{
+extension BufferedStream where Element == [DirectoryEntry] {
     fileprivate static func makeBatchedDirectoryEntryStream(
         handle: SystemFileHandle,
         recursive: Bool,
         entriesPerBatch: Int,
         lowWatermark: Int,
         highWatermark: Int
-    ) -> NIOThrowingAsyncSequenceProducer<
-        [DirectoryEntry], any Error, NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark,
-        DirectoryEntryProducer
-    > {
-        let producer = DirectoryEntryProducer(
-            handle: handle,
-            recursive: recursive,
-            entriesPerBatch: entriesPerBatch
-        )
+    ) -> BufferedStream<[DirectoryEntry]> {
+        let state = DirectoryEnumerator(handle: handle, recursive: recursive)
+        let protectedState = NIOLockedValueBox(state)
 
-        let nioThrowingAsyncSequence = NIOThrowingAsyncSequenceProducer.makeSequence(
-            elementType: [DirectoryEntry].self,
-            backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark(
-                lowWatermark: lowWatermark,
-                highWatermark: highWatermark
-            ),
-            finishOnDeinit: false,
-            delegate: producer
+        var (stream, source) = BufferedStream.makeStream(
+            of: [DirectoryEntry].self,
+            backPressureStrategy: .watermark(low: lowWatermark, high: highWatermark)
         )
 
-        producer.setSequenceProducerSource(nioThrowingAsyncSequence.source)
+        source.onTermination = {
+            guard let threadPool = protectedState.withLockedValue({ $0.threadPoolForClosing() }) else {
+                return
+            }
+
+            threadPool.submit { _ in  // always run, even if cancelled
+                protectedState.withLockedValue { state in
+                    state.closeIfNecessary()
+                }
+            }
+        }
+
+        let producer = DirectoryEntryProducer(
+            state: protectedState,
+            source: source,
+            entriesPerBatch: entriesPerBatch
+        )
+        // Start producing immediately.
+        producer.produceMore()
 
-        return nioThrowingAsyncSequence.sequence
+        return stream
     }
 }
 
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
-private typealias DirectoryEntrySequenceProducer = NIOThrowingAsyncSequenceProducer<
-    [DirectoryEntry], Error, NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark, DirectoryEntryProducer
->
-
-@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
-private final class DirectoryEntryProducer: NIOAsyncSequenceProducerDelegate {
+private struct DirectoryEntryProducer {
     let state: NIOLockedValueBox<DirectoryEnumerator>
+    let source: BufferedStream<[DirectoryEntry]>.Source
     let entriesPerBatch: Int
 
-    init(handle: SystemFileHandle, recursive: Bool, entriesPerBatch: Int) {
-        let state = DirectoryEnumerator(handle: handle, recursive: recursive)
-        self.state = NIOLockedValueBox(state)
-        self.entriesPerBatch = entriesPerBatch
-    }
-
-    func didTerminate() {
-        guard let threadPool = self.state.withLockedValue({ $0.threadPoolForClosing() }) else {
-            return
-        }
-
-        threadPool.submit { _ in  // always run, even if cancelled
-            self.state.withLockedValue { state in
-                state.closeIfNecessary()
-            }
-        }
-    }
-
-    /// sets the source within the producer state
-    func setSequenceProducerSource(_ sequenceProducerSource: DirectoryEntrySequenceProducer.Source) {
-        self.state.withLockedValue { state in
-            switch state.state {
-            case .idle:
-                state.sequenceProducerSource = sequenceProducerSource
-            case .done:
-                sequenceProducerSource.finish()
-            case .open, .openPausedProducing:
-                fatalError()
-            case .modifying:
-                fatalError()
-            }
-        }
-    }
-
-    func clearSource() {
-        self.state.withLockedValue { state in
-            state.sequenceProducerSource = nil
-        }
-    }
-
     /// The 'entry point' for producing elements.
     ///
     /// Calling this function will start producing directory entries asynchronously by dispatching
@@ -253,12 +207,6 @@ private final class DirectoryEntryProducer: NIOAsyncSequenceProducerDelegate {
         }
     }
 
-    func pauseProducing() {
-        self.state.withLockedValue { state in
-            state.pauseProducing()
-        }
-    }
-
     private func nextBatch() throws -> [DirectoryEntry] {
         try self.state.withLockedValue { state in
             try state.next(self.entriesPerBatch)
@@ -273,51 +221,45 @@ private final class DirectoryEntryProducer: NIOAsyncSequenceProducerDelegate {
             // Failed to read more entries: close and notify the stream so consumers receive the
             // error.
             self.close()
-            let source = self.state.withLockedValue { state in
-                state.sequenceProducerSource
-            }
-            source?.finish(error)
-            self.clearSource()
+            self.source.finish(throwing: error)
         }
     }
 
     private func onNextBatch(_ entries: [DirectoryEntry]) {
-        let source = self.state.withLockedValue { state in
-            state.sequenceProducerSource
-        }
-
-        guard let source else {
-            assertionFailure("unexpectedly missing source")
-            return
-        }
-
         // No entries were read: this must be the end (as the batch size must be greater than zero).
         if entries.isEmpty {
-            source.finish()
-            self.clearSource()
+            self.source.finish(throwing: nil)
             return
         }
 
         // Reading short means reading EOF. The enumerator closes itself in that case.
         let readEOF = entries.count < self.entriesPerBatch
 
         // Entries were produced: yield them and maybe produce more.
-        let writeResult = source.yield(contentsOf: CollectionOfOne(entries))
-
-        // Exit early if EOF was read; no use in trying to produce more.
-        if readEOF {
-            source.finish()
-            self.clearSource()
-            return
-        }
+        do {
+            let writeResult = try self.source.write(contentsOf: CollectionOfOne(entries))
+            // Exit early if EOF was read; no use in trying to produce more.
+            if readEOF {
+                self.source.finish(throwing: nil)
+                return
+            }
 
-        switch writeResult {
-        case .produceMore:
-            self.produceMore()
-        case .stopProducing:
-            self.pauseProducing()
-        case .dropped:
-            // The source is finished; mark ourselves as done.
+            switch writeResult {
+            case .produceMore:
+                self.produceMore()
+            case let .enqueueCallback(token):
+                self.source.enqueueCallback(callbackToken: token) {
+                    switch $0 {
+                    case .success:
+                        self.produceMore()
+                    case .failure:
+                        self.close()
+                    }
+                }
+            }
+        } catch {
+            // Failure to write means the source is already done, that's okay we just need to
+            // update our state and stop producing.
             self.close()
         }
     }
@@ -340,30 +282,25 @@ private final class DirectoryEntryProducer: NIOAsyncSequenceProducerDelegate {
 /// Note that this is not a `Sequence` because we allow for errors to be thrown on `next()`.
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
 private struct DirectoryEnumerator: Sendable {
-    internal enum State: @unchecked Sendable {
+    private enum State: @unchecked Sendable {
         case modifying
         case idle(SystemFileHandle.SendableView, recursive: Bool)
         case open(NIOThreadPool, Source, [DirectoryEntry])
-        case openPausedProducing(NIOThreadPool, Source, [DirectoryEntry])
         case done
     }
 
     /// The source of directory entries.
-    internal enum Source {
+    private enum Source {
         case readdir(CInterop.DirPointer)
         case fts(CInterop.FTSPointer)
     }
 
     /// The current state of enumeration.
-    internal var state: State
+    private var state: State
 
     /// The path to the directory being enumerated.
     private let path: FilePath
 
-    /// The route via which directory entry batches are yielded,
-    /// the sourcing end of the `DirectoryEntrySequenceProducer`
-    internal var sequenceProducerSource: DirectoryEntrySequenceProducer.Source?
-
     /// Information about an entry returned by FTS. See 'fts(3)'.
     private enum FTSInfo: Hashable, Sendable {
         case directoryPreOrder
@@ -416,38 +353,22 @@ private struct DirectoryEnumerator: Sendable {
         self.path = handle.path
     }
 
-    internal mutating func produceMore() -> NIOThreadPool? {
+    internal func produceMore() -> NIOThreadPool? {
         switch self.state {
         case let .idle(handle, _):
             return handle.threadPool
         case let .open(threadPool, _, _):
             return threadPool
-        case .openPausedProducing(let threadPool, let source, let array):
-            self.state = .open(threadPool, source, array)
-            return threadPool
         case .done:
             return nil
         case .modifying:
             fatalError()
         }
     }
 
-    internal mutating func pauseProducing() {
-        switch self.state {
-        case .open(let threadPool, let source, let array):
-            self.state = .openPausedProducing(threadPool, source, array)
-        case .idle:
-            ()  // we won't apply back pressure until something has been read
-        case .openPausedProducing, .done:
-            ()  // no-op
-        case .modifying:
-            fatalError()
-        }
-    }
-
     internal func threadPoolForClosing() -> NIOThreadPool? {
         switch self.state {
-        case .open(let threadPool, _, _), .openPausedProducing(let threadPool, _, _):
+        case let .open(threadPool, _, _):
             return threadPool
         case .idle, .done:
             // Don't need to close in the idle state: we don't own the handle.
@@ -476,7 +397,7 @@ private struct DirectoryEnumerator: Sendable {
             // We don't own the handle so don't close it.
             self.state = .done
 
-        case .open(_, let mode, _), .openPausedProducing(_, let mode, _):
+        case let .open(_, mode, _):
             self.state = .done
             switch mode {
             case .readdir(let dir):
@@ -710,9 +631,6 @@ private struct DirectoryEnumerator: Sendable {
                 return result
             }
 
-        case .openPausedProducing:
-            return .yield(.success([]))
-
         case .done:
             return .yield(.success([]))