How wait_group Synchronizes Multiple Fibers and Threads in libfiber

The wait_group primitive uses a 64-bit packed atomic state and a lock-free mailbox to block until all fibers and native threads signal completion, providing a unified synchronization mechanism across heterogeneous execution contexts.

The wait_group class in the iqiyi/libfiber repository eliminates the need for separate synchronization constructs when coordinating mixed workloads. By combining atomic operations with a fiber-aware mailbox system, this lightweight primitive allows a single instance to safely manage concurrency between user-level fibers and OS-level threads.

Core Architecture and State Management

The implementation relies on a compact, lock-free design centered around atomic operations and a specialized mailbox structure.

Packed 64-Bit Atomic State

At the heart of wait_group lies a 64-bit atomic integer (state_) declared in cpp/include/fiber/wait_group.hpp that encodes two distinct counters:

• High 32 bits: Tracks the number of outstanding tasks (incremented via add())
• Low 32 bits: Tracks the number of waiting entities (fibers or threads blocked in wait())

The implementation in cpp/src/wait_group.cpp uses atomic_int64_* operations and compare-and-swap (atomic_int64_cas) to ensure race-free updates. This packing allows the primitive to check and modify both counters in a single atomic operation, eliminating the need for mutex locks.

The fiber_tbox Mailbox

The box_ member is a fiber_tbox<unsigned long>* that serves as a lock-free waiting queue. When wait() blocks, it pushes a token into this mailbox; when tasks complete, the add() method pushes NULL values (box_->push(NULL)) to wake waiting fibers. This mechanism works uniformly for both fibers (which yield to the scheduler) and native threads (which perform a spin-wait yielding the CPU).

Synchronization Methods and Semantics

The public API consists of three primary methods that coordinate the lifecycle of concurrent work.

Adding and Signaling Work

The add(int n) method atomically increments the task counter by shifting n into the high 32 bits. It enforces strict usage rules:

• Concurrent modification protection: If add() is called with a positive n while waiters exist (low 32 bits > 0), the implementation triggers a fatal error: "Add called concurrently with wait"
• Automatic wake-up: When the task counter decrements to zero and waiters are present, add() automatically resets the state to zero and fills the mailbox to unblock all waiting entities

The done() method provides a convenient shortcut that simply calls add(-1) to signal single task completion.

Blocking with wait()

The wait() method implements the blocking logic through a multi-step process:

1. Immediate check: If the task counter is already zero, it returns immediately
2. Waiter registration: Atomically increments the low 32-bit waiter counter using compare-and-swap
3. Mailbox blocking: Calls box_->pop(-1, &found) to suspend execution (yielding for fibers, spinning for threads)
4. Validation: Upon waking, if the task counter is not zero, it aborts with a fatal error indicating reuse before the previous wait cycle completed

Because the mailbox abstracts the underlying scheduling mechanism, the same code path handles both fiber context switches and thread blocking transparently.

Coordinating Fibers and Native Threads

The primitive's key advantage is its ability to synchronize heterogeneous concurrency models without separate code paths.

For fibers: When a fiber calls wait(), it blocks on the fiber_tbox. The fiber scheduler later resumes it when add() pushes a completion token, making the wait non-blocking for the underlying kernel thread.

For native threads: A thread calling wait() also blocks on the same mailbox. Internally, the pop operation performs a spin-wait that yields the CPU, allowing other threads to make progress while maintaining the same API semantics.

This design allows wait_group instances to be shared safely across any mix of fibers and threads, as demonstrated in the library's own fiber pool implementation.

Practical Code Examples

Basic Usage with Mixed Concurrency

The following example from samples/cxx/waite_group/main.cpp demonstrates synchronizing native threads and fibers simultaneously:

#include "fiber/wait_group.hpp"
#include "fiber/go_fiber.hpp"
#include <thread>
#include <iostream>

int main() {
    acl::wait_group wg;
    const int n_threads = 2, n_fibers = 3;

    wg.add(n_threads + n_fibers);

    // Launch native threads
    for (int i = 0; i < n_threads; ++i) {
        std::thread([&wg, i] {
            std::cout << "Thread " << i << " running\n";
            wg.done();
        }).detach();
    }

    // Launch fibers
    for (int i = 0; i < n_fibers; ++i) {
        go[&wg] {
            std::cout << "Fiber " << acl::fiber::self() << " running\n";
            wg.done();
        };
    }

    // Wait from a fiber context
    go[&wg] {
        wg.wait();
        std::cout << "All tasks completed (fiber)\n";
    };

    // Wait from the main thread
    wg.wait();
    std::cout << "All tasks completed (main thread)\n";
}

Integration with Fiber Pools

The wait_group integrates naturally with the fiber pool implementation in cpp/src/fiber_pool.cpp:

#include "fiber/fiber_pool.hpp"
#include "fiber/wait_group.hpp"

void task(acl::wait_group* wg, int id) {
    std::cout << "Task " << id << " running in fiber\n";
    wg->done();
}

int main() {
    acl::fiber_pool pool(4);
    acl::wait_group wg;
    const int tasks = 10;
    
    wg.add(tasks);

    for (int i = 0; i < tasks; ++i) {
        pool.spawn([&wg, i] { task(&wg, i); });
    }

    wg.wait();
    std::cout << "All pool tasks done\n";
}

Summary

• Atomic packing: wait_group uses a 64-bit atomic value to track both outstanding tasks (high 32 bits) and waiting entities (low 32 bits) without locks
• Unified blocking: The fiber_tbox mailbox abstracts fiber scheduling and thread blocking, allowing the same primitive to work across both execution contexts
• Strict lifecycle: The implementation prevents concurrent add() calls during active waits and prohibits reuse before completion, enforcing correct synchronization patterns
• Zero-overhead coordination: By leveraging atomic_int64_cas operations, the primitive coordinates termination without kernel-level synchronization primitives for the common case

Frequently Asked Questions

Can a wait_group instance be reused after wait() returns?

No, the current implementation in cpp/src/wait_group.cpp does not support immediate reuse. If wait() returns and the task counter is not zero, or if you attempt to call add() again before all waiters have cleared, the code triggers a fatal error indicating reuse before the previous wait finished. You should create a new wait_group instance for each synchronization phase.

Is it safe to call add() from multiple threads concurrently?

Only if no fibers or threads are currently blocked in wait(). The code explicitly checks for waiters (low 32 bits > 0) when processing positive increments in add(int n). If it detects active waiters while adding new tasks, it aborts with the error "Add called concurrently with wait". This design prevents race conditions where new work arrives while others are waiting for completion.

How does wait_group handle fiber scheduling differently than thread blocking?

Fibers calling wait() suspend execution through the fiber_tbox::pop() method, which yields control back to the fiber scheduler, allowing other fibers on the same thread to run. Native threads calling the same method enter a spin-wait that yields the CPU but maintains the thread's blocked state. Both mechanisms use the same mailbox tokens to resume execution when done() or add(-1) decrements the task counter to zero.

What happens if done() is called more times than add()?

Calling done() (which invokes add(-1)) when the task counter is already zero results in an underflow in the high 32 bits of the atomic state. While the specific behavior isn't detailed in the header guards, the implementation treats the task counter reaching zero as the completion signal. Negative values would likely corrupt the state tracking and potentially trigger the mailbox wake-up logic unnecessarily, so proper balancing of add() and done() calls is essential.

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