valeriansaliou/sonic · critical · std::io::Error

BrokenPipe

Error message

BrokenPipe

What it means

attach sends a MultiplexerTask::Attach message to the multiplexer task with send_timeout(SEND_TIMEOUT). A timeout means the multiplexer task is not running, so the new connection cannot be registered; it is surfaced as ErrorKind::BrokenPipe. connect() calls this, so a failed attach aborts connection establishment.

Source

Thrown at client/src/multiplexer.rs:56

        let poll_waker = Arc::new(mio::Waker::new(poll.registry(), mio::Token(usize::MAX))?);

        // TODO: Do not auto-start? So one can spawn the task differently or
        //   observe events (e.g. in tests)?
        let event_loop_handle = std::thread::spawn(move || run_event_loop(&mut poll, rx));

        Ok(Self {
            _event_loop_handle: event_loop_handle,
            tx,
            poll_waker,
        })
    }

    pub(crate) fn attach<C: SonicConnectionTrait + 'static>(&self, conn: C) -> std::io::Result<()> {
        if let Err(error) =
            (self.tx).send_timeout(MultiplexerTask::Attach(Box::new(conn)), SEND_TIMEOUT)
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                error.to_string(),
            ));
        };
        self.poll_waker.wake()
    }
}

pub(crate) trait SonicConnectionTrait: AsMut<mio::net::TcpStream> + Send {
    fn wants_to_write(&self) -> bool;

    fn wants_to_read(&self) -> bool;

    fn interest(&self) -> Option<mio::Interest> {
        match (self.wants_to_write(), self.wants_to_read()) {
            (false, false) => None,
            (true, false) => Some(mio::Interest::WRITABLE),
            (false, true) => Some(mio::Interest::READABLE),

View on GitHub (pinned to e6a72da6a5)

Solutions

1. Recreate the multiplexer/client instead of reusing a dead one; attach requires a live multiplexer task.
2. Ensure connect() is not called after the client was shut down/dropped; guard with a state check.
3. Keep the runtime alive while connections are being established.
4. If timeouts are marginal under load, ensure the multiplexer task isn't starved rather than raising SEND_TIMEOUT blindly.

Example fix

// before
let conn = MyConn::connect(addr)?;
multiplexer.attach(conn)?;
// after
match multiplexer.attach(conn) {
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
        // multiplexer task is dead; rebuild the client
        let client = Client::connect(addr)?;
    }
    Ok(()) => {}
}

// Attach only to a live multiplexer:
if multiplexer_task_handle.is_finished() {
    return Err("multiplexer task dead; rebuild the client".into());
}

fn is_broken_pipe(e: &std::io::Error) -> bool {
    e.kind() == std::io::ErrorKind::BrokenPipe
}

match multiplexer.attach(conn) {
    Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
        // multiplexer is gone: rebuild client from scratch
        let client = Client::connect(addr)?;
    }
    other => other?,
}

When it happens

Trigger: Calling connect (which calls attach) when the multiplexer task has already exited — e.g. the previous connection killed it, the runtime shut it down, or attach is invoked after multiplexer shutdown; also when the multiplexer mailbox is full and never drains.

Common situations: Reconnect attempts after the multiplexer task died from a prior connection error; attaching connections from multiple threads during shutdown; runtime teardown racing with new connect() calls.

Related errors

• BrokenPipe
• closing channel
• BrokenPipe
• InvalidData
• unable to initialize japanese tokenizer

AI-assisted analysis of valeriansaliou/sonic@e6a72da6a5 (2026-09-01). Data as JSON: /api/errors/046a889ab5a83bca. Report an issue: GitHub.