ThreadDatastore refactor - synchronous backends

.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

@@ -1,7 +1,7 @@
 import ./datastore/datastore
 import ./datastore/fsds
-import ./datastore/sql
+# import ./datastore/sql
 import ./datastore/mountedds
 import ./datastore/tieredds
 
-export datastore, fsds, mountedds, tieredds, sql
+export datastore, fsds, mountedds, tieredds

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/backend.nim

@@ -0,0 +1,67 @@
+import std/algorithm
+import std/options
+
+import pkg/questionable/results
+
+import ./threads/databuffer
+import ./types
+
+export databuffer, types, SortOrder
+
+## Types for datastore backends.
+## 
+## These should be syncrhonous and work with both GC types 
+## and DataBuffer's. This makes it easier to make them threadsafe.
+## 
+
+type
+  DbQueryResponse*[K, V] = tuple[key: Option[K], data: V]
+
+  DbQuery*[K] = object
+    key*: K         # Key to be queried
+    value*: bool      # Flag to indicate if data should be returned
+    limit*: int       # Max items to return - not available in all backends
+    offset*: int      # Offset from which to start querying - not available in all backends
+    sort*: SortOrder  # Sort order - not available in all backends
+
+  KeyId* = object
+    ## serialized Key ID, equivalent to `key.id()`
+    data*: DataBuffer
+
+  ## Accepted backend key and value types
+  DbKey* = string | KeyId
+  DbVal* = seq[byte] | DataBuffer
+
+  DbBatchEntry*[K, V] = tuple[key: K, data: V]
+
+  DbQueryHandle*[K, V, T] = object
+    query*: DbQuery[K]
+    cancel*: bool
+    closed*: bool
+    env*: T
+
+proc dbQuery*[K](
+    key: K,
+    value = false,
+    sort = SortOrder.Ascending,
+    offset = 0,
+    limit = -1
+): DbQuery[K] =
+  DbQuery[K](key: key, value: value, sort: sort, offset: offset, limit: limit)
+
+proc `$`*(id: KeyId): string = $(id.data)
+
+proc toKey*(tp: typedesc[KeyId], id: cstring): KeyId = KeyId.new(id)
+proc toKey*(tp: typedesc[string], id: cstring): string = $(id)
+
+template toVal*(tp: typedesc[DataBuffer], id: openArray[byte]): DataBuffer = DataBuffer.new(id)
+template toVal*(tp: typedesc[seq[byte]], id: openArray[byte]): seq[byte] = @(id)
+
+proc new*(tp: typedesc[KeyId], id: cstring): KeyId =
+  KeyId(data: DataBuffer.new(id.toOpenArray(0, id.len()-1)))
+
+proc new*(tp: typedesc[KeyId], id: string): KeyId =
+  KeyId(data: DataBuffer.new(id))
+
+template toOpenArray*(x: DbKey): openArray[char] =
+  x.data.toOpenArray(char)

datastore/datastore.nim

@@ -13,32 +13,37 @@ 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.} =
+method query*(self: Datastore,
+              query: Query
+              ): Future[?!QueryIter] {.base, gcsafe, raises: [].} =
 
   raiseAssert("Not implemented!")
 
-proc contains*(self: Datastore, key: Key): Future[bool] {.async.} =
+method queryIter*(self: Datastore,
+              query: Query
+              ): ?!(iterator(): ?!QueryResponse) {.base, gcsafe, raises: [].} =
+  raiseAssert("Not implemented!")
+
+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,27 +1,24 @@
 import pkg/upraises
+import std/algorithm
 import pkg/chronos
 import pkg/questionable
 import pkg/questionable/results
 
 import ./key
 import ./types
+import ./backend
+
+export types
+export options, SortOrder
 
 type
-  SortOrder* {.pure.} = enum
-    Assending,
-    Descending
 
-  Query* = object
-    key*: Key         # Key to be queried
-    value*: bool      # Flag to indicate if data should be returned
-    limit*: int       # Max items to return - not available in all backends
-    offset*: int      # Offset from which to start querying - not available in all backends
-    sort*: SortOrder  # Sort order - not available in all backends
+  ## Front end types
+  Query* = DbQuery[Key]
 
-  QueryResponse* = tuple[key: ?Key, data: seq[byte]]
-  QueryEndedError* = object of DatastoreError
+  QueryResponse* = DbQueryResponse[Key, seq[byte]]
 
-  GetNext* = proc(): Future[?!QueryResponse] {.upraises: [], gcsafe, closure.}
+  GetNext* = proc(): Future[?!QueryResponse] {.upraises: [], gcsafe.}
   IterDispose* = proc(): Future[?!void] {.upraises: [], gcsafe.}
   QueryIter* = ref object
     finished*: bool
