2
2
3
3
use std:: future:: Future ;
4
4
use std:: io;
5
- use std:: marker:: Unpin ;
6
5
use std:: pin:: Pin ;
7
6
use std:: task:: { Context , Poll } ;
8
7
use std:: time:: { Duration , Instant } ;
@@ -15,27 +14,37 @@ use super::Delay;
15
14
///
16
15
/// [`FutureExt.timeout`]: trait.FutureExt.html
17
16
#[ derive( Debug ) ]
18
- pub struct Timeout < F : Future + Unpin > {
17
+ pub struct Timeout < F : Future > {
19
18
future : F ,
20
19
delay : Delay ,
21
20
}
22
21
23
- impl < F : Future + Unpin > Future for Timeout < F > {
22
+ impl < F : Future > Future for Timeout < F > {
24
23
type Output = Result < F :: Output , io:: Error > ;
25
24
26
- fn poll ( mut self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Self :: Output > {
27
- if let Poll :: Ready ( t) = Pin :: new ( & mut self . future ) . poll ( cx) {
25
+ fn poll ( self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Self :: Output > {
26
+ // This pinning projection is safe because:
27
+ // 1. `Timeout` is only Unpin when `F` is Unpin. (Ok for default auto impl)
28
+ // 2. `drop` never moves out of `F`. (No manual `Drop` impl and no `#[repr(packed)]`)
29
+ // 3. `drop` on `F` must be called before overwritten or deallocated. (No manual `Drop` impl)
30
+ // 4. No other operation provided for moving out `F`. (Ok)
31
+ let ( future, delay) = unsafe {
32
+ let Timeout { future, delay } = self . get_unchecked_mut ( ) ;
33
+ ( Pin :: new_unchecked ( future) , Pin :: new ( delay) )
34
+ } ;
35
+
36
+ if let Poll :: Ready ( t) = future. poll ( cx) {
28
37
return Poll :: Ready ( Ok ( t) ) ;
29
38
}
30
39
31
- Pin :: new ( & mut self . delay )
40
+ delay
32
41
. poll ( cx)
33
42
. map ( |_| Err ( io:: Error :: new ( io:: ErrorKind :: TimedOut , "future timed out" ) ) )
34
43
}
35
44
}
36
45
37
46
/// Extend `Future` with methods to time out execution.
38
- pub trait FutureExt : Future + Sized + Unpin {
47
+ pub trait FutureExt : Future + Sized {
39
48
/// Creates a new future which will take at most `dur` time to resolve from
40
49
/// the point at which this method is called.
41
50
///
@@ -52,21 +61,32 @@ pub trait FutureExt: Future + Sized + Unpin {
52
61
/// # #![feature(async_await)]
53
62
/// use futures::prelude::*;
54
63
/// use runtime::prelude::*;
55
- /// use std::time::Duration;
64
+ /// use std::time::{Duration, Instant};
65
+ ///
66
+ /// async fn long_future(dur: Duration) {
67
+ /// // Simulate some network operations...
68
+ /// runtime::time::Delay::new(dur).await;
69
+ /// }
56
70
///
57
- /// # fn long_future() -> impl Future<Output = std::io::Result<()>> {
58
- /// # futures::future::ok(())
59
- /// # }
60
- /// #
61
71
/// #[runtime::main]
62
72
/// async fn main() {
63
- /// let future = long_future();
64
- /// let timed_out = future.timeout(Duration::from_millis(100));
73
+ /// // Fast operation
74
+ /// let begin_inst = Instant::now();
75
+ /// let short = long_future(Duration::from_millis(100))
76
+ /// .timeout(Duration::from_millis(5000)) // Set timeout
77
+ /// .await;
78
+ /// assert!(short.is_ok()); // Success
79
+ /// assert!(begin_inst.elapsed() >= Duration::from_millis(100));
80
+ /// assert!(begin_inst.elapsed() < Duration::from_millis(5000));
65
81
///
66
- /// match timed_out.await {
67
- /// Ok(item) => println!("got {:?} within enough time!", item),
68
- /// Err(_) => println!("took too long to produce the item"),
69
- /// }
82
+ /// // Slow operation
83
+ /// let begin_inst = Instant::now();
84
+ /// let long = long_future(Duration::from_millis(5000))
85
+ /// .timeout(Duration::from_millis(100)) // Set timeout
86
+ /// .await;
87
+ /// assert!(long.is_err()); // Timeout
88
+ /// assert!(begin_inst.elapsed() >= Duration::from_millis(100));
89
+ /// assert!(begin_inst.elapsed() < Duration::from_millis(5000));
70
90
/// }
71
91
/// ```
72
92
fn timeout ( self , dur : Duration ) -> Timeout < Self > {
@@ -116,29 +136,40 @@ pub trait FutureExt: Future + Sized + Unpin {
116
136
}
117
137
}
118
138
119
- impl < T : Future + Unpin > FutureExt for T { }
139
+ impl < T : Future > FutureExt for T { }
120
140
121
141
/// A stream returned by methods in the [`StreamExt`] trait.
122
142
///
123
143
/// [`StreamExt`]: trait.StreamExt.html
124
144
#[ derive( Debug ) ]
125
- pub struct TimeoutStream < S : Stream + Unpin > {
145
+ pub struct TimeoutStream < S : Stream > {
126
146
timeout : Delay ,
127
147
dur : Duration ,
128
148
stream : S ,
129
149
}
130
150
131
- impl < S : Stream + Unpin > Stream for TimeoutStream < S > {
151
+ impl < S : Stream > Stream for TimeoutStream < S > {
132
152
type Item = Result < S :: Item , io:: Error > ;
133
153
134
- fn poll_next ( mut self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Option < Self :: Item > > {
135
- if let Poll :: Ready ( s) = Pin :: new ( & mut self . stream ) . poll_next ( cx) {
136
- self . timeout = Delay :: new ( self . dur ) ;
154
+ fn poll_next ( self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Option < Self :: Item > > {
155
+ // This pinning projection is safe.
156
+ // See detail in `Timeout::poll`.
157
+ let ( mut timeout, dur, stream) = unsafe {
158
+ let TimeoutStream {
159
+ timeout,
160
+ dur,
161
+ stream,
162
+ } = self . get_unchecked_mut ( ) ;
163
+ ( Pin :: new ( timeout) , Pin :: new ( dur) , Pin :: new_unchecked ( stream) )
164
+ } ;
165
+
166
+ if let Poll :: Ready ( s) = stream. poll_next ( cx) {
167
+ timeout. set ( Delay :: new ( * dur) ) ;
137
168
return Poll :: Ready ( Ok ( s) . transpose ( ) ) ;
138
169
}
139
170
140
- Pin :: new ( & mut self . timeout ) . poll ( cx) . map ( |_| {
141
- self . timeout = Delay :: new ( self . dur ) ;
171
+ Pin :: new ( & mut * timeout) . poll ( cx) . map ( |_| {
172
+ timeout. set ( Delay :: new ( * dur) ) ;
142
173
Some ( Err ( io:: Error :: new (
143
174
io:: ErrorKind :: TimedOut ,
144
175
"future timed out" ,
@@ -148,7 +179,7 @@ impl<S: Stream + Unpin> Stream for TimeoutStream<S> {
148
179
}
149
180
150
181
/// Extend `Stream` with methods to time out execution.
151
- pub trait StreamExt : Stream + Sized + Unpin {
182
+ pub trait StreamExt : Stream + Sized {
152
183
/// Creates a new stream which will take at most `dur` time to yield each
153
184
/// item of the stream.
154
185
///
@@ -190,38 +221,49 @@ pub trait StreamExt: Stream + Sized + Unpin {
190
221
}
191
222
}
192
223
193
- impl < S : Stream + Unpin > StreamExt for S { }
224
+ impl < S : Stream > StreamExt for S { }
194
225
195
226
/// A stream returned by methods in the [`StreamExt`] trait.
196
227
///
197
228
/// [`StreamExt`]: trait.StreamExt.html
198
229
#[ derive( Debug ) ]
199
- pub struct TimeoutAsyncRead < S : AsyncRead + Unpin > {
230
+ pub struct TimeoutAsyncRead < S : AsyncRead > {
200
231
timeout : Delay ,
201
232
dur : Duration ,
202
233
stream : S ,
203
234
}
204
235
205
- impl < S : AsyncRead + Unpin > AsyncRead for TimeoutAsyncRead < S > {
236
+ impl < S : AsyncRead > AsyncRead for TimeoutAsyncRead < S > {
206
237
fn poll_read (
207
- mut self : Pin < & mut Self > ,
238
+ self : Pin < & mut Self > ,
208
239
cx : & mut Context < ' _ > ,
209
240
buf : & mut [ u8 ] ,
210
241
) -> Poll < Result < usize , io:: Error > > {
211
- if let Poll :: Ready ( s) = Pin :: new ( & mut self . stream ) . poll_read ( cx, buf) {
212
- self . timeout = Delay :: new ( self . dur ) ;
242
+ // This pinning projection is safe.
243
+ // See detail in `Timeout::poll`.
244
+ let ( mut timeout, dur, stream) = unsafe {
245
+ let TimeoutAsyncRead {
246
+ timeout,
247
+ dur,
248
+ stream,
249
+ } = self . get_unchecked_mut ( ) ;
250
+ ( Pin :: new ( timeout) , Pin :: new ( dur) , Pin :: new_unchecked ( stream) )
251
+ } ;
252
+
253
+ if let Poll :: Ready ( s) = stream. poll_read ( cx, buf) {
254
+ timeout. set ( Delay :: new ( * dur) ) ;
213
255
return Poll :: Ready ( s) ;
214
256
}
215
257
216
- Pin :: new ( & mut self . timeout ) . poll ( cx) . map ( |_| {
217
- self . timeout = Delay :: new ( self . dur ) ;
258
+ Pin :: new ( & mut * timeout) . poll ( cx) . map ( |_| {
259
+ timeout. set ( Delay :: new ( * dur) ) ;
218
260
Err ( io:: Error :: new ( io:: ErrorKind :: TimedOut , "future timed out" ) )
219
261
} )
220
262
}
221
263
}
222
264
223
265
/// Extend `AsyncRead` with methods to time out execution.
224
- pub trait AsyncReadExt : AsyncRead + Sized + Unpin {
266
+ pub trait AsyncReadExt : AsyncRead + Sized {
225
267
/// Creates a new stream which will take at most `dur` time to yield each
226
268
/// item of the stream.
227
269
///
@@ -260,4 +302,4 @@ pub trait AsyncReadExt: AsyncRead + Sized + Unpin {
260
302
}
261
303
}
262
304
263
- impl < S : AsyncRead + Unpin > AsyncReadExt for S { }
305
+ impl < S : AsyncRead > AsyncReadExt for S { }
0 commit comments