Merge pull request #1720 from ahoppen/atomic-instead-of-queue

Use an `AtomicInt32` to count `pendingUnitCount` instead of using `AsyncQueue`

Author: ahoppen

Sources/CAtomics/include/CAtomics.h

@@ -44,4 +44,30 @@ static inline void atomic_uint32_destroy(CAtomicUInt32 *_Nonnull atomic) {
   free(atomic);
 }
 
+typedef struct {
+  _Atomic(int32_t) value;
+} CAtomicInt32;
+
+static inline CAtomicInt32 *_Nonnull atomic_int32_create(int32_t initialValue) {
+  CAtomicInt32 *atomic = malloc(sizeof(CAtomicInt32));
+  atomic->value = initialValue;
+  return atomic;
+}
+
+static inline int32_t atomic_int32_get(CAtomicInt32 *_Nonnull atomic) {
+  return atomic->value;
+}
+
+static inline void atomic_int32_set(CAtomicInt32 *_Nonnull atomic, int32_t newValue) {
+  atomic->value = newValue;
+}
+
+static inline int32_t atomic_int32_fetch_and_increment(CAtomicInt32 *_Nonnull atomic) {
+  return atomic->value++;
+}
+
+static inline void atomic_int32_destroy(CAtomicInt32 *_Nonnull atomic) {
+  free(atomic);
+}
+
 #endif // SOURCEKITLSP_CATOMICS_H

Sources/SourceKitLSP/SourceKitIndexDelegate.swift

@@ -18,60 +18,48 @@ import SwiftExtensions
 
 /// `IndexDelegate` for the SourceKit workspace.
 actor SourceKitIndexDelegate: IndexDelegate {
-
-  let queue = AsyncQueue<Serial>()
-
   /// Registered `MainFilesDelegate`s to notify when main files change.
   var mainFilesChangedCallbacks: [@Sendable () async -> Void] = []
 
   /// The count of pending unit events. Whenever this transitions to 0, it represents a time where
   /// the index finished processing known events. Of course, that may have already changed by the
   /// time we are notified.
-  var pendingUnitCount: Int = 0
+  let pendingUnitCount = AtomicInt32(initialValue: 0)
 
   package init() {}
 
   nonisolated package func processingAddedPending(_ count: Int) {
-    queue.async {
-      await self.addPending(count)
-    }
-  }
-
-  private func addPending(_ count: Int) {
-    pendingUnitCount += count
+    pendingUnitCount.value += Int32(count)
   }
 
   nonisolated package func processingCompleted(_ count: Int) {
-    queue.async {
-      await self.processCompleted(count)
-    }
-  }
-
-  private func processCompleted(_ count: Int) {
-    pendingUnitCount -= count
-    if pendingUnitCount == 0 {
-      indexChanged()
+    pendingUnitCount.value -= Int32(count)
+    if pendingUnitCount.value == 0 {
+      Task {
+        await indexChanged()
+      }
     }
 
-    if pendingUnitCount < 0 {
-      assertionFailure("pendingUnitCount = \(pendingUnitCount) < 0")
-      pendingUnitCount = 0
-      indexChanged()
+    if pendingUnitCount.value < 0 {
+      // Technically this is not data race safe because `pendingUnitCount` might change between the check and us setting
+      // it to 0. But then, this should never happen anyway, so it's fine.
+      logger.fault("pendingUnitCount dropped below zero: \(self.pendingUnitCount.value)")
+      pendingUnitCount.value = 0
+      Task {
+        await indexChanged()
+      }
     }
   }
 
-  private func indexChanged() {
+  private func indexChanged() async {
     logger.debug("IndexStoreDB changed")
     for callback in mainFilesChangedCallbacks {
-      queue.async {
-        await callback()
-      }
+      await callback()
     }
   }
 
   /// Register a delegate to receive notifications when main files change.
   package func addMainFileChangedCallback(_ callback: @escaping @Sendable () async -> Void) {
     mainFilesChangedCallbacks.append(callback)
   }
-
 }

Sources/SwiftExtensions/Atomics.swift

@@ -82,3 +82,28 @@ package final class AtomicUInt32: Sendable {
     return atomic_uint32_fetch_and_increment(atomic)
   }
 }
+
+package final class AtomicInt32: Sendable {
+  private nonisolated(unsafe) let atomic: UnsafeMutablePointer<CAtomicInt32>
+
+  package init(initialValue: Int32) {
+    self.atomic = atomic_int32_create(initialValue)
+  }
+
+  package var value: Int32 {
+    get {
+      atomic_int32_get(atomic)
+    }
+    set {
+      atomic_int32_set(atomic, newValue)
+    }
+  }
+
+  deinit {
+    atomic_int32_destroy(atomic)
+  }
+
+  package func fetchAndIncrement() -> Int32 {
+    return atomic_int32_fetch_and_increment(atomic)
+  }
+}