How to Implement a Producer-Consumer Pattern Using Fiber Channels in libfiber

Use libfiber's acl_channel_create, acl_channel_send, and acl_channel_recv functions to build bounded queues between coroutines that automatically suspend producers when full and consumers when empty, enabling lock-free communication within a single-threaded scheduler.

The iqiyi/libfiber library provides a high-performance coroutine implementation with a channel abstraction specifically designed for the producer-consumer pattern using fiber channels. Unlike traditional thread queues that rely on kernel mutexes, these channels operate within a single-threaded scheduler, allowing fibers to communicate through lock-free buffered transfers defined in c/include/fiber/fiber_channel.h and implemented in c/src/sync/channel.c.

Understanding Fiber Channel Architecture

Libfiber channels function as bounded queues connecting concurrent fibers. The core structure ACL_CHANNEL manages an alternating array (FIBER_ALT) that tracks pending send and receive operations. Because the scheduler runs all fibers within a single thread (unless you explicitly start multiple schedules), channel operations require no locks—only pointer exchanges and fiber queue manipulation via acl_fiber_switch in c/src/sync/channel.c.

For C++ developers, the library provides a thin template wrapper declared in cpp/include/fiber/channel.hpp and implemented in cpp/src/channel.cpp. The acl::channel<T> class encapsulates the raw C API, offering type-safe put() and pop() methods that handle sizeof(T) automatically.

C API Implementation

Creating and Configuring the Channel

Initialize a channel using acl_channel_create(int elemsize, int bufsize), defined in c/include/fiber/fiber_channel.h. The function allocates an ACL_CHANNEL structure with a buffer holding bufsize elements of elemsize bytes each.

#include "fiber/fiber_channel.h"

/* Create a channel for integers with a buffer of 10 items */
ACL_CHANNEL *chan = acl_channel_create(sizeof(int), 10);

Sending and Receiving Data

Producers write data using acl_channel_send(ACL_CHANNEL *c, void *v), which copies the element into the channel buffer. If the buffer is full, the producer fiber automatically suspends until a consumer creates space. Consumers retrieve data using acl_channel_recv(ACL_CHANNEL *c, void *v), which blocks until an element is available.

Complete C Example

The following example creates one producer and two consumers sharing a single channel. The producer sends a sentinel value -1 to signal termination.

/* prod_cons.c */
#include <stdio.h>
#include <stdlib.h>
#include "fiber/lib_fiber.h"
#include "fiber/fiber_channel.h"

#define COUNT 1000

static void prod_fiber(ACL_FIBER *fb, void *ctx)
{
    ACL_CHANNEL *chan = (ACL_CHANNEL *)ctx;
    for (int i = 0; i < COUNT; ++i) {
        acl_channel_send(chan, &i);  /* Blocks if buffer full */
        printf("produced %d\n", i);
    }
    int done = -1;
    acl_channel_send(chan, &done);   /* Sentinel */
}

static void cons_fiber(ACL_FIBER *fb, void *ctx)
{
    ACL_CHANNEL *chan = (ACL_CHANNEL *)ctx;
    int val;
    while (1) {
        acl_channel_recv(chan, &val);  /* Blocks if empty */
        if (val == -1) break;
        printf("consumed %d\n", val);
    }
}

int main(void)
{
    ACL_CHANNEL *chan = acl_channel_create(sizeof(int), 10);
    
    acl_fiber_create(prod_fiber, chan, 128000);
    acl_fiber_create(cons_fiber, chan, 128000);
    acl_fiber_create(cons_fiber, chan, 128000);
    
    acl_fiber_schedule();  /* Run the scheduler */
    
    acl_channel_free(chan);
    return 0;
}

C++ Template Implementation

Type-Safe Channel Interface

The C++ header cpp/include/fiber/channel.hpp defines acl::channel<T>, a template class that automatically calculates element sizes. This wrapper provides put(const T&) and pop(T&) methods that delegate to the underlying C functions in c/src/sync/channel.c.

