Draft: Threadpool refactor fsds

datastore.nimble

@@ -9,12 +9,12 @@ license       = "Apache License 2.0 or MIT"
 requires "nim >= 1.6.14",
          "asynctest >= 0.3.1 & < 0.4.0",
          "chronos#0277b65be2c7a365ac13df002fba6e172be55537",
-         "questionable >= 0.10.3 & < 0.11.0",
+         "questionable#head",
          "sqlite3_abi",
          "stew",
          "unittest2",
          "pretty",
-         "threading",
+         "https://github.com/elcritch/threading#test-smartptrsleak",
          "taskpools",
          "upraises >= 0.1.0 & < 0.2.0",
          "chronicles"

datastore/backend.nim

@@ -51,8 +51,9 @@ proc dbQuery*[K](
 
 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)
+proc toKey*(tp: typedesc[KeyId], id: string|cstring): KeyId = KeyId.new($id)
+proc toKey*(tp: typedesc[string], id: string|cstring): string = $(id)
+# proc toKey*(tp: typedesc[Key], id: string|cstring): KeyId = Key.init($id).expect("valid key")
 
 template toVal*(tp: typedesc[DataBuffer], id: openArray[byte]): DataBuffer = DataBuffer.new(id)
 template toVal*(tp: typedesc[seq[byte]], id: openArray[byte]): seq[byte] = @(id)
@@ -62,6 +63,3 @@ proc new*(tp: typedesc[KeyId], id: cstring): KeyId =
 
 proc new*(tp: typedesc[KeyId], id: string): KeyId =
   KeyId(data: DataBuffer.new(id))
-
-template toOpenArray*(x: DbKey): openArray[char] =
-  x.data.toOpenArray(char)

datastore/fsds.nim

@@ -2,35 +2,36 @@ import std/os
 import std/options
 import std/strutils
 
-import pkg/chronos
 import pkg/questionable
 import pkg/questionable/results
 from pkg/stew/results as stewResults import get, isErr
 import pkg/upraises
 
+import ./backend
 import ./datastore
 
 export datastore
 
 push: {.upraises: [].}
 
 type
-  FSDatastore* = ref object of Datastore
-    root*: string
+  FSDatastore*[K, V] = object
+    root*: DataBuffer
     ignoreProtected: bool
     depth: int
 
+proc isRootSubdir*(root, path: string): bool =
+  path.startsWith(root)
+
 proc validDepth*(self: FSDatastore, key: Key): bool =
   key.len <= self.depth
 
-proc isRootSubdir*(self: FSDatastore, path: string): bool =
-  path.startsWith(self.root)
-
-proc path*(self: FSDatastore, key: Key): ?!string =
+proc findPath*[K,V](self: FSDatastore[K,V], key: K): ?!string =
   ## Return filename corresponding to the key
   ## or failure if the key doesn't correspond to a valid filename
   ##
-
+  let root = $self.root
+  let key = Key.init($key).get()
   if not self.validDepth(key):
     return failure "Path has invalid depth!"
 
@@ -53,22 +54,27 @@ proc path*(self: FSDatastore, key: Key): ?!string =
       segments.add(ns.field / ns.value)
 
   let
-    fullname = (self.root / segments.joinPath())
+    fullname = (root / segments.joinPath())
       .absolutePath()
       .catch()
       .get()
       .addFileExt(FileExt)
 
-  if not self.isRootSubdir(fullname):
+  if not root.isRootSubdir(fullname):
     return failure "Path is outside of `root` directory!"
 
   return success fullname
 
-method has*(self: FSDatastore, key: Key): Future[?!bool] {.async.} =
-  return self.path(key).?fileExists()
+proc has*[K,V](self: FSDatastore[K,V], key: K): ?!bool =
+  without path =? self.findPath(key), error:
+    return failure error
+  success path.fileExists()
+
+proc contains*[K](self: FSDatastore, key: K): bool =
+  return self.has(key).get()
 
-method delete*(self: FSDatastore, key: Key): Future[?!void] {.async.} =
-  without path =? self.path(key), error:
+proc delete*[K,V](self: FSDatastore[K,V], key: K): ?!void =
+  without path =? self.findPath(key), error:
     return failure error
 
   if not path.fileExists():
