Usage: Strange timing behaviour when setting up threads once and sending functions to be run

I'm using `CheapThreads` to setup all threads except thread 1 with a function that listens to a `Channel`, and then runs another function according to what is in the channel. I'm not seeing the speedup over just running it on one thread that I expect.

Here is a MWE of the code I'm using (sorry it's not the prettiest):

```Julia
using CheapThreads
using ThreadingUtilities
using VectorizationBase
using StrideArraysCore: object_and_preserve
using BenchmarkTools

struct BatchClosure{F, A, B}
f::F
end
function (b::BatchClosure{F,A,B})(p::Ptr{UInt}) where {F,A,B}
(offset, args) = ThreadingUtilities.load(p, A, 2*sizeof(UInt))
b.f(args)
B && free_local_threads!()
nothing
end

@inline function batch_closure(f::F, args::A, ::Val{B}) where {F,A,B}
bc = BatchClosure{F,A,B}(f)
@cfunction($bc, Cvoid, (Ptr{UInt},))
end

@inline function setup_batch!(p::Ptr{UInt}, fptr::Ptr{Cvoid}, argtup)
offset = ThreadingUtilities.store!(p, fptr, sizeof(UInt))
offset = ThreadingUtilities.store!(p, argtup, offset)
nothing
end

@generated function _launch_working_threads(
f!::F, threadmask, nthread, args::Vararg{Any,K}
) where {F,K}
q = quote
threads = CheapThreads.UnsignedIteratorEarlyStop(threadmask, nthread)
end
block = quote
i = 0x00000000
tid = 0x00000000
tm = CheapThreads.mask(threads)
while true
VectorizationBase.assume(tm ≠ zero(tm))
tz = trailing_zeros(tm) % UInt32
i += 0x00000001
tz += 0x00000001
tid += tz
tm >>>= tz
launch_batched_thread!(cfunc, tid, argtup)
i == nthread && break
end
nothing
end
gcpr = Expr(:gc_preserve, block, :cfunc)
argt = Expr(:tuple)
for k ∈ 1:K
CheapThreads.add_var!(q, argt, gcpr, args[k], :args, :gcp, k)
end
push!(q.args, :(argtup = $argt), :(cfunc = batch_closure(f!, argtup, Val{false}())), gcpr)
push!(q.args, nothing)
q
end

@inline function launch_batched_thread!(cfunc, tid, argtup)
fptr = Base.unsafe_convert(Ptr{Cvoid}, cfunc)
ThreadingUtilities.launch(tid, fptr, argtup) do p, fptr, argtup
setup_batch!(p, fptr, argtup)
end
end

function launch_threads(p, inchannels, outchannels)
nthreads_to_launch = p.nthreads - 1
torelease_vec = zeros(UInt8, nthreads_to_launch)
threadmask_vec = zeros(UInt8, nthreads_to_launch)
for i in 1:nthreads_to_launch
nthread=1
threads, torelease = CheapThreads.request_threads(Base.Threads.threadid(), nthread)
threadmask = CheapThreads.mask(threads)
_launch_working_threads(thread_workers,
threadmask,
nthread,
p,
inchannels,
outchannels)
torelease_vec[i] = torelease
threadmask_vec[i] = threadmask
end
return torelease_vec, threadmask_vec
end

@generated function _stop_working_threads(threadmask, nthread, torelease)
q = quote
threads = CheapThreads.UnsignedIteratorEarlyStop(threadmask, nthread)
end
block = quote
tm = CheapThreads.mask(threads)
tid = 0x00000000
while true
VectorizationBase.assume(tm ≠ zero(tm))
tz = trailing_zeros(tm) % UInt32
tz += 0x00000001
tm >>>= tz
tid += tz
ThreadingUtilities.wait(tid)
iszero(tm) && break
end
CheapThreads.free_threads!(torelease)
nothing
end
push!(q.args, block)
push!(q.args, nothing)
q
end

function stop_workers(inchannels, outchannels, torelease_vec, threadmask_vec)
for channel in inchannels
put!(channel, :stop_workers)
end
for channel in outchannels
take!(channel)
end
for i in eachindex(torelease_vec)
_stop_working_threads(threadmask_vec[i], 1, torelease_vec[i])
end
end

function f1!(p, start, stop)
for i in start:stop
p.vec1[i] = p.vec2[i] * occursin("1", string(i))
end
end

thread_workers(args::Tuple) = thread_workers(args...)
function thread_workers(p, inchannels, outchannels)
# Set up thread specific variables
tid = Threads.threadid()
start = p.start[tid]
stop = p.stop[tid]
inchannel = inchannels[tid-1] # calling thread doesn't have a channel
outchannel = outchannels[tid-1] # calling thread doesn't have a channel

# Work
spin_down = false
while !spin_down
fname = take!(inchannel)
if fname == :f1!
f1!(p, start, stop)
elseif fname == :stop_workers
spin_down = true
end
put!(outchannel, spin_down)
end
end

function send_to_channels(fname, inchannels, n_channnels)
for channel in @view(inchannels[1:n_channnels])
put!(channel, fname)
end
end

function wait_for_channels(outchannels, n_channnels)
for channel in @view(outchannels[1:n_channnels])
take!(channel)
end
end

function do_local(fname, p)
if fname == :f1!
f1!(p, p.start[1], p.start[2])
end
end

# nchannels is for debugging, allows testing on subset of threads
function send_to_workers(fname::Symbol, p, inchannels, outchannels, n_channnels=num_threads()-1)
send_to_channels(fname, inchannels, n_channnels)
do_local(fname, p)
wait_for_channels(outchannels, n_channnels)
end

struct P1
vec1::Vector{Int64}
vec2::Vector{Int64}
start::Vector{Int64}
stop::Vector{Int64}
nthreads::Int64
end

function get_start_and_stop_indices(total_length, nbatches)
fld_val = fld(total_length, nbatches)
start_indices = collect(1:fld_val:fld_val*nbatches)
stop_indices = vcat(collect(fld_val:fld_val:total_length-fld_val), total_length)
return start_indices, stop_indices
end

len = 8000
p = P1(zeros(Int64, len), ones(Int64, len), get_start_and_stop_indices(len, num_threads())..., num_threads())
inchannels = [Channel{Symbol}() for i in 1:num_threads()-1]
outchannels = [Channel{Bool}() for i in 1:num_threads()-1]

# Launch threads
torelease_vec, threadmask_vec = launch_threads(p, inchannels, outchannels)

# Run code
send_to_workers(:f1!, p, inchannels, outchannels)
```

And here are the strange benchmarks with 8 threads (note setting the last argument of `send_to_workers` affects the number of threads used but not how the range is split):

```Julia
julia> @btime f1!(p, 1, len)
452.311 μs (16000 allocations: 750.00 KiB)
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,0) # no workers, only running 1/8th on main thread
55.999 μs (2002 allocations: 93.84 KiB)
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,1) # 1 worker running
58.974 μs (4003 allocations: 187.63 KiB) # approx 1/8 of full running, expected
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,2)
63.432 μs (6005 allocations: 281.44 KiB)
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,3)
68.613 μs (8008 allocations: 375.28 KiB)
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,4)
116.588 μs (10011 allocations: 469.13 KiB)
julia> @btime send_to_workers(:f1!, p, inchannels, outchannels,7) # All workers running
150.549 μs (16019 allocations: 750.63 KiB)
```

```
# Stop threads - run this after any benchmarks to stop the functions on the threads
stop_workers(inchannels, outchannels, torelease_vec, threadmask_vec)
```
```
I expect the allocations to increase, but expect each thread to finish in approx 60 μs regardless of how many threads are being used. Would be grateful for some help with what is going on here!