Connection pool

.gitignore

@@ -30,6 +30,7 @@ tests/sqlite/tfkpragma_selfref
 tests/sqlite/tfkpragmanotint
 tests/sqlite/tfkpragmanotmodel
 tests/sqlite/tnull
+tests/sqlite/tpool
 tests/sqlite/trelated
 tests/sqlite/trepeatfks
 tests/sqlite/trawselect

src/norm/pool.nim

@@ -0,0 +1,69 @@
+import std/locks
+
+import sqlite
+
+
+type 
+  Pool* = ref object 
+    defaultSize: Natural
+    conns: seq[DbConn]
+    poolExhaustedPolicy: PoolExhaustedPolicy
+    lock: Lock
+  PoolExhaustedError* = object of CatchableError
+  PoolExhaustedPolicy* = enum
+    pepRaise
+    pepExtend
+
+
+func newPool*(defaultSize: Positive, poolExhaustedPolicy = pepRaise): Pool =
+  result = Pool(defaultSize: defaultSize, conns: newSeq[DbConn](defaultSize), poolExhaustedPolicy: poolExhaustedPolicy)
+
+  initLock(result.lock)
+
+  for conn in result.conns.mitems:
+    conn = getDb()
+
+func defaultSize*(pool: Pool): Natural =
+  pool.defaultSize
+
+func size*(pool: Pool): Natural =
+  len(pool.conns)
+
+func pop*(pool: var Pool): DbConn =
+  withLock(pool.lock):
+    if pool.size > 0:
+      result = pool.conns.pop()
+    else:
+      case pool.poolExhaustedPolicy
+      of pepRaise:
+        raise newException(PoolExhaustedError, "Pool exhausted")
+      of pepExtend:
+        result = getDb()
+
+func add*(pool: var Pool, dbConn: DbConn) =
+  withLock(pool.lock):
+    pool.conns.add(dbConn)
+
+func reset*(pool: var Pool) =
+  withLock(pool.lock):
+    while pool.size > pool.defaultSize:
+      var conn = pool.conns.pop()
+      close conn
+
+func close*(pool: var Pool) =
+  withLock(pool.lock):
+    for conn in pool.conns.mitems:
+      close conn
+
+    pool.conns.setLen(0)
+
+template withDb*(pool: var Pool, body: untyped): untyped =
+  block:
+    let db {.inject.} = pool.pop()
+
+    try:
+      body
+
+    finally:
+      pool.add(db)
+

tests/config.nims

@@ -1,3 +1,4 @@
 switch("path", "$projectDir/../src")
 switch("define", "normDebug")
 switch("deepcopy", "on")
+switch("mm", "orc")

tests/sqlite/tenv.nim

@@ -42,3 +42,4 @@ suite "DB config from environment variables":
         @[?0, ?"price", ?"FLOAT", ?1, ?nil, ?0],
         @[?1, ?"id", ?"INTEGER", ?1, ?nil, ?1],
       ]
+

tests/sqlite/tpool.nim

