databendlabs/databend · error · BrokenPipe

memory pool is closed

Error message

memory pool is closed

What it means

A sentinel error raised in AsyncBuffer::alloc_buffer when recv_blocking on the available_buffers channel fails, meaning the memory pool has been shut down while a caller was waiting for a buffer to become available. It fires during spill buffering after close/drop of the pool, signaling that no further allocations can be served.

Solutions

1. Check the pool's shutdown/close state before attempting to allocate, and stop spilling work when the SpillLimiter or owning pool has been closed
2. Propagate the closure as a termination signal: cancel in-flight spill tasks and drain any pending alloc_buffer waiters instead of letting them error
3. Acquire the buffer before beginning a spill operation so closure is detected up front and partial writes are avoided
4. Wrap allocation in retry logic that first verifies pool liveness, returning a clean cancellation error rather than an io::Error on close

When it happens

Trigger: Thrown at src/query/service/src/spillers/async_buffer.rs:128 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of databendlabs/databend@288d84d76e (2026-09-11). Data as JSON: /api/errors/bdb19442c7cced7d. Report an issue: GitHub.

Appendix: source

Thrown at src/query/service/src/spillers/async_buffer.rs:128

        })
    }

    fn alloc_buffer(&self) -> io::Result<BytesMut> {
        if let Ok(buf) = self.available_buffers.try_recv() {
            return Ok(buf);
        }

        let allocated = self.allocated_chunks.fetch_add(1, Ordering::AcqRel);
        if allocated < self.max_chunks {
            return Ok(BytesMut::with_capacity(CHUNK_SIZE));
        }

        self.allocated_chunks.fetch_sub(1, Ordering::AcqRel);

        let start = Instant::now();
        let result = match self.available_buffers.recv_blocking() {
            Ok(buf) => Ok(buf),
            Err(_) => Err(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "memory pool is closed",
            )),
        };
        self.blocking_nanos
            .fetch_add(start.elapsed().as_nanos() as u64, Ordering::Relaxed);
        self.blocking_count.fetch_add(1, Ordering::Relaxed);
        result
    }

    fn release_buffer(&self, buffer: BytesMut) {
        if self.available_buffers_tx.try_send(buffer).is_err() {
            unreachable!("MemoryPool available_buffers need unbounded.");
        }
    }
}

impl Drop for MemoryPool {

View on GitHub (pinned to 288d84d76e)