Batch

pub const Batch = struct

File

Code

pub const Batch = struct {
    storage: []Operation.Storage,
    unused: Operation.List,
    submitted: Operation.List,
    pending: Operation.List,
    completed: Operation.List,
    userdata: ?*anyopaque align(@max(@alignOf(?*anyopaque), 4)),

    /// After calling this, it is safe to unconditionally defer a call to
    /// `cancel`. `storage` is a pre-allocated buffer of undefined memory that
    /// determines the maximum number of active operations that can be
    /// submitted via `add` and `addAt`.
    pub fn init(storage: []Operation.Storage) Batch {
        var prev: Operation.OptionalIndex = .none;
        for (storage, 0..) |*operation, index| {
            operation.* = .{ .unused = .{ .prev = prev, .next = .fromIndex(index + 1) } };
            prev = .fromIndex(index);
        }
        storage[storage.len - 1].unused.next = .none;
        return .{
            .storage = storage,
            .unused = .{
                .head = .fromIndex(0),
                .tail = .fromIndex(storage.len - 1),
            },
            .submitted = .empty,
            .pending = .empty,
            .completed = .empty,
            .userdata = null,
        };
    }

    /// Adds an operation to be performed at the next await call.
    /// Returns the index that will be returned by `next` after the operation completes.
    /// Asserts that no more than `storage.len` operations are active at a time.
    pub fn add(batch: *Batch, operation: Operation) u32 {
        const index = batch.unused.head.toIndex();
        batch.addAt(index, operation);
        return index;
    }

    /// Adds an operation to be performed at the next await call.
    /// After the operation completes, `next` will return `index`.
    /// Asserts that the operation at `index` is not active.
    pub fn addAt(batch: *Batch, index: u32, operation: Operation) void {
        const storage = &batch.storage[index];
        const unused = storage.unused;
        switch (unused.prev) {
            .none => batch.unused.head = unused.next,
            else => |prev_index| batch.storage[prev_index.toIndex()].unused.next = unused.next,
        }
        switch (unused.next) {
            .none => batch.unused.tail = unused.prev,
            else => |next_index| batch.storage[next_index.toIndex()].unused.prev = unused.prev,
        }

        switch (batch.submitted.tail) {
            .none => batch.submitted.head = .fromIndex(index),
            else => |tail_index| batch.storage[tail_index.toIndex()].submission.node.next = .fromIndex(index),
        }
        storage.* = .{ .submission = .{ .node = .{ .next = .none }, .operation = operation } };
        batch.submitted.tail = .fromIndex(index);
    }

    pub const Completion = struct {
        /// The element within the provided operation storage that completed.
        /// `addAt` can be used to re-arm the `Batch` using this `index`.
        index: u32,
        /// The return value of the operation.
        result: Operation.Result,
    };

    /// After calling `awaitAsync`, `awaitConcurrent`, or `cancel`, this
    /// function iterates over the completed operations.
    ///
    /// Each completion returned from this function dequeues from the `Batch`.
    /// It is not required to dequeue all completions before awaiting again.
    pub fn next(batch: *Batch) ?Completion {
        const index = batch.completed.head;
        if (index == .none) return null;
        const storage = &batch.storage[index.toIndex()];
        const completion = storage.completion;
        const next_index = completion.node.next;
        batch.completed.head = next_index;
        if (next_index == .none) batch.completed.tail = .none;

        const tail_index = batch.unused.tail;
        switch (tail_index) {
            .none => batch.unused.head = index,
            else => batch.storage[tail_index.toIndex()].unused.next = index,
        }
        storage.* = .{ .unused = .{ .prev = tail_index, .next = .none } };
        batch.unused.tail = index;
        return .{ .index = index.toIndex(), .result = completion.result };
    }

    /// Waits for at least one of the submitted operations to complete. After
    /// this function returns the completed operations can be iterated with
    /// `next`.
    ///
    /// This function provides opportunity for the implementation to introduce
    /// concurrency into the batched operations, but unlike `awaitConcurrent`,
    /// does not require it, and therefore cannot fail with
    /// `error.ConcurrencyUnavailable`.
    pub fn awaitAsync(batch: *Batch, io: Io) Cancelable!void {
        return io.vtable.batchAwaitAsync(io.userdata, batch);
    }

    pub const AwaitConcurrentError = ConcurrentError || Cancelable || Timeout.Error;

    /// Waits for at least one of the submitted operations to complete. After
    /// this function returns the completed operations can be iterated with
    /// `next`.
    ///
    /// Unlike `awaitAsync`, this function requires the implementation to
    /// perform the operations concurrently and therefore can fail with
    /// `error.ConcurrencyUnavailable`.
    pub fn awaitConcurrent(batch: *Batch, io: Io, timeout: Timeout) AwaitConcurrentError!void {
        return io.vtable.batchAwaitConcurrent(io.userdata, batch, timeout);
    }

    /// Requests all pending operations to be interrupted, then waits for all
    /// pending operations to complete. After this returns, the `Batch` is in a
    /// well-defined state, ready to be iterated with `next`. Successfully
    /// canceled operations will be absent from the iteration. Some operations
    /// may have successfully completed regardless of the cancel request and
    /// will appear in the iteration.
    pub fn cancel(batch: *Batch, io: Io) void {
        { // abort pending submissions
            var tail_index = batch.unused.tail;
            defer batch.unused.tail = tail_index;
            var index = batch.submitted.head;
            errdefer batch.submissions.head = index;
            while (index != .none) {
                const next_index = batch.storage[index.toIndex()].submission.node.next;
                switch (tail_index) {
                    .none => batch.unused.head = index,
                    else => batch.storage[tail_index.toIndex()].unused.next = index,
                }
                batch.storage[index.toIndex()] = .{ .unused = .{ .prev = tail_index, .next = .none } };
                tail_index = index;
                index = next_index;
            }
            batch.submitted = .{ .head = .none, .tail = .none };
        }
        io.vtable.batchCancel(io.userdata, batch);
        assert(batch.submitted.head == .none and batch.submitted.tail == .none);
        assert(batch.pending.head == .none and batch.pending.tail == .none);
        assert(batch.userdata == null); // that was the last chance to deallocate resources
    }
}