as_completed with multiple clients can be error-prone

**Minimal Complete Verifiable Example**:

Consider the following:

```python
import numpy as np
import dask.distributed as dd

def work():
return np.random.randn(1024*1024)

if __name__ == "__main__":
with dd.LocalCluster(n_workers=2, threads_per_worker=1) as cluster:
with dd.Client(cluster.scheduler.address) as c1, dd.Client(cluster.scheduler.address) as c2:
print(c1.loop, c2.loop)
fut = c1.submit(work)

for future, result in dd.as_completed([fut], with_results=True):
print(result)
```

In practice, instead of two local clusters, this mostly happened with a combination of local/remote cluster.

**What happened**:

This code is unsafe, because `as_completed` tries to gather the results of `fut` in the loop/thread of `c2`, but the future was created in the loop/thread of `c1`. This can cause error messages like this, or just simply randomly hang:

```
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/protocol/core.py", line 132, in loads
header = msgpack.loads(header, use_list=False, **msgpack_opts)
File "msgpack/_unpacker.pyx", line 202, in msgpack._cmsgpack.unpackb
msgpack.exceptions.ExtraData: unpack(b) received extra data.
tornado.application - ERROR - Exception in callback functools.partial(>, exception=ExtraData(-15, b"\xbc@\x1c9c\xfb?\x1d\xacO\x96\xaf\xdc\xe4?Sb\xb5\xec\x1a\xc1\x00\xc0\x93D\xc4\x00\x92\xc6\xec?\xdcT\xc0\xcd\xca\x88\xc1\xbf\xf3\x96\x1f\xda\xb7\xa9\xf7?3\xd7|\x0e\xc0\xde\xf9?\x95&]\xe8QJ\xd0\xbf\xcf\x14\x81DI'\xe5?M)\xf4\x02\x07\x04\x01\xc0|\x96\xd6Ig\xcb\xda\xbfa\x97%^.7\xe1\xbf\xe7\xcc\xc2u\x9d\xb0\xe0\xbf\x9e\xf6\xdd\xb2\xe6\xf4\xe4?|{\xc3\xa1\x9a\xc2\xe7\xbf\x85r0y\x97B\xcf?\xe7\x1bj\xa2\xfc3\xd6\xbf\xdc\xcc\x08t\x84\x94\xe3\xbf\xed\xf4Dc\x7f\xa7\xf7?\x12\x92\x1cUN+\xd2\xbf\xa9pa\x1aqc\xf6?\x150\x8brx\x9a\xfb?z\x19\x8d9\n\xdd\xc7\xbfCZ\xaf\x87\x00\x13\xff?\x18\x9a_\xf5\x8a4\xe2?$V\xa1]Lv\xd8\xbf\xf6-\x8f\xce;\t\x06\xc0Fw\t\x01%)\x00@\xf4\x15#\x87\xdd\xcc\xe3\xbfp\xd5\xa6B\xf5\xcb\xcd\xbf1.\x1c\xe1\xd5\x1e\x01\xc0o\x03S2Jp\x96\xbfQE\xec\x81\xc2\xac\xf1\xbf\xb9\xb6\x13\xab\xe4T\xe9\xbfS\xe2\xa3\xcc\xe8\xb7\xc9\xbf\x19O'\x08\xac\xd4")>)
Traceback (most recent call last):
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/tornado/ioloop.py", line 741, in _run_callback
ret = callback()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/tornado/ioloop.py", line 765, in _discard_future_result
future.result()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 4373, in _track_future
result = await future._result(raiseit=False)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 246, in _result
result = await self.client._gather([self])
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1879, in _gather
response = await future
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1930, in _gather_remote
response = await retry_operation(self.scheduler.gather, keys=keys)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/utils_comm.py", line 384, in retry_operation
return await retry(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/utils_comm.py", line 369, in retry
return await coro()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/core.py", line 878, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/core.py", line 661, in send_recv
response = await comm.read(deserializers=deserializers)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/tcp.py", line 212, in read
msg = await from_frames(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/utils.py", line 80, in from_frames
res = _from_frames()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/utils.py", line 63, in _from_frames
return protocol.loads(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/protocol/core.py", line 132, in loads
header = msgpack.loads(header, use_list=False, **msgpack_opts)
File "msgpack/_unpacker.pyx", line 202, in msgpack._cmsgpack.unpackb
msgpack.exceptions.ExtraData: unpack(b) received extra data.
```

