Shift responsibility for in-order message handling from Connection to SourceKitServer

This generally seems like the cleaner design because `SourceKitServer` is actually able to semantically inspect the message and decide whether it can be handled concurrently with other requests.

Author: ahoppen

Sources/LanguageServerProtocol/Connection.swift

@@ -46,15 +46,15 @@ public protocol MessageHandler: AnyObject {
   /// The method should return as soon as the notification has been sufficiently
   /// handled to avoid out-of-order requests, e.g. once the notification has
   /// been forwarded to clangd.
-  func handle(_ params: some NotificationType, from clientID: ObjectIdentifier) async
+  func handle(_ params: some NotificationType, from clientID: ObjectIdentifier)
 
   /// Handle a request and (asynchronously) receive a reply.
   ///
   /// The method should return as soon as the request has been sufficiently
   /// handled to avoid out-of-order requests, e.g. once the corresponding
   /// request has been sent to sourcekitd. The actual semantic computation
   /// should occur after the method returns and report the result via `reply`.
-  func handle<Request: RequestType>(_ params: Request, id: RequestID, from clientID: ObjectIdentifier, reply: @escaping (LSPResult<Request.Response>) -> Void) async
+  func handle<Request: RequestType>(_ params: Request, id: RequestID, from clientID: ObjectIdentifier, reply: @escaping (LSPResult<Request.Response>) -> Void)
 }
 
 /// A connection between two message handlers in the same process.
