QueuePool with more than nthreads?

[kcajf]

Looking at the code, it seems that nthrds is always capped at Threads.nthreads(). Is this restriction necessary? For example, I have some heavily IO bound tasks and I'd like to iterate through them and run, say, 100 of them at once.

[tro3]

Interesting. The cap is because we only want one handler function per thread. But the default handler is what is in your way - it assumes CPU-limited tasks, only allowing one at a time. With a different handler, you can allow more (in the example below, 10):

function bin(fn, xs)
    ys = similar(xs, 0)
    ns = similar(xs, 0)
    for x in xs
        fn(x) ? push!(ys, x) : push!(ns, x)
    end
    return ys, ns
end

function myhandler(pool)
    sleep(0.001)
    tasks = Task[]
    while isready(pool.inq) || length(tasks) > 0
        while isready(pool.inq) && length(tasks) < 10
            push!(tasks, schedule(take!(pool.inq)))
        end
        sleep(0.001)
        done, tasks = bin(istaskdone, tasks)
        for t in done
            put!(pool.outq, t)
            Threads.atomic_sub!(pool.cnt, 1)
        end
    end
end

twith(ThreadPools.QueuePool(1:8, myhandler)) do pool
    tforeach(x -> println((x,Threads.threadid())), pool, 1:100)
end

See if something like the above works for you and let me know.