@@ -0,0 +1,161 @@
+import std/[unittest, os, strutils]
+
+import norm/[sqlite, pool]
+
+import ../models
+
+const dbFile = "test.db"
+
+suite "Connection pool":
+  setup:
+    removeFile dbFile
+    putEnv(dbHostEnv, dbFile)
+
+  teardown:
+    delEnv(dbHostEnv)
+    removeFile dbFile
+
+  test "Create and close pool":
+    var pool = newPool(1)
+
+    check pool.defaultSize == 1
+    check pool.size == 1
+
+    close pool
+
+    check pool.size == 0
+
+  test "Explicit pool connection":
+    var pool = newPool(1)
+    let db = pool.pop()
+
+    db.createTables(newToy())
+
+    let qry = "PRAGMA table_info($#);"
+
+    check db.getAllRows(sql qry % "Toy") == @[
+      @[?0, ?"price", ?"FLOAT", ?1, ?nil, ?0],
+      @[?1, ?"id", ?"INTEGER", ?1, ?nil, ?1],
+    ]
+
+    pool.add(db)
+    close pool
+
+  test "Implicit pool connection":
+    var pool = newPool(1)
+
+    withDb(pool):
+      db.createTables(newToy())
+
+      let qry = "PRAGMA table_info($#);"
+
+      check db.getAllRows(sql qry % "Toy") == @[
+        @[?0, ?"price", ?"FLOAT", ?1, ?nil, ?0],
+        @[?1, ?"id", ?"INTEGER", ?1, ?nil, ?1],
+      ]
+
+    close pool
+
+  test "Concurrent pool connections":
+    var
+      pool = newPool(2)
+      toy1 = newToy(123.45)
+      toy2 = newToy(456.78)
+      threads: array[2, Thread[float]]
+      sum: float
+
+    withDb(pool):
+      db.createTables(toy1)
+      db.insert(toy1)
+      db.insert(toy2)
+
+    proc getToy(price: float) {.thread.} =
+      {.cast(gcsafe).}:
+        var toy = newToy()
+
+        withDb(pool):
+          db.select(toy, "price = ?", price)
+
+        sum += toy.price
+
+    createThread(threads[0], getToy, 123.45)
+    createThread(threads[1], getToy, 456.78)
+
+    joinThreads(threads)
+
+    check sum == toy1.price + toy2.price
+
+    close pool
+
+  test "Pool exhausted, raise exception":
+    var
+      pool = newPool(1, pepRaise)
+      toy1 = newToy(123.45)
+      toy2 = newToy(456.78)
+      threads: array[2, Thread[float]]
+      exceptionRaised: bool
+
+    withDb(pool):
+      db.createTables(toy1)
+      db.insert(toy1)
+      db.insert(toy2)
+
+    proc getToy(price: float) {.thread.} =
+      {.cast(gcsafe).}:
+        var toy = newToy()
+
+        try:
+          withDb(pool):
+            db.select(toy, "price = ?", price)
+            while not exceptionRaised:
+              sleep 100
+        except PoolExhaustedError:
+          exceptionRaised = true 
+
+    createThread(threads[0], getToy, 123.45)
+    createThread(threads[1], getToy, 456.78)
+
+    joinThreads(threads)
+
+    check exceptionRaised
+
+    close pool
+
+  test "Pool exhausted, extend and reset pool":
+    var
+      pool = newPool(1, pepExtend)
+      toy1 = newToy(123.45)
+      toy2 = newToy(456.78)
+      threads: array[2, Thread[float]]
+      maxActiveConnectionCount: Natural
+
+    withDb(pool):
+      db.createTables(toy1)
+      db.insert(toy1)
+      db.insert(toy2)
+
+    proc getToy(price: float) {.thread.} =
+      {.cast(gcsafe).}:
+        var toy = newToy()
+
+        withDb(pool):
+          inc maxActiveConnectionCount
+
+          db.select(toy, "price = ?", price)
+
+          while maxActiveConnectionCount < 2:
+            sleep 100
+
+    createThread(threads[0], getToy, 123.45)
+    createThread(threads[1], getToy, 456.78)
+
+    joinThreads(threads)
+
+    check maxActiveConnectionCount == 2
+
+    reset pool
+
+    check pool.size == pool.defaultSize
+
+    close pool
+

tests/sqlite/tpool.nims

@@ -0,0 +1,2 @@
+switch("threads", "on")
+

tests/sqlite/trows.nim

@@ -19,7 +19,7 @@ suite "Row CRUD":
     removeFile dbFile
 
   test "Insert row":
-    var toy = newtoy(123.45)
+    var toy = newToy(123.45)
 
     dbConn.insert(toy)
 
@@ -31,7 +31,7 @@ suite "Row CRUD":
     check rows[0] == @[?123.45, ?toy.id]
 
   test "Insert row twice":
-    var toy = newtoy(123.45)
+    var toy = newToy(123.45)
 
     dbConn.insert(toy)
     dbConn.insert(toy, force = true, conflictPolicy = cpReplace)
@@ -44,31 +44,31 @@ suite "Row CRUD":
     check rows[^1] == @[?123.45, ?toy.id]
 
   test "Insert with forced id":
-    var toy = newtoy(137.45)
+    var toy = newToy(137.45)
     toy.id = 134
     dbConn.insert(toy, force = true)
     check toy.id == 134
 
   test "Insert row with forced id in non-incremental order":
     block:
-      var toy = newtoy(123.45)
+      var toy = newToy(123.45)
       toy.id = 3
       dbConn.insert(toy, force=true)
       check toy.id == 3
     block:
-      var toy = newtoy(123.45)
+      var toy = newToy(123.45)
       toy.id = 2
       dbConn.insert(toy, force=true)
       check toy.id == 2
     block:
       # Check no id conflict
-      var toy = newtoy(123.45)
+      var toy = newToy(123.45)
       dbConn.insert(toy)
       # SQLite ids starts from the highest ?
       check toy.id == 4
     block:
       # Check no id conflict
-      var toy = newtoy(123.45)
+      var toy = newToy(123.45)
       dbConn.insert(toy)
       # SQLite ids starts from the highest ?
       check toy.id == 5