Or another one:

```
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/protocol/core.py", line 132, in loads
header = msgpack.loads(header, use_list=False, **msgpack_opts)
File "msgpack/_unpacker.pyx", line 195, in msgpack._cmsgpack.unpackb
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xa7 in position 1: invalid start byte
tornado.application - ERROR - Exception in callback functools.partial(>, exception=UnicodeDecodeError('utf-8', b'|\xa7\x15o\xf0\xec\xbf\x9d\xde\xa3\x1b\x0eO\xe3?\t\x9b\x87o\xac', 1, 2, 'invalid start byte')>)
Traceback (most recent call last):
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/tornado/ioloop.py", line 741, in _run_callback
ret = callback()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/tornado/ioloop.py", line 765, in _discard_future_result
future.result()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 4373, in _track_future
result = await future._result(raiseit=False)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 246, in _result
result = await self.client._gather([self])
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1879, in _gather
response = await future
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1930, in _gather_remote
response = await retry_operation(self.scheduler.gather, keys=keys)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/utils_comm.py", line 384, in retry_operation
return await retry(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/utils_comm.py", line 369, in retry
return await coro()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/core.py", line 878, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/core.py", line 661, in send_recv
response = await comm.read(deserializers=deserializers)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/tcp.py", line 212, in read
msg = await from_frames(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/utils.py", line 80, in from_frames
res = _from_frames()
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/comm/utils.py", line 63, in _from_frames
return protocol.loads(
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/protocol/core.py", line 132, in loads
header = msgpack.loads(header, use_list=False, **msgpack_opts)
File "msgpack/_unpacker.pyx", line 195, in msgpack._cmsgpack.unpackb
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xa7 in position 1: invalid start byte
```

Sometimes it also just works, which makes the problem hard to debug. Running with `PYTHONASYNCIODEBUG=1` reveals the problem:

```
distributed.client - ERROR - Non-thread-safe operation invoked on an event loop other than the current one (expected thread 139924263139072, got 139924891268864)
Traceback (most recent call last):
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1251, in _handle_report
result = handler(**msg)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 1273, in _handle_key_in_memory
state.finish(type)
File "/home/alex/.virtualenvs/test/lib/python3.8/site-packages/distributed/client.py", line 449, in finish
self._get_event().set()
File "/home/alex/source/pyenv/versions/3.8.6/lib/python3.8/asyncio/locks.py", line 288, in set
fut.set_result(True)
File "/home/alex/source/pyenv/versions/3.8.6/lib/python3.8/asyncio/base_events.py", line 721, in call_soon
self._check_thread()
File "/home/alex/source/pyenv/versions/3.8.6/lib/python3.8/asyncio/base_events.py", line 758, in _check_thread
raise RuntimeError(
RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one (expected thread 139924263139072, got 139924891268864)
```

(note that I hacked the thread id output into asyncio, so your output will be slightly different)

**What you expected to happen**:

It would be great if `distributed` could warn when using `as_completed` in an unsafe way, for example using something like this:

```patch
diff --git a/distributed/client.py b/distributed/client.py
index e4a31acb..52043681 100644
--- a/distributed/client.py
+++ b/distributed/client.py
@@ -4418,6 +4418,8 @@ class as_completed:
for f in futures:
if not isinstance(f, Future):
raise TypeError("Input must be a future, got %s" % f)
+ if f.client.loop != self.loop:
+ raise RuntimeError("Can't add futures from different loop")
self.futures[f] += 1
self.loop.add_callback(self._track_future, f)
```

It would also be possible to infer the correct `loop` from the futures that were passed in, but that would be a bit more "magical".

Users of `as_completed` can then fix their usage by explicitly passing in the correct `loop`. It's possible that this also affects different APIs that use the default `Client`, `distributed.client.wait` is one example I found which triggers an exception with `asyncio` debugging enabled.

**Environment**:

- Dask version: `distributed==2020.12.0`
- Python version: 3.8.6 (via pyenv)
- Operating System: Linux Debian sid
- Install method (conda, pip, source): pip