Implement connection limit.

Author: nickmshelley

SQLite/Core/ConnectionPool.swift

@@ -39,6 +39,7 @@ public final class ConnectionPool {
     private var unavailableReadConnections = [DirectConnection]()
     private let lockQueue : dispatch_queue_t
     private var writeConnection : DirectConnection!
+    private let connectionSemaphore = dispatch_semaphore_create(5)
 
     public var foreignKeys : Bool {
         get {
@@ -91,6 +92,7 @@ public final class ConnectionPool {
                     self.pool.unavailableReadConnections.removeAtIndex(index)
                 }
                 self.pool.availableReadConnections.append(self.connection)
+                dispatch_semaphore_signal(self.pool.connectionSemaphore)
             }
         }
 
@@ -151,44 +153,41 @@ public final class ConnectionPool {
 
         var borrowed : BorrowedConnection!
 
-        repeat {
+        dispatch_semaphore_wait(connectionSemaphore, DISPATCH_TIME_FOREVER)
+        dispatch_sync(lockQueue) {
 
-            dispatch_sync(lockQueue) {
-                
-                // Ensure database is open
-                self.writable
+            // Ensure database is open
+            self.writable
+            
+            let connection : DirectConnection
+            
+            if let availableConnection = self.availableReadConnections.popLast() {
+                connection = availableConnection
+            }
+            else {
 
-                let connection : DirectConnection
+                let flags = SQLITE_OPEN_READONLY | SQLITE_OPEN_WAL | SQLITE_OPEN_NOMUTEX
 
-                if let availableConnection = self.availableReadConnections.popLast() {
-                    connection = availableConnection
-                }
-                else {
-        
-                    let flags = SQLITE_OPEN_READONLY | SQLITE_OPEN_WAL | SQLITE_OPEN_NOMUTEX
-                    
-                    connection = try! DirectConnection(self.location, flags: flags, dispatcher: ImmediateDispatcher(), vfsName: vfsName)
-                    connection.busyTimeout = 2
+                connection = try! DirectConnection(self.location, flags: flags, dispatcher: ImmediateDispatcher(), vfsName: vfsName)
+                connection.busyTimeout = 2
 
-                    for (type, setupProcessor) in self.internalSetup {
-                        if type == .WriteAheadLogging {
-                            continue
-                        }
-                        try! setupProcessor(connection)
-                    }
-                    
-                    for setupProcessor in self.setup {
-                        try! setupProcessor(connection)
+                for (type, setupProcessor) in self.internalSetup {
+                    if type == .WriteAheadLogging {
+                        continue
                     }
-                    
+                    try! setupProcessor(connection)
                 }
 
-                self.unavailableReadConnections.append(connection)
+                for setupProcessor in self.setup {
+                    try! setupProcessor(connection)
+                }
 
-                borrowed = BorrowedConnection(pool: self, connection: connection)
             }
 
-        } while borrowed == nil
+            self.unavailableReadConnections.append(connection)
+            
+            borrowed = BorrowedConnection(pool: self, connection: connection)
+        }
 
         return borrowed
     }

SQLiteTests/ConnectionPoolTests.swift

@@ -22,33 +22,32 @@ class ConnectionPoolTests : SQLiteTestCase {
 
     func testConcurrentAccess2() {
 
+        let threadCount = 20
         let conn = pool.writable
         try! conn.execute("DROP TABLE IF EXISTS test; CREATE TABLE test(id INTEGER PRIMARY KEY, name TEXT);")
         try! conn.execute("DELETE FROM test")
-        try! conn.execute("INSERT INTO test(id,name) VALUES(0, 'test0')")
-        try! conn.execute("INSERT INTO test(id,name) VALUES(1, 'test1')")
-        try! conn.execute("INSERT INTO test(id,name) VALUES(2, 'test2')")
-        try! conn.execute("INSERT INTO test(id,name) VALUES(3, 'test3')")
-        try! conn.execute("INSERT INTO test(id,name) VALUES(4, 'test4')")
+        for threadNumber in 0..<threadCount {
+            try! conn.execute("INSERT INTO test(id,name) VALUES(\(threadNumber), 'test\(threadNumber)')")
+        }
 
         var quit = false
         let queue = dispatch_queue_create("Readers", DISPATCH_QUEUE_CONCURRENT)
-        for x in 0..<5 {
+        for threadNumber in 0..<threadCount {
             var reads = 0
 
-            let ex = expectationWithDescription("thread" + String(x))
+            let ex = expectationWithDescription("thread" + String(threadNumber))
 
             dispatch_async(queue) {
 
-                print("started", x)
+                print("started", threadNumber)
 
                 let conn = self.pool.readable
 
                 let stmt = try! conn.prepare("SELECT name FROM test WHERE id = ?")
-                var curr = stmt.scalar(x) as! String
+                var curr = stmt.scalar(threadNumber) as! String
                 while !quit {
 
-                    let now = stmt.scalar(x) as! String
+                    let now = stmt.scalar(threadNumber) as! String
                     if now != curr {
                         //print(now)
                         curr = now
@@ -90,7 +89,7 @@ class ConnectionPoolTests : SQLiteTestCase {
         let q = dispatch_queue_create("Readers/Writers", DISPATCH_QUEUE_CONCURRENT);
         var finished = false
 
-        for _ in 0..<5 {
+        for _ in 0..<20 {
 
             dispatch_async(q) {