Too many dangling CLOSE_WAIT connections

Hey Wangle team,

thank you guys for the wonderful work. We have been using wangle to build a RPC layer pretty well in our projects except too many dangling CLOSE_WAIT connections on Ubuntu 16, eventually halting the server for any responses. With much efforts of debugging, we resort to get hints from here. Note that all destructors and close methods are called properly. I'd highlight the code structure here. Your help on how to debug this are greatly appreciated!!!!

Btw, wangle v2018.10.22.00 is used in this case.

Specifically, client sock is in FIN_WAIT2 status, and server sock is in CLOSE_WAIT status. Client socks soon disappear (prob. being forced to close by OS), however, server socks in CLOSE_WAIT accumulate, turning into dangling socks leaving CLOSE_WAIT status.

netstat output of client socks,
```
tcp 0 0 172.31.38.97:60445 54.118.66.170:15515 FIN_WAIT2
tcp 0 0 172.31.38.97:60447 54.118.66.170:15515 FIN_WAIT2
tcp 0 0 172.31.38.97:60449 54.118.66.170:15515 FIN_WAIT2
```
netstat output of server socks
```
tcp6 1 0 54.118.66.170:15515 172.31.38.97:60445 CLOSE_WAIT
tcp6 1 0 54.118.66.170:15515 172.31.38.97:60447 CLOSE_WAIT
tcp6 1 0 54.118.66.170:15515 172.31.38.97:60449 CLOSE_WAIT
```
Waiting for a while, CLOSE_WAIT socks transition to dangling status,
lsof -p 5489 | grep TCPv6
```
rpc-server 5489 ubuntu 20u sock 0,8 0t0 2007789 protocol: TCPv6
rpc-server 5489 ubuntu 25u sock 0,8 0t0 2006998 protocol: TCPv6
rpc-server 5489 ubuntu 26u sock 0,8 0t0 2009216 protocol: TCPv6
rpc-server 5489 ubuntu 28u sock 0,8 0t0 2009218 protocol: TCPv6
```

Code wise, for client side, RpcClient forwards request to RpcConnection maintained by ConnectionPool based on connection id (e.g. host and port). RpcConnection internally has RpcService which is ClientDispatcher created by ConnectionFactory.

```
class RpcClient {
public:
// internally calls RpcConnection::SendRequest
virtual ... AsyncCall() {
cp_->GetConnection(remote_id)->SendRequest(...)
}

private:
std::shared_ptr cp_;
std::shared_ptr io_executor_;
std::shared_ptr cpu_executor_;
};
```

```
class RpcConnection {
public:
virtual folly::Future> SendRequest(std::unique_ptr req);

private:
std::recursive_mutex mutex_;
std::shared_ptr io_executor_;
std::shared_ptr cpu_executor_;

// ConnectionId used by ConnectionPool
std::shared_ptr connection_id_;

// initialized by ConnectionFactory::connect, it's ClientDispatcher indeed where
// Promise and Future are handled
std::shared_ptr rpc_service_;

std::shared_ptr cf_;
std::shared_ptr> client_bootstrap_;
};
```

```
class ConnectionPool {
...
private:
std::shared_ptr cf_;
std::shared_ptr conf_;
std::unordered_map, std::shared_ptr> connections_;
}
```

```
class ConnectionFactory {
virtual std::shared_ptr Connect(
std::shared_ptr> client_bootstrap,
const std::string &hostname, uint16_t port);
};
```
```
class RpcClientPipelineFactory {

public:
RpcClientSerializePipeline::Ptr RpcClientPipelineFactory::newPipeline() {
auto pipeline = RpcClientSerializePipeline::create();
pipeline->setTransport(sock);
pipeline->addBack(wangle::AsyncSocketHandler{sock});
pipeline->addBack(wangle::EventBaseHandler{});
pipeline->addBack(wangle::LengthFieldBasedFrameDecoder{});
pipeline->addBack(RpcClientSeralizeHandler....)
}

};
```

For server side, it's very straightforward,

```
class RpcServer {
public:
void StartListening(int port) {
auto factory = std::make_shared();
auto server = std::make_shared();
server->childPipeline(factory);
server->bind(port);
}
};
```

```
class RpcServerPipelineFactory {
public:
RpcServerSerializePipeline::Ptr newPipeline(
std::shared_ptr sock) {

auto pipeline = RpcServerSerializePipeline::create();
pipeline->addBack(wangle::AsyncSocketHandler(sock));
pipeline->addBack(wangle::EventBaseHandler());
pipeline->addBack(wangle::LengthFieldBasedFrameDecoder());
pipeline->addBack(RpcServerSerializeHandler());
pipeline->addBack(wangle::MultiplexServerDispatcher<
std::unique_ptr, std::unique_ptr>(
service_.get()));
pipeline->finalize();
return pipeline;
}
};
```