1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351
//! A one-shot, futures-aware channel //! //! This channel is similar to that in `sync::oneshot` but cannot be sent across //! threads. use std::cell::{Cell, RefCell}; use std::fmt; use std::rc::{Rc, Weak}; use {Future, Poll, Async}; use future::{Executor, IntoFuture, Lazy, lazy}; use task::{self, Task}; /// Creates a new futures-aware, one-shot channel. /// /// This function is the same as `sync::oneshot::channel` except that the /// returned values cannot be sent across threads. pub fn channel<T>() -> (Sender<T>, Receiver<T>) { let inner = Rc::new(RefCell::new(Inner { value: None, tx_task: None, rx_task: None, })); let tx = Sender { inner: Rc::downgrade(&inner), }; let rx = Receiver { state: State::Open(inner), }; (tx, rx) } /// Represents the completion half of a oneshot through which the result of a /// computation is signaled. /// /// This is created by the `unsync::oneshot::channel` function and is equivalent /// in functionality to `sync::oneshot::Sender` except that it cannot be sent /// across threads. #[derive(Debug)] pub struct Sender<T> { inner: Weak<RefCell<Inner<T>>>, } /// A future representing the completion of a computation happening elsewhere in /// memory. /// /// This is created by the `unsync::oneshot::channel` function and is equivalent /// in functionality to `sync::oneshot::Receiver` except that it cannot be sent /// across threads. #[derive(Debug)] #[must_use = "futures do nothing unless polled"] pub struct Receiver<T> { state: State<T>, } #[derive(Debug)] enum State<T> { Open(Rc<RefCell<Inner<T>>>), Closed(Option<T>), } pub use sync::oneshot::Canceled; #[derive(Debug)] struct Inner<T> { value: Option<T>, tx_task: Option<Task>, rx_task: Option<Task>, } impl<T> Sender<T> { /// Completes this oneshot with a successful result. /// /// This function will consume `self` and indicate to the other end, the /// `Receiver`, that the error provided is the result of the computation this /// represents. /// /// If the value is successfully enqueued for the remote end to receive, /// then `Ok(())` is returned. If the receiving end was deallocated before /// this function was called, however, then `Err` is returned with the value /// provided. pub fn send(self, val: T) -> Result<(), T> { if let Some(inner) = self.inner.upgrade() { inner.borrow_mut().value = Some(val); Ok(()) } else { Err(val) } } /// Polls this `Sender` half to detect whether the `Receiver` this has /// paired with has gone away. /// /// This function can be used to learn about when the `Receiver` (consumer) /// half has gone away and nothing will be able to receive a message sent /// from `complete`. /// /// Like `Future::poll`, this function will panic if it's not called from /// within the context of a task. In other words, this should only ever be /// called from inside another future. /// /// If `Ready` is returned then it means that the `Receiver` has disappeared /// and the result this `Sender` would otherwise produce should no longer /// be produced. /// /// If `NotReady` is returned then the `Receiver` is still alive and may be /// able to receive a message if sent. The current task, however, is /// scheduled to receive a notification if the corresponding `Receiver` goes /// away. pub fn poll_cancel(&mut self) -> Poll<(), ()> { match self.inner.upgrade() { Some(inner) => { inner.borrow_mut().tx_task = Some(task::current()); Ok(Async::NotReady) } None => Ok(().into()), } } /// Tests to see whether this `Sender`'s corresponding `Receiver` /// has gone away. /// /// This function can be used to learn about when the `Receiver` (consumer) /// half has gone away and nothing will be able to receive a message sent /// from `send`. /// /// Note that this function is intended to *not* be used in the context of a /// future. If you're implementing a future you probably want to call the /// `poll_cancel` function which will block the current task if the /// cancellation hasn't happened yet. This can be useful when working on a /// non-futures related thread, though, which would otherwise panic if /// `poll_cancel` were called. pub fn is_canceled(&self) -> bool { !self.inner.upgrade().is_some() } } impl<T> Drop for Sender<T> { fn drop(&mut self) { let inner = match self.inner.upgrade() { Some(inner) => inner, None => return, }; let rx_task = { let mut borrow = inner.borrow_mut(); borrow.tx_task.take(); borrow.rx_task.take() }; if let Some(task) = rx_task { task.notify(); } } } impl<T> Receiver<T> { /// Gracefully close this receiver, preventing sending any future messages. /// /// Any `send` operation which happens after this method returns is /// guaranteed to fail. Once this method is called the normal `poll` method /// can be used to determine whether a message was actually sent or not. If /// `Canceled` is returned from `poll` then no message was sent. pub fn close(&mut self) { let (item, task) = match self.state { State::Open(ref inner) => { let mut inner = inner.borrow_mut(); drop(inner.rx_task.take()); (inner.value.take(), inner.tx_task.take()) } State::Closed(_) => return, }; self.state = State::Closed(item); if let Some(task) = task { task.notify(); } } } impl<T> Future for Receiver<T> { type Item = T; type Error = Canceled; fn poll(&mut self) -> Poll<Self::Item, Self::Error> { let inner = match self.state { State::Open(ref mut inner) => inner, State::Closed(ref mut item) => { match item.take() { Some(item) => return Ok(item.into()), None => return Err(Canceled), } } }; // If we've got a value, then skip the logic below as we're done. if let Some(val) = inner.borrow_mut().value.take() { return Ok(Async::Ready(val)) } // If we can get mutable access, then the sender has gone away. We // didn't see a value above, so we're canceled. Otherwise we park // our task and wait for a value to come in. if Rc::get_mut(inner).is_some() { Err(Canceled) } else { inner.borrow_mut().rx_task = Some(task::current()); Ok(Async::NotReady) } } } impl<T> Drop for Receiver<T> { fn drop(&mut self) { self.close(); } } /// Handle returned from the `spawn` function. /// /// This handle is a future representing the completion of a different future on /// a separate executor. Created through the `oneshot::spawn` function this /// handle will resolve when the future provided to `spawn` resolves on the /// `Executor` instance provided to that function. /// /// If this handle is dropped then the future will automatically no longer be /// polled and is scheduled to be dropped. This can be canceled with the /// `forget` function, however. pub struct SpawnHandle<T, E> { rx: Receiver<Result<T, E>>, keep_running: Rc<Cell<bool>>, } /// Type of future which `Spawn` instances below must be able to spawn. pub struct Execute<F: Future> { future: F, tx: Option<Sender<Result<F::Item, F::Error>>>, keep_running: Rc<Cell<bool>>, } /// Spawns a `future` onto the instance of `Executor` provided, `executor`, /// returning a handle representing the completion of the future. /// /// The `SpawnHandle` returned is a future that is a proxy for `future` itself. /// When `future` completes on `executor` then the `SpawnHandle` will itself be /// resolved. Internally `SpawnHandle` contains a `oneshot` channel and is /// thus not safe to send across threads. /// /// The `future` will be canceled if the `SpawnHandle` is dropped. If this is /// not desired then the `SpawnHandle::forget` function can be used to continue /// running the future to completion. /// /// # Panics /// /// This function will panic if the instance of `Spawn` provided is unable to /// spawn the `future` provided. /// /// If the provided instance of `Spawn` does not actually run `future` to /// completion, then the returned handle may panic when polled. Typically this /// is not a problem, though, as most instances of `Spawn` will run futures to /// completion. pub fn spawn<F, E>(future: F, executor: &E) -> SpawnHandle<F::Item, F::Error> where F: Future, E: Executor<Execute<F>>, { let flag = Rc::new(Cell::new(false)); let (tx, rx) = channel(); executor.execute(Execute { future: future, tx: Some(tx), keep_running: flag.clone(), }).expect("failed to spawn future"); SpawnHandle { rx: rx, keep_running: flag, } } /// Spawns a function `f` onto the `Spawn` instance provided `s`. /// /// For more information see the `spawn` function in this module. This function /// is just a thin wrapper around `spawn` which will execute the closure on the /// executor provided and then complete the future that the closure returns. pub fn spawn_fn<F, R, E>(f: F, executor: &E) -> SpawnHandle<R::Item, R::Error> where F: FnOnce() -> R, R: IntoFuture, E: Executor<Execute<Lazy<F, R>>>, { spawn(lazy(f), executor) } impl<T, E> SpawnHandle<T, E> { /// Drop this future without canceling the underlying future. /// /// When `SpawnHandle` is dropped, the spawned future will be canceled as /// well if the future hasn't already resolved. This function can be used /// when to drop this future but keep executing the underlying future. pub fn forget(self) { self.keep_running.set(true); } } impl<T, E> Future for SpawnHandle<T, E> { type Item = T; type Error = E; fn poll(&mut self) -> Poll<T, E> { match self.rx.poll() { Ok(Async::Ready(Ok(t))) => Ok(t.into()), Ok(Async::Ready(Err(e))) => Err(e), Ok(Async::NotReady) => Ok(Async::NotReady), Err(_) => panic!("future was canceled before completion"), } } } impl<T: fmt::Debug, E: fmt::Debug> fmt::Debug for SpawnHandle<T, E> { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { f.debug_struct("SpawnHandle") .finish() } } impl<F: Future> Future for Execute<F> { type Item = (); type Error = (); fn poll(&mut self) -> Poll<(), ()> { // If we're canceled then we may want to bail out early. // // If the `forget` function was called, though, then we keep going. if self.tx.as_mut().unwrap().poll_cancel().unwrap().is_ready() { if !self.keep_running.get() { return Ok(().into()) } } let result = match self.future.poll() { Ok(Async::NotReady) => return Ok(Async::NotReady), Ok(Async::Ready(t)) => Ok(t), Err(e) => Err(e), }; drop(self.tx.take().unwrap().send(result)); Ok(().into()) } } impl<F: Future + fmt::Debug> fmt::Debug for Execute<F> { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { f.debug_struct("Execute") .field("future", &self.future) .finish() } }