Add parallel removal of items

Sources/NIOFileSystem/FileSystem.swift

@@ -438,29 +438,37 @@ public struct FileSystem: Sendable, FileSystemProtocol {
         }
     }
 
-    @discardableResult
-    private func removeItemSequentially(
-        at path: FilePath
-    ) async throws -> Int {
-        var (subdirectories, filesRemoved) = try await self.withDirectoryHandle(
+    func _collectItemsInDirectory(at path: FilePath) async throws -> ([FilePath], [FilePath]) {
+        try await self.withDirectoryHandle(
             atPath: path
         ) { directory in
             var subdirectories = [FilePath]()
-            var filesRemoved = 0
+            var nonDirectoryItems = [FilePath]()
 
             for try await batch in directory.listContents().batched() {
                 for entry in batch {
                     switch entry.type {
                     case .directory:
                         subdirectories.append(entry.path)
-
                     default:
-                        filesRemoved += try await self.removeOneItem(at: entry.path)
+                        nonDirectoryItems.append(entry.path)
                     }
                 }
             }
 
-            return (subdirectories, filesRemoved)
+            return (subdirectories, nonDirectoryItems)
+        }
+    }
+
+    @discardableResult
+    private func removeItemSequentially(
+        at path: FilePath
+    ) async throws -> Int {
+        var filesRemoved = 0
+        let (subdirectories, itemsToRemove) = try await self._collectItemsInDirectory(at: path)
+
+        for item in itemsToRemove {
+            filesRemoved += try await self.removeOneItem(at: item)
         }
 
         for subdirectory in subdirectories {

Sources/NIOFileSystem/Internal/Concurrency Primitives/TokenBucket.swift

@@ -33,7 +33,7 @@ import NIOConcurrencyHelpers
 /// Instead of using an ``actor``, we define a class and limit access through
 /// ``NIOLock``.
 @available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
-final class TokenBucket {
+final class TokenBucket: @unchecked Sendable {
     private var tokens: Int
     private var waiters: Deque<CheckedContinuation<Void, Never>>
     private let lock: NIOLock

Sources/NIOFileSystem/Internal/ParallelRemoval.swift

@@ -26,23 +26,7 @@ extension FileSystem {
         // Discover current directory and find all files/directories. Free up
         // the handle as fast as possible.
         let (directoriesToRecurseInto, itemsToDelete) = try await bucket.withToken {
-            try await self.withDirectoryHandle(atPath: path) { directory in
-                var subdirectories: [FilePath] = []
-                var itemsInDirectory: [FilePath] = []
-
-                for try await batch in directory.listContents().batched() {
-                    for entry in batch {
-                        switch entry.type {
-                        case .directory:
-                            subdirectories.append(entry.path)
-                        default:
-                            itemsInDirectory.append(entry.path)
-                        }
-                    }
-                }
-
-                return (subdirectories, itemsInDirectory)
-            }
+            try await self._collectItemsInDirectory(at: path)
         }
 
         return try await withThrowingTaskGroup(of: Int.self) { group in