We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Newly frequent WorkQueueTaskFailure in CI
**Describe the bug**

I'm seeing this WorkQueueExecutor heisenbug happen in CI a lot recently: I'm not clear what has changed to make it happen more - for example in https://github.com/Parsl/parsl/actions/runs/6518865549/job/17704749713

```
ERROR    parsl.dataflow.dflow:dflow.py:350 Task 207 failed after 0 retry attempts
Traceback (most recent call last):
  File "/home/runner/work/parsl/parsl/parsl/dataflow/dflow.py", line 301, in handle_exec_update
    res = self._unwrap_remote_exception_wrapper(future)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/runner/work/parsl/parsl/parsl/dataflow/dflow.py", line 571, in _unwrap_remote_exception_wrapper
    result = future.result()
             ^^^^^^^^^^^^^^^
  File "/opt/hostedtoolcache/Python/3.11.5/x64/lib/python3.11/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/opt/hostedtoolcache/Python/3.11.5/x64/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
parsl.executors.workqueue.errors.WorkQueueTaskFailure: ('work queue result: The result file was not transfered from the worker.\nThis usually means that there is a problem with the python setup,\nor the wrapper that executes the function.\nTrace:\n', FileNotFoundError(2, 'No such file or directory'))
INFO     parsl.dataflow.dflow:dflow.py:1390 Standard output for task 207 available at std.out
```

I'm don't have any immediate strong ideas about what is going on - I've had a little poke but can't see anything that sticks out right away.

I've opened:
* PR #2912 to try a newer cctools
* draft PR #2910 to try to capture more FileNotFoundError information in output - there is more stuff in that FileNotFoundError (such as the actual filename) that isn't rendered by the above error reporting

I haven't been successful in recreating this on my laptop. However I have seen a related error on perlmutter under certain high load / high concurrency conditions which is a bit more recreatable and maybe I can debug from there.

cc @dthain

