Rework handle cache so that we never switch tasks in finalizers.

[maleadt]

Should fix #1923

[maleadt]

Hmm, this doesn't seem sufficient. The following MWE, on 1.8, quickly runs violates my assumption that if you surround locks with GC.enable(...) we shouldn't ever enter a finalizer while the lock is taken:

struct Cache{K,V}
    handles::Dict{K,V}
    lock::ReentrantLock

    function Cache{K,V}() where {K,V}
        return new{K,V}(Dict{K,V}(), ReentrantLock())
    end
end

macro safe_lock(l, ex)
    quote
        GC.enable(false)
        lock($(esc(l)))
        try
            $(esc(ex))
        finally
            unlock($(esc(l)))
            GC.enable(true)
        end
    end
end

# get or create value
function Base.pop!(ctor::Function, cache::Cache{K,V}, key) where {K,V}
    # lookup
    function check_cache()
        @safe_lock cache.lock begin
            if !haskey(cache.handles, key)
                nothing
            else
                cache.handles[key]
            end
        end
    end
    handle = check_cache()
    if handle === nothing
        GC.gc(false)
        handle = check_cache()
    end

    # create
    if handle === nothing
        handle = ctor()::V
    end

    return handle
end

# put in cache or destroy value
function Base.push!(dtor::Function, cache::Cache{K,V}, key::K, handle::V) where {K,V}
    # we're running from a finalizer, so we can't lock
    @assert !islocked(cache.lock)

    # cache
    should_destroy = @safe_lock cache.lock begin
        if haskey(cache.handles, key)
            true
        else
            cache.handles[key] = handle
            false
        end
    end

    # destroy
    if should_destroy
        dtor(handle)
    end

    return
end

const handles = Cache{Int, Int}()
mutable struct Foo
    handle

    function Foo()
        x = rand(1:100)
        handle = pop!(handles, x) do
            x
        end
        obj = new(handle)
        finalizer(obj) do _
            push!(handles, handle, handle) do _
                # do nothing
            end
        end
    end
end

function test()
    try
        for nn=1:10
            tids = [Threads.@spawn Foo() for nn=1:10]
            while !all(istaskdone.(tids))
                yield()
            end
        end
    catch e
        error(e)
    end
end

while true
    test()
end

Running this with -t32 quickly shows error in running finalizer: AssertionError(msg="!(islocked(cache.lock))").
@vchuravy Any thoughts?

On 1.9 it runs into #1921 instead, so this is a good MWE for that too.

[maleadt]

Superseded by #2352