mirror of
https://github.com/logos-storage/nim-datastore.git
synced 2026-01-07 16:13:07 +00:00
verifying types
This commit is contained in:
parent
cf3f1defd6
commit
7f921901d7
@ -112,7 +112,7 @@ method query*(
|
|||||||
iter.finished = true
|
iter.finished = true
|
||||||
return success (Key.none, EmptyBytes)
|
return success (Key.none, EmptyBytes)
|
||||||
|
|
||||||
let key = kb.toKey().expect("should not fail")
|
let key = kb.toKey()
|
||||||
var ds: ValueBuffer
|
var ds: ValueBuffer
|
||||||
if query.value:
|
if query.value:
|
||||||
ds = self.store[kb]
|
ds = self.store[kb]
|
||||||
|
|||||||
@ -116,7 +116,7 @@ proc toBuffer*(q: Query): QueryBuffer =
|
|||||||
proc toQuery*(qb: QueryBuffer): Query =
|
proc toQuery*(qb: QueryBuffer): Query =
|
||||||
## convert QueryBuffer to regular Query
|
## convert QueryBuffer to regular Query
|
||||||
Query(
|
Query(
|
||||||
key: qb.key.toKey().expect("key expected"),
|
key: qb.key.toKey(),
|
||||||
value: qb.value,
|
value: qb.value,
|
||||||
limit: qb.limit,
|
limit: qb.limit,
|
||||||
offset: qb.offset,
|
offset: qb.offset,
|
||||||
@ -138,7 +138,7 @@ proc toQueryResponse*(qb: QueryResponseBuffer): QueryResponse =
|
|||||||
## convert QueryReponseBuffer to regular QueryResponse
|
## convert QueryReponseBuffer to regular QueryResponse
|
||||||
let key =
|
let key =
|
||||||
if qb.key.isNil: none(Key)
|
if qb.key.isNil: none(Key)
|
||||||
else: some qb.key.toKey().expect("key response should work")
|
else: some qb.key.toKey()
|
||||||
let data =
|
let data =
|
||||||
if qb.data.isNil: EmptyBytes
|
if qb.data.isNil: EmptyBytes
|
||||||
else: qb.data.toSeq(byte)
|
else: qb.data.toSeq(byte)
|
||||||
|
|||||||
@ -87,13 +87,13 @@ method put*(
|
|||||||
data: seq[byte]
|
data: seq[byte]
|
||||||
): Future[?!void] {.async.} =
|
): Future[?!void] {.async.} =
|
||||||
|
|
||||||
echo "put new request thr: ", $getThreadId()
|
echoed "put new request thr: ", $getThreadId()
|
||||||
var ret = await newThreadResult(void)
|
var ret = await newThreadResult(void)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
put(ret, self.tds, key, data)
|
put(ret, self.tds, key, data)
|
||||||
echo "\n"
|
echo "\n"
|
||||||
echo "wait put thr: ", $getThreadId()
|
echoed "wait put thr: ", $getThreadId()
|
||||||
echo "\n"
|
echo "\n"
|
||||||
await sleepAsync(400.milliseconds)
|
await sleepAsync(400.milliseconds)
|
||||||
await wait(ret)
|
await wait(ret)
|
||||||
@ -104,7 +104,7 @@ method put*(
|
|||||||
finally:
|
finally:
|
||||||
echo "\n"
|
echo "\n"
|
||||||
await sleepAsync(400.milliseconds)
|
await sleepAsync(400.milliseconds)
|
||||||
echo "PUT RELEASE"
|
echoed "PUT RELEASE"
|
||||||
ret.release()
|
ret.release()
|
||||||
|
|
||||||
method put*(
|
method put*(
|
||||||
|
|||||||
@ -42,20 +42,21 @@ proc `==`*(a, b: DataBuffer): bool =
|
|||||||
elif a[].buf == b[].buf: return true
|
elif a[].buf == b[].buf: return true
|
||||||
else: a.hash() == b.hash()
|
else: a.hash() == b.hash()
|
||||||
|
|
||||||
proc new*(tp: typedesc[DataBuffer], size: int = 0): DataBuffer =
|
proc new*[D: DataBuffer](tp: typedesc[D], size: int = 0): D =
|
||||||
## allocate new buffer with given size
|
## allocate new buffer with given size
|
||||||
result = newSharedPtr(DataBufferHolder(
|
result = newSharedPtr(DataBufferHolder(
|
||||||
buf: cast[typeof(result[].buf)](allocShared0(size)),
|
buf: cast[typeof(result[].buf)](allocShared0(size)),
|
||||||
size: size,
|
size: size,
|
||||||
))
|
))
|
||||||
echoed "DataBuffer:new: ", result.unsafeRawPtr().repr,
|
echoed "DataBuffer:new: ", result.unsafeRawPtr().repr,
|
||||||
|
" tp ", $(typeof(D)),
|
||||||
" @ ", result[].buf.pointer.repr,
|
" @ ", result[].buf.pointer.repr,
|
||||||
" -> ", result.toString().repr
|
" -> ", result.toString().repr
|
||||||
|
|
||||||
proc new*[T: byte | char](tp: typedesc[DataBuffer], data: openArray[T]): DataBuffer =
|
proc new*[T: byte | char; D: DataBuffer](tp: typedesc[D], data: openArray[T]): D =
|
||||||
## allocate new buffer and copies indata from openArray
|
## allocate new buffer and copies indata from openArray
|
||||||
##
|
##
|
||||||
result = DataBuffer.new(data.len)
|
result = D.new(data.len)
|
||||||
if data.len() > 0:
|
if data.len() > 0:
|
||||||
copyMem(result[].buf, unsafeAddr data[0], data.len)
|
copyMem(result[].buf, unsafeAddr data[0], data.len)
|
||||||
|
|
||||||
@ -85,6 +86,8 @@ import ../key
|
|||||||
import stew/results
|
import stew/results
|
||||||
|
|
||||||
proc new*(tp: typedesc[KeyBuffer], key: Key): KeyBuffer =
|
proc new*(tp: typedesc[KeyBuffer], key: Key): KeyBuffer =
|
||||||
KeyBuffer.new(key.id())
|
result = KeyBuffer.new(key.id())
|
||||||
proc toKey*(kb: KeyBuffer): Result[Key, ref CatchableError] =
|
echoed "KeyBuffer:new: ", $result
|
||||||
Key.init(kb.toString())
|
proc toKey*(kb: KeyBuffer): Key =
|
||||||
|
let res = Key.init(kb.toString())
|
||||||
|
res.expect("should always be valid")
|
||||||
|
|||||||
@ -72,8 +72,8 @@ proc hasTask*(
|
|||||||
tds: ThreadDatastorePtr,
|
tds: ThreadDatastorePtr,
|
||||||
kb: KeyBuffer,
|
kb: KeyBuffer,
|
||||||
) =
|
) =
|
||||||
without key =? kb.toKey(), err:
|
|
||||||
ret.failure(err)
|
let key = kb.toKey()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
let res = waitFor tds[].ds.has(key)
|
let res = waitFor tds[].ds.has(key)
|
||||||
@ -90,7 +90,7 @@ proc has*(
|
|||||||
tds: ThreadDatastorePtr,
|
tds: ThreadDatastorePtr,
|
||||||
key: Key,
|
key: Key,
|
||||||
) =
|
) =
|
||||||
let bkey = StringBuffer.new(key.id())
|
let bkey = KeyBuffer.new(key)
|
||||||
tds[].tp.spawn hasTask(ret, tds, bkey)
|
tds[].tp.spawn hasTask(ret, tds, bkey)
|
||||||
|
|
||||||
proc getTask*(
|
proc getTask*(
|
||||||
@ -98,8 +98,7 @@ proc getTask*(
|
|||||||
tds: ThreadDatastorePtr,
|
tds: ThreadDatastorePtr,
|
||||||
kb: KeyBuffer,
|
kb: KeyBuffer,
|
||||||
) =
|
) =
|
||||||
without key =? kb.toKey(), err:
|
let key = kb.toKey()
|
||||||
ret.failure(err)
|
|
||||||
try:
|
try:
|
||||||
let res = waitFor tds[].ds.get(key)
|
let res = waitFor tds[].ds.get(key)
|
||||||
if res.isErr:
|
if res.isErr:
|
||||||
@ -117,7 +116,7 @@ proc get*(
|
|||||||
tds: ThreadDatastorePtr,
|
tds: ThreadDatastorePtr,
|
||||||
key: Key,
|
key: Key,
|
||||||
) =
|
) =
|
||||||
let bkey = StringBuffer.new(key.id())
|
let bkey = KeyBuffer.new(key)
|
||||||
tds[].tp.spawn getTask(ret, tds, bkey)
|
tds[].tp.spawn getTask(ret, tds, bkey)
|
||||||
|
|
||||||
import std/os
|
import std/os
|
||||||
@ -131,12 +130,12 @@ proc putTask*(
|
|||||||
|
|
||||||
os.sleep(400)
|
os.sleep(400)
|
||||||
var ret = ret
|
var ret = ret
|
||||||
echo "putTask: ", $getThreadId()
|
echo "\n"
|
||||||
echo "putTask:kb: ", kb.toString
|
echoed "putTask: ", $getThreadId()
|
||||||
echo "putTask:db: ", db.toString
|
echoed "putTask:kb: ", kb.toString
|
||||||
|
echoed "putTask:db: ", db.toString
|
||||||
|
|
||||||
without key =? kb.toKey(), err:
|
let key = kb.toKey()
|
||||||
ret.failure(err)
|
|
||||||
|
|
||||||
let data = db.toSeq(byte)
|
let data = db.toSeq(byte)
|
||||||
let res = (waitFor tds[].ds.put(key, data)).catch
|
let res = (waitFor tds[].ds.put(key, data)).catch
|
||||||
@ -148,7 +147,7 @@ proc putTask*(
|
|||||||
|
|
||||||
discard ret.fireSync()
|
discard ret.fireSync()
|
||||||
ret.release()
|
ret.release()
|
||||||
echo "putTask: FINISH\n"
|
echoed "putTask: FINISH\n"
|
||||||
|
|
||||||
proc put*(
|
proc put*(
|
||||||
ret: TResult[void],
|
ret: TResult[void],
|
||||||
@ -156,11 +155,11 @@ proc put*(
|
|||||||
key: Key,
|
key: Key,
|
||||||
data: seq[byte]
|
data: seq[byte]
|
||||||
) =
|
) =
|
||||||
echo "put request args: ", $getThreadId()
|
echoed "put request args: ", $getThreadId()
|
||||||
let bkey = StringBuffer.new(key.id())
|
let bkey = KeyBuffer.new(key)
|
||||||
let bval = DataBuffer.new(data)
|
let bval = DataBuffer.new(data)
|
||||||
|
|
||||||
echo "spawn put request: ", $getThreadId()
|
echoed "spawn put request: ", $getThreadId()
|
||||||
tds[].tp.spawn putTask(ret, tds, bkey, bval)
|
tds[].tp.spawn putTask(ret, tds, bkey, bval)
|
||||||
|
|
||||||
|
|
||||||
@ -170,8 +169,7 @@ proc deleteTask*(
|
|||||||
kb: KeyBuffer,
|
kb: KeyBuffer,
|
||||||
) =
|
) =
|
||||||
|
|
||||||
without key =? kb.toKey(), err:
|
let key = kb.toKey()
|
||||||
ret.failure(err)
|
|
||||||
|
|
||||||
let res = (waitFor tds[].ds.delete(key)).catch
|
let res = (waitFor tds[].ds.delete(key)).catch
|
||||||
# print "thrbackend: putTask: fire", ret[].signal.fireSync().get()
|
# print "thrbackend: putTask: fire", ret[].signal.fireSync().get()
|
||||||
@ -189,7 +187,7 @@ proc delete*(
|
|||||||
tds: ThreadDatastorePtr,
|
tds: ThreadDatastorePtr,
|
||||||
key: Key,
|
key: Key,
|
||||||
) =
|
) =
|
||||||
let bkey = StringBuffer.new(key.id())
|
let bkey = KeyBuffer.new(key)
|
||||||
tds[].tp.spawn deleteTask(ret, tds, bkey)
|
tds[].tp.spawn deleteTask(ret, tds, bkey)
|
||||||
|
|
||||||
# import os
|
# import os
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user