Threaded IO datastore with minimal modifications

.github/workflows/tests.yml

@@ -13,7 +13,7 @@ jobs:
       fail-fast: false
       matrix:
         cache_nonce: [ 1 ]
-        nim_version: [ 1.2.18, 1.6.6 ]
+        nim_version: [ 1.6.14 ]
         platform:
           - {
             icon: 🐧,

.gitignore

@@ -9,3 +9,6 @@ coverage
 datastore.nims
 nimcache
 TODO
+nim.cfg
+nimble.develop
+nimble.paths

config.nims

@@ -10,3 +10,7 @@ when (NimMajor, NimMinor) == (1, 2):
 
 when (NimMajor, NimMinor) > (1, 2):
   switch("hint", "XCannotRaiseY:off")
+# begin Nimble config (version 2)
+when withDir(thisDir(), system.fileExists("nimble.paths")):
+  include "nimble.paths"
+# end Nimble config

datastore.nim

@@ -3,5 +3,6 @@ import ./datastore/fsds
 import ./datastore/sql
 import ./datastore/mountedds
 import ./datastore/tieredds
+import ./datastore/threads/threadproxyds
 
-export datastore, fsds, mountedds, tieredds, sql
+export datastore, fsds, mountedds, tieredds, sql, threadproxyds

datastore.nimble

@@ -6,14 +6,18 @@ author        = "Status Research & Development GmbH"
 description   = "Simple, unified API for multiple data stores"
 license       = "Apache License 2.0 or MIT"
 
-requires "nim >= 1.2.0",
+requires "nim >= 1.6.14",
          "asynctest >= 0.3.1 & < 0.4.0",
-         "chronos",
+         "chronos#0277b65be2c7a365ac13df002fba6e172be55537",
          "questionable >= 0.10.3 & < 0.11.0",
          "sqlite3_abi",
          "stew",
          "unittest2",
-         "upraises >= 0.1.0 & < 0.2.0"
+         "pretty",
+         "threading",
+         "taskpools",
+         "upraises >= 0.1.0 & < 0.2.0",
+         "chronicles"
 
 task coverage, "generates code coverage report":
   var (output, exitCode) = gorgeEx("which lcov")

datastore/datastore.nim

@@ -13,32 +13,32 @@ push: {.upraises: [].}
 type
   BatchEntry* = tuple[key: Key, data: seq[byte]]
 
-method has*(self: Datastore, key: Key): Future[?!bool] {.base, locks: "unknown".} =
+method has*(self: Datastore, key: Key): Future[?!bool] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method delete*(self: Datastore, key: Key): Future[?!void] {.base, locks: "unknown".} =
+method delete*(self: Datastore, key: Key): Future[?!void] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method delete*(self: Datastore, keys: seq[Key]): Future[?!void] {.base, locks: "unknown".} =
+method delete*(self: Datastore, keys: seq[Key]): Future[?!void] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method get*(self: Datastore, key: Key): Future[?!seq[byte]] {.base, locks: "unknown".} =
+method get*(self: Datastore, key: Key): Future[?!seq[byte]] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method put*(self: Datastore, key: Key, data: seq[byte]): Future[?!void] {.base, locks: "unknown".} =
+method put*(self: Datastore, key: Key, data: seq[byte]): Future[?!void] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method put*(self: Datastore, batch: seq[BatchEntry]): Future[?!void] {.base, locks: "unknown".} =
+method put*(self: Datastore, batch: seq[BatchEntry]): Future[?!void] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
-method close*(self: Datastore): Future[?!void] {.base, async, locks: "unknown".} =
+method close*(self: Datastore): Future[?!void] {.base, locks: "unknown", raises: [].} =
   raiseAssert("Not implemented!")
 
 method query*(
   self: Datastore,
-  query: Query): Future[?!QueryIter] {.base, gcsafe.} =
+  query: Query): Future[?!QueryIter] {.base, gcsafe, raises: [].} =
 
   raiseAssert("Not implemented!")
 
-proc contains*(self: Datastore, key: Key): Future[bool] {.async.} =
+proc contains*(self: Datastore, key: Key): Future[bool] {.async, raises: [].} =
   return (await self.has(key)) |? false

datastore/fsds.nim

