We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
"ZMQError: Operation cannot be accomplished in current state" triggers shutdown
## Brief description and request
I've been seeing an error crop up in longer running Parsl DFK's which trigger a graceful, but unwanted shutdown. I can't find a reliable/minimal example to cause this trigger other than "run a while." So, I'm going to post as much debug info I can about our setup and am seeking advice on things to test or other information which would be helpful in debugging this error.

I'm happy to test whatever is needed, I can also make direct code modifications 

## Backtrace of error thrown
 The final error is in the title but here is the backtrace, it reliably fails on this line:

```
... (there are higher functions in this stack from our program)
  File "/home/lnaden/github/qcfractal/qcfractal/queue/parsl_adapter.py", line 94, in count_running_workers
    if hasattr(executor, 'connected_workers'):
  File "/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/parsl/executors/high_throughput/executor.py", line 463, in connected_workers
    workers = self.command_client.run("WORKERS")
  File "/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/parsl/executors/high_throughput/zmq_pipes.py", line 40, in run
    self.zmq_socket.send_pyobj(message, copy=True)
  File "/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/zmq/sugar/socket.py", line 606, in send_pyobj
    return self.send(msg, flags=flags, **kwargs)
  File "/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/zmq/sugar/socket.py", line 395, in send
    return super(Socket, self).send(data, flags=flags, copy=copy, track=track)
  File "zmq/backend/cython/socket.pyx", line 725, in zmq.backend.cython.socket.Socket.send
  File "zmq/backend/cython/socket.pyx", line 772, in zmq.backend.cython.socket.Socket.send
  File "zmq/backend/cython/socket.pyx", line 247, in zmq.backend.cython.socket._send_copy
  File "zmq/backend/cython/socket.pyx", line 242, in zmq.backend.cython.socket._send_copy
  File "zmq/backend/cython/checkrc.pxd", line 25, in zmq.backend.cython.checkrc._check_rc
zmq.error.ZMQError: Operation cannot be accomplished in current state
```

Below this it behaves as though a standard shutdown signal and the next log item is `[I 190716 15:38:08 dflow:821] DFK cleanup initiated`

## Additional info

`if hasattr(executor, 'connected_workers'):` is inside a loop over the values: `parsl.dataflow.dflow.DataFlowKernel.executors.values()` 

Near as I can tell, this error is only thrown and logged at the higher level programs which are running Parsl and this does not appear in any of the logs or files from the `runinfo` folder. 

Leading up to the crash, there are multiple "loss of Manager" errors. The first set of lost managers happened ~6 hours after the processes were started, which is expected given the wall time of the cluster would have stopped them. However, after that initial set, we started loosing managers every 5-30 minutes seemingly randomly, all due to "too many missed heartbeats." This behavior is the same on 2 separate clusters (similar hardware, same SLURM scheduler). Not sure if this is related, but it's an observation.

Config file:
```python
Config(
    app_cache=True,
    checkpoint_files=None,
    checkpoint_mode=None,
    checkpoint_period=None,
    data_management_max_threads=10,
    executors=[HighThroughputExecutor(
        address='calogin1',
        cores_per_worker=16,
        heartbeat_period=30,
        heartbeat_threshold=120,
        interchange_port_range=(55000, 56000),
        label='QCFractal_Parsl_Slurm_Executor',
        launch_cmd='process_worker_pool.py {debug} {max_workers} -p {prefetch_capacity} -c {cores_per_worker} -m {mem_per_worker} --poll {poll_period} --task_url={task_url} --result_url={result_url} --logdir={logdir} --block_id={{block_id}} --hb_period={heartbeat_period} --hb_threshold={heartbeat_threshold} ',
        managed=True,
        max_workers=48,
        mem_per_worker=None,
        poll_period=10,
        prefetch_capacity=0,
        provider=SlurmProvider(
            'normal_q',
            channel=LocalChannel(envs={}, script_dir=None, userhome='/home/lnaden/qcarchive/run_cc'),
            cmd_timeout=10,
            exclusive=True,
            init_blocks=0,
            launcher=SingleNodeLauncher(),
            max_blocks=24,
            min_blocks=0,
            move_files=True,
            nodes_per_block=1,
            parallelism=1,
            scheduler_options='#SBATCH --exclusive\n#SBATCH -A themolssi\n',
            walltime='06:00:00',
            worker_init='source /home/lnaden/qcarchive/boot_qcarchive_env.sh'
        ),
        storage_access=[],
        suppress_failure=False,
        worker_debug=False,
        worker_logdir_root=None,
        worker_port_range=(54000, 55000),
        worker_ports=None,
        working_dir=None
    )],
    lazy_errors=True,
    monitoring=None,
    retries=0,
    run_dir='runinfo',
    strategy='simple',
    usage_tracking=False
)

```

