Memory leak on long lived RpcChannel

[jsdanielh]

Multiple calls to the async call method in the RpcChannel class for the same instance seems to be leaking memory.
I have been tracking a client instance that lives forever in a machine and the client seems to be leaking memory.

Tracking down the memory leaks with pympler I noticed that each call to the method mentioned leaks at least these objects:

- set
- collections.deque
- _asyncio.Task
- _asyncio.Future
- coroutine
- asyncio.locks.Event
- function (_on_completion)

Digging a little more I think there are two explanations for these leaks:

1. The default argument value of None to timeout in the WebSocketRpcClient's call method would cause a leak (and in general when passing None to the RpcChannel's wait_for_response method) since asyncio.as_completed wouldn't have a timeout and since it waits for the self._close Event, that future may not end up marked as completed (since the close event rarely happens in a long lived RpcChannel instance) or timed-out (since there is no timeout) leaking the internal structures used in the method implementation such as the set, collections.deque. So, my personal conclusion is that None as a timeout leaks memory and should be avoided and documented somewhere.
2. wait_for_reponse method in RpcChannel when calling asyncio.as_completed actually creates a task to wait for the self._close event. In a long lived RpcChannel that event never happens and thus that future continues to run until the close method is called. And since the task is created with each RpcChannel's call method, this could end up leaking objects such as _asyncio.Task, _asyncio.Future and coroutine. Here I think the best would be to use a wait_for future with timeout instead of the current Event'swait.

[orweis]

Hi @jsdanielh this seems like a very important find! Thank you for reporting and for the matching PR.

As it's late here, I'll do my best to find time tomorrow to thoroughly review this.

CC: @asafc @roekatz

[orweis]

This is indeed an important find; kudos for finding and reporting this - @jsdanielh
It comes from a rather unexpected angle - RPC call promises frequently failing, and significant resource allocations within asyncio core functions.

In the as_completed I think the main expectation is for the promise to resolve a lot sooner than anything else;
An RPC channel for which the other side is not sending or failing to send replies frequently is a pretty useless RPC channel. That being said, I would agree that we should not expect every RPC implementation on top of the package to behave correctly.

We should probably change the default - to something long enough but not forever - say 300 seconds (5 minutes)
As a default that comes with a memory leak, is a bad default.

We can also look at using asyncio.wait with FIRST_COMPLETED instead of asyncio.as_completed- which is probably more correct, and reduces the potential for a leak a little, though objects are still created on the stack.

I've set some of these points as change requests in your PR, I hope you can apply them as well :)