Add async-friendly WritableStream, create _AsyncFileSystem module (#7831)

Existing `WritableByteStream` protocol used by `AsyncProcess` is not
thread-safe. It's new `WritableStream` replacement is, with an added
conformance to the existing `OpenWritableFile`. That one is built on top
`AsyncFS`, which has been moved to the `Basics` module. That allows us
to adopt it in `AsyncProcess` in the future.

This is NFC, new API is adopted solely by `QueryEngine`, which is
exercised only in tests.

Author: MaxDesiatov

Package.swift

@@ -174,9 +174,23 @@ let package = Package(
 
         .systemLibrary(name: "SPMSQLite3", pkgConfig: systemSQLitePkgConfig),
 
+        .target(
+            name: "_AsyncFileSystem",
+            dependencies: [
+                .product(name: "SystemPackage", package: "swift-system"),
+            ],
+            exclude: ["CMakeLists.txt"],
+            swiftSettings: [
+                .enableExperimentalFeature("StrictConcurrency"),
+                .enableExperimentalFeature("AccessLevelOnImport"),
+                .enableExperimentalFeature("InternalImportsByDefault")
+            ]
+        ),
+
         .target(
             name: "Basics",
             dependencies: [
+                "_AsyncFileSystem",
                 "SPMSQLite3",
                 .product(name: "DequeModule", package: "swift-collections"),
                 .product(name: "OrderedCollections", package: "swift-collections"),
@@ -482,6 +496,7 @@ let package = Package(
         .target(
             name: "QueryEngine",
             dependencies: [
+                "_AsyncFileSystem",
                 "Basics",
                 .product(name: "Crypto", package: "swift-crypto"),
             ],
@@ -603,6 +618,14 @@ let package = Package(
 
         // MARK: SwiftPM tests
 
+        .testTarget(
+            name: "_AsyncFileSystemTests",
+            dependencies: [
+                "_AsyncFileSystem",
+                "_InternalTestSupport",
+            ]
+        ),
+
         .testTarget(
             name: "SourceKitLSPAPITests",
             dependencies: [

Sources/Basics/CMakeLists.txt

@@ -11,13 +11,13 @@ add_library(Basics
   Archiver/TarArchiver.swift
   Archiver/ZipArchiver.swift
   Archiver/UniversalArchiver.swift
-  AsyncProcess.swift
   AuthorizationProvider.swift
   Cancellator.swift
   Collections/ByteString+Extensions.swift
   Collections/Dictionary+Extensions.swift
   Collections/IdentifiableSet.swift
   Collections/String+Extensions.swift
+  Concurrency/AsyncProcess.swift
   Concurrency/ConcurrencyHelpers.swift
   Concurrency/NSLock+Extensions.swift
   Concurrency/SendableBox.swift
@@ -77,6 +77,7 @@ add_library(Basics
   Vendor/Triple.swift
   Vendor/Triple+Platforms.swift)
 target_link_libraries(Basics PUBLIC
+  _AsyncFileSystem
   SwiftCollections::DequeModule
   SwiftCollections::OrderedCollections
   SwiftSystem::SystemPackage

Sources/Basics/AsyncProcess.swift renamed to Sources/Basics/Concurrency/AsyncProcess.swift

@@ -452,7 +452,7 @@ package final class AsyncProcess {
     /// stdin. If needed, the stream can be closed using the close() API. Otherwise, the stream will be closed
     /// automatically.
     @discardableResult
-    package func launch() throws -> WritableByteStream {
+    package func launch() throws -> any WritableByteStream {
         precondition(
             self.arguments.count > 0 && !self.arguments[0].isEmpty,
             "Need at least one argument to launch the process."

Sources/Basics/Concurrency/ConcurrencyHelpers.swift

@@ -52,3 +52,17 @@ extension DispatchQueue {
 // As of Swift 5.7 and 5.8 swift-corelibs-foundation doesn't have `Sendable` annotations yet.
 extension URL: @unchecked Sendable {}
 #endif
+
+extension DispatchQueue {
+    package func scheduleOnQueue<T>(work: @escaping @Sendable () throws -> T) async throws -> T {
+        try await withCheckedThrowingContinuation { continuation in
+            self.async {
+                do {
+                    continuation.resume(returning: try work())
+                } catch {
+                    continuation.resume(throwing: error)
+                }
+            }
+        }
+    }
+}

Sources/CMakeLists.txt

@@ -1,13 +1,14 @@
 # This source file is part of the Swift open source project
 #
-# Copyright (c) 2014 - 2021 Apple Inc. and the Swift project authors
+# Copyright (c) 2014 - 2024 Apple Inc. and the Swift project authors
 # Licensed under Apache License v2.0 with Runtime Library Exception
 #
 # See http://swift.org/LICENSE.txt for license information
 # See http://swift.org/CONTRIBUTORS.txt for Swift project authors
 
 add_compile_definitions(USE_IMPL_ONLY_IMPORTS)
 
+add_subdirectory(_AsyncFileSystem)
 add_subdirectory(Basics)
 add_subdirectory(Build)
 add_subdirectory(Commands)

Sources/QueryEngine/FileSystem/FileCacheRecord.swift renamed to Sources/QueryEngine/FileCacheRecord.swift

@@ -10,6 +10,8 @@
 //
 //===----------------------------------------------------------------------===//
 
+import struct _AsyncFileSystem.OpenReadableFile
+
 // FIXME: need a new swift-system tag to remove `@preconcurrency`
 @preconcurrency import struct SystemPackage.FilePath
 
@@ -37,3 +39,13 @@ extension FileCacheRecord: Codable {
         try container.encode(self.hash, forKey: .hash)
     }
 }
+
+extension OpenReadableFile {
+    func hash(with hashFunction: inout some HashFunction) async throws {
+        let stream = try await self.read()
+
+        for try await bytes in stream {
+            hashFunction.update(data: bytes)
+        }
+    }
+}