Repository: netty/netty
File: transport-classes-io_uring/src/main/java/io/netty/channel/uring/CompletionBuffer.java (Java)

Code before:
final class CompletionBuffer {
    private final CompletionCallback callback = this::add;
    // long[(tail + 1) % capacity] holds res and flags (packed as long) and long[(tail + 2) % capacity] the udata.
    private final long[] array;
    private final int capacity;
    private final int mask;
    private final long tombstone;
    private int size;
    private int head;
    private int tail = -1;

    CompletionBuffer(int numCompletions, long tombstone) {
        capacity = MathUtil.findNextPositivePowerOfTwo(numCompletions * 2);
        array = new long[capacity];
        mask = capacity - 1;
        for (int i = 0; i < capacity; i += 2) {
            array[i] = tombstone;
        }
        this.tombstone = tombstone;
    }

    private boolean add(int res, int flags, long udata) {
        if (size == capacity) {
            // The buffer is already full.
            return false;
        }
        if (udata == tombstone) {
            throw new IllegalStateException("udata can't be the same as the tombstone");
        }
        // Pack res and flag into the first slot.
        array[combinedIdx(tail + 1)] = (((long) res) << 32) | (flags & 0xffffffffL);
        array[udataIdx(tail + 1)] = udata;

        tail += 2;
        size += 2;
        return size < capacity;
    }

    /**
     * Drain completions from the given {@link CompletionQueue}.
     *
     * @param queue the queue to drain from.
     * @return      {@code true} if the whole queue was drained, {@code false} otherwise.
     */
    boolean drain(CompletionQueue queue) {
        queue.process(callback);
        return !queue.hasCompletions();
    }

    /**
     * Process buffered completions via the given {@link CompletionCallback}.

Review comment (suggestion):
I think we still have a problem with full `CompletionBuffers`. The `CompletionQueue` calls `handle` _after_ it's incremented the `head` counter, so the boolean return value we have here is really expected to mean "we've accepted this value but can't take any more".

I think either the `CompletionBuffer::drain` method should not call `queue.process()` if we're already full.
If we instead make the `CompletionQueue` not increment the head counter unless the `handle` call, then we might get reentrancy problems, and it also might not work with how other tasks want to signal to stop processing.

Code after:
final class CompletionBuffer {
    private final CompletionCallback callback = this::add;
    // long[(tail + 1) % capacity] holds res and flags (packed as long) and long[(tail + 2) % capacity] the udata.
    private final long[] array;
    private final int capacity;
    private final int mask;
    private final long tombstone;
    private int size;
    private int head;
    private int tail = -1;

    CompletionBuffer(int numCompletions, long tombstone) {
        capacity = MathUtil.findNextPositivePowerOfTwo(numCompletions * 2);
        array = new long[capacity];
        mask = capacity - 1;
        for (int i = 0; i < capacity; i += 2) {
            array[i] = tombstone;
        }
        this.tombstone = tombstone;
    }

    private boolean add(int res, int flags, long udata) {
        if (udata == tombstone) {
            throw new IllegalStateException("udata can't be the same as the tombstone");
        }
        // Pack res and flag into the first slot.
        array[combinedIdx(tail + 1)] = (((long) res) << 32) | (flags & 0xffffffffL);
        array[udataIdx(tail + 1)] = udata;

        tail += 2;
        size += 2;
        return size < capacity;
    }

    /**
     * Drain completions from the given {@link CompletionQueue}.
     *
     * @param queue the queue to drain from.
     * @return      {@code true} if the whole queue was drained, {@code false} otherwise.
     */
    boolean drain(CompletionQueue queue) {
        if (size == capacity) {
            // The buffer is already full.
            return false;
        }
        queue.process(callback);
        return !queue.hasCompletions();
    }

    /**
     * Process buffered completions via the given {@link CompletionCallback}.