Add test_phased_push_pop

test/ConcurrentCollectionsTests/Project.toml

@@ -9,6 +9,7 @@ ConcurrentCollectionsBenchmarks = "172d371b-ba13-4e7e-a6ad-f169f5735fdf"
 Documenter = "e30172f5-a6a5-5a46-863b-614d45cd2de4"
 Formatting = "59287772-0a20-5a39-b81b-1366585eb4c0"
 ProgressLogging = "33c8b6b6-d38a-422a-b730-caa89a2f386c"
+SyncBarriers = "3986aa12-c984-439b-887a-f8545bea0e93"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 
 [compat]

test/ConcurrentCollectionsTests/src/test_dict.jl

@@ -2,6 +2,7 @@ module TestDict
 
 using ConcurrentCollections
 using ConcurrentCollections.Implementations: clusters, migrate!
+using SyncBarriers: Barrier, cycle!
 using Test
 
 function test_expand_and_shrink(n = 17)
@@ -205,4 +206,57 @@ function test_random_mutation(; kwargs...)
     @test actual == desired
 end
 
+function phased_push_pop!(
+    dict;
+    nkeys = 16,
+    repeat = 2^10,
+    phases = 2^10,
+    ntasks = Threads.nthreads(),
+)
+    locals = [
+        (
+            popped = zeros(valtype(dict), nkeys * phases),  # sum of popped values
+            added = zeros(valtype(dict), nkeys * phases),   # sum of all inserted values
+        ) for _ in 1:ntasks
+    ]
+    barrier = Barrier(ntasks)
+    @sync for (itask, (; popped, added)) in enumerate(locals)
+        Threads.@spawn begin
+            for p in 1:phases
+                k0 = (p - 1) * nkeys + 1
+                ks = k0:k0+nkeys-1
+                for _ in 1:repeat
+                    k = rand(ks)
+                    added[k] += 1
+                    modify!(dict, k) do ref
+                        Base.@_inline_meta
+                        Some(ref === nothing ? 1 : ref[] + 1)
+                    end
+                end
+                spin = 10_000  # spin for a few μs
+                cycle!(barrier[itask], spin)
+                for k in ks
+                    popped[k] += something(trypop!(dict, k), 0)
+                end
+            end
+        end
+    end
+    return locals
+end
+
+function test_phased_push_pop(; nkeys = 16, phases = 2^10, kwargs...)
+    dict = ConcurrentDict{Int,Int}()
+    locals = phased_push_pop!(dict; kwargs..., nkeys, phases)
+    actual = zeros(valtype(dict), nkeys * phases)
+    desired = zeros(valtype(dict), nkeys * phases)
+    for k in eachindex(actual)
+        actual[k] = get(dict, k, 0)
+    end
+    for (; popped, added) in locals
+        actual .+= popped
+        desired .+= added
+    end
+    @test actual == desired
+end
+
 end  # module

test/Project.toml

@@ -5,5 +5,6 @@ ConcurrentCollections = "5060bff5-0b44-40c5-b522-fcd3ca5cecdd"
 Documenter = "e30172f5-a6a5-5a46-863b-614d45cd2de4"
 Formatting = "59287772-0a20-5a39-b81b-1366585eb4c0"
 ProgressLogging = "33c8b6b6-d38a-422a-b730-caa89a2f386c"
+SyncBarriers = "3986aa12-c984-439b-887a-f8545bea0e93"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 TestFunctionRunner = "792026f5-ac9a-4a19-adcb-47b0ce2deb5d"