@@ -38,17 +35,13 @@ proc defaultDispose(): Future[?!void] {.upraises: [], gcsafe, async.} =
 proc new*(T: type QueryIter, dispose = defaultDispose): T =
   QueryIter(dispose: dispose)
 
-proc init*(
-  T: type Query,
-  key: Key,
-  value = true,
-  sort = SortOrder.Assending,
-  offset = 0,
-  limit = -1): T =
-
-  T(
-    key: key,
-    value: value,
-    sort: sort,
-    offset: offset,
-    limit: limit)
+proc init*(T: type Query,
+           key: Key,
+           value = true,
+           sort = SortOrder.Ascending,
+           offset = 0,
+           limit = -1): Query =
+  dbQuery[Key](key, value, sort, offset, limit)
+
+proc toKey*(key: KeyId): Key {.inline, raises: [].} =
+  Key.init($key.data).expect("expected valid key here for but got `" & $key.data & "`")

datastore/sql.nim

@@ -1,3 +1,102 @@
+import std/times
+import std/options
+
+import pkg/chronos
+import pkg/questionable
+import pkg/questionable/results
+import pkg/sqlite3_abi
+from pkg/stew/results as stewResults import isErr
+import pkg/upraises
+
+import std/sequtils
+import ../datastore
+import ./backend
 import ./sql/sqliteds
 
-export sqliteds
+export datastore, sqliteds
+
+push: {.upraises: [].}
+
+type
+  SQLiteDatastore* = ref object of Datastore
+    db: SQLiteBackend[KeyId, DataBuffer]
+
+proc path*(self: SQLiteDatastore): string =
+  self.db.path()
+
+proc readOnly*(self: SQLiteDatastore): bool =
+  self.db.readOnly()
+
+method has*(self: SQLiteDatastore,
+            key: Key): Future[?!bool] {.async.} =
+  return self.db.has(KeyId.new key.id())
+
+method delete*(self: SQLiteDatastore,
+               key: Key): Future[?!void] {.async.} =
+  return self.db.delete(KeyId.new key.id())
+
+method delete*(self: SQLiteDatastore,
+               keys: seq[Key]): Future[?!void] {.async.} =
+  let dkeys = keys.mapIt(KeyId.new it.id())
+  return self.db.delete(dkeys)
+
+method get*(self: SQLiteDatastore,
+            key: Key): Future[?!seq[byte]] {.async.} =
+  self.db.get(KeyId.new key.id()).map() do(d: DataBuffer) -> seq[byte]:
+    d.toSeq()
+
+method put*(self: SQLiteDatastore,
+            key: Key,
+            data: seq[byte]): Future[?!void] {.async.} =
+  self.db.put(KeyId.new key.id(), DataBuffer.new data)
+
+method put*(self: SQLiteDatastore,
+            batch: seq[BatchEntry]): Future[?!void] {.async.} =
+  var dbatch: seq[tuple[key: KeyId, data: DataBuffer]]
+  for entry in batch:
+    dbatch.add((KeyId.new entry.key.id(), DataBuffer.new entry.data))
+  self.db.put(dbatch)
+
+method close*(self: SQLiteDatastore): Future[?!void] {.async.} =
+  self.db.close()
+
+method queryIter*(
+  self: SQLiteDatastore,
+  query: Query
+): ?!(iterator(): ?!QueryResponse) =
+
+  let dbquery = dbQuery(
+    key= KeyId.new query.key.id(),
+    value= query.value,
+    limit= query.limit,
+    offset= query.offset,
+    sort= query.sort,
+  )
+  var qhandle = ? self.db.query(dbquery)
+
+  let iter = iterator(): ?!QueryResponse =
+    for resp in qhandle.iter():
+      without qres =? resp, err:
+        yield QueryResponse.failure err
+      let k = qres.key.map() do(k: KeyId) -> Key:
+        Key.init($k).expect("valid key")
+      let v: seq[byte] = qres.data.toSeq()
+      yield success (k, v)
+
+  success iter
+
+proc new*(
+  T: type SQLiteDatastore,
+  path: string,
+  readOnly = false): ?!SQLiteDatastore =
+
+  success SQLiteDatastore(
+    db: ? newSQLiteBackend[KeyId, DataBuffer](path, readOnly))
+
+proc new*(
+  T: type SQLiteDatastore,
+  db: SQLiteBackend[KeyId, DataBuffer]): ?!T =
+
+  success T(
+    db: db,
+    readOnly: db.readOnly)