Fix failing tests for number of threads created by DatabasePool

Author: groue

GRDB/Core/DatabasePool.swift

@@ -330,22 +330,14 @@ extension DatabasePool: DatabaseReader {
     }
 
     public func asyncRead(_ value: @escaping (Result<Database, Error>) -> Void) {
-        // First async jump in order to grab a reader connection.
-        // Honor configuration dispatching (qos/targetQueue).
-        let label = configuration.identifier(
-            defaultLabel: "GRDB.DatabasePool",
-            purpose: "asyncRead")
-        configuration
-            .makeReaderDispatchQueue(label: label)
-            .async {
+        do {
+            guard let readerPool = self.readerPool else {
+                throw DatabaseError(resultCode: .SQLITE_MISUSE, message: "Connection is closed")
+            }
+            readerPool.async { result in
                 do {
-                    guard let readerPool = self.readerPool else {
-                        throw DatabaseError.connectionIsClosed()
-                    }
-                    let (reader, releaseReader) = try readerPool.get()
-                    
-                    // Second async jump because sync could deadlock if
-                    // configuration has a serial targetQueue.
+                    let (reader, releaseReader) = try result.get()
+                    // Second async jump because that's how `Pool.async` has to be used.
                     reader.async { db in
                         defer {
                             try? db.commit() // Ignore commit error
@@ -364,6 +356,9 @@ extension DatabasePool: DatabaseReader {
                     value(.failure(error))
                 }
             }
+        } catch {
+            value(.failure(error))
+        }
     }
 
     @_disfavoredOverload // SR-15150 Async overloading in protocol implementation fails
@@ -381,22 +376,14 @@ extension DatabasePool: DatabaseReader {
     }
 
     public func asyncUnsafeRead(_ value: @escaping (Result<Database, Error>) -> Void) {
-        // First async jump in order to grab a reader connection.
-        // Honor configuration dispatching (qos/targetQueue).
-        let label = configuration.identifier(
-            defaultLabel: "GRDB.DatabasePool",
-            purpose: "asyncUnsafeRead")
-        configuration
-            .makeReaderDispatchQueue(label: label)
-            .async {
+        do {
+            guard let readerPool = self.readerPool else {
+                throw DatabaseError(resultCode: .SQLITE_MISUSE, message: "Connection is closed")
+            }
+            readerPool.async { result in
                 do {
-                    guard let readerPool = self.readerPool else {
-                        throw DatabaseError(resultCode: .SQLITE_MISUSE, message: "Connection is closed")
-                    }
-                    let (reader, releaseReader) = try readerPool.get()
-                    
-                    // Second async jump because sync could deadlock if
-                    // configuration has a serial targetQueue.
+                    let (reader, releaseReader) = try result.get()
+                    // Second async jump because that's how `Pool.async` has to be used.
                     reader.async { db in
                         defer {
                             releaseReader()
@@ -413,6 +400,9 @@ extension DatabasePool: DatabaseReader {
                     value(.failure(error))
                 }
             }
+        } catch {
+            value(.failure(error))
+        }
     }
 
     public func unsafeReentrantRead<T>(_ value: (Database) throws -> T) throws -> T {

GRDB/Utils/Pool.swift

@@ -51,13 +51,15 @@ final class Pool<T> {
     private let itemsSemaphore: DispatchSemaphore // limits the number of elements
     private let itemsGroup: DispatchGroup         // knows when no element is used
     private let barrierQueue: DispatchQueue
+    private let semaphoreWaitingQueue: DispatchQueue // Inspired by https://khanlou.com/2016/04/the-GCD-handbook/
 
     init(maximumCount: Int, makeElement: @escaping () throws -> T) {
         GRDBPrecondition(maximumCount > 0, "Pool size must be at least 1")
         self.makeElement = makeElement
         self.itemsSemaphore = DispatchSemaphore(value: maximumCount)
         self.itemsGroup = DispatchGroup()
         self.barrierQueue = DispatchQueue(label: "GRDB.Pool.barrier", attributes: [.concurrent])
+        self.semaphoreWaitingQueue = DispatchQueue(label: "GRDB.Pool.wait")
     }
 
     /// Returns a tuple (element, release)
@@ -87,6 +89,25 @@ final class Pool<T> {
         }
     }
 
+    /// Eventually produces a tuple (element, release), where element is
+    /// intended to be used asynchronously.
+    ///
+    /// Client must call release(), only once, after the element has been used.
+    ///
+    /// - important: The `execute` argument is executed in a serial dispatch
+    ///   queue, so make sure you use the element asynchronously.
+    func async(_ execute: @escaping (Result<(element: T, release: () -> Void), Error>) -> Void) {
+        // Inspired by https://khanlou.com/2016/04/the-GCD-handbook/
+        // > We wait on the semaphore in the serial queue, which means that
+        // > we’ll have at most one blocked thread when we reach maximum
+        // > executing blocks on the concurrent queue. Any other tasks the user
+        // > enqueues will sit inertly on the serial queue waiting to be
+        // > executed, and won’t cause new threads to be started.
+        semaphoreWaitingQueue.async {
+            execute(Result { try self.get() })
+        }
+    }
+    
     /// Performs a synchronous block with an element. The element turns
     /// available after the block has executed.
     func get<U>(block: (T) throws -> U) throws -> U {