diff options
Diffstat (limited to 'cli')
-rw-r--r-- | cli/resources.rs | 37 | ||||
-rw-r--r-- | cli/tokio_util.rs | 21 |
2 files changed, 2 insertions, 56 deletions
diff --git a/cli/resources.rs b/cli/resources.rs index 66a2ebdb3..3a7121d4c 100644 --- a/cli/resources.rs +++ b/cli/resources.rs @@ -171,49 +171,12 @@ impl Resource { } } - /// Track the current task (for TcpListener resource). - /// Throws an error if another task is already tracked. - pub fn track_task(&mut self) -> Result<(), std::io::Error> { - let mut table = RESOURCE_TABLE.lock().unwrap(); - // Only track if is TcpListener. - if let Some(Repr::TcpListener(_, t)) = table.get_mut(&self.rid) { - // Currently, we only allow tracking a single accept task for a listener. - // This might be changed in the future with multiple workers. - // Caveat: TcpListener by itself also only tracks an accept task at a time. - // See https://github.com/tokio-rs/tokio/issues/846#issuecomment-454208883 - if t.is_some() { - return Err(std::io::Error::new( - std::io::ErrorKind::Other, - "Another accept task is ongoing", - )); - } - t.replace(futures::task::current()); - } - Ok(()) - } - - /// Stop tracking a task (for TcpListener resource). - /// Happens when the task is done and thus no further tracking is needed. - pub fn untrack_task(&mut self) { - let mut table = RESOURCE_TABLE.lock().unwrap(); - // Only untrack if is TcpListener. - if let Some(Repr::TcpListener(_, t)) = table.get_mut(&self.rid) { - assert!(t.is_some()); - t.take(); - } - } - // close(2) is done by dropping the value. Therefore we just need to remove // the resource from the RESOURCE_TABLE. pub fn close(&self) { let mut table = RESOURCE_TABLE.lock().unwrap(); let r = table.remove(&self.rid); assert!(r.is_some()); - // If TcpListener, we must kill all pending accepts! - if let Repr::TcpListener(_, Some(t)) = r.unwrap() { - // Call notify on the tracked task, so that they would error out. - t.notify(); - } } pub fn shutdown(&mut self, how: Shutdown) -> Result<(), DenoError> { diff --git a/cli/tokio_util.rs b/cli/tokio_util.rs index a57cbbd2e..e1f8587c3 100644 --- a/cli/tokio_util.rs +++ b/cli/tokio_util.rs @@ -78,31 +78,14 @@ pub fn accept(r: Resource) -> Accept { pub struct Accept { state: AcceptState, } + impl Future for Accept { type Item = (TcpStream, SocketAddr); type Error = io::Error; fn poll(&mut self) -> Poll<Self::Item, Self::Error> { let (stream, addr) = match self.state { - // Similar to try_ready!, but also track/untrack accept task - // in TcpListener resource. - // In this way, when the listener is closed, the task can be - // notified to error out (instead of stuck forever). - AcceptState::Pending(ref mut r) => match r.poll_accept() { - Ok(futures::prelude::Async::Ready(t)) => { - r.untrack_task(); - t - } - Ok(futures::prelude::Async::NotReady) => { - // Would error out if another accept task is being tracked. - r.track_task()?; - return Ok(futures::prelude::Async::NotReady); - } - Err(e) => { - r.untrack_task(); - return Err(From::from(e)); - } - }, + AcceptState::Pending(ref mut r) => try_ready!(r.poll_accept()), AcceptState::Empty => panic!("poll Accept after it's done"), }; |