AtomicChannels

AtomicChannels provides thread-safe, lock-free and task-free channel/queue for Julia.

Features

• AtomicChannel: faster lock-free multi-producer multi-consumer (MPMC) channel; a single-threaded variant is also provided.
• Up to 30 times faster than Base.Channel in contention scenarios (details in benchmark section).
• ReusePool{T}: allows efficient reuse of objects, reducing the overhead of frequent object creation and destruction.
• Blocking and non-blocking APIs.

Installation

using Pkg
Pkg.add("AtomicChannels")

Quick start

AtomicChannel

The core usage of AtomicChannel is similar to a buffered Base.Channel.

• Constructor:
  • AtomicChannel{eltype}(channel_size) (auto-choose thread-safe or -unsafe version)
• Blocking operations:
  • put!(::AtomicChannel{T}, item::T): push item to channel
  • take!(::AtomicChannel{T}): take the next available item
• Non-blocking operations:
  • tryput!(::AtomicChannel{T}, item::T): return ::Bool
  • trytake!(::AtomicChannel{T}): return item or nothing

using AtomicChannels

chnl = AtomicChannel{Int}(2)  # element type=Int, size=2
put!(chnl, 1)
tryput!(chnl, 2)

take!(chnl)          # 1
trytake!(chnl)       # 2
trytake!(chnl)       # nothing

ReusePool

ReusePool is a lightweight object pool implemented on top of AtomicChannel.

The pool allows efficient reuse of objects, reducing the overhead of frequent object creation and destruction.

• Constructor:
  • ReusePool{T}(create::Function, size::Int = 1, reset::Function = identity)
    • create has to be a thread-safe function that creates a new instance of T.
    • reset (optional) should be a function to in-place edit an item that is put back in the pool.
• Blocking operations:
  • take!(::ReusePool): take an object, blocks when empty
  • put!(::ReusePool, item): reset and return an object, blocks when full
• Non-blocking operations:
  • acquire!(::ReusePool): take an object if available, otherwise create a new one
  • release!(ReusePool{T}, item::T): try to return an object to the pool, returns true on success

using AtomicChannels

pool = ReusePool(() -> Vector{Int}(undef, 1024), 4, x -> fill!(x, 0))

buf = acquire!(pool)
buf[1] = 42
release!(pool, buf)

Benchmark

The benchmark reports in benchmark compare AtomicChannel with Base.Channel on Linux (AMD Ryzen Threadripper PRO 7985WX, 128 cores).

Method summary:

• Benchmark target: MPMC put/take throughput (items=100_000) for AtomicChannel vs Base.Channel.
• Metric: speedup is computed as Channel ÷ AtomicChannel (elapsed time in ms).
• Three scenario groups are measured:
  • Case 1: low-capacity contention (capacity=4 or 1).
  • Case 2: higher-capacity, low data contention (capacity=256).
  • Case 3: varying worker/task counts at capacity=256.

Representative speedups from the latest mpmc_result files:

Julia	Case 1: 64 threads, 64 workers, cap=4	Case 2: 64 threads, 64 workers, cap=256	Case 3: 32 threads, 256 workers, cap=256
1.12.6	33.8x	3.023x	27.638x
1.11.9	9.751x	4.4x	2.165x
1.10.11	13.103x	4.876x	2.288x
1.9.4	17.579x	6.155x	2.367x
1.8.5	6.311x	1.994x	1.709x

Conclusion:

• AtomicChannel are faster than Base.Channel in most test scenarios; drastically faster with heavy task switch and multiple concurrent data operations.

Raw benchmark reports:

• benchmark/mpmc_result__1.12.6.md
• benchmark/mpmc_result__1.11.9.md
• benchmark/mpmc_result__1.10.11.md
• benchmark/mpmc_result__1.9.4.md
• benchmark/mpmc_result__1.8.5.md

Reproduce Benchmark

Run from the repository root, and pin the Julia version with juliaup:

cd AtomicChannels.jl

julia +1.12 --project benchmark/mpmc.jl
julia +1.11 --project benchmark/mpmc.jl
julia +1.10 --project benchmark/mpmc.jl
julia +1.9 --project benchmark/mpmc.jl
julia +1.8 --project benchmark/mpmc.jl

Each run generates a report to the benchmark directory.

Documentation

• Stable docs: https://cihga39871.github.io/AtomicChannels.jl/stable/
• Dev docs: https://cihga39871.github.io/AtomicChannels.jl/dev/