Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Tests for concurrent queries. Finishes disposed iterators. #67

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+144 −3
Open

Tests for concurrent queries. Finishes disposed iterators. #67

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e005690
Adds tests for concurrent queries. Finishes disposed iterators.
benbierens May 28, 2024
996878e
Merge branch 'master' into tests/concurrent-queries
benbierens Aug 19, 2024
660aa6d
Disposed iters stop yielding data.
benbierens Aug 19, 2024
10c115f
Merge branch 'master' into tests/concurrent-queries
benbierens Aug 19, 2024
6b02676
Merge branch 'master' into tests/concurrent-queries
benbierens Aug 21, 2024
3704f26
Merge branch 'master' into tests/concurrent-queries
benbierens Aug 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions datastore/fsds.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,9 @@ method query*(
iter = QueryIter.new()

proc next(): Future[?!QueryResponse] {.async.} =
if iter.finished:
return success (Key.none, EmptyBytes)

let
path = walker()

Expand All @@ -216,6 +219,10 @@ method query*(
return success (key.some, data)

iter.next = next
iter.dispose = proc(): Future[?!void] {.async.} =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that disposing an inter should follow iter being finished, not the other way around.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm sorry, I'm not sure what you mean. :o

Copy link
Contributor

@tbekas tbekas Aug 15, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So setting finished to true, should be an internal to iterator event that's signaling that all items were yielded (line 197).

With this change, flag finished would be set to true whenever dispose() would be called (even if all items were not yielded). So as a client of this API you can finish iteration prematurely and call dispose(), but that's not the same as yielding all elements.

Copy link
Member

@gmega gmega Aug 15, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Well the client won't even finish iteration because next will still happily return the next element as next doesn't check the finished flag to decide what to do. So if you're relying on it returning success (Key.none, ...) to know when to stop, this will not have the intended effect.

I suppose that to get the right semantics (assuming this is the right one) you need to check the state of the finished flag in addition to the state of the walker iterator (e.g if finished(walker) or iter.finished).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I understand the desired behavior and I'm gonne write a test for it.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update: Finished iters will always return None when asked for next. There's a test to ensure this, and ensure that disposed iters become finished.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure why we're (incorrectly) changing the semantics of the iterators now?

as @tbekas is saying:

So setting finished to true, should be an internal to iterator event that's signaling that all items were yielded (line 197).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the goal is to allow dispose to be called before the iterator returns all elements. I suppose this is useful because it's not always that you will want to consume the whole set of results from a query, and in that case you'll want to free up the resources (dispose) anyway. Is that the use case @benbierens?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah OK the concern I had is fixed by this.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we can't assume the iterator will always run through all elements. I've added a test to show that you can properly clean up an iterator early by calling dispose on it. Any nexts would then yield nothing and the underlying resources can be closed.

iter.finished = true
return success()

return success iter

method modifyGet*(
Expand Down
3 changes: 2 additions & 1 deletion datastore/leveldb/leveldbds.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ method query*(

proc next(): Future[?!QueryResponse] {.async.} =
if iter.finished:
return failure(newException(QueryEndedError, "Calling next on a finished query!"))
return success (Key.none, EmptyBytes)

try:
let (keyStr, valueStr) = dbIter.next()
Expand All @@ -116,6 +116,7 @@ method query*(

proc dispose(): Future[?!void] {.async.} =
dbIter.dispose()
iter.finished = true
return success()

iter.next = next
Expand Down
4 changes: 2 additions & 2 deletions datastore/sql/sqliteds.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,7 +269,7 @@ method query*(

proc next(): Future[?!QueryResponse] {.async.} =
if iter.finished:
return failure(newException(QueryEndedError, "Calling next on a finished query!"))
return success (Key.none, EmptyBytes)

let
v = sqlite3_step(s)
Expand Down Expand Up @@ -324,7 +324,7 @@ method query*(
iter.dispose = proc(): Future[?!void] {.async.} =
discard sqlite3_reset(s)
discard sqlite3_clear_bindings(s)
iter.next = nil
iter.finished = true
return success()

iter.next = next
Expand Down
133 changes: 133 additions & 0 deletions tests/datastore/querycommontests.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,139 @@ template queryTests*(ds: Datastore, testLimitsAndOffsets = true, testSortOrder =

(await iter.dispose()).tryGet

test "Concurrent queries":
let
q1 = Query.init(key1)
q2 = Query.init(key1)

(await ds.put(key1, val1)).tryGet
(await ds.put(key2, val2)).tryGet
(await ds.put(key3, val3)).tryGet

let
iter1 = (await ds.query(q1)).tryGet
iter2 = (await ds.query(q1)).tryGet

let one1 = (await iter1.next()).tryGet
check not iter1.finished

let one2 = (await iter2.next()).tryGet
check not iter2.finished

let two1 = (await iter1.next()).tryGet
check not iter1.finished
let three1 = (await iter1.next()).tryGet
check not iter1.finished

let two2 = (await iter2.next()).tryGet
check not iter2.finished
let three2 = (await iter2.next()).tryGet
check not iter2.finished

let four1 = (await iter1.next()).tryGet
check iter1.finished
let four2 = (await iter2.next()).tryGet
check iter2.finished

check:
one1[0].get == key1
one1[1] == val1
two1[0].get == key2
two1[1] == val2
three1[0].get == key3
three1[1] == val3
four1[0] == Key.none

one2[0].get == key1
one2[1] == val1
two2[0].get == key2
two2[1] == val2
three2[0].get == key3
three2[1] == val3
four2[0] == Key.none

(await iter1.dispose()).tryGet
(await iter2.dispose()).tryGet

test "Concurrent queries - dispose":
let
q1 = Query.init(key1)
q2 = Query.init(key1)

(await ds.put(key1, val1)).tryGet
(await ds.put(key2, val2)).tryGet
(await ds.put(key3, val3)).tryGet

let
iter1 = (await ds.query(q1)).tryGet
iter2 = (await ds.query(q1)).tryGet

let one1 = (await iter1.next()).tryGet
check not iter1.finished

let one2 = (await iter2.next()).tryGet
check not iter2.finished

(await iter1.dispose()).tryGet
check iter1.finished

let two2 = (await iter2.next()).tryGet
check not iter2.finished
let three2 = (await iter2.next()).tryGet
check not iter2.finished

let four2 = (await iter2.next()).tryGet
check iter2.finished

check:
one1[0].get == key1
one1[1] == val1

one2[0].get == key1
one2[1] == val1
two2[0].get == key2
two2[1] == val2
three2[0].get == key3
three2[1] == val3
four2[0] == Key.none

(await iter1.dispose()).tryGet
(await iter2.dispose()).tryGet

test "Dispose should discontinue iteration":
let
q1 = Query.init(key1)

(await ds.put(key1, val1)).tryGet
(await ds.put(key2, val2)).tryGet
(await ds.put(key3, val3)).tryGet

let
iter = (await ds.query(q1)).tryGet

let one = (await iter.next()).tryGet
check not iter.finished

let two = (await iter.next()).tryGet
check not iter.finished

(await iter.dispose()).tryGet
check iter.finished

let three = (await iter.next()).tryGet
check iter.finished

let four = (await iter.next()).tryGet
check iter.finished

check:
one[0].get == key1
one[1] == val1
two[0].get == key2
two[1] == val2
three[0] == Key.none
four[0] == Key.none

test "Key should query all keys without values":
let
q = Query.init(key1, value = false)
Expand Down