Infinite loop on failure to unpickle

distributed 2.11
Python 3.8
Linux x64

#### Use case
I'm using distributed and pint in an application that uses custom units.
In order for a python interpreter to be able to unpickle custom units, it needs to explicitly invoke a registration function (pint.set_application_registry) ahead of unpickling.
Read: https://pint.readthedocs.io/en/latest/serialization.html

My application contained a bug where
1. a pint.Quantity with custom units was created in a worker
2. it was returned by the dask task, serialized, and sent to a different worker where I accidentally had not invoked pint.set_application_registry, thus causing pickle.loads to fail

#### Expected behaviour
Graph resolution crashes; ``Future.result`` on the client re-raises whatever exception was raised by ``pickle.loads`` on the worker

#### Actual behaviour
dask.distributed enters an infinite loop.

#### POC
```python
import distributed
import time

class C:
def __reduce__(self):
return C.unpickle, ()

@classmethod
def unpickle(cls):
assert False

def dummy(x):
time.sleep(1)

def main():
cluster = distributed.LocalCluster(threads_per_worker=1, n_workers=8)
client = distributed.Client(cluster)
f1 = client.submit(C)
f2s = [client.submit(dummy, f1, pure=False) for _ in range(8)]
client.gather(f2s)

if __name__ == "__main__":
main()
```
The issue disappears if I replace ``range(8)`` with ``range(2)`` or replace ``time.sleep(1)`` with ``pass``. In both cases, the underlying reason is that the worker executing dummy ends up being the same worker that created C, thus it never gets to unpickle it.