Complete C++ Example

This example passes custom Message structures between fibers using the C++ API:

/* prod_cons.cpp */
#include <cstdio>
#include <string>
#include "fiber/libfiber.hpp"
#include "fiber/channel.hpp"

struct Message {
    int id;
    std::string text;
};

static void producer(acl::fiber* fb, void* ctx)
{
    acl::channel<Message> *chan = static_cast<acl::channel<Message>*>(ctx);
    for (int i = 0; i < 20; ++i) {
        Message m{ i, "msg#" + std::to_string(i) };
        chan->put(m);  /* Suspends if buffer (size 5 default) is full */
        printf("produced %d: %s\n", m.id, m.text.c_str());
    }
    Message term{-1, ""};
    chan->put(term);
}

static void consumer(acl::fiber* fb, void* ctx)
{
    acl::channel<Message> *chan = static_cast<acl::channel<Message>*>(ctx);
    Message m;
    while (true) {
        chan->pop(m);  /* Suspends until message arrives */
        if (m.id == -1) break;
        printf("consumed %d: %s\n", m.id, m.text.c_str());
    }
}

int main()
{
    acl::channel<Message> chan;  /* Default buffer length applies */
    
    acl::fiber::create(producer, &chan, 128000);
    acl::fiber::create(consumer, &chan, 128000);
    acl::fiber::create(consumer, &chan, 128000);
    
    acl::fiber::schedule();
    return 0;
}

Internal Blocking and Scheduling Mechanics

When acl_channel_send executes in c/src/sync/channel.c, it registers the operation in the channel's FIBER_ALT alternating array. If a consumer is already blocked on acl_channel_recv, the scheduler matches the pair immediately and invokes acl_fiber_switch to resume the consumer while suspending the producer until the copy completes.

If no consumer is waiting and the buffer has space, the data copies directly into the ring buffer. When the buffer reaches capacity, the producer fiber yields control entirely, sleeping in the channel's wait queue until a subsequent receive creates room. This mechanism eliminates spin-locking and kernel syscalls, reducing context switch overhead to a simple stack swap within the fiber scheduler.

Summary

• Create channels with acl_channel_create (C) or acl::channel<T> (C++) to establish typed, bounded queues between fibers.
• Blocking operations (acl_channel_send/acl_channel_recv) automatically suspend fibers using the internal FIBER_ALT array and acl_fiber_switch, eliminating manual synchronization.
• Lock-free design relies on the single-threaded scheduler assumption; no mutexes are acquired during channel operations in c/src/sync/channel.c.
• Multiple producers and consumers can share one channel instance safely within the same scheduler instance.

Frequently Asked Questions

How do fiber channels differ from thread-based blocking queues?

Fiber channels use cooperative scheduling within a single thread, eliminating kernel context switches and lock contention. When a producer blocks on a full buffer, libfiber's scheduler in c/src/sync/channel.c immediately switches to a ready consumer via acl_fiber_switch, whereas thread queues rely on mutexes and condition variables that trap into kernel space.

What happens when a channel buffer reaches capacity?

The calling fiber automatically suspends. In acl_channel_send, the operation registers with the internal alternating array and yields control until a matching receive creates space. Both fibers resume once the data transfers, without explicit locking or polling.

Can multiple producers and consumers share a single channel?

Yes. The FIBER_ALT structure in c/src/sync/channel.c supports multiple pending send and receive operations. You can start several producer fibers and consumer fibers—all sharing one ACL_CHANNEL instance—and the scheduler pairs them as data becomes available.

Are fiber channels thread-safe across different OS threads?

No. Channels assume all fibers run under a single-threaded scheduler (acl_fiber_schedule). They are lock-free only because no preemption occurs during channel operations. For cross-thread communication, you must either run separate schedulers in each thread with appropriate isolation or use external synchronization primitives.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}