tokio-rs/tokio-core

Tokio does not appear to drop chained future on disconnect

cetra3 opened this issue · 14 comments

cetra3 commented

This is a weird one, and I'm not sure where the problem lies exactly but I am seeing that connections are hanging around when doing a hyper proxy-like configuration, so it might be a bug in hyper too.

What I'm seeing though is when I'm trying to stream a proxy request (like in this example), the TCP connection drops off from the client -> proxy, but from the proxy -> server, if the client prematurely disconnects, then this second connection stays open.

This was seen by proxying Partial Content requests, especially with a video or other media, and skipping around.

What I'd like to know is when a tcp connection is cut off, is this meant to propagate down the future chain? I would've thought it would cause the whole thing to disconnect, but I'm also not sure when hyper actually disconnects.

alexcrichton commented

When making a proxy there typically needs to be some form of the shutdown syscall, but maybe that's not happening?

cetra3 commented

I'm not sure where the issue is. I've raised this with hyper as well just to cover bases, but it does seem to be a tokio-level bug. They have alleviated one issue by decoupling tokio-proto, but that has caused some new ones (namely successful connections don't drop now).

It's might be because tokio is only expecting one socket per future to be active, and just forgets about the other one. I can't really find a way to print out the state of a tokio reactor periodically which would be useful.

cramertj commented

I think I might have tracked this down here.

cramertj commented

@cetra3 Can you confirm that tokio-rs/tokio-proto#195 fixes your leak?

cetra3 commented

@cramertj I have tried that PR, but this doesn't resolve the issue.

I added the following to my Cargo.toml:

[patch.crates-io]
tokio-proto = {git = "https://github.com/cramertj/tokio-proto"}

This also doesn't appear to resolve hyperium/hyper#1353 which I have tested by using the following Cargo.toml with the pasted code there:

[package]
name = "hyper_test"
version = "0.1.0"

[dependencies]
hyper = {git = "https://github.com/hyperium/hyper"}
env_logger = "*"
tokio-core = "*"
tokio-io = "*"
futures = "*"

[patch.crates-io]
tokio-proto = {git = "https://github.com/cramertj/tokio-proto"}
cramertj commented

@cetra3 Shucks! I figured they must be the same issue. I'll try and figure out what's going on here when I have another moment-- perhaps there's a related fix that needs to happen somewhere else.

cetra3 commented

@cramertj No worries. There is a workaround in hyper by not even using the tokio-proto at all that appears to work, but it'd be nice if it was fixed for tokio-proto too.

carllerche commented

I don't think that this is a tokio-core issue. If you think it is, could you provide a reproduction that only uses tokio-core (no additional dependencies). Thanks!

cramertj commented

@carllerche Yes, it probably belongs on tokio-io or hyper.

carllerche commented

Ok, I'm going to close this now. Please re-open the issue on the repo & cross reference 👍

cetra3 commented

@cramertj @carllerche Should I raise an issue there?

carllerche commented

Probably hyper, that said i would be surprised if there isn't already one.

YetAnotherMinion commented

I tracked down why the sockets are not being shutdown to a seemingly bizarre code in tokio-core, where if you are using a reference to a TcpStream, then shutdown is a no-op. It does not touch the self.io at all.

impl<'a> AsyncWrite for &'a TcpStream {
    fn shutdown(&mut self) -> Poll<(), io::Error> {
        Ok(().into())
    }

Which is called by

impl AsyncWrite for TcpStream {
    fn shutdown(&mut self) -> Poll<(), io::Error> {
        <&TcpStream>::shutdown(&mut &*self)
    }

Which comes from tokio-io using calling the AsyncWrite version of shutdown because of its generic implementation.

#1  0x00005555555d6d6c in tokio_io::framed::{{impl}}::shutdown<tokio_core::net::tcp::TcpStream,websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>> (self=0x7ffff509a0f8)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-io-0.1.4/src/framed.rs:190
190	        self.0.shutdown()
(gdb) l
185	    }
186	}
187	
188	impl<T: AsyncWrite, U> AsyncWrite for Fuse<T, U> {
189	    fn shutdown(&mut self) -> Poll<(), io::Error> {
190	        self.0.shutdown()
191	    }
192	}
193	
194	impl<T, U: Decoder> Decoder for Fuse<T, U> {

Why is there a different version of shutdown for the AsyncWrite trait bounds? Did someone just forget to fill in the method?

Stack trace:

#0  tokio_core::net::tcp::{{impl}}::shutdown (self=0x7ffff509a0f8)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-core-0.1.10/src/net/tcp.rs:525
#1  0x00005555555d6d6c in tokio_io::framed::{{impl}}::shutdown<tokio_core::net::tcp::TcpStream,websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>> (self=0x7ffff509a0f8)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-io-0.1.4/src/framed.rs:190
#2  0x00005555555ade1e in tokio_io::framed_write::{{impl}}::close<tokio_io::framed::Fuse<tokio_core::net::tcp::TcpStream, websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>>> (self=0x7ffff509a0f8)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-io-0.1.4/src/framed_write.rs:229
#3  0x00005555555d6f74 in tokio_io::framed::{{impl}}::close<tokio_core::net::tcp::TcpStream,websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>> (self=0x7ffff509a0f8)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-io-0.1.4/src/framed.rs:148
#4  0x00005555555abd9b in futures::stream::split::{{impl}}::close<tokio_io::framed::Framed<tokio_core::net::tcp::TcpStream, websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>>> (self=0x7ffff49fc738)
    at /home/user/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.17/src/stream/split.rs:70
#5  0x00005555555d817c in parallel_server::main::{{closure}}::{{closure}}::{{closure}}::{{closure}} (
    sink=SplitSink<tokio_io::framed::Framed<tokio_core::net::tcp::TcpStream, websocket::codec::ws::MessageCodec<websocket::message::OwnedMessage>>> = {...}) at src/bin/parallel-server.rs:90
carllerche commented

Ok, this is due to AsyncWrite::shutdown being named confusingly: tokio-rs/tokio-io#80

It looks like there is no bug and the confusing name of AsyncWrite::shutdown is a known issue that will be fixed in the next release, so I am going to close this. Please let me know if I am wrong.

Thansk for digging into this.