Channel: Add batched append! and take! operations

[adienes]

Add append!(c::Channel, iter) and take!(c::Channel, n[, buffer]) for moving many items through a channel at once.

For a buffered channel, both operations acquire the channel lock once and copy a whole run of items into or out of the internal buffer before releasing it, rather than locking, notifying, and updating bookkeeping per item as a put!/take! loop does. This drastically reduces the number of lock and notify operations and is substantially faster when throughput matters. Bulk transfer is restricted to the cases where it is safe and beneficial: append! only batches when the source is an AbstractArray whose elements already have an acceptable type, and otherwise falls back to a plain put! loop with identical semantics; take!(c, n) batches drains from any buffered channel.

The operations are not atomic: items may become visible to consumers before the call returns, and a batched take! returns however many items were available when the channel closes. Both are exception-safe, leaving the channel's buffer and availability count consistent if iterating the source or copying an element throws.

This revives the idea explored in JuliaLang#56473.

Benchmarks

Measured on this branch (julia 1.14.0-DEV, single thread). Each figure is the minimum wall time over 20 trials, each on a fresh channel sized so nothing blocks — isolating the per-item overhead the bulk path removes.

Producer — append!(c, data) vs a per-item put! loop:

N	put! loop	append!	speedup
1,000	13.4 µs	1.5 µs	8.9×
10,000	125.3 µs	3.5 µs	35.4×
100,000	1227 µs	13.5 µs	90.9×

Consumer — take!(c, n) vs a per-item take! loop:

N	take! loop	take!(c, n)	speedup
1,000	11.1 µs	1.9 µs	5.8×
10,000	112.5 µs	3.5 µs	32.2×
100,000	1122 µs	13.0 µs	86.1×

Even versus the best you can already do without these methods — manually grabbing the lock once and push!-ing in a loop — the bulk copy is still several times faster, because it skips the per-item notify/bookkeeping and uses a single copyto!:

N	hand-rolled lock-once loop	append!	speedup
1,000	4.5 µs	1.5 µs	2.9×
10,000	29.8 µs	2.7 µs	11.2×
100,000	265.3 µs	14.2 µs	18.7×

The headline gains are on throughput in an uncontended buffer, which is what the per-item overhead dominates. Under heavy consumer contention the producer-side change still helps by cutting wakeup traffic (one notify per chunk instead of per item): an 8-thread test feeding 50k items through a 64-slot buffer to M competing consumers (which still take per-item) showed a steady ~1.2–1.3× from M=2 to M=32.

Reproduce

bestof(reps, setup, body) = minimum(((c = setup(); GC.gc(); @elapsed body(c)) for _ in 1:reps))

fill_loop!(c, data) = (for x in data; put!(c, x); end)
drain_loop!(c, n)   = (for _ in 1:n; take!(c); end)

for N in (10^3, 10^4, 10^5)
    data = collect(1:N)
    tloop = bestof(20, () -> Channel{Int}(N), c -> fill_loop!(c, data))
    tapp  = bestof(20, () -> Channel{Int}(N), c -> append!(c, data))
    mk()  = (c = Channel{Int}(N); append!(c, data); c)
    tdl   = bestof(20, mk, c -> drain_loop!(c, N))
    tdb   = bestof(20, mk, c -> take!(c, N))
    println("N=$N  put!-loop=$(round(tloop*1e6;digits=1))µs append!=$(round(tapp*1e6;digits=1))µs ($(round(tloop/tapp;digits=1))x)",
            "  take!-loop=$(round(tdl*1e6;digits=1))µs take!(c,n)=$(round(tdb*1e6;digits=1))µs ($(round(tdl/tdb;digits=1))x)")
end

---

This pull request was written with the assistance of generative AI (Claude Code).