Setup info: SLURM Cluster, executing locally on the head node in a background process. Directly calling and manipulating the `DataFlowKernel` after providing it a `Config` 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/executors/high_throughput/zmq_pipes.py`
Content:
```
1 #!/usr/bin/env python3
2 
3 import zmq
4 import time
5 import pickle
6 import logging
7 
8 logger = logging.getLogger(__name__)
9 
10 
11 class CommandClient(object):
12     """ CommandClient
13     """
14     def __init__(self, ip_address, port_range):
15         """
16         Parameters
17         ----------
18 
19         ip_address: str
20            IP address of the client (where Parsl runs)
21         port_range: tuple(int, int)
22            Port range for the comms between client and interchange
23 
24         """
25         self.context = zmq.Context()
26         self.zmq_socket = self.context.socket(zmq.REQ)
27         self.port = self.zmq_socket.bind_to_random_port("tcp://{}".format(ip_address),
28                                                         min_port=port_range[0],
29                                                         max_port=port_range[1])
30 
31     def run(self, message):
32         """ This function needs to be fast at the same time aware of the possibility of
33         ZMQ pipes overflowing.
34 
35         The timeout increases slowly if contention is detected on ZMQ pipes.
36         We could set copy=False and get slightly better latency but this results
37         in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.
38         This issue can be magnified if each the serialized buffer itself is larger.
39         """
40         self.zmq_socket.send_pyobj(message, copy=True)
41         reply = self.zmq_socket.recv_pyobj()
42         return reply
43 
44     def close(self):
45         self.zmq_socket.close()
46         self.context.term()
47 
48 
49 class TasksOutgoing(object):
50     """ Outgoing task queue from the executor to the Interchange
51     """
52     def __init__(self, ip_address, port_range):
53         """
54         Parameters
55         ----------
56 
57         ip_address: str
58            IP address of the client (where Parsl runs)
59         port_range: tuple(int, int)
60            Port range for the comms between client and interchange
61 
62         """
63         self.context = zmq.Context()
64         self.zmq_socket = self.context.socket(zmq.DEALER)
65         self.zmq_socket.set_hwm(0)
66         self.port = self.zmq_socket.bind_to_random_port("tcp://{}".format(ip_address),
67                                                         min_port=port_range[0],
68                                                         max_port=port_range[1])
69         self.poller = zmq.Poller()
70         self.poller.register(self.zmq_socket, zmq.POLLOUT)
71 
72     def put(self, message):
73         """ This function needs to be fast at the same time aware of the possibility of
74         ZMQ pipes overflowing.
75 
76         The timeout increases slowly if contention is detected on ZMQ pipes.
77         We could set copy=False and get slightly better latency but this results
78         in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.
79         This issue can be magnified if each the serialized buffer itself is larger.
80         """
81         timeout_ms = 0
82         while True:
83             socks = dict(self.poller.poll(timeout=timeout_ms))
84             if self.zmq_socket in socks and socks[self.zmq_socket] == zmq.POLLOUT:
85                 # The copy option adds latency but reduces the risk of ZMQ overflow
86                 self.zmq_socket.send_pyobj(message, copy=True)
87                 return
88             else:
89                 timeout_ms += 1
90                 logger.debug("Not sending due to full zmq pipe, timeout: {} ms".format(timeout_ms))
91 
92     def close(self):
93         self.zmq_socket.close()
94         self.context.term()
95 
96 
97 class ResultsIncoming(object):
98     """ Incoming results queue from the Interchange to the executor
99     """
100 
101     def __init__(self, ip_address, port_range):
102         """
103         Parameters
104         ----------
105 
106         ip_address: str
107            IP address of the client (where Parsl runs)
108         port_range: tuple(int, int)
109            Port range for the comms between client and interchange
110 
111         """
112         self.context = zmq.Context()
113         self.results_receiver = self.context.socket(zmq.DEALER)
114         self.results_receiver.set_hwm(0)
115         self.port = self.results_receiver.bind_to_random_port("tcp://{}".format(ip_address),
116                                                               min_port=port_range[0],
117                                                               max_port=port_range[1])
118 
119     def get(self, block=True, timeout=None):
120         return self.results_receiver.recv_multipart()
121 
122     def request_close(self):
123         status = self.results_receiver.send(pickle.dumps(None))
124         time.sleep(0.1)
125         return status
126 
127     def close(self):
128         self.results_receiver.close()
129         self.context.term()
130
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/executors/high_throughput/zmq_pipes.py b/parsl/executors/high_throughput/zmq_pipes.py
--- a/parsl/executors/high_throughput/zmq_pipes.py
+++ b/parsl/executors/high_throughput/zmq_pipes.py
@@ -4,6 +4,7 @@
 import time
 import pickle
 import logging
