opendal_layer_timeout/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Timeout layer implementation for Apache OpenDAL.
19
20#![cfg_attr(docsrs, feature(doc_cfg))]
21#![deny(missing_docs)]
22
23use std::future::Future;
24use std::sync::Arc;
25use std::time::Duration;
26
27use opendal_core::raw::*;
28use opendal_core::*;
29
30/// Add timeout for every operation to avoid slow or unexpected hang operations.
31///
32/// For example, a dead connection could hang a databases sql query. TimeoutLayer
33/// will break this connection and returns an error so users can handle it by
34/// retrying or print to users.
35///
36/// # Notes
37///
38/// `TimeoutLayer` treats all operations in two kinds:
39///
40/// - Non IO Operation like `stat`, `delete` they operate on a single file. We control
41///   them by setting `timeout`.
42/// - IO Operation like `read`, `Reader::read` and `Writer::write`, they operate on data directly, we
43///   control them by setting `io_timeout`.
44///
45/// # Default
46///
47/// - timeout: 60 seconds
48/// - io_timeout: 10 seconds
49///
50/// # Panics
51///
52/// TimeoutLayer will drop the future if the timeout is reached. This might cause the internal state
53/// of the future to be broken. If underlying future moves ownership into the future, it will be
54/// dropped and will neven return back.
55///
56/// For example, while using `TimeoutLayer` with `RetryLayer` at the same time, please make sure
57/// timeout layer showed up before retry layer.
58///
59/// ```no_run
60/// # use std::time::Duration;
61/// #
62/// # use opendal_core::services;
63/// # use opendal_core::Operator;
64/// # use opendal_core::Result;
65/// # use opendal_layer_retry::RetryLayer;
66/// # use opendal_layer_timeout::TimeoutLayer;
67/// #
68/// # fn main() -> Result<()> {
69/// let op = Operator::new(services::Memory::default())?
70///     // This is fine, since timeout happen during retry.
71///     .layer(TimeoutLayer::default().with_io_timeout(Duration::from_nanos(1)))
72///     .layer(RetryLayer::default())
73///     // This is wrong. Since timeout layer will drop future, leaving retry layer in a bad state.
74///     .layer(TimeoutLayer::default().with_io_timeout(Duration::from_nanos(1)))
75///     .finish();
76/// # Ok(())
77/// # }
78/// ```
79///
80/// # Examples
81///
82/// The following examples will create a timeout layer with 10 seconds timeout for all non-io
83/// operations, 3 seconds timeout for all io operations.
84///
85/// ```no_run
86/// # use std::time::Duration;
87/// #
88/// # use opendal_core::services;
89/// # use opendal_core::Operator;
90/// # use opendal_core::Result;
91/// # use opendal_layer_timeout::TimeoutLayer;
92/// #
93/// # fn main() -> Result<()> {
94/// let _ = Operator::new(services::Memory::default())?
95///     .layer(
96///         TimeoutLayer::default()
97///             .with_timeout(Duration::from_secs(10))
98///             .with_io_timeout(Duration::from_secs(3)),
99///     )
100///     .finish();
101/// # Ok(())
102/// # }
103/// ```
104///
105/// # Implementation Notes
106///
107/// TimeoutLayer is using [`tokio::time::timeout`] to implement timeout for operations. And IO
108/// Operations insides `reader`, `writer` will use `Pin<Box<tokio::time::Sleep>>` to track the
109/// timeout.
110///
111/// This might introduce a bit overhead for IO operations, but it's the only way to implement
112/// timeout correctly. We used to implement timeout layer in zero cost way that only stores
113/// a [`Instant`] and check the timeout by comparing the instant with current time.
114/// However, it doesn't work for all cases.
115///
116/// For examples, users TCP connection could be in [Busy ESTAB](https://blog.cloudflare.com/when-tcp-sockets-refuse-to-die) state. In this state, no IO event will be emitted. The runtime
117/// will never poll our future again. From the application side, this future is hanging forever
118/// until this TCP connection is closed for reaching the linux [net.ipv4.tcp_retries2](https://man7.org/linux/man-pages/man7/tcp.7.html) times.
119#[derive(Clone)]
120pub struct TimeoutLayer {
121    timeout: Duration,
122    io_timeout: Duration,
123}
124
125impl Default for TimeoutLayer {
126    fn default() -> Self {
127        Self {
128            timeout: Duration::from_secs(60),
129            io_timeout: Duration::from_secs(10),
130        }
131    }
132}
133
134impl TimeoutLayer {
135    /// Create a new [`TimeoutLayer`] with default settings.
136    pub fn new() -> Self {
137        Self::default()
138    }
139
140    /// Set timeout for TimeoutLayer with given value.
141    ///
142    /// This timeout is for all non-io operations like `stat`, `delete`.
143    pub fn with_timeout(mut self, timeout: Duration) -> Self {
144        self.timeout = timeout;
145        self
146    }
147
148    /// Set io timeout for TimeoutLayer with given value.
149    ///
150    /// This timeout is for all io operations like `read`, `Reader::read` and `Writer::write`.
151    pub fn with_io_timeout(mut self, timeout: Duration) -> Self {
152        self.io_timeout = timeout;
153        self
154    }
155}
156
157impl<A: Access> Layer<A> for TimeoutLayer {
158    type LayeredAccess = TimeoutAccessor<A>;
159
160    fn layer(&self, inner: A) -> Self::LayeredAccess {
161        let info = inner.info();
162        info.update_executor(|exec| {
163            Executor::with(TimeoutExecutor::new(exec.into_inner(), self.io_timeout))
164        });
165
166        TimeoutAccessor {
167            inner,
168
169            timeout: self.timeout,
170            io_timeout: self.io_timeout,
171        }
172    }
173}
174
175#[doc(hidden)]
176#[derive(Debug)]
177pub struct TimeoutAccessor<A: Access> {
178    inner: A,
179
180    timeout: Duration,
181    io_timeout: Duration,
182}
183
184impl<A: Access> TimeoutAccessor<A> {
185    async fn timeout<F: Future<Output = Result<T>>, T>(&self, op: Operation, fut: F) -> Result<T> {
186        tokio::time::timeout(self.timeout, fut).await.map_err(|_| {
187            Error::new(ErrorKind::Unexpected, "operation timeout reached")
188                .with_operation(op)
189                .with_context("timeout", self.timeout.as_secs_f64().to_string())
190                .set_temporary()
191        })?
192    }
193
194    async fn io_timeout<F: Future<Output = Result<T>>, T>(
195        &self,
196        op: Operation,
197        fut: F,
198    ) -> Result<T> {
199        tokio::time::timeout(self.io_timeout, fut)
200            .await
201            .map_err(|_| {
202                Error::new(ErrorKind::Unexpected, "io timeout reached")
203                    .with_operation(op)
204                    .with_context("timeout", self.io_timeout.as_secs_f64().to_string())
205                    .set_temporary()
206            })?
207    }
208}
209
210impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
211    type Inner = A;
212    type Reader = TimeoutWrapper<A::Reader>;
213    type Writer = TimeoutWrapper<A::Writer>;
214    type Lister = TimeoutWrapper<A::Lister>;
215    type Deleter = TimeoutWrapper<A::Deleter>;
216    type Copier = TimeoutWrapper<A::Copier>;
217
218    fn inner(&self) -> &Self::Inner {
219        &self.inner
220    }
221
222    async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
223        self.timeout(Operation::CreateDir, self.inner.create_dir(path, args))
224            .await
225    }
226
227    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
228        self.io_timeout(Operation::Read, self.inner.read(path, args))
229            .await
230            .map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
231    }
232
233    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
234        self.io_timeout(Operation::Write, self.inner.write(path, args))
235            .await
236            .map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
237    }
238
239    async fn copy(
240        &self,
241        from: &str,
242        to: &str,
243        args: OpCopy,
244        opts: OpCopier,
245    ) -> Result<(RpCopy, Self::Copier)> {
246        self.timeout(
247            Operation::Copy,
248            self.inner.copy(from, to, args, opts.clone()),
249        )
250        .await
251        .map(|(rp, c)| (rp, TimeoutWrapper::new(c, self.io_timeout)))
252    }
253
254    async fn rename(&self, from: &str, to: &str, args: OpRename) -> Result<RpRename> {
255        self.timeout(Operation::Rename, self.inner.rename(from, to, args))
256            .await
257    }
258
259    async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
260        self.timeout(Operation::Stat, self.inner.stat(path, args))
261            .await
262    }
263
264    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
265        self.timeout(Operation::Delete, self.inner.delete())
266            .await
267            .map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
268    }
269
270    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
271        self.io_timeout(Operation::List, self.inner.list(path, args))
272            .await
273            .map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
274    }
275
276    async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
277        self.timeout(Operation::Presign, self.inner.presign(path, args))
278            .await
279    }
280}
281
282struct TimeoutExecutor {
283    exec: Arc<dyn Execute>,
284    timeout: Duration,
285}
286
287impl TimeoutExecutor {
288    fn new(exec: Arc<dyn Execute>, timeout: Duration) -> Self {
289        Self { exec, timeout }
290    }
291}
292
293impl Execute for TimeoutExecutor {
294    fn execute(&self, f: BoxedStaticFuture<()>) {
295        self.exec.execute(f)
296    }
297
298    fn timeout(&self) -> Option<BoxedStaticFuture<()>> {
299        Some(Box::pin(tokio::time::sleep(self.timeout)))
300    }
301}
302
303#[doc(hidden)]
304pub struct TimeoutWrapper<R> {
305    inner: R,
306
307    timeout: Duration,
308}
309
310impl<R> TimeoutWrapper<R> {
311    fn new(inner: R, timeout: Duration) -> Self {
312        Self { inner, timeout }
313    }
314
315    #[inline]
316    async fn io_timeout<F: Future<Output = Result<T>>, T>(
317        timeout: Duration,
318        op: &'static str,
319        fut: F,
320    ) -> Result<T> {
321        tokio::time::timeout(timeout, fut).await.map_err(|_| {
322            Error::new(ErrorKind::Unexpected, "io operation timeout reached")
323                .with_operation(op)
324                .with_context("timeout", timeout.as_secs_f64().to_string())
325                .set_temporary()
326        })?
327    }
328}
329
330impl<R: oio::Read> oio::Read for TimeoutWrapper<R> {
331    async fn read(&mut self) -> Result<Buffer> {
332        let fut = self.inner.read();
333        Self::io_timeout(self.timeout, Operation::Read.into_static(), fut).await
334    }
335}
336
337impl<R: oio::Write> oio::Write for TimeoutWrapper<R> {
338    async fn write(&mut self, bs: Buffer) -> Result<()> {
339        let fut = self.inner.write(bs);
340        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
341    }
342
343    async fn close(&mut self) -> Result<Metadata> {
344        let fut = self.inner.close();
345        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
346    }
347
348    async fn abort(&mut self) -> Result<()> {
349        let fut = self.inner.abort();
350        Self::io_timeout(self.timeout, Operation::Write.into_static(), fut).await
351    }
352}
353
354impl<R: oio::List> oio::List for TimeoutWrapper<R> {
355    async fn next(&mut self) -> Result<Option<oio::Entry>> {
356        let fut = self.inner.next();
357        Self::io_timeout(self.timeout, Operation::List.into_static(), fut).await
358    }
359}
360
361impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> {
362    async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
363        let fut = self.inner.delete(path, args);
364        Self::io_timeout(self.timeout, Operation::Delete.into_static(), fut).await
365    }
366
367    async fn close(&mut self) -> Result<()> {
368        let fut = self.inner.close();
369        Self::io_timeout(self.timeout, Operation::Delete.into_static(), fut).await
370    }
371}
372
373impl<C: oio::Copy> oio::Copy for TimeoutWrapper<C> {
374    async fn next(&mut self) -> Result<Option<usize>> {
375        let fut = self.inner.next();
376        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
377    }
378
379    async fn close(&mut self) -> Result<Metadata> {
380        let fut = self.inner.close();
381        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
382    }
383
384    async fn abort(&mut self) -> Result<()> {
385        let fut = self.inner.abort();
386        Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use std::future::pending;
393
394    use futures::StreamExt;
395    use tokio::time::sleep;
396    use tokio::time::timeout;
397
398    use super::*;
399
400    #[derive(Debug, Clone, Default)]
401    struct MockService;
402
403    impl Access for MockService {
404        type Reader = oio::Reader;
405        type Writer = oio::Writer;
406        type Lister = oio::Lister;
407        type Deleter = oio::Deleter;
408        type Copier = oio::Copier;
409
410        fn info(&self) -> Arc<AccessorInfo> {
411            let am = AccessorInfo::default();
412            am.set_native_capability(Capability {
413                read: true,
414                delete: true,
415                ..Default::default()
416            });
417
418            am.into()
419        }
420
421        /// This function will build a reader that always return pending.
422        async fn read(&self, _: &str, _: OpRead) -> Result<(RpRead, Self::Reader)> {
423            Ok((
424                RpRead::new(Metadata::new(EntryMode::FILE).with_content_length(0)),
425                Box::new(MockReader),
426            ))
427        }
428
429        /// This function will never return.
430        async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
431            sleep(Duration::from_secs(u64::MAX)).await;
432
433            Ok((RpDelete::default(), Box::new(())))
434        }
435
436        async fn list(&self, _: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
437            Ok((RpList::default(), Box::new(MockLister)))
438        }
439
440        async fn copy(
441            &self,
442            _: &str,
443            _: &str,
444            _: OpCopy,
445            _: OpCopier,
446        ) -> Result<(RpCopy, Self::Copier)> {
447            Ok((RpCopy::default(), Box::new(MockCopier)))
448        }
449    }
450
451    #[derive(Debug, Clone, Default)]
452    struct MockReader;
453
454    impl oio::Read for MockReader {
455        fn read(&mut self) -> impl Future<Output = Result<Buffer>> {
456            pending()
457        }
458    }
459
460    #[derive(Debug, Clone, Default)]
461    struct MockLister;
462
463    impl oio::List for MockLister {
464        fn next(&mut self) -> impl Future<Output = Result<Option<oio::Entry>>> {
465            pending()
466        }
467    }
468
469    #[derive(Debug, Clone, Default)]
470    struct MockDeleter;
471
472    impl oio::Delete for MockDeleter {
473        fn delete(&mut self, _: &str, _: OpDelete) -> impl Future<Output = Result<()>> {
474            pending()
475        }
476
477        async fn close(&mut self) -> Result<()> {
478            Ok(())
479        }
480    }
481
482    #[derive(Debug, Clone, Default)]
483    struct MockCopier;
484
485    impl oio::Copy for MockCopier {
486        fn next(&mut self) -> impl Future<Output = Result<Option<usize>>> {
487            pending()
488        }
489
490        fn close(&mut self) -> impl Future<Output = Result<Metadata>> {
491            pending()
492        }
493
494        fn abort(&mut self) -> impl Future<Output = Result<()>> {
495            pending()
496        }
497    }
498
499    #[tokio::test]
500    async fn test_operation_timeout() {
501        let srv = MockService;
502        let op = Operator::from_inner(Arc::new(srv))
503            .layer(TimeoutLayer::default().with_timeout(Duration::from_secs(1)));
504
505        let fut = async {
506            let res = op.delete("test").await;
507            assert!(res.is_err());
508            let err = res.unwrap_err();
509            assert_eq!(err.kind(), ErrorKind::Unexpected);
510            assert!(err.to_string().contains("timeout"))
511        };
512
513        timeout(Duration::from_secs(2), fut)
514            .await
515            .expect("this test should not exceed 2 seconds")
516    }
517
518    #[tokio::test]
519    async fn test_io_timeout() {
520        let srv = MockService;
521        let op = Operator::from_inner(Arc::new(srv))
522            .layer(TimeoutLayer::default().with_io_timeout(Duration::from_secs(1)));
523
524        let reader = op.reader("test").await.unwrap();
525
526        let res = reader.read(0..4).await;
527        assert!(res.is_err());
528        let err = res.unwrap_err();
529        assert_eq!(err.kind(), ErrorKind::Unexpected);
530        assert!(err.to_string().contains("timeout"))
531    }
532
533    #[tokio::test]
534    async fn test_list_timeout() {
535        let srv = MockService;
536        let op = Operator::from_inner(Arc::new(srv)).layer(
537            TimeoutLayer::default()
538                .with_timeout(Duration::from_secs(1))
539                .with_io_timeout(Duration::from_secs(1)),
540        );
541
542        let mut lister = op.lister("test").await.unwrap();
543
544        let res = lister.next().await.unwrap();
545        assert!(res.is_err());
546        let err = res.unwrap_err();
547        assert_eq!(err.kind(), ErrorKind::Unexpected);
548        assert!(err.to_string().contains("timeout"))
549    }
550
551    #[tokio::test]
552    async fn test_delete_io_timeout() {
553        use oio::Delete;
554
555        let mut deleter = TimeoutWrapper::new(MockDeleter, Duration::from_secs(1));
556
557        let res = deleter.delete("test", OpDelete::default()).await;
558        assert!(res.is_err());
559        let err = res.unwrap_err();
560        assert_eq!(err.kind(), ErrorKind::Unexpected);
561        assert!(err.to_string().contains("timeout"));
562    }
563
564    #[tokio::test]
565    async fn test_copy_io_timeout() {
566        use oio::Copy;
567
568        let acc = TimeoutLayer::default()
569            .with_io_timeout(Duration::from_millis(100))
570            .layer(MockService);
571        let (_, mut copier) = Access::copy(&acc, "f", "t", OpCopy::default(), OpCopier::default())
572            .await
573            .unwrap();
574
575        let err = copier.next().await.unwrap_err();
576        assert!(err.to_string().contains("timeout"));
577    }
578
579    #[tokio::test]
580    async fn test_list_timeout_raw() {
581        use oio::List;
582
583        let acc = MockService;
584        let timeout_layer = TimeoutLayer::default()
585            .with_timeout(Duration::from_secs(1))
586            .with_io_timeout(Duration::from_secs(1));
587        let timeout_acc = timeout_layer.layer(acc);
588
589        let (_, mut lister) = Access::list(&timeout_acc, "test", OpList::default())
590            .await
591            .unwrap();
592
593        let res = lister.next().await;
594        assert!(res.is_err());
595        let err = res.unwrap_err();
596        assert_eq!(err.kind(), ErrorKind::Unexpected);
597        assert!(err.to_string().contains("timeout"));
598    }
599}
opendal_layer_timeout/lib.rs

opendal_layer_timeout/
lib.rs
