How to handle long running external function calls such as blocking I/O in Rust?

Question

Editor's note: This question is from a version of Rust prior to 1.0 and uses terms and functions that do not exist in Rust 1.0 code. The concepts expressed are still relevant.

I need to read data provided by an external process via a POSIX file descriptor in my Rust program. The file descriptor connection is kept up a very long time (hours) and the other side passes data to me from time to time. I need to read and process the data stream continuously.

To do so, I wrote a loop that calls libc::read() (readv actually) to read the data and processes it when received. Since this would block the whole scheduler, I'm spawning a task on a new scheduler (task::spawn_sched(SingleThreaded)). This works fine as long as it runs, but I can't find a way to cleanly shut down the loop.

Since the loop is blocking most of the time, I can't use a port/channel to notify the loop to exit.

I tried to kill the loop task by taking it down using a failing linked task (spawn the loop task supervised, spawn a linked task within it and wait for a signal on a port to happen before fail!()ing and taking down the loop task with it). It works well in tests, but the libc::read() isn't interrupted (the task doesn't fail before read finishes and it hits task::yield() at some time.

I learned a lot looking at libcore sources, but I can't seem to find a proper solution.

1. Is there a way to kill a (child) task in Rust even if it's doing some long external function call like a blocking read?
2. Is there a way to do non-blocking reads on a POSIX file descriptor so that Rust keeps control over the task?
3. How can I react to signals, e.g. SIGTERMif the user terminates my program? There doesn't seem to be something like sigaction() in Rust yet.

Answer 1

1. According to mozila, killing a task is no more possible, for now, let alone blocking read.
2. It will be possible to do so after mozilla/rust/pull/11410, see also my other issue report for rust-zmq erickt/rust-zmq/issues/24 which also depends on this. (sorry about the links)
3. Maybe the signal listener will work for you.

Answer 2

Is there a way to kill a (child) task in Rust even if it's doing some long external function call like a blocking read?

No.

See also:

• How does Rust handle killing threads?
• How to terminate or suspend a Rust thread from another thread?
• What is the standard way to get a Rust thread out of blocking operations?

Is there a way to do non-blocking reads [...] so that Rust keeps control over the task?

Yes.

See also:

• How can I read non-blocking from stdin?
• How do I read the output of a child process without blocking in Rust?
• How can I force a thread that is blocked reading from a file to resume in Rust?
• Force non blocking read with TcpStream

on a POSIX file descriptor

Yes.

See also:

• How can I read from a specific raw file descriptor in Rust?
• How do I write to a specific raw file descriptor from Rust?
• How to get tokio-io's async_read for a File handle
• How to asynchronously read a file?

How can I react to signals

Decide your desired platform support, then pick an appropriate crate.

See also:

• How to catch signals in Rust
• Is there a way to listen to signals on Windows
• How to handle SIGSEGV signal in userspace using Rust?

Putting it all together

use future::Either;
use signal_hook::iterator::Signals;
use std::os::unix::io::FromRawFd;
use tokio::{fs::File, io, prelude::*};

type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;

fn main() -> Result<()> {
    let signals = Signals::new(&[signal_hook::SIGUSR1])?;
    let signals = signals.into_async()?;

    let input = unsafe { std::fs::File::from_raw_fd(5) };
    let input = File::from_std(input);
    let lines = io::lines(std::io::BufReader::new(input));

    let signals = signals.map(Either::A);
    let lines = lines.map(Either::B);

    let combined = signals.select(lines);

    tokio::run({
        combined
            .map_err(|e| panic!("Early error: {}", e))
            .for_each(|v| match v {
                Either::A(signal) => {
                    println!("Got signal: {:?}", signal);
                    Err(())
                }
                Either::B(data) => {
                    println!("Got data: {:?}", data);
                    Ok(())
                }
            })
    });

    Ok(())
}

Cargo.toml

[package]
name = "future_example"
version = "0.1.0"
authors = ["An Devloper <an.devloper@example.com>"]
edition = "2018"

[dependencies]
tokio = "0.1.22"
signal-hook = { version = "0.1.9", features = ["tokio-support"] }

shim.sh

#!/bin/bash

set -eu

exec 5< /tmp/testpipe
exec ./target/debug/future_example

Execution

cargo build
mkfifo /tmp/testpipe
./shim.sh

Another terminal

printf 'hello\nthere\nworld' > /tmp/testpipe
kill -s usr1 $PID_OF_THE_PROCESS