TcpStream sockets leaked by websocket::client::async::Framed::close()

When you call close on a Framed stream, it does not actually close or shutdown the underlying socket fd.
See (https://github.com/tokio-rs/tokio-io/issues/80). I tested the parallel-server from the examples/ folder by setting the file descriptor limit per process to 1024 and used https://github.com/observing/thor to connect the the server 2000 times and send 1 message with 100 concurrently active connections.

./bin/thor --amount 2000 --concurrent 100 --masked --messages 1 ws://127.0.0.1:8081

The example server will crash in the accept call because it hits the file descriptor limit. View the open file descriptors using ls /proc/$PID/fd of the example parallel-server, and you will see that none of the socket connections were closed (there will be 1024 sockets open even though the client program only has 100 open)

This is the shutdown that gets called by calling close on the Framed. Note that it does not call shutdown on the socket, unlike TcpStream::shutdown

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 Framed impl using 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> {

See https://github.com/tokio-rs/tokio-io/issues/80#issuecomment-352207760

@P-E-Meunier Because "AsyncWrite::shutdown" and TCP shutdown are two completely separate functions with different behaviors.

This is why AsyncWrite::shutdown needs to be renamed.

I solved the TcpStream not being closed by creating a custom split function that returned a custom SplitSink struct that returned mutable access to the contained stream.

/// A `Sink` part of the split pair                                                                 
#[derive(Debug)]                                                                                    
pub struct MySplitSink<S>(BiLock<S>);                                                               
                                                                                                    
impl<S> MySplitSink<S> {                                                                            
    /// Attempts to put the two "halves" of a split `Stream + Sink` back                            
    /// together. Succeeds only if the `MySplitStream<S>` and `MySplitSink<S>` are                  
    /// a matching pair originating from the same call to `Stream::split`.                          
    pub fn reunite(self, other: MySplitStream<S>) -> Result<S, ReuniteError<S>> {                   
        self.0.reunite(other.0).map_err(|err| {                                                     
            ReuniteError(MySplitSink(err.0), MySplitStream(err.1))                                  
        })                                                                                          
    }                                                                                               
                                                                                                    
    pub fn inner(&mut self) -> &mut BiLock<S> {                                                     
        &mut self.0                                                                                 
    }                                                                                               
                                                                                                    
}

Which can then be used with a custom future that calls the shutdown method on the TcpStream instance directly, not through the AsyncWrite trait.

pub struct MyShutdown {                                                                               
    pub stream: MySplitSink<Framed<TcpStream, MessageCodec<OwnedMessage>>>,                         
}                                                                                                   
                                                                                                    
impl Future for MyShutdown {                                                                          
    type Item = ();                                                                                 
    type Error = WebSocketError;                                                                    
                                                                                                    
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {                                           
        match self.stream.inner().poll_lock() {                                                     
            Async::Ready(mut framed) => {                                                           
                let tcp_stream: &mut TcpStream = framed.get_mut();                                                                                                                
                match tcp_stream.shutdown(::std::net::Shutdown::Both) {                           
                    Ok(()) => Ok(Async::Ready(())),                                               
                    Err(error) => Err(WebSocketError::IoError(error))                             
                }                                                                                 
                Ok(Async::Ready(()))                                                                
            },                                                                                      
            Async::NotReady => Ok(Async::NotReady),                                                 
        }                                                                                           
    }                                                                                               
}

I propose that the websocket library include this custom version of SplitSink, split, and the Shutdown future because the generic futures implementation for a stream does not handle cleaning up any resources the generic stream may have.
Dropping both the sink and stream of a split stream does not close the socket either.

Errata: I also copied the implementation of SplitStream into MySplitStream so I could make reunite work with the new MySplitSink.

Reproduce the leak with examples/parallel-server and the provided thor command, then inspect websocket::client::async::Framed::close() and tokio-io's framed.rs shutdown path. Compare this behavior with TcpStream::shutdown and the issue's proposed split and shutdown types. Done means repeated connections release their socket descriptors instead of exhausting the process limit.