Add threadsafe Pool data structure

[quinnj]

Also split the now-large ConcurrentUtilities.jl file into more manageable individual files. For reviewing, mainly worry about the src/pools.jl and test/pools.jl file, since they're the only net-new code here.

[Drvi]

Left a couple of comments about optimization opportunities. Would it make sense to specialize the Pool on being keyed/non-keyed and Semaphore-bounded vs not?

[Drvi]

Wouldn't the macro version of locking be preferred to the anonymous function one?
I've made a little benchmark:

julia> function foo()
           pool = Pool(Int; max=10000)
           for i in 1:9999
               x = acquire(() -> i, pool)
           end

           for i in 10000:15000
               x = acquire(() -> i, pool)
               release(pool, i)
               release(pool, i-9999)
           end
           return pool
       end

And here are the results:

julia> @btime foo(); # uses Base.@lock
  1.673 ms (10016 allocations: 313.28 KiB)

julia> @btime foo(); # uses do syntax
  1.743 ms (10016 allocations: 313.28 KiB)

julia> @btime foo(); # uses do syntax
  1.735 ms (10016 allocations: 313.28 KiB)

julia> @btime foo(); # uses Base.@lock
  1.675 ms (10016 allocations: 313.28 KiB)

[quinnj]

Yeah, let's do it!

[quinnj]

On my updated PR (refactored pool) using the above benchmark, I'm seeing:

julia> @btime foo()
  673.667 μs (8 allocations: 132.39 KiB)

[Drvi]

Very cool:) Still, I think the macro is preferable

[Drvi]

So in this dummy benchmark:

function foo()
    pool = Pool(Int; max=10000)
    for i in 1:10000
        x = acquire(() -> i, pool)
    end

    for i in 10001:20000
        release(pool, i-10000)
        x = acquire(() -> i, pool)
        release(pool, i)
    end

    for i in 1:10000
        x = acquire(() -> i, pool)
    end

    for i in 10001:20000
        release(pool, i-10000)
        x = acquire(() -> i, pool)
        release(pool, i)
    end

    for i in 1:10000
        x = acquire(() -> i, pool)
    end

    for i in 10001:20000
        release(pool, i-10000)
        x = acquire(() -> i, pool)
        release(pool, i)
    end
    return pool
end

we seem to pay 20% penalty for using a Dict{Nothing,Stack} instead of a Stack directly:

Here is a quick comparison where I hacked a version that used the Stack directly iff K === Nothing:

julia> @btime foo();
  5.792 ms (60010 allocations: 1.83 MiB)

This is the original:

julia> @btime foo();
  8.205 ms (60014 allocations: 1.83 MiB)

[Drvi]

This makes me wonder if the use case in HTTP would benefit from ditching the Dict and use e.g. a NamedTuple of Stacks or something like that? EDIT: Just read the HTTP PR and noticed that we don't key based on OpenSSL/MbedTLS/TCP so the NamedTuple idea is probably not applicable.

[quinnj]

We might be able to do something faster in the non-key case, which I think is worth exploring (like having an extra field for a single stack that we use if the key type is Nothing)

[quinnj]

Updated timing post-refactor of the above benchmark:

julia> @btime foo()
  3.740 ms (8 allocations: 132.39 KiB)
Pool{Nothing, Int64}(Base.GenericCondition{ReentrantLock}(Base.IntrusiveLinkedList{Task}(nothing, nothing), ReentrantLock(nothing, 0x00000000, 0x00, Base.GenericCondition{Base.Threads.SpinLock}(Base.IntrusiveLinkedList{Task}(nothing, nothing), Base.Threads.SpinLock(0)), (12800, 0, 141733920768))), 10000, 0, #undef, [10001, 10002, 10003, 10004, 10005, 10006, 10007, 10008, 10009, 10010  …  19991, 19992, 19993, 19994, 19995, 19996, 19997, 19998, 19999, 20000])

[Drvi]

Love it! This is much more lean:)
I hope this is as performant in your HTTP.jl benchmarks -- the dummy benchmarks I used were not stressing any concurrent accesses. I think we're now spending more time behind lock, which originally caused you to look into lockfree datastructures.

[quinnj]

That was mainly because we were calling connect while holding the lock, which then depends on network conditions and can easily take milliseconds. I think we should be ok here. The only slight worry is that we do call isvalid while holding the lock, so users have to be careful that their it isn't expensive to call that (it should be fine for how we use it in HTTP)

[nickrobinson251]

Should this release all the permits too?

Also i think it should return the pool, rather than the pool.values[]

Suggested change

	Base.empty!(pool::Pool) = Base.@lock pool.values empty!(pool.values[])
	function Base.empty!(pool::Pool)
	Base.@lock pool.values begin
	foreach(_ -> release(pool.sem), pool.values[])
	empty!(pool.values[])
	end
	return pool
	end

without this, a testset like this would hang:

@testset "empty!" begin
    pool = Pool(Int; max=2)
    # initially populate the pool with 1 object
    x0 = acquire(() -> 0, pool)
    @test x0 == 0
    empty!(pool)
    # we should now be able to refill pool with 2 objects (up to capacity)
    x1 = acquire(() -> 1, pool)
    x2 = acquire(() -> 2, pool)
    @test x1 == 1
    @test x2 == 2
end

[quinnj]

As you noted elsewhere, I think this is just not named properly; we can't release the permits, because then if the tasks that actually did the acquires tried to release, things get all out of sorts. This is just a means to ensure that all existing objects in the pool are evicted so nothing is reused (like forcenew, but more of a global eviction).

[nickrobinson251]

why do we support empty! ?

At first i thought Pool was supporting the exact same interface as Semaphone, which is just acquire/release... but then there's empty!

i think i'm confused by what the contract of empty! is supposed to be here (see my comment above).
what does "removes all objects for reuse" mean? I wonder if empty! is even the right name.

It also makes me wonder if there should also be a release_all(pool) that just releases all the permits? probably not

And if there should be other Dict-like interface functions (e.g. keys/haskey or length)?

[quinnj]

Yeah, let's come up with a better name; HTTP has this function closeall, but that's not quite the right thing I think for a generic pool. Maybe evictall? empty! felt the most right to me, but as you pointed out there's this tension between whether a Pool is just a semaphore or a Dict storing non-checked-out objects.

[quinnj]

Changed the function name to drain! in the PR

[Drvi]

Let's assume noone holds the lock, but task (1) is waiting on the pool.lock condition. Is it racy when task (2) releases from the pool, just as task (3) also tries to acquire? Task 1 would wake up, task 3 would hold the lock and both of them might now pass the pool.cur >= pool.max check?

[quinnj]

Task 1 would wake up, task 3 would hold the lock

This isn't possible. The design of Threads.Condition means that when you "wake up", you are guaranteed exclusive access to the lock.