--- 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/workqueue/parsl_coprocess.py`
Content:
```
1 #! /usr/bin/env python3
2 
3 import sys
4 from parsl.app.errors import RemoteExceptionWrapper
5 
6 import socket
7 import json
8 import os
9 import sys
10 import threading
11 import queue
12 def remote_execute(func):
13     def remote_wrapper(event, q=None):
14         if q:
15             event = json.loads(event)
16         kwargs = event["fn_kwargs"]
17         args = event["fn_args"]
18         try:
19             response = {
20                 "Result": func(*args, **kwargs),
21                 "StatusCode": 200
22             }
23         except Exception as e:
24             response = { 
25                 "Result": str(e),
26                 "StatusCode": 500 
27             }
28         if not q:
29             return response
30         q.put(response)
31     return remote_wrapper
32     
33 read, write = os.pipe() 
34 def send_configuration(config):
35     config_string = json.dumps(config)
36     config_cmd = f"{len(config_string) + 1}\n{config_string}\n"
37     sys.stdout.write(config_cmd)
38     sys.stdout.flush()
39 def main():
40     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
41     try:
42         # modify the port argument to be 0 to listen on an arbitrary port
43         s.bind(('localhost', 0))
44     except Exception as e:
45         s.close()
46         print(e)
47         exit(1)
48     # information to print to stdout for worker
49     config = {
50             "name": name(),
51             "port": s.getsockname()[1],
52             }
53     send_configuration(config)
54     while True:
55         s.listen()
56         conn, addr = s.accept()
57         print('Network function: connection from {}'.format(addr), file=sys.stderr)
58         while True:
59             # peek at message to find newline to get the size
60             event_size = None
61             line = conn.recv(100, socket.MSG_PEEK)
62             eol = line.find(b'\n')
63             if eol >= 0:
64                 size = eol+1
65                 # actually read the size of the event
66                 input_spec = conn.recv(size).decode('utf-8').split()
67                 function_name = input_spec[0]
68                 task_id = int(input_spec[1])
69                 event_size = int(input_spec[2])
70             try:
71                 if event_size:
72                     # receive the bytes containing the event and turn it into a string
73                     event_str = conn.recv(event_size).decode("utf-8")
74                     # turn the event into a python dictionary
75                     event = json.loads(event_str)
76                     # see if the user specified an execution method
77                     exec_method = event.get("remote_task_exec_method", None)
78                     print('Network function: recieved event: {}'.format(event), file=sys.stderr)
79                     os.chdir(f"t.{task_id}")
80                     if exec_method == "thread":
81                         # create a forked process for function handler
82                         q = queue.Queue()
83                         p = threading.Thread(target=globals()[function_name], args=(event_str, q))
84                         p.start()
85                         p.join()
86                         response = json.dumps(q.get()).encode("utf-8")
87                     elif exec_method == "direct":
88                         response = json.dumps(globals()[function_name](event)).encode("utf-8")
89                     else:
90                         p = os.fork()
91                         if p == 0:
92                             response =globals()[function_name](event)
93                             os.write(write, json.dumps(response).encode("utf-8"))
94                             os._exit(0)
95                         elif p < 0:
96                             print('Network function: unable to fork', file=sys.stderr)
97                             response = { 
98                                 "Result": "unable to fork",
99                                 "StatusCode": 500 
100                             }
101                         else:
102                             chunk = os.read(read, 65536).decode("utf-8")
103                             all_chunks = [chunk]
104                             while (len(chunk) >= 65536):
105                                 chunk = os.read(read, 65536).decode("utf-8")
106                                 all_chunks.append(chunk)
107                             response = "".join(all_chunks).encode("utf-8")
108                             os.waitid(os.P_PID, p, os.WEXITED)
109                     response_size = len(response)
110                     size_msg = "{}\n".format(response_size)
111                     # send the size of response
112                     conn.sendall(size_msg.encode('utf-8'))
113                     # send response
114                     conn.sendall(response)
115                     os.chdir("..")
116                     break
117             except Exception as e:
118                 print("Network function encountered exception ", str(e), file=sys.stderr)
119     return 0
120 def name():
121     return 'parsl_coprocess'
122 @remote_execute
123 def run_parsl_task(a, b, c):
124     import parsl.executors.workqueue.exec_parsl_function as epf
125     try:
126         map_file, function_file, result_file = (a, b, c)
127         try:
128             namespace, function_code, result_name = epf.load_function(map_file, function_file)
129         except Exception:
130             raise
131         try:
132             result = epf.execute_function(namespace, function_code, result_name)
133         except Exception:
134             raise
135     except Exception:
136         result = RemoteExceptionWrapper(*sys.exc_info())
137     epf.dump_result_to_file(result_file, result)
138     return None
139 if __name__ == "__main__":
140 	main()
141 
142
```
--- 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/workqueue/parsl_coprocess.py b/parsl/executors/workqueue/parsl_coprocess.py
--- a/parsl/executors/workqueue/parsl_coprocess.py
+++ b/parsl/executors/workqueue/parsl_coprocess.py
@@ -51,6 +51,7 @@
             "port": s.getsockname()[1],
             }
     send_configuration(config)
+    abs_working_dir = os.getcwd()
     while True:
         s.listen()
         conn, addr = s.accept()
@@ -76,7 +77,7 @@
                     # see if the user specified an execution method
                     exec_method = event.get("remote_task_exec_method", None)
                     print('Network function: recieved event: {}'.format(event), file=sys.stderr)
-                    os.chdir(f"t.{task_id}")
+                    os.chdir(os.path.join(abs_working_dir, f't.{task_id}'))
                     if exec_method == "thread":
                         # create a forked process for function handler
                         q = queue.Queue()
@@ -112,10 +113,22 @@
                     conn.sendall(size_msg.encode('utf-8'))
                     # send response
                     conn.sendall(response)
-                    os.chdir("..")
                     break
             except Exception as e:
                 print("Network function encountered exception ", str(e), file=sys.stderr)
+                response = {
+                    'Result': f'network function encountered exception {e}',
+                    'Status Code': 500
+                }
+                response = json.dumps(response).encode('utf-8')
+                response_size = len(response)
+                size_msg = "{}\n".format(response_size)
+                # send the size of response
+                conn.sendall(size_msg.encode('utf-8'))
+                # send response
+                conn.sendall(response)
+            finally:
+                os.chdir(abs_working_dir)
     return 0
 def name():
     return 'parsl_coprocess'

{"golden_diff": "diff --git a/parsl/executors/workqueue/parsl_coprocess.py b/parsl/executors/workqueue/parsl_coprocess.py\n--- a/parsl/executors/workqueue/parsl_coprocess.py\n+++ b/parsl/executors/workqueue/parsl_coprocess.py\n@@ -51,6 +51,7 @@\n             \"port\": s.getsockname()[1],\n             }\n     send_configuration(config)\n+    abs_working_dir = os.getcwd()\n     while True:\n         s.listen()\n         conn, addr = s.accept()\n@@ -76,7 +77,7 @@\n                     # see if the user specified an execution method\n                     exec_method = event.get(\"remote_task_exec_method\", None)\n                     print('Network function: recieved event: {}'.format(event), file=sys.stderr)\n-                    os.chdir(f\"t.{task_id}\")\n+                    os.chdir(os.path.join(abs_working_dir, f't.{task_id}'))\n                     if exec_method == \"thread\":\n                         # create a forked process for function handler\n                         q = queue.Queue()\n@@ -112,10 +113,22 @@\n                     conn.sendall(size_msg.encode('utf-8'))\n                     # send response\n                     conn.sendall(response)\n-                    os.chdir(\"..\")\n                     break\n             except Exception as e:\n                 print(\"Network function encountered exception \", str(e), file=sys.stderr)\n+                response = {\n+                    'Result': f'network function encountered exception {e}',\n+                    'Status Code': 500\n+                }\n+                response = json.dumps(response).encode('utf-8')\n+                response_size = len(response)\n+                size_msg = \"{}\\n\".format(response_size)\n+                # send the size of response\n+                conn.sendall(size_msg.encode('utf-8'))\n+                # send response\n+                conn.sendall(response)\n+            finally:\n+                os.chdir(abs_working_dir)\n     return 0\n def name():\n     return 'parsl_coprocess'\n", "issue": "Newly frequent WorkQueueTaskFailure in CI\n**Describe the bug**\r\n\r\nI'm seeing this WorkQueueExecutor heisenbug happen in CI a lot recently: I'm not clear what has changed to make it happen more - for example in https://github.com/Parsl/parsl/actions/runs/6518865549/job/17704749713\r\n\r\n```\r\nERROR    parsl.dataflow.dflow:dflow.py:350 Task 207 failed after 0 retry attempts\r\nTraceback (most recent call last):\r\n  File \"/home/runner/work/parsl/parsl/parsl/dataflow/dflow.py\", line 301, in handle_exec_update\r\n    res = self._unwrap_remote_exception_wrapper(future)\r\n          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/home/runner/work/parsl/parsl/parsl/dataflow/dflow.py\", line 571, in _unwrap_remote_exception_wrapper\r\n    result = future.result()\r\n             ^^^^^^^^^^^^^^^\r\n  File \"/opt/hostedtoolcache/Python/3.11.5/x64/lib/python3.11/concurrent/futures/_base.py\", line 449, in result\r\n    return self.__get_result()\r\n           ^^^^^^^^^^^^^^^^^^^\r\n  File \"/opt/hostedtoolcache/Python/3.11.5/x64/lib/python3.11/concurrent/futures/_base.py\", line 401, in __get_result\r\n    raise self._exception\r\nparsl.executors.workqueue.errors.WorkQueueTaskFailure: ('work queue result: The result file was not transfered from the worker.\\nThis usually means that there is a problem with the python setup,\\nor the wrapper that executes the function.\\nTrace:\\n', FileNotFoundError(2, 'No such file or directory'))\r\nINFO     parsl.dataflow.dflow:dflow.py:1390 Standard output for task 207 available at std.out\r\n```\r\n\r\nI'm don't have any immediate strong ideas about what is going on - I've had a little poke but can't see anything that sticks out right away.\r\n\r\nI've opened:\r\n* PR #2912 to try a newer cctools\r\n* draft PR #2910 to try to capture more FileNotFoundError information in output - there is more stuff in that FileNotFoundError (such as the actual filename) that isn't rendered by the above error reporting\r\n\r\nI haven't been successful in recreating this on my laptop. However I have seen a related error on perlmutter under certain high load / high concurrency conditions which is a bit more recreatable and maybe I can debug from there.\r\n\r\ncc @dthain\n", "before_files": [{"content": "#! /usr/bin/env python3\n\nimport sys\nfrom parsl.app.errors import RemoteExceptionWrapper\n\nimport socket\nimport json\nimport os\nimport sys\nimport threading\nimport queue\ndef remote_execute(func):\n    def remote_wrapper(event, q=None):\n        if q:\n            event = json.loads(event)\n        kwargs = event[\"fn_kwargs\"]\n        args = event[\"fn_args\"]\n        try:\n            response = {\n                \"Result\": func(*args, **kwargs),\n                \"StatusCode\": 200\n            }\n        except Exception as e:\n            response = { \n                \"Result\": str(e),\n                \"StatusCode\": 500 \n            }\n        if not q:\n            return response\n        q.put(response)\n    return remote_wrapper\n    \nread, write = os.pipe() \ndef send_configuration(config):\n    config_string = json.dumps(config)\n    config_cmd = f\"{len(config_string) + 1}\\n{config_string}\\n\"\n    sys.stdout.write(config_cmd)\n    sys.stdout.flush()\ndef main():\n    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n    try:\n        # modify the port argument to be 0 to listen on an arbitrary port\n        s.bind(('localhost', 0))\n    except Exception as e:\n        s.close()\n        print(e)\n        exit(1)\n    # information to print to stdout for worker\n    config = {\n            \"name\": name(),\n            \"port\": s.getsockname()[1],\n            }\n    send_configuration(config)\n    while True:\n        s.listen()\n        conn, addr = s.accept()\n        print('Network function: connection from {}'.format(addr), file=sys.stderr)\n        while True:\n            # peek at message to find newline to get the size\n            event_size = None\n            line = conn.recv(100, socket.MSG_PEEK)\n            eol = line.find(b'\\n')\n            if eol >= 0:\n                size = eol+1\n                # actually read the size of the event\n                input_spec = conn.recv(size).decode('utf-8').split()\n                function_name = input_spec[0]\n                task_id = int(input_spec[1])\n                event_size = int(input_spec[2])\n            try:\n                if event_size:\n                    # receive the bytes containing the event and turn it into a string\n                    event_str = conn.recv(event_size).decode(\"utf-8\")\n                    # turn the event into a python dictionary\n                    event = json.loads(event_str)\n                    # see if the user specified an execution method\n                    exec_method = event.get(\"remote_task_exec_method\", None)\n                    print('Network function: recieved event: {}'.format(event), file=sys.stderr)\n                    os.chdir(f\"t.{task_id}\")\n                    if exec_method == \"thread\":\n                        # create a forked process for function handler\n                        q = queue.Queue()\n                        p = threading.Thread(target=globals()[function_name], args=(event_str, q))\n                        p.start()\n                        p.join()\n                        response = json.dumps(q.get()).encode(\"utf-8\")\n                    elif exec_method == \"direct\":\n                        response = json.dumps(globals()[function_name](event)).encode(\"utf-8\")\n                    else:\n                        p = os.fork()\n                        if p == 0:\n                            response =globals()[function_name](event)\n                            os.write(write, json.dumps(response).encode(\"utf-8\"))\n                            os._exit(0)\n                        elif p < 0:\n                            print('Network function: unable to fork', file=sys.stderr)\n                            response = { \n                                \"Result\": \"unable to fork\",\n                                \"StatusCode\": 500 \n                            }\n                        else:\n                            chunk = os.read(read, 65536).decode(\"utf-8\")\n                            all_chunks = [chunk]\n                            while (len(chunk) >= 65536):\n                                chunk = os.read(read, 65536).decode(\"utf-8\")\n                                all_chunks.append(chunk)\n                            response = \"\".join(all_chunks).encode(\"utf-8\")\n                            os.waitid(os.P_PID, p, os.WEXITED)\n                    response_size = len(response)\n                    size_msg = \"{}\\n\".format(response_size)\n                    # send the size of response\n                    conn.sendall(size_msg.encode('utf-8'))\n                    # send response\n                    conn.sendall(response)\n                    os.chdir(\"..\")\n                    break\n            except Exception as e:\n                print(\"Network function encountered exception \", str(e), file=sys.stderr)\n    return 0\ndef name():\n    return 'parsl_coprocess'\n@remote_execute\ndef run_parsl_task(a, b, c):\n    import parsl.executors.workqueue.exec_parsl_function as epf\n    try:\n        map_file, function_file, result_file = (a, b, c)\n        try:\n            namespace, function_code, result_name = epf.load_function(map_file, function_file)\n        except Exception:\n            raise\n        try:\n            result = epf.execute_function(namespace, function_code, result_name)\n        except Exception:\n            raise\n    except Exception:\n        result = RemoteExceptionWrapper(*sys.exc_info())\n    epf.dump_result_to_file(result_file, result)\n    return None\nif __name__ == \"__main__\":\n\tmain()\n\n", "path": "parsl/executors/workqueue/parsl_coprocess.py"}], "after_files": [{"content": "#! /usr/bin/env python3\n\nimport sys\nfrom parsl.app.errors import RemoteExceptionWrapper\n\nimport socket\nimport json\nimport os\nimport sys\nimport threading\nimport queue\ndef remote_execute(func):\n    def remote_wrapper(event, q=None):\n        if q:\n            event = json.loads(event)\n        kwargs = event[\"fn_kwargs\"]\n        args = event[\"fn_args\"]\n        try:\n            response = {\n                \"Result\": func(*args, **kwargs),\n                \"StatusCode\": 200\n            }\n        except Exception as e:\n            response = { \n                \"Result\": str(e),\n                \"StatusCode\": 500 \n            }\n        if not q:\n            return response\n        q.put(response)\n    return remote_wrapper\n    \nread, write = os.pipe() \ndef send_configuration(config):\n    config_string = json.dumps(config)\n    config_cmd = f\"{len(config_string) + 1}\\n{config_string}\\n\"\n    sys.stdout.write(config_cmd)\n    sys.stdout.flush()\ndef main():\n    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n    try:\n        # modify the port argument to be 0 to listen on an arbitrary port\n        s.bind(('localhost', 0))\n    except Exception as e:\n        s.close()\n        print(e)\n        exit(1)\n    # information to print to stdout for worker\n    config = {\n            \"name\": name(),\n            \"port\": s.getsockname()[1],\n            }\n    send_configuration(config)\n    abs_working_dir = os.getcwd()\n    while True:\n        s.listen()\n        conn, addr = s.accept()\n        print('Network function: connection from {}'.format(addr), file=sys.stderr)\n        while True:\n            # peek at message to find newline to get the size\n            event_size = None\n            line = conn.recv(100, socket.MSG_PEEK)\n            eol = line.find(b'\\n')\n            if eol >= 0:\n                size = eol+1\n                # actually read the size of the event\n                input_spec = conn.recv(size).decode('utf-8').split()\n                function_name = input_spec[0]\n                task_id = int(input_spec[1])\n                event_size = int(input_spec[2])\n            try:\n                if event_size:\n                    # receive the bytes containing the event and turn it into a string\n                    event_str = conn.recv(event_size).decode(\"utf-8\")\n                    # turn the event into a python dictionary\n                    event = json.loads(event_str)\n                    # see if the user specified an execution method\n                    exec_method = event.get(\"remote_task_exec_method\", None)\n                    print('Network function: recieved event: {}'.format(event), file=sys.stderr)\n                    os.chdir(os.path.join(abs_working_dir, f't.{task_id}'))\n                    if exec_method == \"thread\":\n                        # create a forked process for function handler\n                        q = queue.Queue()\n                        p = threading.Thread(target=globals()[function_name], args=(event_str, q))\n                        p.start()\n                        p.join()\n                        response = json.dumps(q.get()).encode(\"utf-8\")\n                    elif exec_method == \"direct\":\n                        response = json.dumps(globals()[function_name](event)).encode(\"utf-8\")\n                    else:\n                        p = os.fork()\n                        if p == 0:\n                            response =globals()[function_name](event)\n                            os.write(write, json.dumps(response).encode(\"utf-8\"))\n                            os._exit(0)\n                        elif p < 0:\n                            print('Network function: unable to fork', file=sys.stderr)\n                            response = { \n                                \"Result\": \"unable to fork\",\n                                \"StatusCode\": 500 \n                            }\n                        else:\n                            chunk = os.read(read, 65536).decode(\"utf-8\")\n                            all_chunks = [chunk]\n                            while (len(chunk) >= 65536):\n                                chunk = os.read(read, 65536).decode(\"utf-8\")\n                                all_chunks.append(chunk)\n                            response = \"\".join(all_chunks).encode(\"utf-8\")\n                            os.waitid(os.P_PID, p, os.WEXITED)\n                    response_size = len(response)\n                    size_msg = \"{}\\n\".format(response_size)\n                    # send the size of response\n                    conn.sendall(size_msg.encode('utf-8'))\n                    # send response\n                    conn.sendall(response)\n                    break\n            except Exception as e:\n                print(\"Network function encountered exception \", str(e), file=sys.stderr)\n                response = {\n                    'Result': f'network function encountered exception {e}',\n                    'Status Code': 500\n                }\n                response = json.dumps(response).encode('utf-8')\n                response_size = len(response)\n                size_msg = \"{}\\n\".format(response_size)\n                # send the size of response\n                conn.sendall(size_msg.encode('utf-8'))\n                # send response\n                conn.sendall(response)\n            finally:\n                os.chdir(abs_working_dir)\n    return 0\ndef name():\n    return 'parsl_coprocess'\n@remote_execute\ndef run_parsl_task(a, b, c):\n    import parsl.executors.workqueue.exec_parsl_function as epf\n    try:\n        map_file, function_file, result_file = (a, b, c)\n        try:\n            namespace, function_code, result_name = epf.load_function(map_file, function_file)\n        except Exception:\n            raise\n        try:\n            result = epf.execute_function(namespace, function_code, result_name)\n        except Exception:\n            raise\n    except Exception:\n        result = RemoteExceptionWrapper(*sys.exc_info())\n    epf.dump_result_to_file(result_file, result)\n    return None\nif __name__ == \"__main__\":\n\tmain()\n\n", "path": "parsl/executors/workqueue/parsl_coprocess.py"}]}