CommClosedError: in <closed TCP>: Stream is closed (Client.run)

Hi,
I'm trying to send a function to all the workers in my distributed cluster, but I'm getting the following error:
- What am I doing wrong?

```python
from dask.distributed import Client
cli = Client("172.19.18.88:8786")
cli
```

Client

• Scheduler: tcp://172.19.18.88:8786
  
• Dashboard: http://172.19.18.88:8787/status
  

Cluster

• Workers: 2


• Cores: 16


• Memory: 8.14 GB

## Example 1

```python
import os
cli.run(os.getpid)
```

{'tcp://172.19.18.61:35829': 8711, 'tcp://172.19.18.90:46199': 9018}

## Example 2

```python
def get_status(dask_worker):
return dask_worker.status

def get_data():
return range(10)

# the two functions return the same error
```

```python
cli.run(get_status)
```

---------------------------------------------------------------------------

StreamClosedError Traceback (most recent call last)

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/comm/tcp.py in read(self, deserializers)
176 try:
--> 177 n_frames = yield stream.read_bytes(8)
178 n_frames = struct.unpack('Q', n_frames)[0]

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1132 try:
-> 1133 value = future.result()
1134 except Exception:

StreamClosedError: Stream is closed

During handling of the above exception, another exception occurred:

CommClosedError Traceback (most recent call last)

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
573 comm = yield self.live_comm()
--> 574 result = yield send_recv(comm=comm, op=key, **kwargs)
575 except (RPCClosed, CommClosedError) as e:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1132 try:
-> 1133 value = future.result()
1134 except Exception:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1140 try:
-> 1141 yielded = self.gen.throw(*exc_info)
1142 finally:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/core.py in send_recv(comm, reply, serializers, deserializers, **kwargs)
450 if reply:
--> 451 response = yield comm.read(deserializers=deserializers)
452 else:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1132 try:
-> 1133 value = future.result()
1134 except Exception:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1140 try:
-> 1141 yielded = self.gen.throw(*exc_info)
1142 finally:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/comm/tcp.py in read(self, deserializers)
197 if not shutting_down():
--> 198 convert_stream_closed_error(self, e)
199 else:

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/comm/tcp.py in convert_stream_closed_error(obj, exc)
125 else:
--> 126 raise CommClosedError("in %s: %s" % (obj, exc))
127

CommClosedError: in : Stream is closed

During handling of the above exception, another exception occurred:

CommClosedError Traceback (most recent call last)

in
----> 1 cli.run(get_status)

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/client.py in run(self, function, *args, **kwargs)
2130 '192.168.0.101:9000': 'running}
2131 """
-> 2132 return self.sync(self._run, function, *args, **kwargs)
2133
2134 @gen.coroutine

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
663 return future
664 else:
--> 665 return sync(self.loop, func, *args, **kwargs)
666
667 def __repr__(self):

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
275 e.wait(10)
276 if error[0]:
--> 277 six.reraise(*error[0])
278 else:
279 return result[0]

~/virtualEnv/daskEnv/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/utils.py in f()
260 if timeout is not None:
261 future = gen.with_timeout(timedelta(seconds=timeout), future)
--> 262 result[0] = yield future
263 except Exception as exc:
264 error[0] = sys.exc_info()

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1131
1132 try:
-> 1133 value = future.result()
1134 except Exception:
1135 self.had_exception = True

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1139 if exc_info is not None:
1140 try:
-> 1141 yielded = self.gen.throw(*exc_info)
1142 finally:
1143 # Break up a reference to itself

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/client.py in _run(self, function, *args, **kwargs)
2078 args=dumps(args),
2079 kwargs=dumps(kwargs)),
-> 2080 workers=workers, nanny=nanny)
2081 results = {}
2082 for key, resp in responses.items():

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1131
1132 try:
-> 1133 value = future.result()
1134 except Exception:
1135 self.had_exception = True

~/virtualEnv/daskEnv/lib/python3.6/site-packages/tornado/gen.py in run(self)
1139 if exc_info is not None:
1140 try:
-> 1141 yielded = self.gen.throw(*exc_info)
1142 finally:
1143 # Break up a reference to itself

~/virtualEnv/daskEnv/lib/python3.6/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
575 except (RPCClosed, CommClosedError) as e:
576 raise e.__class__("%s: while trying to call remote method %r"
--> 577 % (e, key,))
578
579 self.comms[comm] = True # mark as open

CommClosedError: in : Stream is closed: while trying to call remote method 'broadcast'

```python

```

Thanks in advance!