make concurrent (but don't need anymore)

This commit is contained in:
Dmitriy Ryajov 2023-09-13 14:42:22 -06:00
parent 776c58f091
commit 7a9bc11c33
No known key found for this signature in database
GPG Key ID: DA8C680CE7C657A4

View File

@ -1,8 +1,12 @@
import std/tables import std/tables
import std/sharedtables import std/sharedtables
import std/sharedlist
import std/sequtils import std/sequtils
import std/strutils import std/strutils
import std/algorithm import std/algorithm
import std/locks
import std/atomics
import pkg/chronos import pkg/chronos
import pkg/questionable import pkg/questionable
@ -20,21 +24,36 @@ push: {.upraises: [].}
type type
MemoryDatastore* = ref object of Datastore MemoryDatastore* = ref object of Datastore
store*: SharedTable[Key, seq[byte]] store*: SharedTable[Key, seq[byte]]
keys*: SharedList[Key]
lock: Lock # yes, we need the lock since we need to update both the table and the list :facepalm:
template withLock(lock: Lock, body: untyped) =
try:
lock.acquire()
body
finally:
lock.release()
method has*( method has*(
self: MemoryDatastore, self: MemoryDatastore,
key: Key): Future[?!bool] {.async.} = key: Key): Future[?!bool] {.async.} =
let
keys = toSeq(self.keys)
self.store.withValue(key, value): for k in keys:
return success true if k == key:
do: return success true
return success(false)
return success false
method delete*( method delete*(
self: MemoryDatastore, self: MemoryDatastore,
key: Key): Future[?!void] {.async.} = key: Key): Future[?!void] {.async.} =
self.store.del(key) withLock(self.lock):
self.keys.iterAndMutate(proc(k: Key): bool = k == key)
self.store.del(key)
return success() return success()
method delete*( method delete*(
@ -51,18 +70,25 @@ method get*(
self: MemoryDatastore, self: MemoryDatastore,
key: Key): Future[?!seq[byte]] {.async.} = key: Key): Future[?!seq[byte]] {.async.} =
self.store.withValue(key, value): withLock(self.lock):
return success value[] let
do: has = (await self.has(key))
return failure (ref DatastoreError)(msg: "no such key")
if has.isOk and has.get:
return self.store.mget(key).catch()
return failure (ref DatastoreError)(msg: "not found")
method put*( method put*(
self: MemoryDatastore, self: MemoryDatastore,
key: Key, key: Key,
data: seq[byte]): Future[?!void] {.async.} = data: seq[byte]): Future[?!void] {.async.} =
self.store[key] = data withLock(self.lock):
return success() if not self.store.hasKeyOrPut(key, data):
self.keys.add(key)
else:
self.store[key] = data
method put*( method put*(
self: MemoryDatastore, self: MemoryDatastore,
@ -74,50 +100,50 @@ method put*(
return success() return success()
# proc keyIterator(self: MemoryDatastore, queryKey: string): iterator: KeyBuffer {.gcsafe.} = method query*(
# return iterator(): KeyBuffer {.closure.} = self: MemoryDatastore,
# var keys = self.store.keys().toSeq() query: Query,
# keys.sort(proc (x, y: KeyBuffer): int = cmp(x.toString, y.toString)) ): Future[?!QueryIter] {.async.} =
# for key in keys:
# if key.toString().startsWith(queryKey):
# yield key
# method query*( let
# self: MemoryDatastore, queryKey = query.key.id()
# query: Query, keys = toSeq(self.keys)
# ): Future[?!QueryIter] {.async.} =
# let var
# queryKey = query.key.id() iter = QueryIter.new()
# walker = keyIterator(self, queryKey) pos = 0
# var
# iter = QueryIter.new()
# proc next(): Future[?!QueryResponse] {.async.} = proc next(): Future[?!QueryResponse] {.async.} =
# let kb = walker() defer:
pos.inc
# if finished(walker): if iter.finished:
# iter.finished = true return failure (ref QueryEndedError)(msg: "Calling next on a finished query!")
# return success (Key.none, EmptyBytes)
# let key = kb.toKey().expect("should not fail") if pos > keys.len - 1:
# var ds: ValueBuffer iter.finished = true
# if query.value: return success (Key.none, EmptyBytes)
# ds = self.store[kb]
# let data = if ds.isNil: EmptyBytes else: ds.toSeq(byte)
# return success (key.some, data) return success (
Key.init(keys[pos]).expect("Should not fail!").some,
if query.value: self.store.mget(keys[pos]) else: EmptyBytes)
# iter.next = next iter.next = next
# return success iter return success iter
method close*(self: MemoryDatastore): Future[?!void] {.async.} = method close*(self: MemoryDatastore): Future[?!void] {.async.} =
self.store.deinitSharedTable() self.store.deinitSharedTable()
self.keys.deinitSharedList()
self.lock.deinitLock()
return success() return success()
proc new*(tp: type MemoryDatastore): MemoryDatastore = proc new*(tp: type MemoryDatastore): MemoryDatastore =
var var
table: SharedTable[Key, seq[byte]] table: SharedTable[Key, seq[byte]]
keys: SharedList[Key]
lock: Lock
table.init() table.init()
MemoryDatastore(store: table) keys.init()
lock.initLock()
MemoryDatastore(store: table, keys: keys, lock: lock)