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

Add test_random_mutation and fix a bug in shrink #40

Merged
merged 1 commit into from
Sep 28, 2021
Merged

Add test_random_mutation and fix a bug in shrink #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
64 changes: 55 additions & 9 deletions src/dict.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,7 +339,12 @@ function migrate!(dict::LinearProbingDict, oldslots, expand; basesize = nothing)
if expand
expand_parallel!(newslots, slots, basesize)
else
migrate_serial!(newslots, slots)
if migrate_serial!(newslots, slots, Val(true)) === nothing
# Shrinking the slots have failed and rolled back. Keep using
# the old slots:
@atomic dict.slots = slots
return slots
end
end
# TODO: parallelize `shrink!`

Expand All @@ -355,10 +360,13 @@ end
function finishmove!(dict::LinearProbingDict, oldslots)
lock(dict.migration) do
slots = (@atomic dict.slots)::slots_type(dict)
# The caller observed `Moved` which only sets inside the `migration`
# lock. Thus, the migration should be finished once this lock is
# acquired:
# TODO: `oldslots` was used for a sanity check that `slots` is updated.
# However, now that shrink can rollback migration, it's not correct any
# more. Still keeping this code since it might make sense to actually
# not support shrinking.
#=
@assert oldslots !== slots
=#
return slots
end
end
Expand Down Expand Up @@ -417,8 +425,7 @@ function expand_parallel_basecase!(newslots, slots, basesize, start0, ichunk, fi
return (0, false)
end

migrate_between(start, stop) =
migrate_serial_nofill!(newslots, slots, start, stop, Val(false))::Int
migrate_between(start, stop) = migrate_serial_nofill!(newslots, slots, start, stop)::Int

# An empty slot is observed. There is at least one cluster started within
# this chunk.
Expand Down Expand Up @@ -500,8 +507,23 @@ function expand_parallel!(newslots, slots, basesize)
end
end

function migrate_serial!(newslots, slots)
nadded = migrate_serial_nofill!(newslots, slots, 1, length(slots), Val(false))
function migrate_serial!(
newslots,
slots,
rollback_on_error::Union{Val{false},Val{true}} = Val(false),
)
nadded = migrate_serial_nofill!(
newslots,
slots,
1,
length(slots),
Val(false),
rollback_on_error,
)
if nadded === nothing
rollback_on_error::Val{true}
return nothing
end
nadded = nadded::Int
fill_undef!(newslots)
return nadded
Expand Down Expand Up @@ -530,7 +552,9 @@ function migrate_serial_nofill!(
slots::AbstractVector{Slot},
start::Int,
stop::Int,
stop_on_empty::Union{Val{false},Val{true}},
# TODO: use custom singletong rather than Val:
stop_on_empty::Union{Val{false},Val{true}} = Val(false),
rollback_on_error::Union{Val{false},Val{true}} = Val(false),
) where {K,V,Slot<:DictSlot{K,V}}
nadded = 0
for i in start:stop
Expand Down Expand Up @@ -577,13 +601,35 @@ function migrate_serial_nofill!(

nprobes += 1
if nprobes > c
if rollback_on_error === Val(true)
rollback_migration!(slots, start, i)
return nothing
end
@static_error "unreachable: too many probings during migration"
end
end
end
return nadded
end

function rollback_migration!(slots, start, stop)
for i in start:stop
ref = slots[i]
value = @atomic ref.value
while true
if value isa Moved{Empty}
value, ok = @atomicreplace ref.value value => Empty()
ok && break
# TODO: maybe simple store is OK, since no one should be touching it?
elseif value isa Moved
@static_error("unexpected Moved{Value}")
else
break
end
end
end
end

Base.IteratorSize(::Type{<:LinearProbingDict}) = Base.SizeUnknown()
Base.IteratorSize(::Type{<:Base.KeySet{<:Any,<:LinearProbingDict}}) = Base.SizeUnknown()
Base.IteratorSize(::Type{<:Base.ValueIterator{<:LinearProbingDict}}) = Base.SizeUnknown()
Expand Down
46 changes: 46 additions & 0 deletions test/ConcurrentCollectionsTests/src/test_dict.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,4 +159,50 @@ function test_shrink()
end
end

function random_mutation!(dict; nkeys = 8, repeat = 2^20, ntasks = Threads.nthreads())
ks = 1:nkeys
locals = [
(
popped = zeros(valtype(dict), nkeys), # sum of popped values
added = zeros(valtype(dict), nkeys), # sum of all inserted values
) for _ in 1:ntasks
]
@sync for (; popped, added) in locals
Threads.@spawn begin
for _ in 1:repeat
k = rand(ks)
if rand(Bool)
y = trypop!(dict, k)
if y !== nothing
popped[k] += something(y)
end
else
added[k] += 1
modify!(dict, k) do ref
Base.@_inline_meta
Some(ref === nothing ? 1 : ref[] + 1)
end
end
end
end
end
return locals
end

function test_random_mutation(; kwargs...)
dict = ConcurrentDict{Int,Int}()
nkeys = 16
locals = random_mutation!(dict; kwargs..., nkeys)
actual = zeros(valtype(dict), nkeys)
desired = zeros(valtype(dict), nkeys)
for (k, v) in dict
actual[k] = v
end
for (; popped, added) in locals
actual .+= popped
desired .+= added
end
@test actual == desired
end

end # module