Channels in Action

channel.jl

using Random
using ProgressMeter

function myproducer(ch::Channel, seed)
    Random.seed!(seed)
    while true
        r = rand()
        if r > 0.5
            put!(ch, r)
        else
            sleep(r)
        end
    end
end

function mypipe(inC::Channel, outC::Channel)
    while isopen(inC) && isopen(outC)
        data = take!(inC)
        result = 1 - data
        sleep(result) # do something that takes some time
        put!(outC, result)
    end
end

function myconsumer(c::Channel, n::Int)
    s = 0.
    @showprogress for _ in 1:n
        s += take!(c)
    end
    @show s
end

function main(n=50; seed=54321)
    sz = 10
    i = Channel(c -> myproducer(c, seed), sz)
    o = Channel(sz)
    task = @async mypipe(i, o)
    @time myconsumer(o, n)
    close(i); close(o)
end

main()

naive.jl

using Random
using ProgressMeter

Base.@kwdef struct NaiveProducer
    n::Int
    seed::Int
end

Base.length(lc::NaiveProducer) = lc.n

function Base.iterate(lc::NaiveProducer, state=1)
    state == 1 && Random.seed!(lc.seed)
    state > lc.n && return nothing
    r = 0.
    while true
        r = rand()
        if r > 0.5
            break
        else
            sleep(r)
        end
    end
    data = r
    result = 1 - data
    sleep(result)
    return (data, state+1)
end

function straightforward(n, seed)
    s = 0.
    @showprogress for data in NaiveProducer(;n, seed)
        result = 1 - data
        sleep(result)
        s += result
    end
    @show s
end

function main(n=50; seed=54321)
    @info "doing $(nameof(straightforward))"
    @time straightforward(n, seed)
end

main()

Ref: https://twitter.com/MathSorcerer/status/1454708261689565186?s=20