Struct opendal::layers::TimeoutLayer

pub struct TimeoutLayer { /* private fields */ }

Add timeout for every operation to avoid slow or unexpected hang operations.

For example, a dead connection could hang a databases sql query. TimeoutLayer will break this connection and returns an error so users can handle it by retrying or print to users.

Notes

TimeoutLayer treats all operations in two kinds:

• Non IO Operation like stat, delete they operate on a single file. We control them by setting timeout.
• IO Operation like read, Reader::read and Writer::write, they operate on data directly, we control them by setting io_timeout.

Default

• timeout: 60 seconds
• io_timeout: 10 seconds

Panics

TimeoutLayer will drop the future if the timeout is reached. This might cause the internal state of the future to be broken. If underlying future moves ownership into the future, it will be dropped and will neven return back.

For example, while using TimeoutLayer with RetryLayer at the same time, please make sure timeout layer showed up before retry layer.

let op = Operator::new(services::Memory::default())?
    // This is fine, since timeout happen during retry.
    .layer(TimeoutLayer::new().with_io_timeout(Duration::from_nanos(1)))
    .layer(RetryLayer::new())
    // This is wrong. Since timeout layer will drop future, leaving retry layer in a bad state.
    .layer(TimeoutLayer::new().with_io_timeout(Duration::from_nanos(1)))
    .finish();
Ok(())

Examples

The following examples will create a timeout layer with 10 seconds timeout for all non-io operations, 3 seconds timeout for all io operations.

let _ = Operator::new(services::Memory::default())?
    .layer(
        TimeoutLayer::default()
            .with_timeout(Duration::from_secs(10))
            .with_io_timeout(Duration::from_secs(3)),
    )
    .finish();
Ok(())

Implementation Notes

TimeoutLayer is using [tokio::time::timeout] to implement timeout for operations. And IO Operations insides reader, writer will use Pin<Box<tokio::time::Sleep>> to track the timeout.

This might introduce a bit overhead for IO operations, but it’s the only way to implement timeout correctly. We used to implement timeout layer in zero cost way that only stores a std::time::Instant and check the timeout by comparing the instant with current time. However, it doesn’t works for all cases.

For examples, users TCP connection could be in Busy ESTAB state. In this state, no IO event will be emit. The runtime will never poll our future again. From the application side, this future is hanging forever until this TCP connection is closed for reaching the linux net.ipv4.tcp_retries2 times.

Implementations

impl TimeoutLayer

pub fn new() -> Self

Create a new TimeoutLayer with default settings.

pub fn with_timeout(self, timeout: Duration) -> Self

Set timeout for TimeoutLayer with given value.

This timeout is for all non-io operations like stat, delete.

pub fn with_io_timeout(self, timeout: Duration) -> Self

Set io timeout for TimeoutLayer with given value.

This timeout is for all io operations like read, Reader::read and Writer::write.

pub fn with_speed(self, _: u64) -> Self

👎Deprecated: with speed is not supported anymore, please use with_io_timeout instead

Set speed for TimeoutLayer with given value.

Notes

The speed should be the lower bound of the IO speed. Set this value too large could result in all write operations failing.

Panics

This function will panic if speed is 0.

Trait Implementations

impl Clone for TimeoutLayer

fn clone(&self) -> TimeoutLayer

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Default for TimeoutLayer

fn default() -> Self

Returns the “default value” for a type.

impl<A: Access> Layer<A> for TimeoutLayer

type LayeredAccess = TimeoutAccessor<A>

The layered accessor that returned by this layer.

fn layer(&self, inner: A) -> Self::LayeredAccess

Intercept the operations on the underlying storage.