+import threading
 
 logger = logging.getLogger(__name__)
 
@@ -23,12 +24,27 @@
 
         """
         self.context = zmq.Context()
-        self.zmq_socket = self.context.socket(zmq.REQ)
-        self.port = self.zmq_socket.bind_to_random_port("tcp://{}".format(ip_address),
-                                                        min_port=port_range[0],
-                                                        max_port=port_range[1])
+        self.ip_address = ip_address
+        self.port_range = port_range
+        self.port = None
+        self.create_socket_and_bind()
+        self._lock = threading.Lock()
 
-    def run(self, message):
+    def create_socket_and_bind(self):
+        """ Creates socket and binds to a port.
+
+        Upon recreating the socket, we bind to the same port.
+        """
+        self.zmq_socket = self.context.socket(zmq.REQ)
+        self.zmq_socket.setsockopt(zmq.LINGER, 0)
+        if self.port is None:
+            self.port = self.zmq_socket.bind_to_random_port("tcp://{}".format(self.ip_address),
+                                                            min_port=self.port_range[0],
+                                                            max_port=self.port_range[1])
+        else:
+            self.zmq_socket.bind("tcp://{}:{}".format(self.ip_address, self.port))
+
+    def run(self, message, max_retries=3):
         """ This function needs to be fast at the same time aware of the possibility of
         ZMQ pipes overflowing.
 