@@ -77,18 +77,7 @@ public final class LocalConnection {
 
   /// The queue guarding `_nextRequestID`.
   let queue: DispatchQueue = DispatchQueue(label: "local-connection-queue")
-
-  /// The queue on which all messages (notifications, requests, responses) are
-  /// handled.
-  ///
-  /// The queue is blocked until the message has been sufficiently handled to
-  /// avoid out-of-order handling of messages. For sourcekitd, this means that
-  /// a request has been sent to sourcekitd and for clangd, this means that we
-  /// have forwarded the request to clangd.
-  ///
-  /// The actual semantic handling of the message happens off this queue.
-  let messageHandlingQueue: AsyncQueue = AsyncQueue(.serial)
-
+  
   var _nextRequestID: Int = 0
 
   var state: State = .ready
@@ -125,9 +114,7 @@ public final class LocalConnection {
 
 extension LocalConnection: Connection {
   public func send<Notification>(_ notification: Notification) where Notification: NotificationType {
-    messageHandlingQueue.async {
-      await self.handler?.handle(notification, from: ObjectIdentifier(self))
-    }
+    self.handler?.handle(notification, from: ObjectIdentifier(self))
   }
 
   public func send<Request: RequestType>(
@@ -137,19 +124,17 @@ extension LocalConnection: Connection {
   ) -> RequestID {
     let id = nextRequestID()
 
-    messageHandlingQueue.async {
-      guard let handler = self.handler else {
-        queue.async {
-          reply(.failure(.serverCancelled))
-        }
-        return
+    guard let handler = self.handler else {
+      queue.async {
+        reply(.failure(.serverCancelled))
       }
+      return id
+    }
 
-      precondition(self.state == .started)
-      await handler.handle(request, id: id, from: ObjectIdentifier(self)) { result in
-        queue.async {
-          reply(result)
-        }
+    precondition(self.state == .started)
+    handler.handle(request, id: id, from: ObjectIdentifier(self)) { result in
+      queue.async {
+        reply(result)
       }
     }
 

Sources/LanguageServerProtocol/Message.swift

@@ -28,7 +28,7 @@ public protocol _RequestType: MessageType {
     id: RequestID,
     connection: Connection,
     reply: @escaping (LSPResult<ResponseType>, RequestID) -> Void
-  ) async
+  )
 }
 
 /// A request, which must have a unique `method` name as well as an associated response type.
@@ -54,16 +54,16 @@ extension RequestType {
     id: RequestID,
     connection: Connection,
     reply: @escaping (LSPResult<ResponseType>, RequestID) -> Void
-  ) async {
-    await handler.handle(self, id: id, from: ObjectIdentifier(connection)) { response in
+  ) {
+    handler.handle(self, id: id, from: ObjectIdentifier(connection)) { response in
       reply(response.map({ $0 as ResponseType }), id)
     }
   }
 }
 
 extension NotificationType {
-  public func _handle(_ handler: MessageHandler, connection: Connection) async {
-    await handler.handle(self, from: ObjectIdentifier(connection))
+  public func _handle(_ handler: MessageHandler, connection: Connection) {
+    handler.handle(self, from: ObjectIdentifier(connection))
   }
 }
 

Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift

@@ -31,16 +31,6 @@ public final class JSONRPCConnection {
   /// The queue on which we send data.
   let sendQueue: DispatchQueue = DispatchQueue(label: "jsonrpc-send-queue", qos: .userInitiated)
 
-  /// The queue on which all messages (notifications, requests, responses) are
-  /// handled.
-  ///
-  /// The queue is blocked until the message has been sufficiently handled to
-  /// avoid out-of-order handling of messages. For sourcekitd, this means that
-  /// a request has been sent to sourcekitd and for clangd, this means that we
-  /// have forwarded the request to clangd.
-  ///
-  /// The actual semantic handling of the message happens off this queue.
-  let messageHandlingQueue: AsyncQueue = AsyncQueue(.serial)
   let receiveIO: DispatchIO
   let sendIO: DispatchIO
   let messageRegistry: MessageRegistry
@@ -282,17 +272,14 @@ public final class JSONRPCConnection {
   func handle(_ message: JSONRPCMessage) {
     switch message {
     case .notification(let notification):
-      messageHandlingQueue.async {
-        await notification._handle(self.receiveHandler!, connection: self)
-      }
+      notification._handle(self.receiveHandler!, connection: self)
     case .request(let request, id: let id):
       let semaphore: DispatchSemaphore? = syncRequests ? .init(value: 0) : nil
-      messageHandlingQueue.async {
-        await request._handle(self.receiveHandler!, id: id, connection: self) { (response, id) in
-          self.sendReply(response, id: id)
-          semaphore?.signal()
-        }
+      request._handle(self.receiveHandler!, id: id, connection: self) { (response, id) in
+        self.sendReply(response, id: id)
+        semaphore?.signal()
       }
+
       semaphore?.wait()
 
     case .response(let response, id: let id):

Sources/SKCore/BuildServerBuildSystem.swift

@@ -51,6 +51,17 @@ public actor BuildServerBuildSystem: MessageHandler {
 
   var buildServer: JSONRPCConnection?
 
+  /// The queue on which all messages that originate from the build server are
+  /// handled.
+  ///
+  /// These are requests and notifications sent *from* the build server,
+  /// not replies from the build server.
+  ///
+  /// This ensures that messages from the build server are handled in the order
+  /// they were received. Swift concurrency does not guarentee in-order
+  /// execution of tasks.
+  public let bspMessageHandlingQueue = AsyncQueue(.serial)
+
   let searchPaths: [AbsolutePath]
 
   public private(set) var indexDatabasePath: AbsolutePath?
@@ -167,18 +178,20 @@ public actor BuildServerBuildSystem: MessageHandler {
   /// the build server has sent us a notification.
   ///
   /// We need to notify the delegate about any updated build settings.
-  public func handle(_ params: some NotificationType, from clientID: ObjectIdentifier) async {
-    if let params = params as? BuildTargetsChangedNotification {
-      await self.handleBuildTargetsChanged(Notification(params, clientID: clientID))
-    } else if let params = params as? FileOptionsChangedNotification {
-      await self.handleFileOptionsChanged(Notification(params, clientID: clientID))
+  public nonisolated func handle(_ params: some NotificationType, from clientID: ObjectIdentifier) {
+    bspMessageHandlingQueue.async {
+      if let params = params as? BuildTargetsChangedNotification {
+        await self.handleBuildTargetsChanged(Notification(params, clientID: clientID))
+      } else if let params = params as? FileOptionsChangedNotification {
+        await self.handleFileOptionsChanged(Notification(params, clientID: clientID))
+      }
     }
   }
 
   /// Handler for requests received **from** the build server.
   ///
   /// We currently can't handle any requests sent from the build server to us.
-  public func handle<R: RequestType>(
+  public nonisolated func handle<R: RequestType>(
     _ params: R,
     id: RequestID,
     from clientID: ObjectIdentifier,

Sources/SourceKitLSP/Clang/ClangLanguageServer.swift

@@ -45,6 +45,19 @@ actor ClangLanguageServerShim: ToolchainLanguageServer, MessageHandler {
   /// The queue on which clangd calls us back.
   public let clangdCommunicationQueue: DispatchQueue = DispatchQueue(label: "language-server-queue", qos: .userInitiated)
 
+  /// The queue on which all messages that originate from clangd are handled.
+  ///
+  /// This includes requests and notifications sent *from* clangd and does not
+  /// include replies from clangd.
+  ///
+  /// These are requests and notifications sent *from* clangd, not replies from
+  /// clangd.
+  ///
+  /// Since we are blindly forwarding requests from clangd to the editor, we
+  /// cannot allow concurrent requests. This should be fine since the number of
+  /// requests and notifications sent from clangd to the client is quite small.
+  public let clangdMessageHandlingQueue = AsyncQueue(.serial)
+
   /// The ``SourceKitServer`` instance that created this `ClangLanguageServerShim`.
   ///
   /// Used to send requests and notifications to the editor.
@@ -255,41 +268,41 @@ actor ClangLanguageServerShim: ToolchainLanguageServer, MessageHandler {
   /// sending a notification that's intended for the editor.
   ///
   /// We should either handle it ourselves or forward it to the editor.
-  func handle(_ params: some NotificationType, from clientID: ObjectIdentifier) async {
-    switch params {
-    case let publishDiags as PublishDiagnosticsNotification:
-      await self.publishDiagnostics(Notification(publishDiags, clientID: clientID))
-    default:
-      // We don't know how to handle any other notifications and ignore them.
-      log("Ignoring unknown notification \(type(of: params))", level: .warning)
-      break
+  nonisolated func handle(_ params: some NotificationType, from clientID: ObjectIdentifier) {
+    clangdMessageHandlingQueue.async {
+      switch params {
+      case let publishDiags as PublishDiagnosticsNotification:
+        await self.publishDiagnostics(Notification(publishDiags, clientID: clientID))
+      default:
+        // We don't know how to handle any other notifications and ignore them.
+        log("Ignoring unknown notification \(type(of: params))", level: .warning)
+        break
+      }
     }
   }
 
   /// Handler for requests received **from** clangd, ie. **clangd** is
   /// sending a notification that's intended for the editor.
   ///
   /// We should either handle it ourselves or forward it to the client.
-  func handle<R: RequestType>(
+  nonisolated func handle<R: RequestType>(
     _ params: R,
     id: RequestID,
     from clientID: ObjectIdentifier,
     reply: @escaping (LSPResult<R.Response>) -> Void
-  ) async {
-    let request = Request(params, id: id, clientID: clientID, cancellation: CancellationToken(), reply: { result in
-      reply(result)
-    })
-    guard let sourceKitServer else {
-      // `SourceKitServer` has been destructed. We are tearing down the language
-      // server. Nothing left to do.
-      request.reply(.failure(.unknown("Connection to the editor closed")))
-      return
-    }
+  ) {
+    clangdMessageHandlingQueue.async {
+      let request = Request(params, id: id, clientID: clientID, cancellation: CancellationToken(), reply: { result in
+        reply(result)
+      })
+      guard let sourceKitServer = await self.sourceKitServer else {
+        // `SourceKitServer` has been destructed. We are tearing down the language
+        // server. Nothing left to do.
+        request.reply(.failure(.unknown("Connection to the editor closed")))
+        return
+      }
 
-    if request.clientID == ObjectIdentifier(self.clangd) {
       await sourceKitServer.sendRequestToClient(request.params, reply: request.reply)
-    } else {
-      request.reply(.failure(ResponseError.methodNotFound(R.method)))
     }
   }