Calloop: A callback-based event loop

When building a complex EventSource that is composed of multiple sub-sources (e.g., a channel, a ping source, and a socket), the process_events implementation should recursively call the process_events method of each sub-source.

The Role of the Token

The token passed into the top-level process_events is the mechanism used to determine which sub-source is responsible for the current wakeup. Each sub-source compares the provided token against its own registered token. If they match, the sub-source executes its logic and calls the provided callback.

Workflow

1. The loop wakes up and calls process_events on the composed source.
2. The composed source calls sub_source.process_events(readiness, token, callback) for each component.
3. If token == sub_source.token, the sub-source processes its internal events.
4. The composed source then returns a PostAction (usually PostAction::Continue) to the loop.

When adding a Timer to a calloop event loop, you provide a callback function. This callback receives three arguments:

1. Instant: The exact time the timer was scheduled to expire. This allows for precise time tracking even if the event loop was busy and the callback is invoked slightly late.
2. &mut (): A placeholder argument (timers do not use the EventSource functionality).
3. SharedData: The shared data passed to your event loop.

Your callback must return a TimeoutAction to tell calloop how to handle the timer after the event fires:

• TimeoutAction::Drop: Disables and destroys the timer.
• TimeoutAction::ToInstant(Instant): Reschedules the timer to fire at a specific Instant. This is ideal for maintaining a consistent frequency (e.g., key repetition) because you can calculate the next deadline relative to the previous one, preventing drift even if the loop lags.
• TimeoutAction::ToDuration(Duration): Reschedules the timer to fire after a specific Duration from the current moment. This is useful for background tasks where catching up on missed intervals is not required.

// Example of a callback returning a TimeoutAction
use calloop::timer::TimeoutAction;
use std::time::{Duration, Instant};

let callback = |instant: Instant, _: &mut (), data: &mut MyData| {
    // Perform task...
    
    // Reschedule to fire again in 1 second
    TimeoutAction::ToDuration(Duration::from_secs(1))
};

A common pattern for creating custom event sources in Calloop is to return a tuple containing the source itself and a handle used to interact with it (such as a channel sender). This allows the caller to own the source for registration while maintaining a way to trigger events or send data to the source.

Example pattern:

pub fn from_socket(socket: zmq::Socket) -> io::Result<(Self, calloop::channel::Sender<T>)> {
    // ... initialization logic ...
    Ok((Self { ... }, mpsc_sender))
}

pub fn from_socket(socket: zmq::Socket) -> io::Result<(Self, calloop::channel::Sender<T>)> {
    let (mpsc_sender, mpsc_receiver) = calloop::channel::channel();
    let (wake_ping_sender, wake_ping_receiver) = calloop::ping::make_ping()?;

    let fd = socket.get_fd()?;

    let socket_source =
        calloop::generic::Generic::from_fd(fd, calloop::Interest::READ, calloop::Mode::Edge);

    Ok((
        Self {
            socket,
            socket_source,
            mpsc_receiver,
            wake_ping_receiver,
            wake_ping_sender,
        },
        mpsc_sender,
    ))
}

The Ping event source is used to manually wake up the calloop event loop. This is necessary when you have events that cannot wake the loop themselves (for example, when using a standard Rust MPSC channel for a Channel source).

To use it, you must manage two parts:

1. PingSource: The event source component that is registered with the event loop.
2. Ping: The sending end used to trigger the wake-up.

To wake the loop, call the ping() method on the Ping handle.

CRITICAL: You must ensure that the PingSource events are processed. Even though the events carry no data (they are ()), you must call process_events() on the PingSource to "reset" it. If you fail to process the events, the event loop will enter a busy-loop, continually waking up without doing useful work.

// Conceptual usage pattern:
// 1. Register PingSource in the event loop
// 2. Keep the Ping handle to trigger wake-ups
// 3. When an external event occurs (e.g. MPSC message):
//    ping.ping();
// 4. Ensure the loop calls process_events() on the PingSource to avoid busy-looping

When implementing an EventSource, you should categorize errors into two distinct types to ensure the event loop behaves correctly:

1. Event-level errors: These are errors that occur during normal operation but do not stop the source from functioning. These should be part of your Event associated type (e.g., as a variant in an enum or a Result).

   • Example: A channel being closed is an event (Closed) because you may still need to process messages already in the queue.
   • Example: A subprocess exiting with a non-zero status code is an event.

2. Source-level errors: These are errors indicating the event source is no longer capable of processing events. These should be returned via the Error associated type in process_events().

   • Example: An internal failure in the underlying mechanism (like a failed file descriptor) that makes further polling impossible.