Cluster dies with "failed to deserialize" and "IndexError: pop from empty list"

**What happened**:
I use dask to dispatch embarrassingly parallel vtk processing jobs. If the vtk files are large-ish, the dask cluster dies halfway through processing the files.

**What you expected to happen**:
The files should be processed and when all of them are done the dask cluster exits cleanly.

**Minimal Complete Verifiable Example**:
Minimal example is difficult because of the complexity of my dispatch system, plus the function called in parallel must be part of an installed package or else I get errors saying that vtk objects are not pickle-able. Nonetheless, I've been able to produce a working example that produces the error fairly simply and consistently.
Download and unzip the attached files. Install the vtktest package (and dependencies) using pip and run the test scripts to see the failure.
[mcve.zip](https://github.com/dask/distributed/files/5183344/mcve.zip)
e.g.

```
$ unzip mcve.zip
$ cd mcve
$ conda create -n vtktest python=3.8 vtk=8 dask distributed psutil tqdm tornado asyncssh
$ conda activate vtktest
$ pip install vtktest/
$ python run_vtk_test_succeeds.py
$ python run_vtk_test_fails.py
```

The two scripts are identical except that the "succeeds" script processes the small vtk file and the "fails" script processes the big vtk file. On my mac, running the above proceedure I get these outputs:

Success

```
$ python run_vtk_test_succeeds.py
INFO:vtktest.parallel:launching dask LocalCluster with 4 workers
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0003.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0010.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0011.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0006.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0008.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0001.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0009.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0000.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0007.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0005.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0004.vtp)
Success!

```

Failure

```
$ python run_vtk_test_fails.py
INFO:vtktest.parallel:launching dask LocalCluster with 4 workers
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:dask.vtktest.convert:vtk converted in 4.3 s (isoSurfaceCell-big_0001.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 4.3 s (isoSurfaceCell-big_0006.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 4.4 s (isoSurfaceCell-big_0010.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 4.4 s (isoSurfaceCell-big_0008.vtp)
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/protocol/core.py", line 125, in loads
small_header = frames.pop()
IndexError: pop from empty list
distributed.worker - WARNING - Compute Failed
Function: inner
args: ('0.6/isoSurfaceCell-big.vtk', 'isoSurfaceCell-big_0006.vtp')
kwargs: {}
Exception: IndexError('pop from empty list')

distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/protocol/core.py", line 125, in loads
small_header = frames.pop()
IndexError: pop from empty list
distributed.worker - WARNING - Compute Failed
Function: inner
args: ('0.8/isoSurfaceCell-big.vtk', 'isoSurfaceCell-big_0008.vtp')
kwargs: {}
Exception: IndexError('pop from empty list')

/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py:252: RuntimeWarning: Closing semaphore semaphore-96153c55c35744e2b1bbc3e17376b4c2 but there remain unreleased leases ['4300952401ee46de90639f352105afb0', '77db9a1d0db74b99a69522788dcbca08', 'db72c24cde334261a2d967ec970cbd93', 'dfae5aff6a434731a22854c89b8dfe8e']
warnings.warn(
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
Traceback (most recent call last):
File "run_vtk_test_fails.py", line 34, in
ret = dmap(convert_vtk, in_out, n_procs=mp.cpu_count(), star=True, p_bar=False)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 135, in dmap
return _dmap_local(func, args, n_procs, star, p_bar, batch, change_cwd, kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 189, in _dmap_local
return _dmap_client(func, args, client, star, p_bar, batch, change_cwd, kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 364, in _dmap_client
return client.gather(futures)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 1985, in gather
return self.sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 833, in sync
return sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 339, in sync
raise exc.with_traceback(tb)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 323, in f
result[0] = yield future
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 1850, in _gather
raise exception.with_traceback(traceback)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 417, in inner
return ret
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py", line 491, in __exit__
self.release()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py", line 476, in release
return self.client.sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 833, in sync
return sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 339, in sync
raise exc.with_traceback(tb)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 323, in f
result[0] = yield future
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/core.py", line 883, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/core.py", line 666, in send_recv
response = await comm.read(deserializers=deserializers)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/tcp.py", line 204, in read
msg = await from_frames(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/utils.py", line 87, in from_frames
res = _from_frames()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/utils.py", line 65, in _from_frames
return protocol.loads(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/protocol/core.py", line 125, in loads
small_header = frames.pop()
IndexError: pop from empty list
```

Weirdly, sometimes if I then rerun the succeeds script, it will pick up the big vtk files that didn't get deleted when the previous script failed, and successfully process both the small and big files with no errors. Other times it will error again and I have no idea why this happens seemingly randomly. I'd say this is about 50/50.

Second Run Success

```
$ python run_vtk_test_succeeds.py
INFO:vtktest.parallel:launching dask LocalCluster with 4 workers
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0003.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0011.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0001.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0006.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0008.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0009.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0000.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.7 s (isoSurfaceCell-small_0007.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0002.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0005.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0010.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.6 s (isoSurfaceCell-small_0004.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.8 s (isoSurfaceCell-big_0011.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 4.3 s (isoSurfaceCell-big_0003.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.5 s (isoSurfaceCell-big_0006.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.5 s (isoSurfaceCell-big_0000.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.6 s (isoSurfaceCell-big_0008.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.9 s (isoSurfaceCell-big_0005.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.8 s (isoSurfaceCell-big_0007.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.3 s (isoSurfaceCell-big_0010.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 5.2 s (isoSurfaceCell-big_0001.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 5.3 s (isoSurfaceCell-big_0004.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.6 s (isoSurfaceCell-big_0002.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 3.6 s (isoSurfaceCell-big_0009.vtp)
Success!
```

Second Run Failure

```
$ python run_vtk_test_succeeds.py
INFO:vtktest.parallel:launching dask LocalCluster with 4 workers
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:asyncio:Using selector: KqueueSelector
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0010.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0006.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0001.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0008.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0000.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0009.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0007.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0002.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0005.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0004.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0003.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 0.5 s (isoSurfaceCell-small_0011.vtp)
DEBUG:dask.vtktest.convert:vtk converted in 5.3 s (isoSurfaceCell-big_0011.vtp)
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/protocol/core.py", line 125, in loads
small_header = frames.pop()
IndexError: pop from empty list
distributed.worker - WARNING - Compute Failed
Function: inner
args: ('1.1/isoSurfaceCell-big.vtk', 'isoSurfaceCell-big_0011.vtp')
kwargs: {}
Exception: IndexError('pop from empty list')

DEBUG:dask.vtktest.convert:vtk converted in 5.3 s (isoSurfaceCell-big_0010.vtp)
/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py:252: RuntimeWarning: Closing semaphore semaphore-66fc4b9b312a4b0a8803f6d0bf0faf2e but there remain unreleased leases ['24718da6ebf04627a0dde8bf82dbb1a5', '47ff5f43ba144deb87ae1cb155798202', '51817c2438d24072a7fdaf8184d79b1e', '8fe931500fd14881b42087e90735aaef']
warnings.warn(
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 3 seconds, killing
distributed.nanny - WARNING - Worker process still alive after 4 seconds, killing
Traceback (most recent call last):
File "run_vtk_test_succeeds.py", line 35, in
ret = dmap(convert_vtk, in_out, n_procs=mp.cpu_count(), star=True, p_bar=False)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 135, in dmap
return _dmap_local(func, args, n_procs, star, p_bar, batch, change_cwd, kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 189, in _dmap_local
return _dmap_client(func, args, client, star, p_bar, batch, change_cwd, kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 364, in _dmap_client
return client.gather(futures)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 1985, in gather
return self.sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 833, in sync
return sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 339, in sync
raise exc.with_traceback(tb)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 323, in f
result[0] = yield future
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 1850, in _gather
raise exception.with_traceback(traceback)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/vtktest/parallel.py", line 417, in inner
return ret
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py", line 491, in __exit__
self.release()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/semaphore.py", line 476, in release
return self.client.sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/client.py", line 833, in sync
return sync(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 339, in sync
raise exc.with_traceback(tb)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/utils.py", line 323, in f
result[0] = yield future
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/core.py", line 883, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/core.py", line 666, in send_recv
response = await comm.read(deserializers=deserializers)
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/tcp.py", line 204, in read
msg = await from_frames(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/utils.py", line 87, in from_frames
res = _from_frames()
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/comm/utils.py", line 65, in _from_frames
return protocol.loads(
File "/Users/shannon/opt/miniconda3/lib/python3.8/site-packages/distributed/protocol/core.py", line 125, in loads
small_header = frames.pop()
IndexError: pop from empty list
```

**Anything else we need to know?**:

If I run the "fails" script repeatedly, around 25-30% of the time it succeeds with no errors.
I also see the same errors on CentOS 6 and 7, and I see the same errors (or close enough) dispatching to a distributed cluster using an SSHCLuster.

**Environment**:

- Dask version: 2.25.0
- Python version: 3.8.5
- Operating System: macOS, CentOS 6, CentOS 7
- Install method (conda, pip, source): conda