scatter broadcast fails for large numpy array

**What happened**:

When dealing with large NumPy arrays, scatter starts to fail with a strange error: `Exception: too many values to unpack (expected 1)` rather than telling me the array is too large.

**What you expected to happen**: Either for it to work or give me some kind of memory error.

**Minimal Complete Verifiable Example**:

The example below might obviously not throw an error if you have enough memory. I am on a machine with 12gb memory.
```python
import numpy as np

X = np.random.random(size=[150_000, 100])

from joblib import parallel_backend, Parallel, delayed
from dask.distributed import Client

client = Client(processes=True, threads_per_worker=2,
n_workers=4)

client.scatter(X, broadcast=True)
```

outputs:
```
distributed.core - ERROR - Exception while handling op scatter
Traceback (most recent call last):
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py", line 501, in handle_comm
result = await result
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/scheduler.py", line 5037, in scatter
keys, who_has, nbytes = await scatter_to_workers(
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils_comm.py", line 144, in scatter_to_workers
out = await All(
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils.py", line 237, in All
result = await tasks.next()
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py", line 862, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/home/alexis/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py", line 663, in send_recv
raise Exception(response["text"])
Exception: too many values to unpack (expected 1)
---------------------------------------------------------------------------
Exception Traceback (most recent call last)
in
----> 1 client.scatter(X, broadcast=True)

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/client.py in scatter(self, data, workers, broadcast, direct, hash, timeout, asynchronous)
2175 else:
2176 local_worker = None
-> 2177 return self.sync(
2178 self._scatter,
2179 data,

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
842 return future
843 else:
--> 844 return sync(
845 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
846 )

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
351 if error[0]:
352 typ, exc, tb = error[0]
--> 353 raise exc.with_traceback(tb)
354 else:
355 return result[0]

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils.py in f()
334 if callback_timeout is not None:
335 future = asyncio.wait_for(future, callback_timeout)
--> 336 result[0] = yield future
337 except Exception as exc:
338 error[0] = sys.exc_info()

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/tornado/gen.py in run(self)
760
761 try:
--> 762 value = future.result()
763 except Exception:
764 exc_info = sys.exc_info()

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/client.py in _scatter(self, data, workers, broadcast, direct, local_worker, timeout, hash)
2066 )
2067 else:
-> 2068 await self.scheduler.scatter(
2069 data=data2,
2070 workers=workers,

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
860 name, comm.name = comm.name, "ConnectionPool." + key
861 try:
--> 862 result = await send_recv(comm=comm, op=key, **kwargs)
863 finally:
864 self.pool.reuse(self.addr, comm)

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py in send_recv(comm, reply, serializers, deserializers, **kwargs)
659 if comm.deserialize:
660 typ, exc, tb = clean_exception(**response)
--> 661 raise exc.with_traceback(tb)
662 else:
663 raise Exception(response["text"])

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py in handle_comm()
499 result = asyncio.ensure_future(result)
500 self._ongoing_coroutines.add(result)
--> 501 result = await result
502 except (CommClosedError, CancelledError) as e:
503 if self.status == Status.running:

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/scheduler.py in scatter()
5035 assert isinstance(data, dict)
5036
-> 5037 keys, who_has, nbytes = await scatter_to_workers(
5038 nthreads, data, rpc=self.rpc, report=False
5039 )

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils_comm.py in scatter_to_workers()
142 rpcs = {addr: rpc(addr) for addr in d}
143 try:
--> 144 out = await All(
145 [
146 rpcs[address].update_data(

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/utils.py in All()
235 while not tasks.done():
236 try:
--> 237 result = await tasks.next()
238 except Exception:
239

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py in send_recv_from_rpc()
860 name, comm.name = comm.name, "ConnectionPool." + key
861 try:
--> 862 result = await send_recv(comm=comm, op=key, **kwargs)
863 finally:
864 self.pool.reuse(self.addr, comm)

~/anaconda3/envs/wsl_ml/lib/python3.8/site-packages/distributed/core.py in send_recv()
661 raise exc.with_traceback(tb)
662 else:
--> 663 raise Exception(response["text"])
664 return response
665

Exception: too many values to unpack (expected 1)
```

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

**Environment**:

- Dask version: 2021.03.1
- Python version: 3.8.5 (default, Sep 4 2020, 07:30:14)
- Operating System: Ubuntu 18.04 WSL2 on Windows 10
- Install method (conda, pip, source): conda