Channel (programming)

libthread, a multithreading library for C, which was first created for the operating system Plan 9, offers inter-thread communication based on fixed-size channels.

In Bjarne Stroustrup's "C with Classes" language (which would later become C++), inter-task communication was done through queue classes provided by <task.h>, with a producer task putting data into qhead and a consumer task pulling it out from qhead.^[1]

In C#, the System.Threading.Channels namespace provides bounded/unbounded channels.

using System;
using System.Threading.Channels;
using System.Threading.Tasks;

Channel<string> channel = Channel.CreateBounded<string>(capacity: 10);

// Producer task
Task producer = Task.Run(async () =>
{
    for (int i = 0; i < 20; i++)
    {
        await channel.Writer.WriteAsync($"Message {i}");
    }
    channel.Writer.Complete();
});

// Consumer task
Task consumer = Task.Run(async () =>
{
    await foreach (string message in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine(message);
    }
});

await Task.WhenAll(producer, consumer);

C++ has no standardised channel class in the standard library, but C++29 has a proposal for a concurrent queue class, std::bounded_queue<T, Alloc>, to support C++'s existing coroutine and std::execution framework.^[2]

import std;

using std::bounded_queue;
using std::jthread;
using std::string;

bounded_queue<string> q(10);

// Producer thread
jthread producer([&q] -> void {
    for (int i = 0; i < 20; ++i) {
        q.push_wait(std::format("Message {}", i));
    }
    q.close(); // Signals consumer to finish
});

// Consumer thread
jthread consumer([&q] -> void {
    string message;
    while (q.pop_wait(message) { // Blocks if empty
        std::println("{}", message);
    }
});

This snippet of Go code performs similarly to the XC code. First the channel c is created, then a goroutine is spawned which sends 42 through the channel. When the number is put in the channel x is set to 42. Go allows channels to buffer contents, as well as non blocking receiving through the use of a select block.^[3]

c := make(chan int)

go func() {c <- 42}()

x := <- c

In Java, the java.util.concurrent.BlockingQueue interface and its implementing classes are essentially channels:

• java.util.concurrent.SynchronousQueue is an unbuffered channel.
• java.util.concurrent.ArrayBlockingQueue and java.util.concurrent.LinkedBlockingQueue are buffered channels.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

BlockingQueue<String> q = new ArrayBlockingQueue<>(10); // capacity of 10

// Producer thread
Thread producer = new Thread(() -> {
    try {
        for (int i = 0; i < 20; i++) {
            q.put("Message " + i); // Blocks safely until consumer makes space
        }
        q.put("DONE"); // Poison pill to signal completion
    } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
    }
});

// Consumer thread
Thread consumer = new Thread(() -> {
    try {
        while (true) {
            String message = q.take(); // Blocks safely until item arrives
            if (message.equals("DONE")) {
                break;
            }
            System.out.println(message);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
});

producer.start();
consumer.start();

The package java.nio.channels, though sharing the name "channels", are not actually related; java.nio.channels are in fact buffer-oriented I/O classes.

The Kotlin Channel API provides channels; while a kotlinx.coroutines.Deferred allows passing a single value between coroutines, a kotlinx.coroutines.channels.Channel provides a way to transfer a continuous stream of values safely. It is conceptually similar to java.util.concurrent.BlockingQueue, but has suspending operations.

• kotlinx.coroutines.channels.SendChannel is used by the producer coroutine to push data using the suspending send() method.
• kotlinx.coroutines.channels.ReceiveChannel is used by the consumer coroutine to pull data using the suspending receive() method, or by iterating over the channel with a for loop.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.produce

fun main() = runBlocking {
    val channel: ReceiveChannel<String> = produce(Dispatchers.Default, capacity = 10) {
        for (i in 0 until 20) {
            channel.send("Message $i")
        }
    }

    // Consumer coroutine
    launch(Dispatchers.Default) {
        for (message in channel) {
            println(message)
        }
    }
}

The Love2D library which uses the Lua programming language implements channels with push and pop operations similar to stacks. The pop operation will block so as long as there is data resident on the stack. A demand operation is equivalent to pop, except it will block until there is data on the stack

-- A string containing code which will be interpreted by a function such as loadstring(),
-- but on the C side to start a native thread.

local threadCode = [[
    love.thread.getChannel("test"):push("Hello world!")
]]


function love.load()
    -- Start the thread.
    
    thread = love.thread.newThread(threadCode)
    thread:start()
    
    -- The thread will block until "Hello world!" is popped off channel test's stack.
    -- Because the channel can be popped from before the thread first executes, there may not be data on the stack.
    -- in that case use :demand() instead of :pop() because :demand() will block until there is data on the stack and then return the data.
    
    print(love.thread.getChannel("test"):demand())
    
    -- The thread can now finish.
end

The OCaml Event module offers typed channels for synchronization, with 'a channel and 'a event. When the module's send and receive functions are called, they create corresponding send and receive events which can be synchronized.

Rust provides asynchronous channels for communication between threads. Channels allow a unidirectional flow of information between two endpoints: the std::sync::mpsc::Sender and the std::sync::mpsc::Receiver.^[4]

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        tx.send(123).unwrap();
    });

    let result = rx.recv();
    println!("{result:?}");
}

The XMOS programming language XC provides a primitive type "Chan" and two operators "<:" and ":>" for sending and receiving data from a channel.^[5]

In this example, two hardware threads are started on the XMOS, running the two lines in the "par" block. The first line transmits the number 42 through the channel while the second waits until it is received and sets the value of x. The XC language also allows asynchronous receiving on channels through a select statement.

chan c;
int x;
par {
  c <: 42;
  c :> x;
}