@@ -81,33 +87,39 @@ method delete*(self: FSDatastore, key: Key): Future[?!void] {.async.} =
 
   return success()
 
-method delete*(self: FSDatastore, keys: seq[Key]): Future[?!void] {.async.} =
+proc delete*[K,V](self: FSDatastore[K,V], keys: openArray[K]): ?!void =
   for key in keys:
-    if err =? (await self.delete(key)).errorOption:
+    if err =? self.delete(key).errorOption:
       return failure err
 
   return success()
 
-proc readFile*(self: FSDatastore, path: string): ?!seq[byte] =
+proc readFile[V](self: FSDatastore, path: string): ?!V =
   var
     file: File
 
   defer:
     file.close
 
   if not file.open(path):
-    return failure "unable to open file!"
+    return failure "unable to open file! path: " & path
 
   try:
     let
-      size = file.getFileSize
+      size = file.getFileSize().int
 
+    when V is seq[byte]:
+      var bytes = newSeq[byte](size)
+    elif V is V:
+      var bytes = V.new(size=size)
+    else:
+      {.error: "unhandled result type".}
     var
-      bytes = newSeq[byte](size)
       read = 0
 
+    # echo "BYTES: ", bytes.repr
     while read < size:
-      read += file.readBytes(bytes, read, size)
+      read += file.readBytes(bytes.toOpenArray(0, size-1), read, size)
 
     if read < size:
       return failure $read & " bytes were read from " & path &
@@ -118,61 +130,71 @@ proc readFile*(self: FSDatastore, path: string): ?!seq[byte] =
   except CatchableError as e:
     return failure e
 
-method get*(self: FSDatastore, key: Key): Future[?!seq[byte]] {.async.} =
-  without path =? self.path(key), error:
+proc get*[K,V](self: FSDatastore[K,V], key: K): ?!V =
+  without path =? self.findPath(key), error:
     return failure error
 
   if not path.fileExists():
     return failure(
       newException(DatastoreKeyNotFound, "Key doesn't exist"))
 
-  return self.readFile(path)
+  return readFile[V](self, path)
 
-method put*(
-  self: FSDatastore,
-  key: Key,
-  data: seq[byte]): Future[?!void] {.async.} =
+proc put*[K,V](self: FSDatastore[K,V],
+               key: K,
+               data: V
+              ): ?!void =
 
-  without path =? self.path(key), error:
+  without path =? self.findPath(key), error:
     return failure error
 
   try:
+    var data = data
     createDir(parentDir(path))
-    writeFile(path, data)
+    writeFile(path, data.toOpenArray(0, data.len()-1))
   except CatchableError as e:
     return failure e
 
   return success()
 
-method put*(
+proc put*[K,V](
   self: FSDatastore,
-  batch: seq[BatchEntry]): Future[?!void] {.async.} =
+  batch: seq[DbBatchEntry[K, V]]): ?!void =
 
   for entry in batch:
-    if err =? (await self.put(entry.key, entry.data)).errorOption:
+    if err =? self.put(entry.key, entry.data).errorOption:
       return failure err
 
   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):
-        yield p
-    except CatchableError as exc:
-      raise newException(Defect, exc.msg)
+iterator dirIter(path: string): string {.gcsafe.} =
+  try:
+    for p in path.walkDirRec(yieldFilter = {pcFile}, relative = true):
+      yield p
+  except CatchableError as exc:
+    raise newException(Defect, exc.msg)
 
-method close*(self: FSDatastore): Future[?!void] {.async.} =
+proc close*[K,V](self: FSDatastore[K,V]): ?!void =
   return success()
 
-method query*(
-  self: FSDatastore,
-  query: Query): Future[?!QueryIter] {.async.} =
-
-  without path =? self.path(query.key), error:
+type
+  FsQueryHandle*[K, V] = object
+    query*: DbQuery[K]
+    cancel*: bool
+    closed*: bool
+    env*: FsQueryEnv[K,V]
+
+  FsQueryEnv*[K,V] = object
+    self: FSDatastore[K,V]
+    basePath: DataBuffer
+
+proc query*[K,V](
+  self: FSDatastore[K,V],
+  query: DbQuery[K],
+): Result[FsQueryHandle[K, V], ref CatchableError] =
+
+  let key = query.key
+  without path =? self.findPath(key), error:
     return failure error
 
   let basePath =
