TcpListener does not notify non-ready tasks on drop

[kevinkassimo]

Version

0.1.14

Platform

MacOS High Sierra (but probably also on Linux and Windows)

Description

(Sorry for any confusing description, since I am not very familiar with Tokio and its internals...)

We have a scenario where accept tasks are first put to sleep due to TcpListener::poll_accept() returns Async::NotReady, and then we drop the TcpListener. For some reason, such tasks are not notified on dropping of the listener and thus hangs in NotReady state forever (instead we want the tasks to poll on drop so that the tasks could error out)

The example scenario is described in denoland/deno#1516 and a current workaround is in denoland/deno#1517 . Currently I have to keep references to the accept tasks and manually call notify() on each of the them when the listener is dropped. Not quite sure if this could/should be handled by TcpListener itself?

[tobz]

Hmm, I'm confused on what's going on under the hood that allows your code to let this happen.

Do you have some more code -- the actual Rust/Tokio code -- showing how you're managing to drop the listener from a task other than the task actually using it?

[kevinkassimo]

@tobz
It is somehow a bit complicated structure (Sorry for piling up code here...)

We have a global hashmap called a RESOURCE_TABLE to keep references of open resources, such as TcpListener (wrapped in enum). From deno/src/resources.rs

  // 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());
  }

close() removes, for example, a wrapped TcpListener out of the table and thus dropping it. It is possible that this close would be called by some user on the main thread anytime.

And there is another struct called Resource, which contains an id and could be used to retrieve the actual resource (TcpListener) from the RESOURCE_TABLE.

There is a method poll_accept implemented on Resource. It only retrieves the wrapped TcpListener from the RESOURCE_TABLE entry and calls poll_accept on the listener:

  pub fn poll_accept(&mut self) -> Poll<(TcpStream, SocketAddr), Error> {
    let mut table = RESOURCE_TABLE.lock().unwrap();
    let maybe_repr = table.get_mut(&self.rid); // Use the id of `Resource` to retrieve the actual TcpListener resource
    match maybe_repr {
      None => panic!("bad rid"),
      Some(repr) => match repr {
        Repr::TcpListener(ref mut s) => s.poll_accept(), // call poll_accept on TcpListener
        _ => panic!("Cannot accept"),
      },
    }
  }

With these basic setup, at some random point (after the creation of the listener), a user might call a method called accept():

pub fn accept(r: Resource) -> Accept {
  Accept {
    state: AcceptState::Pending(r), // This `r` happens to be a Resource linking to the target TcpListener
  }
}

which creates a special struct Accept and save a Resource (as defined above) indirectly linking to the TcpListener inside of it. Trait Future is implemented on Accept:

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 {
      AcceptState::Pending(ref mut r) => try_ready!(r.poll_accept()), // <- Marked
      AcceptState::Empty => panic!("poll Accept after it's done"),
    };

    match mem::replace(&mut self.state, AcceptState::Empty) {
      AcceptState::Pending(_) => Ok((stream, addr).into()),
      AcceptState::Empty => panic!("invalid internal state"),
    }
  }
}

So at the marked line inside this poll, we call the poll_accept defined above for Resource, and thus indirectly calling TcpListener::poll_accept if the listener still survives in the RESOURCE_TABLE.

Since we are using try_ready!, I assume some of the tasks might be put in suspense due to Async::NotReady from TcpListener::poll_accept.

It is possible that there might be multiple tasks associated with Accept futures at the moment Resource::close() is called upon a TcpListener resource. It seems that when TcpListener is dropped, none of the tasks held in suspense are notified, and they would just hang forever.

Instead, we hope that a Task::notify could be called on these tasks, so that they would attempt one more poll and find out that the TcpListener resource is gone, either panicking or returning an error result.

Hope this would be clear... Sorry for any confusing description since I am not that familiar with Tokio internals...

[tobz]

So, generally speaking, a future will never hold on to more than one task calling it. If you're expecting all tasks that ever poll a single TcpListener -- when it returns NotReady -- to be notified on drop, or even on ready, then I'm pretty confident your code will not ever work as expected.

I have a better understanding on your use case here now, though. I think having a Drop impl would be reasonable, although as mentioned, there is no way it can notify multiple tasks for you.

[kevinkassimo]

@tobz Thanks for the explanation. I believe I can enforce single accept at a time myself here, and a Drop impl for the currently tracked accept task would be good enough.

[carllerche]

This is an interesting case and it may be a question of defining guarantees.

The task is intentionally not notified on drop. The intent is that, when poll_accept returns NotReady, the task is notified when a call to poll_accept might return Ready. If the TcpListener instance is dropped, poll_accept cannot return Ready because the TcpListener is gone thus poll_accept cannot be called.

It sounds like, in your case, you are implementing your own custom "resource" . You will probably want to implement the notification logic you need yourself.

In your own poll_accept implementation, if you return NotReady, track the current task yourself. Then, when close is called, notify that task yourself. The best way to do that depends on your concurrency story. AtomicTask in the futures crate provides a lock-free way to do thread-safe task notification.

Does this make sense?

[carllerche]

I'm going to close the issue, but please let me know if you think this is incorrect.

[ry]

Yes - makes sense - thanks for explanations.

[kevinkassimo]

@carllerche Got it. Have implemented something ourselves to track the current task and seems working fine.