@@ -37,8 +53,24 @@
         in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.
         This issue can be magnified if each the serialized buffer itself is larger.
         """
-        self.zmq_socket.send_pyobj(message, copy=True)
-        reply = self.zmq_socket.recv_pyobj()
+        reply = '__PARSL_ZMQ_PIPES_MAGIC__'
+        with self._lock:
+            for i in range(max_retries):
+                try:
+                    self.zmq_socket.send_pyobj(message, copy=True)
+                    reply = self.zmq_socket.recv_pyobj()
+                except zmq.ZMQError:
+                    logger.exception("Potential ZMQ REQ-REP deadlock caught")
+                    logger.info("Trying to reestablish context")
+                    self.zmq_socket.close()
+                    self.context.destroy()
+                    self.context = zmq.Context()
+                    self.create_socket_and_bind()
+
+        if reply == '__PARSL_ZMQ_PIPES_MAGIC__':
+            logger.error("Command channel run retries exhausted. Unable to run command")
+            raise Exception("Command Channel retries exhausted")
+
         return reply
 
     def close(self):

{"golden_diff": "diff --git a/parsl/executors/high_throughput/zmq_pipes.py b/parsl/executors/high_throughput/zmq_pipes.py\n--- a/parsl/executors/high_throughput/zmq_pipes.py\n+++ b/parsl/executors/high_throughput/zmq_pipes.py\n@@ -4,6 +4,7 @@\n import time\n import pickle\n import logging\n+import threading\n \n logger = logging.getLogger(__name__)\n \n@@ -23,12 +24,27 @@\n \n         \"\"\"\n         self.context = zmq.Context()\n-        self.zmq_socket = self.context.socket(zmq.REQ)\n-        self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(ip_address),\n-                                                        min_port=port_range[0],\n-                                                        max_port=port_range[1])\n+        self.ip_address = ip_address\n+        self.port_range = port_range\n+        self.port = None\n+        self.create_socket_and_bind()\n+        self._lock = threading.Lock()\n \n-    def run(self, message):\n+    def create_socket_and_bind(self):\n+        \"\"\" Creates socket and binds to a port.\n+\n+        Upon recreating the socket, we bind to the same port.\n+        \"\"\"\n+        self.zmq_socket = self.context.socket(zmq.REQ)\n+        self.zmq_socket.setsockopt(zmq.LINGER, 0)\n+        if self.port is None:\n+            self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(self.ip_address),\n+                                                            min_port=self.port_range[0],\n+                                                            max_port=self.port_range[1])\n+        else:\n+            self.zmq_socket.bind(\"tcp://{}:{}\".format(self.ip_address, self.port))\n+\n+    def run(self, message, max_retries=3):\n         \"\"\" This function needs to be fast at the same time aware of the possibility of\n         ZMQ pipes overflowing.\n \n@@ -37,8 +53,24 @@\n         in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.\n         This issue can be magnified if each the serialized buffer itself is larger.\n         \"\"\"\n-        self.zmq_socket.send_pyobj(message, copy=True)\n-        reply = self.zmq_socket.recv_pyobj()\n+        reply = '__PARSL_ZMQ_PIPES_MAGIC__'\n+        with self._lock:\n+            for i in range(max_retries):\n+                try:\n+                    self.zmq_socket.send_pyobj(message, copy=True)\n+                    reply = self.zmq_socket.recv_pyobj()\n+                except zmq.ZMQError:\n+                    logger.exception(\"Potential ZMQ REQ-REP deadlock caught\")\n+                    logger.info(\"Trying to reestablish context\")\n+                    self.zmq_socket.close()\n+                    self.context.destroy()\n+                    self.context = zmq.Context()\n+                    self.create_socket_and_bind()\n+\n+        if reply == '__PARSL_ZMQ_PIPES_MAGIC__':\n+            logger.error(\"Command channel run retries exhausted. Unable to run command\")\n+            raise Exception(\"Command Channel retries exhausted\")\n+\n         return reply\n \n     def close(self):\n", "issue": "\"ZMQError: Operation cannot be accomplished in current state\" triggers shutdown\n## Brief description and request\r\nI've been seeing an error crop up in longer running Parsl DFK's which trigger a graceful, but unwanted shutdown. I can't find a reliable/minimal example to cause this trigger other than \"run a while.\" So, I'm going to post as much debug info I can about our setup and am seeking advice on things to test or other information which would be helpful in debugging this error.\r\n\r\nI'm happy to test whatever is needed, I can also make direct code modifications \r\n\r\n## Backtrace of error thrown\r\n The final error is in the title but here is the backtrace, it reliably fails on this line:\r\n\r\n```\r\n... (there are higher functions in this stack from our program)\r\n  File \"/home/lnaden/github/qcfractal/qcfractal/queue/parsl_adapter.py\", line 94, in count_running_workers\r\n    if hasattr(executor, 'connected_workers'):\r\n  File \"/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/parsl/executors/high_throughput/executor.py\", line 463, in connected_workers\r\n    workers = self.command_client.run(\"WORKERS\")\r\n  File \"/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/parsl/executors/high_throughput/zmq_pipes.py\", line 40, in run\r\n    self.zmq_socket.send_pyobj(message, copy=True)\r\n  File \"/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/zmq/sugar/socket.py\", line 606, in send_pyobj\r\n    return self.send(msg, flags=flags, **kwargs)\r\n  File \"/home/lnaden/miniconda3/envs/qca/lib/python3.7/site-packages/zmq/sugar/socket.py\", line 395, in send\r\n    return super(Socket, self).send(data, flags=flags, copy=copy, track=track)\r\n  File \"zmq/backend/cython/socket.pyx\", line 725, in zmq.backend.cython.socket.Socket.send\r\n  File \"zmq/backend/cython/socket.pyx\", line 772, in zmq.backend.cython.socket.Socket.send\r\n  File \"zmq/backend/cython/socket.pyx\", line 247, in zmq.backend.cython.socket._send_copy\r\n  File \"zmq/backend/cython/socket.pyx\", line 242, in zmq.backend.cython.socket._send_copy\r\n  File \"zmq/backend/cython/checkrc.pxd\", line 25, in zmq.backend.cython.checkrc._check_rc\r\nzmq.error.ZMQError: Operation cannot be accomplished in current state\r\n```\r\n\r\nBelow this it behaves as though a standard shutdown signal and the next log item is `[I 190716 15:38:08 dflow:821] DFK cleanup initiated`\r\n\r\n## Additional info\r\n\r\n`if hasattr(executor, 'connected_workers'):` is inside a loop over the values: `parsl.dataflow.dflow.DataFlowKernel.executors.values()` \r\n\r\nNear as I can tell, this error is only thrown and logged at the higher level programs which are running Parsl and this does not appear in any of the logs or files from the `runinfo` folder. \r\n\r\nLeading up to the crash, there are multiple \"loss of Manager\" errors. The first set of lost managers happened ~6 hours after the processes were started, which is expected given the wall time of the cluster would have stopped them. However, after that initial set, we started loosing managers every 5-30 minutes seemingly randomly, all due to \"too many missed heartbeats.\" This behavior is the same on 2 separate clusters (similar hardware, same SLURM scheduler). Not sure if this is related, but it's an observation.\r\n\r\nConfig file:\r\n```python\r\nConfig(\r\n    app_cache=True,\r\n    checkpoint_files=None,\r\n    checkpoint_mode=None,\r\n    checkpoint_period=None,\r\n    data_management_max_threads=10,\r\n    executors=[HighThroughputExecutor(\r\n        address='calogin1',\r\n        cores_per_worker=16,\r\n        heartbeat_period=30,\r\n        heartbeat_threshold=120,\r\n        interchange_port_range=(55000, 56000),\r\n        label='QCFractal_Parsl_Slurm_Executor',\r\n        launch_cmd='process_worker_pool.py {debug} {max_workers} -p {prefetch_capacity} -c {cores_per_worker} -m {mem_per_worker} --poll {poll_period} --task_url={task_url} --result_url={result_url} --logdir={logdir} --block_id={{block_id}} --hb_period={heartbeat_period} --hb_threshold={heartbeat_threshold} ',\r\n        managed=True,\r\n        max_workers=48,\r\n        mem_per_worker=None,\r\n        poll_period=10,\r\n        prefetch_capacity=0,\r\n        provider=SlurmProvider(\r\n            'normal_q',\r\n            channel=LocalChannel(envs={}, script_dir=None, userhome='/home/lnaden/qcarchive/run_cc'),\r\n            cmd_timeout=10,\r\n            exclusive=True,\r\n            init_blocks=0,\r\n            launcher=SingleNodeLauncher(),\r\n            max_blocks=24,\r\n            min_blocks=0,\r\n            move_files=True,\r\n            nodes_per_block=1,\r\n            parallelism=1,\r\n            scheduler_options='#SBATCH --exclusive\\n#SBATCH -A themolssi\\n',\r\n            walltime='06:00:00',\r\n            worker_init='source /home/lnaden/qcarchive/boot_qcarchive_env.sh'\r\n        ),\r\n        storage_access=[],\r\n        suppress_failure=False,\r\n        worker_debug=False,\r\n        worker_logdir_root=None,\r\n        worker_port_range=(54000, 55000),\r\n        worker_ports=None,\r\n        working_dir=None\r\n    )],\r\n    lazy_errors=True,\r\n    monitoring=None,\r\n    retries=0,\r\n    run_dir='runinfo',\r\n    strategy='simple',\r\n    usage_tracking=False\r\n)\r\n\r\n```\r\n\r\nSetup info: SLURM Cluster, executing locally on the head node in a background process. Directly calling and manipulating the `DataFlowKernel` after providing it a `Config` \n", "before_files": [{"content": "#!/usr/bin/env python3\n\nimport zmq\nimport time\nimport pickle\nimport logging\n\nlogger = logging.getLogger(__name__)\n\n\nclass CommandClient(object):\n    \"\"\" CommandClient\n    \"\"\"\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.zmq_socket = self.context.socket(zmq.REQ)\n        self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(ip_address),\n                                                        min_port=port_range[0],\n                                                        max_port=port_range[1])\n\n    def run(self, message):\n        \"\"\" This function needs to be fast at the same time aware of the possibility of\n        ZMQ pipes overflowing.\n\n        The timeout increases slowly if contention is detected on ZMQ pipes.\n        We could set copy=False and get slightly better latency but this results\n        in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.\n        This issue can be magnified if each the serialized buffer itself is larger.\n        \"\"\"\n        self.zmq_socket.send_pyobj(message, copy=True)\n        reply = self.zmq_socket.recv_pyobj()\n        return reply\n\n    def close(self):\n        self.zmq_socket.close()\n        self.context.term()\n\n\nclass TasksOutgoing(object):\n    \"\"\" Outgoing task queue from the executor to the Interchange\n    \"\"\"\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.zmq_socket = self.context.socket(zmq.DEALER)\n        self.zmq_socket.set_hwm(0)\n        self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(ip_address),\n                                                        min_port=port_range[0],\n                                                        max_port=port_range[1])\n        self.poller = zmq.Poller()\n        self.poller.register(self.zmq_socket, zmq.POLLOUT)\n\n    def put(self, message):\n        \"\"\" This function needs to be fast at the same time aware of the possibility of\n        ZMQ pipes overflowing.\n\n        The timeout increases slowly if contention is detected on ZMQ pipes.\n        We could set copy=False and get slightly better latency but this results\n        in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.\n        This issue can be magnified if each the serialized buffer itself is larger.\n        \"\"\"\n        timeout_ms = 0\n        while True:\n            socks = dict(self.poller.poll(timeout=timeout_ms))\n            if self.zmq_socket in socks and socks[self.zmq_socket] == zmq.POLLOUT:\n                # The copy option adds latency but reduces the risk of ZMQ overflow\n                self.zmq_socket.send_pyobj(message, copy=True)\n                return\n            else:\n                timeout_ms += 1\n                logger.debug(\"Not sending due to full zmq pipe, timeout: {} ms\".format(timeout_ms))\n\n    def close(self):\n        self.zmq_socket.close()\n        self.context.term()\n\n\nclass ResultsIncoming(object):\n    \"\"\" Incoming results queue from the Interchange to the executor\n    \"\"\"\n\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.results_receiver = self.context.socket(zmq.DEALER)\n        self.results_receiver.set_hwm(0)\n        self.port = self.results_receiver.bind_to_random_port(\"tcp://{}\".format(ip_address),\n                                                              min_port=port_range[0],\n                                                              max_port=port_range[1])\n\n    def get(self, block=True, timeout=None):\n        return self.results_receiver.recv_multipart()\n\n    def request_close(self):\n        status = self.results_receiver.send(pickle.dumps(None))\n        time.sleep(0.1)\n        return status\n\n    def close(self):\n        self.results_receiver.close()\n        self.context.term()\n", "path": "parsl/executors/high_throughput/zmq_pipes.py"}], "after_files": [{"content": "#!/usr/bin/env python3\n\nimport zmq\nimport time\nimport pickle\nimport logging\nimport threading\n\nlogger = logging.getLogger(__name__)\n\n\nclass CommandClient(object):\n    \"\"\" CommandClient\n    \"\"\"\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.ip_address = ip_address\n        self.port_range = port_range\n        self.port = None\n        self.create_socket_and_bind()\n        self._lock = threading.Lock()\n\n    def create_socket_and_bind(self):\n        \"\"\" Creates socket and binds to a port.\n\n        Upon recreating the socket, we bind to the same port.\n        \"\"\"\n        self.zmq_socket = self.context.socket(zmq.REQ)\n        self.zmq_socket.setsockopt(zmq.LINGER, 0)\n        if self.port is None:\n            self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(self.ip_address),\n                                                            min_port=self.port_range[0],\n                                                            max_port=self.port_range[1])\n        else:\n            self.zmq_socket.bind(\"tcp://{}:{}\".format(self.ip_address, self.port))\n\n    def run(self, message, max_retries=3):\n        \"\"\" This function needs to be fast at the same time aware of the possibility of\n        ZMQ pipes overflowing.\n\n        The timeout increases slowly if contention is detected on ZMQ pipes.\n        We could set copy=False and get slightly better latency but this results\n        in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.\n        This issue can be magnified if each the serialized buffer itself is larger.\n        \"\"\"\n        reply = '__PARSL_ZMQ_PIPES_MAGIC__'\n        with self._lock:\n            for i in range(max_retries):\n                try:\n                    self.zmq_socket.send_pyobj(message, copy=True)\n                    reply = self.zmq_socket.recv_pyobj()\n                except zmq.ZMQError:\n                    logger.exception(\"Potential ZMQ REQ-REP deadlock caught\")\n                    logger.info(\"Trying to reestablish context\")\n                    self.zmq_socket.close()\n                    self.context.destroy()\n                    self.context = zmq.Context()\n                    self.create_socket_and_bind()\n\n        if reply == '__PARSL_ZMQ_PIPES_MAGIC__':\n            logger.error(\"Command channel run retries exhausted. Unable to run command\")\n            raise Exception(\"Command Channel retries exhausted\")\n\n        return reply\n\n    def close(self):\n        self.zmq_socket.close()\n        self.context.term()\n\n\nclass TasksOutgoing(object):\n    \"\"\" Outgoing task queue from the executor to the Interchange\n    \"\"\"\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.zmq_socket = self.context.socket(zmq.DEALER)\n        self.zmq_socket.set_hwm(0)\n        self.port = self.zmq_socket.bind_to_random_port(\"tcp://{}\".format(ip_address),\n                                                        min_port=port_range[0],\n                                                        max_port=port_range[1])\n        self.poller = zmq.Poller()\n        self.poller.register(self.zmq_socket, zmq.POLLOUT)\n\n    def put(self, message):\n        \"\"\" This function needs to be fast at the same time aware of the possibility of\n        ZMQ pipes overflowing.\n\n        The timeout increases slowly if contention is detected on ZMQ pipes.\n        We could set copy=False and get slightly better latency but this results\n        in ZMQ sockets reaching a broken state once there are ~10k tasks in flight.\n        This issue can be magnified if each the serialized buffer itself is larger.\n        \"\"\"\n        timeout_ms = 0\n        while True:\n            socks = dict(self.poller.poll(timeout=timeout_ms))\n            if self.zmq_socket in socks and socks[self.zmq_socket] == zmq.POLLOUT:\n                # The copy option adds latency but reduces the risk of ZMQ overflow\n                self.zmq_socket.send_pyobj(message, copy=True)\n                return\n            else:\n                timeout_ms += 1\n                logger.debug(\"Not sending due to full zmq pipe, timeout: {} ms\".format(timeout_ms))\n\n    def close(self):\n        self.zmq_socket.close()\n        self.context.term()\n\n\nclass ResultsIncoming(object):\n    \"\"\" Incoming results queue from the Interchange to the executor\n    \"\"\"\n\n    def __init__(self, ip_address, port_range):\n        \"\"\"\n        Parameters\n        ----------\n\n        ip_address: str\n           IP address of the client (where Parsl runs)\n        port_range: tuple(int, int)\n           Port range for the comms between client and interchange\n\n        \"\"\"\n        self.context = zmq.Context()\n        self.results_receiver = self.context.socket(zmq.DEALER)\n        self.results_receiver.set_hwm(0)\n        self.port = self.results_receiver.bind_to_random_port(\"tcp://{}\".format(ip_address),\n                                                              min_port=port_range[0],\n                                                              max_port=port_range[1])\n\n    def get(self, block=True, timeout=None):\n        return self.results_receiver.recv_multipart()\n\n    def request_close(self):\n        status = self.results_receiver.send(pickle.dumps(None))\n        time.sleep(0.1)\n        return status\n\n    def close(self):\n        self.results_receiver.close()\n        self.context.term()\n", "path": "parsl/executors/high_throughput/zmq_pipes.py"}]}