@@ -184,61 +206,58 @@ method query*(
       path.parentDir
     else:
       path.changeFileExt("")
+
+  let env = FsQueryEnv[K,V](self: self, basePath: DataBuffer.new(basePath))
+  success FsQueryHandle[K, V](query: query, env: env)
 
-  let
-    walker = dirWalker(basePath)
-
-  var
-    iter = QueryIter.new()
+proc close*[K,V](handle: var FsQueryHandle[K,V]) =
+  if not handle.closed:
+    handle.closed = true
 
-  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()
+iterator queryIter*[K, V](
+    handle: var FsQueryHandle[K, V]
+): ?!DbQueryResponse[K, V] =
+  let root = $(handle.env.self.root)
+  let basePath = $(handle.env.basePath)
 
-    if iter.finished:
-      return failure "iterator is finished"
-
-    await lock.acquire()
-
-    if finished(walker):
-      iter.finished = true
-      return success (Key.none, EmptyBytes)
+  for path in basePath.dirIter():
+    if handle.cancel:
+      break
 
     var
+      basePath = $handle.env.basePath
       keyPath = basePath
 
-    keyPath.removePrefix(self.root)
+    keyPath.removePrefix(root)
     keyPath = keyPath / path.changeFileExt("")
     keyPath = keyPath.replace("\\", "/")
 
     let
-      key = Key.init(keyPath).expect("should not fail")
+      flres = (basePath / path).absolutePath().catch
+    if flres.isErr():
+      yield DbQueryResponse[K,V].failure flres.error()
+      continue
+
+    let
+      key = K.toKey($Key.init(keyPath).expect("valid key"))
       data =
-        if query.value:
-          self.readFile((basePath / path).absolutePath)
-            .expect("Should read file")
+        if handle.query.value:
+          let res = readFile[V](handle.env.self, flres.get)
+          if res.isErr():
+            yield DbQueryResponse[K,V].failure res.error()
+            continue
+          res.get()
         else:
-          @[]
-
-    return success (key.some, data)
+          V.new()
 
-  iter.next = next
-  return success iter
+    yield success (key.some, data)
+  handle.close()
 
-proc new*(
-  T: type FSDatastore,
-  root: string,
-  depth = 2,
-  caseSensitive = true,
-  ignoreProtected = false): ?!T =
+proc newFSDatastore*[K,V](root: string,
+                          depth = 2,
+                          caseSensitive = true,
+                          ignoreProtected = false
+                         ): ?!FSDatastore[K,V] =
 
   let root = ? (
     block:
@@ -248,7 +267,7 @@ proc new*(
   if not dirExists(root):
     return failure "directory does not exist: " & root
 
-  success T(
-    root: root,
+  success FSDatastore[K,V](
+    root: DataBuffer.new root,
     ignoreProtected: ignoreProtected,
     depth: depth)

datastore/sql.nim

@@ -60,10 +60,9 @@ method put*(self: SQLiteDatastore,
 method close*(self: SQLiteDatastore): Future[?!void] {.async.} =
   self.db.close()
 
-method queryIter*(
-  self: SQLiteDatastore,
-  query: Query
-): ?!(iterator(): ?!QueryResponse) =
+method queryIter*(self: SQLiteDatastore,
+                  query: Query
+                 ): ?!(iterator(): ?!QueryResponse) =
 
   let dbquery = dbQuery(
     key= KeyId.new query.key.id(),
@@ -75,7 +74,7 @@ method queryIter*(
   var qhandle = ? self.db.query(dbquery)
 
   let iter = iterator(): ?!QueryResponse =
-    for resp in qhandle.iter():
+    for resp in qhandle.queryIter():
       without qres =? resp, err:
         yield QueryResponse.failure err
       let k = qres.key.map() do(k: KeyId) -> Key: