Struct Reactor 

pub struct Reactor { /* private fields */ }

A lightweight epoll reactor using edge-triggered monitoring (EPOLLET).

Edge-Triggered Contract

Because this reactor uses EPOLLET, all handlers MUST drain their respective read or write sources until they receive an EAGAIN / EWOULDBLOCK error (represented as Ok(None) in the Fd helpers).

Failure to drain a source will result in missing future readiness events for that file descriptor until it is re-registered or another event occurs.

Example

let mut reactor = Reactor::new()?;
let token = reactor.add(&fd, true, false)?;

let mut events = Vec::new();
loop {
    reactor.wait(&mut events, 64, -1)?;
    for ev in &events {
        if ev.token == token {
            // Drain fd...
        }
    }
}

Implementations

impl Reactor

pub fn new() -> Result<Self, CoreError>

Create a new epoll reactor.

Errors

Returns CoreError if epoll_create1 fails.

Examples found in repository

examples/watch_temp.rs (line 6)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut reactor = Reactor::new()?;
    let (inotify_fd, inotify_token) = reactor.setup_inotify()?;

    let temp_dir = std::env::temp_dir();
    let temp_dir_str = temp_dir.to_string_lossy().into_owned();

    println!("Watching {} for modifications...", temp_dir_str);
    add_watch(&inotify_fd, &temp_dir_str, MODIFY_MASK)?;

    let mut events = Vec::new();
    // Monitor for 10 seconds
    let start = std::time::Instant::now();
    while start.elapsed() < Duration::from_secs(10) {
        let n = reactor.wait(&mut events, 64, 1000)?;
        if n > 0 {
            for ev in &events {
                if ev.token == inotify_token {
                    let in_events = read_events(&inotify_fd)?;
                    for ie in in_events {
                        if let Some(name) = ie.name {
                            println!("File modified: {}", String::from_utf8_lossy(&name));
                        } else {
                            println!("Directory modified (wd={})", ie.wd);
                        }
                    }
                }
            }
        }
    }

    println!("Finished watching.");
    Ok(())
}

pub fn setup_inotify(&mut self) -> Result<(Fd, Token), CoreError>

Initialize inotify and add it to the reactor.

Errors

Returns CoreError if inotify_init1 or epoll_ctl fails.

Examples found in repository

examples/watch_temp.rs (line 7)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut reactor = Reactor::new()?;
    let (inotify_fd, inotify_token) = reactor.setup_inotify()?;

    let temp_dir = std::env::temp_dir();
    let temp_dir_str = temp_dir.to_string_lossy().into_owned();

    println!("Watching {} for modifications...", temp_dir_str);
    add_watch(&inotify_fd, &temp_dir_str, MODIFY_MASK)?;

    let mut events = Vec::new();
    // Monitor for 10 seconds
    let start = std::time::Instant::now();
    while start.elapsed() < Duration::from_secs(10) {
        let n = reactor.wait(&mut events, 64, 1000)?;
        if n > 0 {
            for ev in &events {
                if ev.token == inotify_token {
                    let in_events = read_events(&inotify_fd)?;
                    for ie in in_events {
                        if let Some(name) = ie.name {
                            println!("File modified: {}", String::from_utf8_lossy(&name));
                        } else {
                            println!("Directory modified (wd={})", ie.wd);
                        }
                    }
                }
            }
        }
    }

    println!("Finished watching.");
    Ok(())
}

pub fn setup_signalfd(&mut self) -> Result<Token, CoreError>

Initialize signalfd for SIGCHLD and add it to the reactor.

Errors

Returns CoreError if pthread_sigmask, signalfd, or epoll_ctl fails.

The previous current-thread signal mask is restored when the reactor is dropped.

pub fn drain_signalfd(&self) -> Result<(), CoreError>

Drain the internal signalfd buffer.

pub fn add( &mut self, fd: &Fd, readable: bool, writable: bool, ) -> Result<Token, CoreError>

Register a file descriptor with the reactor.

This assigns a new unique token for the descriptor and enables edge-triggered monitoring.

pub fn add_priority(&mut self, fd: &Fd) -> Result<Token, CoreError>

Register a file descriptor for priority readiness (EPOLLPRI).

pub fn del(&self, fd: &Fd) -> Result<(), CoreError>

Remove a file descriptor from the reactor.

pub fn wait( &mut self, buffer: &mut Vec<Event>, max_events: usize, timeout: i32, ) -> Result<usize, CoreError>

Wait for events.

This function blocks until at least one event is ready or the timeout expires. Ready events are appended to the buffer.

Returns the number of events received.

Examples found in repository

examples/watch_temp.rs (line 19)

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut reactor = Reactor::new()?;
    let (inotify_fd, inotify_token) = reactor.setup_inotify()?;

    let temp_dir = std::env::temp_dir();
    let temp_dir_str = temp_dir.to_string_lossy().into_owned();

    println!("Watching {} for modifications...", temp_dir_str);
    add_watch(&inotify_fd, &temp_dir_str, MODIFY_MASK)?;

    let mut events = Vec::new();
    // Monitor for 10 seconds
    let start = std::time::Instant::now();
    while start.elapsed() < Duration::from_secs(10) {
        let n = reactor.wait(&mut events, 64, 1000)?;
        if n > 0 {
            for ev in &events {
                if ev.token == inotify_token {
                    let in_events = read_events(&inotify_fd)?;
                    for ie in in_events {
                        if let Some(name) = ie.name {
                            println!("File modified: {}", String::from_utf8_lossy(&name));
                        } else {
                            println!("Directory modified (wd={})", ie.wd);
                        }
                    }
                }
            }
        }
    }

    println!("Finished watching.");
    Ok(())
}

Trait Implementations

impl Drop for Reactor

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.