@@ -155,6 +155,9 @@ method put*(
   return success()
 
 proc dirWalker(path: string): iterator: string {.gcsafe.} =
+  var localPath {.threadvar.}: string
+
+  localPath = path
   return iterator(): string =
     try:
       for p in path.walkDirRec(yieldFilter = {pcFile}, relative = true):
@@ -188,10 +191,23 @@ method query*(
   var
     iter = QueryIter.new()
 
+  var lock = newAsyncLock() # serialize querying under threads
   proc next(): Future[?!QueryResponse] {.async.} =
+    defer:
+      if lock.locked:
+        lock.release()
+
+    if lock.locked:
+      return failure (ref DatastoreError)(msg: "Should always await query features")
+
     let
       path = walker()
 
+    if iter.finished:
+      return failure "iterator is finished"
+
+    await lock.acquire()
+
     if finished(walker):
       iter.finished = true
       return success (Key.none, EmptyBytes)

datastore/query.nim

@@ -1,16 +1,16 @@
 import pkg/upraises
+import std/algorithm
 import pkg/chronos
 import pkg/questionable
 import pkg/questionable/results
 
 import ./key
 import ./types
 
-type
-  SortOrder* {.pure.} = enum
-    Assending,
-    Descending
+export types
+export options, SortOrder
 
+type
   Query* = object
     key*: Key         # Key to be queried
     value*: bool      # Flag to indicate if data should be returned
@@ -19,11 +19,10 @@ type
     sort*: SortOrder  # Sort order - not available in all backends
 
   QueryResponse* = tuple[key: ?Key, data: seq[byte]]
-  QueryEndedError* = object of DatastoreError
 
-  GetNext* = proc(): Future[?!QueryResponse] {.upraises: [], gcsafe, closure.}
+  GetNext* = proc(): Future[?!QueryResponse] {.upraises: [], gcsafe.}
   IterDispose* = proc(): Future[?!void] {.upraises: [], gcsafe.}
-  QueryIter* = ref object
+  QueryIter* {.acyclic.} = ref object
     finished*: bool
     next*: GetNext
     dispose*: IterDispose
@@ -42,7 +41,7 @@ proc init*(
   T: type Query,
   key: Key,
   value = true,
-  sort = SortOrder.Assending,
+  sort = SortOrder.Ascending,
   offset = 0,
   limit = -1): T =
 

datastore/sql/sqliteds.nim

@@ -151,9 +151,19 @@ method query*(
     if not (v == SQLITE_OK):
       return failure newException(DatastoreError, $sqlite3_errstr(v))
 
+  let lock = newAsyncLock()
   proc next(): Future[?!QueryResponse] {.async.} =
+    defer:
+      if lock.locked:
+        lock.release()
+
+    if lock.locked:
+      return failure (ref DatastoreError)(msg: "Should always await query features")
+
     if iter.finished:
-      return failure(newException(QueryEndedError, "Calling next on a finished query!"))
+      return failure((ref QueryEndedError)(msg: "Calling next on a finished query!"))
+
+    await lock.acquire()
 
     let
       v = sqlite3_step(s)

datastore/sql/sqliteutils.nim

@@ -50,7 +50,8 @@ proc bindParam(
       # to the return from sqlite3_bind_*(). The object and pointer to it
       # must remain valid until then. SQLite will then manage the lifetime of
       # its private copy."
-      sqlite3_bind_blob(s, n.cint, unsafeAddr val[0], val.len.cint,
+      var val = val
+      sqlite3_bind_blob(s, n.cint, addr val[0], val.len.cint,
         SQLITE_TRANSIENT)
     else:
       sqlite3_bind_null(s, n.cint)

datastore/threads/asyncsemaphore.nim

@@ -0,0 +1,96 @@
+# Nim-Libp2p
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+{.push raises: [].}
+
+import sequtils
+import chronos, chronicles
+
+# TODO: this should probably go in chronos
+
+logScope:
+  topics = "datastore semaphore"
+
+type
+  AsyncSemaphore* = ref object of RootObj
+    size*: int
+    count: int
+    queue: seq[Future[void]]
+
+func new*(_: type AsyncSemaphore, size: int): AsyncSemaphore =
+  AsyncSemaphore(size: size, count: size)
+
+proc `count`*(s: AsyncSemaphore): int = s.count
+
+proc tryAcquire*(s: AsyncSemaphore): bool =
+  ## Attempts to acquire a resource, if successful
+  ## returns true, otherwise false
+  ##
+
+  if s.count > 0 and s.queue.len == 0:
+    s.count.dec
+    trace "Acquired slot", available = s.count, queue = s.queue.len
+    return true
+
+proc acquire*(s: AsyncSemaphore): Future[void] =
+  ## Acquire a resource and decrement the resource
+  ## counter. If no more resources are available,
+  ## the returned future will not complete until
+  ## the resource count goes above 0.
+  ##
+
+  let fut = newFuture[void]("AsyncSemaphore.acquire")
+  if s.tryAcquire():
+    fut.complete()
+    return fut
+
+  proc cancellation(udata: pointer) {.gcsafe.} =
+    fut.cancelCallback = nil
+    if not fut.finished:
+      s.queue.keepItIf( it != fut )
+
+  fut.cancelCallback = cancellation
+
+  s.queue.add(fut)
+
+  trace "Queued slot", available = s.count, queue = s.queue.len
+  return fut
+
+proc forceAcquire*(s: AsyncSemaphore) =
+  ## ForceAcquire will always succeed,
+  ## creating a temporary slot if required.
+  ## This temporary slot will stay usable until
+  ## there is less `acquire`s than `release`s
+  s.count.dec
+
+proc release*(s: AsyncSemaphore) =
+  ## Release a resource from the semaphore,
+  ## by picking the first future from the queue
+  ## and completing it and incrementing the
+  ## internal resource count
+  ##
+
+  doAssert(s.count <= s.size)
+
+  if s.count < s.size:
+    trace "Releasing slot", available = s.count,
+                            queue = s.queue.len
+
+    s.count.inc
+    while s.queue.len > 0:
+      var fut = s.queue[0]
+      s.queue.delete(0)
+      if not fut.finished():
+        s.count.dec
+        fut.complete()
+        break
+
+    trace "Released slot", available = s.count,
+                           queue = s.queue.len
+    return

datastore/threads/databuffer.nim

@@ -0,0 +1,92 @@
+import threading/smartptrs
+import std/hashes
+import pkg/stew/ptrops
+
+export hashes
+
+type
+  DataBufferHolder* = object
+    buf: ptr UncheckedArray[byte]
+    size: int
+
+  DataBuffer* = SharedPtr[DataBufferHolder] ##\
+    ## A fixed length data buffer using a SharedPtr.
+    ## It is thread safe even with `refc` since
+    ## it doesn't use string or seq types internally.
+    ##
+
+proc `=destroy`*(x: var DataBufferHolder) =
+  ## copy pointer implementation
+  ##
+
+  if x.buf != nil:
+    # echo "buffer: FREE: ", repr x.buf.pointer
+    deallocShared(x.buf)
+
+proc len*(a: DataBuffer): int = a[].size
+
+proc isNil*(a: DataBuffer): bool = smartptrs.isNil(a)
+
+proc hash*(a: DataBuffer): Hash =
+  a[].buf.toOpenArray(0, a[].size-1).hash()
+
+proc `==`*(a, b: DataBuffer): bool =
+  if a.isNil and b.isNil: return true
+  elif a.isNil or b.isNil: return false
+  elif a[].size != b[].size: return false
+  elif a[].buf == b[].buf: return true
+  else: a.hash() == b.hash()
+
+proc new*(tp: type DataBuffer, size: int = 0): DataBuffer =
+  ## allocate new buffer with given size
+  ##
+
+  newSharedPtr(DataBufferHolder(
+    buf: cast[typeof(result[].buf)](allocShared0(size)),
+    size: size,
+  ))
+
+proc new*[T: byte | char](tp: type DataBuffer, data: openArray[T]): DataBuffer =
+  ## allocate new buffer and copies indata from openArray
+  ##
+  result = DataBuffer.new(data.len)
+  if data.len() > 0:
+    # TODO: we might want to copy data, otherwise the GC might
+    # release it on stack-unwind
+    copyMem(result[].buf, baseAddr data, data.len)
+
+converter toSeq*(self: DataBuffer): seq[byte] =
+  ## convert buffer to a seq type using copy and either a byte or char
+  ##
+
+  result = newSeq[byte](self.len)
+  if self.len() > 0:
+    copyMem(addr result[0], addr self[].buf[0], self.len)
+
+proc `@`*(self: DataBuffer): seq[byte] =
+  ## Convert a buffer to a seq type using copy and
+  ## either a byte or char
+  ##
+
+  self.toSeq()
+
+converter toString*(data: DataBuffer): string =
+  ## convert buffer to string type using copy
+  ##
+
+  if data.isNil: return ""
+  result = newString(data.len())
+  if data.len() > 0:
+    copyMem(addr result[0], addr data[].buf[0], data.len)
+
+proc `$`*(data: DataBuffer): string =
+  ## convert buffer to string type using copy
+  ##
+
+  data.toString()
+
+converter toBuffer*(err: ref CatchableError): DataBuffer =
+  ## convert exception to an object with StringBuffer
+  ##
+
+  return DataBuffer.new(err.msg)