improve RepositoryManager::lookup

motivation: improved concurency control and cleaer code

changes:
* use a seperate queue to schedule the lookup tasks to ensure oncurrency control is done correctly
* use Result::catching to help remove boilerplate error handling

Author: tomerd

Sources/SourceControl/RepositoryManager.swift

@@ -34,7 +34,9 @@ public class RepositoryManager {
     private let delegate: Delegate?
 
     /// DispatchSemaphore to restrict concurrent operations on manager.
-    private let lookupSemaphore: DispatchSemaphore
+    private let concurrencySemaphore: DispatchSemaphore
+    /// DispatchQueue to park pending lookups
+    private let lookupQueue = DispatchQueue(label: "org.swift.swiftpm.repository-manager", attributes: .concurrent)
 
     /// The filesystem to operate on.
     private let fileSystem: FileSystem
@@ -71,7 +73,7 @@ public class RepositoryManager {
         self.provider = provider
         self.delegate = delegate
 
-        self.lookupSemaphore = DispatchSemaphore(value: Swift.min(3, Concurrency.maxOperations))
+        self.concurrencySemaphore = DispatchSemaphore(value: Swift.min(3, Concurrency.maxOperations))
     }
 
     /// Get a handle to a repository.
@@ -98,24 +100,30 @@ public class RepositoryManager {
         callbackQueue: DispatchQueue,
         completion: @escaping (Result<RepositoryHandle, Error>) -> Void
     ) {
-        // wrap the callback in the requested queue
+        // wrap the callback in the requested queue and cleanup operations
         let originalCompletion = completion
         let completion: (Result<RepositoryHandle, Error>) -> Void = { result in
-            self.lookupSemaphore.signal()
+            // free concurrency control semaphore
+            self.concurrencySemaphore.signal()
+            // remove any pending lookup
+            self.pendingLookupsLock.lock()
+            self.pendingLookups[repository]?.leave()
+            self.pendingLookups[repository] = nil
+            self.pendingLookupsLock.unlock()
+            // call back on the request queue
             callbackQueue.async { originalCompletion(result) }
         }
 
-        self.lookupSemaphore.wait()
         let relativePath = repository.storagePath()
         let repositoryPath = self.path.appending(relativePath)
-        let handle = RepositoryManager.RepositoryHandle(manager: self, repository: repository, subpath: relativePath)
+        let handle = RepositoryHandle(manager: self, repository: repository, subpath: relativePath)
 
         // check if there is a pending lookup
         self.pendingLookupsLock.lock()
         if let pendingLookup = self.pendingLookups[repository] {
             self.pendingLookupsLock.unlock()
             // chain onto the pending lookup
-            return pendingLookup.notify(queue: callbackQueue) {
+            return pendingLookup.notify(queue: self.lookupQueue) {
                 // at this point the previous lookup should be complete and we can re-lookup
                 self.lookup(
                     package: package,
@@ -127,91 +135,77 @@ public class RepositoryManager {
                     completion: originalCompletion
                 )
             }
+        } else {
+            // record the pending lookup
+            assert(self.pendingLookups[repository] == nil)
+            let group = DispatchGroup()
+            group.enter()
+            self.pendingLookups[repository] = group
+            self.pendingLookupsLock.unlock()
         }
 
-        // record the pending lookup
-        assert(self.pendingLookups[repository] == nil)
-        let group = DispatchGroup()
-        group.enter()
-        self.pendingLookups[repository] = group
-        self.pendingLookupsLock.unlock()
-
-        // check if a repository already exists
-        // errors when trying to check if a repository already exists are legitimate
-        // and recoverable, and as such can be ignored
-        if (try? self.provider.repositoryExists(at: repositoryPath)) ?? false {
-            let result = Result<RepositoryHandle, Error>(catching: {
-                // skip update if not needed
-                if skipUpdate {
+        // we must not block the calling thread (for concurrency control) so nesting this in a queue
+        self.lookupQueue.async {
+            // park the lookup thread based on the max concurrency allowed
+            self.concurrencySemaphore.wait()
+
+            // check if a repository already exists
+            // errors when trying to check if a repository already exists are legitimate
+            // and recoverable, and as such can be ignored
+            if (try? self.provider.repositoryExists(at: repositoryPath)) ?? false {
+                // update if necessary and return early
+                return completion(.init(catching: {
+                    // skip update if not needed
+                    if skipUpdate {
+                        return handle
+                    }
+                    // Update the repository when it is being looked up.
+                    let start = DispatchTime.now()
+                    delegateQueue.async {
+                        self.delegate?.willUpdate(package: package, repository: handle.repository)
+                    }
+                    let repository = try handle.open()
+                    try repository.fetch()
+                    let duration = start.distance(to: .now())
+                    delegateQueue.async {
+                        self.delegate?.didUpdate(package: package, repository: handle.repository, duration: duration)
+                    }
                     return handle
-                }
-                // Update the repository when it is being looked up.
-                let start = DispatchTime.now()
-                delegateQueue.async {
-                    self.delegate?.willUpdate(package: package, repository: handle.repository)
-                }
-                let repository = try handle.open()
-                try repository.fetch()
-                let duration = start.distance(to: .now())
-                delegateQueue.async {
-                    self.delegate?.didUpdate(package: package, repository: handle.repository, duration: duration)
-                }
-                return handle
-            })
-
-            // remove the pending lookup
-            self.pendingLookupsLock.lock()
-            self.pendingLookups[repository]?.leave()
-            self.pendingLookups[repository] = nil
-            self.pendingLookupsLock.unlock()
-            // and done
-            return completion(result)
-        }
+                }))
+            }
 
-        // perform the fetch
-        // inform delegate that we are starting to fetch
-        // calculate if cached (for delegate call) outside queue as it may change while queue is processing
-        let isCached = self.cachePath.map{ self.fileSystem.exists($0.appending(handle.subpath)) } ?? false
-        delegateQueue.async {
-            let details = FetchDetails(fromCache: isCached, updatedCache: false)
-            self.delegate?.willFetch(package: package, repository: handle.repository, details: details)
-        }
+            // inform delegate that we are starting to fetch
+            // calculate if cached (for delegate call) outside queue as it may change while queue is processing
+            let isCached = self.cachePath.map{ self.fileSystem.exists($0.appending(handle.subpath)) } ?? false
+            delegateQueue.async {
+                let details = FetchDetails(fromCache: isCached, updatedCache: false)
+                self.delegate?.willFetch(package: package, repository: handle.repository, details: details)
+            }
 
-        let start = DispatchTime.now()
-        let lookupResult: Result<RepositoryHandle, Error>
-        let delegateResult: Result<FetchDetails, Error>
+            // perform the fetch
+            let start = DispatchTime.now()
+            let result = Result<FetchDetails, Error>(catching: {
+                // make sure destination is free.
+                try? self.fileSystem.removeFileTree(repositoryPath)
+                // Fetch the repo.
+                return try self.fetchAndPopulateCache(
+                    package: package,
+                    handle: handle,
+                    repositoryPath: repositoryPath,
+                    observabilityScope: observabilityScope,
+                    delegateQueue: delegateQueue
+                )
+            })
 
-        do {
-            // make sure destination is free.
-            try? self.fileSystem.removeFileTree(repositoryPath)
-            // Fetch the repo.
-            let details = try self.fetchAndPopulateCache(
-                package: package,
-                handle: handle,
-                repositoryPath: repositoryPath,
-                observabilityScope: observabilityScope,
-                delegateQueue: delegateQueue
-            )
-            lookupResult = .success(handle)
-            delegateResult = .success(details)
-        } catch {
-            lookupResult = .failure(error)
-            delegateResult = .failure(error)
-        }
+            // Inform delegate fetch is done
+            let duration = start.distance(to: .now())
+            delegateQueue.async {
+                self.delegate?.didFetch(package: package, repository: handle.repository, result: result, duration: duration)
+            }
 
-        // Inform delegate.
-        let duration = start.distance(to: .now())
-        delegateQueue.async {
-            self.delegate?.didFetch(package: package, repository: handle.repository, result: delegateResult, duration: duration)
+            // and done
+            completion(result.map{ _ in handle })
         }
-
-        // remove the pending lookup
-        self.pendingLookupsLock.lock()
-        self.pendingLookups[repository]?.leave()
-        self.pendingLookups[repository] = nil
-        self.pendingLookupsLock.unlock()
-        // and done
-        completion(lookupResult)
     }
 
     /// Fetches the repository into the cache. If no `cachePath` is set or an error occurred fall back to fetching the repository without populating the cache.