# Concurrency Study – SPMC

ref:
[linux2021 Concurrency Study -- SPMC](/Lexlfim5Ri-ptGWFJ6NgHw)
[spmc](https://github.com/sysprog21/concurrent-programs/blob/master/spmc/spmc.c#L128)

## spmc_enqueue

if (!IS_WRITABLE(idx, node)) {
        spmc_node_t *const next =
            atomic_load_explicit(&node->next, memory_order_relaxed);
        /* Never move to write on top of the node that is currently being read;
         * In that case, items would be read out of order they were enqueued.
         */
        if (next !=
            atomic_load_explicit(&spmc->curr_dequeue, memory_order_relaxed)) {
            node = next;
            goto retry;
        }

        const uint8_t power = ++spmc->last_power;
        assert(power < sizeof(size_t) * CHAR_BIT);
        const size_t cap = 1 << power;
        spmc_node_t *new_node = malloc(SIZE_FROM_CAP(cap, sizeof(spmc_node_t)));
        if (!new_node)
            return false;

        init_node(new_node, next, cap);
        atomic_store_explicit(&node->next, new_node, memory_order_release);
        idx = 0;
        node = new_node;
    }

IS_WRITABLE ensures that the node is not full.

If the node is full, there are two circumstances:

1. The ringbuffer is not full.
2. The ringbuffer is now full.(The consumers are now handling the next node.)

In the first situation, producer only moves to the next available node and store data. Shown as below.

if (next !=
    atomic_load_explicit(&spmc->curr_dequeue, memory_order_relaxed)) {
    node = next;
    goto retry;
        }

When the the ringbuffer is full, we allocate even bigger capacity for the new node. Store new data. As the rest of the above code.