Merge pull request #1382 from ahoppen/stream-index-log

Instead of sending a message to the index log when an indexing task finishes, stream results as they come in

Author: ahoppen

Sources/CAtomics/include/CAtomics.h

@@ -70,7 +70,7 @@ typedef struct {
 } AtomicUInt32;
 
 __attribute__((swift_name("AtomicUInt32.init(initialValue:)")))
-static inline AtomicUInt32 atomic_int_create(uint8_t initialValue) {
+static inline AtomicUInt32 atomic_int_create(uint32_t initialValue) {
   AtomicUInt32 atomic;
   atomic.value = initialValue;
   return atomic;

Sources/SKCore/BuildServerBuildSystem.swift

@@ -293,7 +293,7 @@ extension BuildServerBuildSystem: BuildSystem {
 
   public func prepare(
     targets: [ConfiguredTarget],
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws {
     throw PrepareNotSupportedError()
   }

Sources/SKCore/BuildSystem.swift

@@ -169,7 +169,7 @@ public protocol BuildSystem: AnyObject, Sendable {
   /// dependencies.
   func prepare(
     targets: [ConfiguredTarget],
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @escaping @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws
 
   /// If the build system has knowledge about the language that this document should be compiled in, return it.

Sources/SKCore/BuildSystemManager.swift

@@ -237,9 +237,9 @@ extension BuildSystemManager {
 
   public func prepare(
     targets: [ConfiguredTarget],
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @escaping @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws {
-    try await buildSystem?.prepare(targets: targets, indexProcessDidProduceResult: indexProcessDidProduceResult)
+    try await buildSystem?.prepare(targets: targets, logMessageToIndexLog: logMessageToIndexLog)
   }
 
   public func registerForChangeNotifications(for uri: DocumentURI, language: Language) async {

Sources/SKCore/CMakeLists.txt

@@ -11,7 +11,7 @@ add_library(SKCore STATIC
   ExperimentalFeatures.swift
   FallbackBuildSystem.swift
   FileBuildSettings.swift
-  IndexProcessResult.swift
+  IndexTaskID.swift
   MainFilesProvider.swift
   PathPrefixMapping.swift
   SplitShellCommand.swift

Sources/SKCore/CompilationDatabaseBuildSystem.swift

@@ -125,7 +125,7 @@ extension CompilationDatabaseBuildSystem: BuildSystem {
 
   public func prepare(
     targets: [ConfiguredTarget],
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws {
     throw PrepareNotSupportedError()
   }

Sources/SKCore/IndexProcessResult.swift

@@ -0,0 +1,46 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2014 - 2020 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import struct TSCBasic.ProcessResult
+
+/// The ID of a preparation or update indexstore task. This allows us to log messages from multiple concurrently running
+/// indexing tasks to the index log while still being able to differentiate them.
+public enum IndexTaskID: Sendable {
+  case preparation(id: UInt32)
+  case updateIndexStore(id: UInt32)
+
+  private static func numberToEmojis(_ number: Int, numEmojis: Int) -> String {
+    let emojis = ["🟥", "🟩", "🟦", "🟧", "⬜️", "🟪", "⬛️", "🟨", "🟫"]
+    var number = abs(number)
+    var result = ""
+    for _ in 0..<numEmojis {
+      let (quotient, remainder) = number.quotientAndRemainder(dividingBy: emojis.count)
+      result += emojis[remainder]
+      number = quotient
+    }
+    return result
+  }
+
+  /// Returns a two-character emoji string that allows easy differentiation between different task IDs.
+  ///
+  /// This marker is prepended to every line in the index log.
+  public var emojiRepresentation: String {
+    // Multiply by 2 and optionally add 1 to make sure preparation and update index store have distinct IDs.
+    // Run .hashValue to make sure we semi-randomly pick new emoji markers for new tasks
+    switch self {
+    case .preparation(id: let id):
+      return Self.numberToEmojis((id * 2).hashValue, numEmojis: 2)
+    case .updateIndexStore(id: let id):
+      return Self.numberToEmojis((id * 2 + 1).hashValue, numEmojis: 2)
+    }
+  }
+}

Sources/SKCore/IndexTaskID.swift

@@ -11,6 +11,7 @@ add_library(SKSupport STATIC
   DocumentURI+CustomLogStringConvertible.swift
   FileSystem.swift
   LineTable.swift
+  PipeAsStringHandler.swift
   Process+LaunchWithWorkingDirectoryIfPossible.swift
   Process+WaitUntilExitWithCancellation.swift
   Random.swift

Sources/SKSupport/CMakeLists.txt

@@ -0,0 +1,49 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the Swift.org open source project
+//
+// Copyright (c) 2014 - 2024 Apple Inc. and the Swift project authors
+// Licensed under Apache License v2.0 with Runtime Library Exception
+//
+// See https://swift.org/LICENSE.txt for license information
+// See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
+//
+//===----------------------------------------------------------------------===//
+
+import Foundation
+
+/// Gathers data from a stdout or stderr pipe. When it has accumulated a full line, calls the handler to handle the
+/// string.
+public actor PipeAsStringHandler {
+  /// Queue on which all data from the pipe will be handled. This allows us to have a
+  /// nonisolated `handle` function but ensure that data gets processed in order.
+  private let queue = AsyncQueue<Serial>()
+  private var buffer = Data()
+
+  /// The closure that actually handles
+  private let handler: @Sendable (String) -> Void
+
+  public init(handler: @escaping @Sendable (String) -> Void) {
+    self.handler = handler
+  }
+
+  private func handleDataFromPipeImpl(_ newData: Data) {
+    self.buffer += newData
+    while let newlineIndex = self.buffer.firstIndex(of: UInt8(ascii: "\n")) {
+      // Output a separate log message for every line in the pipe.
+      // The reason why we don't output multiple lines in a single log message is that
+      //  a) os_log truncates log messages at about 1000 bytes. The assumption is that a single line is usually less
+      //     than 1000 bytes long but if we merge multiple lines into one message, we might easily exceed this limit.
+      //  b) It might be confusing why sometimes a single log message contains one line while sometimes it contains
+      //     multiple.
+      handler(String(data: self.buffer[...newlineIndex], encoding: .utf8) ?? "<invalid UTF-8>")
+      buffer = buffer[buffer.index(after: newlineIndex)...]
+    }
+  }
+
+  public nonisolated func handleDataFromPipe(_ newData: Data) {
+    queue.async {
+      await self.handleDataFromPipeImpl(newData)
+    }
+  }
+}

Sources/SKSupport/PipeAsStringHandler.swift

@@ -26,6 +26,7 @@ extension Process {
     arguments: [String],
     environmentBlock: ProcessEnvironmentBlock = ProcessEnv.block,
     workingDirectory: AbsolutePath?,
+    outputRedirection: OutputRedirection = .collect,
     startNewProcessGroup: Bool = true,
     loggingHandler: LoggingHandler? = .none
   ) throws -> Process {
@@ -35,13 +36,15 @@ extension Process {
           arguments: arguments,
           environmentBlock: environmentBlock,
           workingDirectory: workingDirectory,
+          outputRedirection: outputRedirection,
           startNewProcessGroup: startNewProcessGroup,
           loggingHandler: loggingHandler
         )
       } else {
         Process(
           arguments: arguments,
           environmentBlock: environmentBlock,
+          outputRedirection: outputRedirection,
           startNewProcessGroup: startNewProcessGroup,
           loggingHandler: loggingHandler
         )
@@ -57,6 +60,7 @@ extension Process {
         arguments: arguments,
         environmentBlock: environmentBlock,
         workingDirectory: nil,
+        outputRedirection: outputRedirection,
         startNewProcessGroup: startNewProcessGroup,
         loggingHandler: loggingHandler
       )

Sources/SKSupport/Process+LaunchWithWorkingDirectoryIfPossible.swift

@@ -13,7 +13,9 @@
 import Basics
 import Build
 import BuildServerProtocol
+import CAtomics
 import Dispatch
+import Foundation
 import LSPLogging
 import LanguageServerProtocol
 import PackageGraph
@@ -71,6 +73,9 @@ fileprivate extension ConfiguredTarget {
   static let forPackageManifest = ConfiguredTarget(targetID: "", runDestinationID: "")
 }
 
+/// `nonisolated(unsafe)` is fine because `preparationTaskID` is atomic.
+fileprivate nonisolated(unsafe) var preparationTaskID: AtomicUInt32 = AtomicUInt32(initialValue: 0)
+
 /// Swift Package Manager build system and workspace support.
 ///
 /// This class implements the `BuildSystem` interface to provide the build settings for a Swift
@@ -516,20 +521,20 @@ extension SwiftPMBuildSystem: SKCore.BuildSystem {
 
   public func prepare(
     targets: [ConfiguredTarget],
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @escaping @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws {
     // TODO (indexing): Support preparation of multiple targets at once.
     // https://github.com/apple/sourcekit-lsp/issues/1262
     for target in targets {
-      try await prepare(singleTarget: target, indexProcessDidProduceResult: indexProcessDidProduceResult)
+      try await prepare(singleTarget: target, logMessageToIndexLog: logMessageToIndexLog)
     }
     let filesInPreparedTargets = targets.flatMap { self.targets[$0]?.buildTarget.sources ?? [] }
     await fileDependenciesUpdatedDebouncer.scheduleCall(Set(filesInPreparedTargets.map(DocumentURI.init)))
   }
 
   private func prepare(
     singleTarget target: ConfiguredTarget,
-    indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+    logMessageToIndexLog: @escaping @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
   ) async throws {
     if target == .forPackageManifest {
       // Nothing to prepare for package manifests.
@@ -586,15 +591,28 @@ extension SwiftPMBuildSystem: SKCore.BuildSystem {
       return
     }
     let start = ContinuousClock.now
-    let process = try Process.launch(arguments: arguments, workingDirectory: nil)
-    let result = try await process.waitUntilExitSendingSigIntOnTaskCancellation()
-    indexProcessDidProduceResult(
-      IndexProcessResult(
-        taskDescription: "Preparing \(target.targetID) for \(target.runDestinationID)",
-        processResult: result,
-        start: start
+
+    let logID = IndexTaskID.preparation(id: preparationTaskID.fetchAndIncrement())
+    logMessageToIndexLog(
+      logID,
+      """
+      Preparing \(target.targetID) for \(target.runDestinationID)
+      \(arguments.joined(separator: " "))
+      """
+    )
+    let stdoutHandler = PipeAsStringHandler { logMessageToIndexLog(logID, $0) }
+    let stderrHandler = PipeAsStringHandler { logMessageToIndexLog(logID, $0) }
+
+    let process = try Process.launch(
+      arguments: arguments,
+      workingDirectory: nil,
+      outputRedirection: .stream(
+        stdout: { stdoutHandler.handleDataFromPipe(Data($0)) },
+        stderr: { stderrHandler.handleDataFromPipe(Data($0)) }
       )
     )
+    let result = try await process.waitUntilExitSendingSigIntOnTaskCancellation()
+    logMessageToIndexLog(logID, "Finished in \(start.duration(to: .now))")
     switch result.exitStatus.exhaustivelySwitchable {
     case .terminated(code: 0):
       break

Sources/SKSwiftPMWorkspace/SwiftPMBuildSystem.swift

@@ -258,12 +258,12 @@ public final class TestSourceKitLSPClient: MessageHandler {
   /// Ignores any notifications that are of a different type or that don't satisfy the predicate.
   public func nextNotification<ExpectedNotificationType: NotificationType>(
     ofType: ExpectedNotificationType.Type,
-    satisfying predicate: (ExpectedNotificationType) -> Bool = { _ in true },
+    satisfying predicate: (ExpectedNotificationType) throws -> Bool = { _ in true },
     timeout: Duration = .seconds(defaultTimeout)
   ) async throws -> ExpectedNotificationType {
     while true {
       let nextNotification = try await nextNotification(timeout: timeout)
-      if let notification = nextNotification as? ExpectedNotificationType, predicate(notification) {
+      if let notification = nextNotification as? ExpectedNotificationType, try predicate(notification) {
         return notification
       }
     }

Sources/SKTestSupport/TestSourceKitLSPClient.swift

@@ -37,8 +37,8 @@ public struct PreparationTaskDescription: IndexTaskDescription {
 
   private let preparationUpToDateTracker: UpToDateTracker<ConfiguredTarget>
 
-  /// See `SemanticIndexManager.indexProcessDidProduceResult`
-  private let indexProcessDidProduceResult: @Sendable (IndexProcessResult) -> Void
+  /// See `SemanticIndexManager.logMessageToIndexLog`.
+  private let logMessageToIndexLog: @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void
 
   /// Test hooks that should be called when the preparation task finishes.
   private let testHooks: IndexTestHooks
@@ -60,13 +60,13 @@ public struct PreparationTaskDescription: IndexTaskDescription {
     targetsToPrepare: [ConfiguredTarget],
     buildSystemManager: BuildSystemManager,
     preparationUpToDateTracker: UpToDateTracker<ConfiguredTarget>,
-    indexProcessDidProduceResult: @escaping @Sendable (IndexProcessResult) -> Void,
+    logMessageToIndexLog: @escaping @Sendable (_ taskID: IndexTaskID, _ message: String) -> Void,
     testHooks: IndexTestHooks
   ) {
     self.targetsToPrepare = targetsToPrepare
     self.buildSystemManager = buildSystemManager
     self.preparationUpToDateTracker = preparationUpToDateTracker
-    self.indexProcessDidProduceResult = indexProcessDidProduceResult
+    self.logMessageToIndexLog = logMessageToIndexLog
     self.testHooks = testHooks
   }
 
@@ -105,7 +105,7 @@ public struct PreparationTaskDescription: IndexTaskDescription {
       do {
         try await buildSystemManager.prepare(
           targets: targetsToPrepare,
-          indexProcessDidProduceResult: indexProcessDidProduceResult
+          logMessageToIndexLog: logMessageToIndexLog
         )
       } catch {
         logger.error(