We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
readFileStream deadlocks when not read through completely
I think I was aware of this at one point, but using readFileStream without reading all the way to EOF will totally deadlock the worker process on anything but the file jobStore. This is pretty unintuitive (surely this is one of the only use cases for streaming the file?). It may be especially confusing to new users who aren't used to toil deadlocks.

Unfortunately, it's not easy to fix. The writing thread will be blocked on write, so there's no way to wake it up and tell the writing thread to quit. Python threads can't (safely) be killed from their parents. The thread *should* have dropped the GIL while blocked on the write, so we might barely get away with killing its PID. But I'd suggest adjusting the ReadableThread/WritableThread classes to use processes rather than threads. They already use pipes, so very little would need to change. There would be a bit of added overhead from the extra fork(), but surely that's worth getting rid of some of these GIL nightmares.
readFileStream deadlocks when not read through completely
I think I was aware of this at one point, but using readFileStream without reading all the way to EOF will totally deadlock the worker process on anything but the file jobStore. This is pretty unintuitive (surely this is one of the only use cases for streaming the file?). It may be especially confusing to new users who aren't used to toil deadlocks.

Unfortunately, it's not easy to fix. The writing thread will be blocked on write, so there's no way to wake it up and tell the writing thread to quit. Python threads can't (safely) be killed from their parents. The thread *should* have dropped the GIL while blocked on the write, so we might barely get away with killing its PID. But I'd suggest adjusting the ReadableThread/WritableThread classes to use processes rather than threads. They already use pipes, so very little would need to change. There would be a bit of added overhead from the extra fork(), but surely that's worth getting rid of some of these GIL nightmares.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/toil/jobStores/utils.py`
Content:
```
1 import logging
2 import os
3 from abc import ABCMeta
4 from abc import abstractmethod
5 
6 from bd2k.util.threading import ExceptionalThread
7 
8 log = logging.getLogger(__name__)
9 
10 class WritablePipe(object):
11     """
12     An object-oriented wrapper for os.pipe. Clients should subclass it, implement
13     :meth:`.readFrom` to consume the readable end of the pipe, then instantiate the class as a
14     context manager to get the writable end. See the example below.
15 
16     >>> import sys, shutil
17     >>> class MyPipe(WritablePipe):
18     ...     def readFrom(self, readable):
19     ...         shutil.copyfileobj(readable, sys.stdout)
20     >>> with MyPipe() as writable:
21     ...     writable.write('Hello, world!\\n')
22     Hello, world!
23 
24     Each instance of this class creates a thread and invokes the readFrom method in that thread.
25     The thread will be join()ed upon normal exit from the context manager, i.e. the body of the
26     `with` statement. If an exception occurs, the thread will not be joined but a well-behaved
27     :meth:`.readFrom` implementation will terminate shortly thereafter due to the pipe having
28     been closed.
29 
30     Now, exceptions in the reader thread will be reraised in the main thread:
31 
32     >>> class MyPipe(WritablePipe):
33     ...     def readFrom(self, readable):
34     ...         raise RuntimeError('Hello, world!')
35     >>> with MyPipe() as writable:
36     ...     pass
37     Traceback (most recent call last):
38     ...
39     RuntimeError: Hello, world!
40 
41     More complicated, less illustrative tests:
42 
43     Same as above, but provving that handles are closed:
44 
45     >>> x = os.dup(0); os.close(x)
46     >>> class MyPipe(WritablePipe):
47     ...     def readFrom(self, readable):
48     ...         raise RuntimeError('Hello, world!')
49     >>> with MyPipe() as writable:
50     ...     pass
51     Traceback (most recent call last):
52     ...
53     RuntimeError: Hello, world!
54     >>> y = os.dup(0); os.close(y); x == y
55     True
56 
57     Exceptions in the body of the with statement aren't masked, and handles are closed:
58 
59     >>> x = os.dup(0); os.close(x)
60     >>> class MyPipe(WritablePipe):
61     ...     def readFrom(self, readable):
62     ...         pass
63     >>> with MyPipe() as writable:
64     ...     raise RuntimeError('Hello, world!')
65     Traceback (most recent call last):
66     ...
67     RuntimeError: Hello, world!
68     >>> y = os.dup(0); os.close(y); x == y
69     True
70     """
71 
72     __metaclass__ = ABCMeta
73 
74     @abstractmethod
75     def readFrom(self, readable):
76         """
77         Implement this method to read data from the pipe.
78 
79         :param file readable: the file object representing the readable end of the pipe. Do not
80         explicitly invoke the close() method of the object, that will be done automatically.
81         """
82         raise NotImplementedError()
83 
84     def _reader(self):
85         with os.fdopen(self.readable_fh, 'r') as readable:
86             # FIXME: another race here, causing a redundant attempt to close in the main thread
87             self.readable_fh = None  # signal to parent thread that we've taken over
88             self.readFrom(readable)
89 
90     def __init__(self):
91         super(WritablePipe, self).__init__()
92         self.readable_fh = None
93         self.writable = None
94         self.thread = None
95 
96     def __enter__(self):
97         self.readable_fh, writable_fh = os.pipe()
98         self.writable = os.fdopen(writable_fh, 'w')
99         self.thread = ExceptionalThread(target=self._reader)
100         self.thread.start()
101         return self.writable
102 
103     def __exit__(self, exc_type, exc_val, exc_tb):
104         try:
105             self.writable.close()
106             # Closeing the writable end will send EOF to the readable and cause the reader thread
107             # to finish.
108             if exc_type is None:
109                 if self.thread is not None:
110                     # reraises any exception that was raised in the thread
111                     self.thread.join()
112         finally:
113             # The responsibility for closing the readable end is generally that of the reader
114             # thread. To cover the small window before the reader takes over we also close it here.
115             readable_fh = self.readable_fh
116             if readable_fh is not None:
117                 # FIXME: This is still racy. The reader thread could close it now, and someone
118                 # else may immediately open a new file, reusing the file handle.
119                 os.close(readable_fh)
120 
121 
122 # FIXME: Unfortunately these two classes are almost an exact mirror image of each other.
123 # Basically, read and write are swapped. The only asymmetry lies in how shutdown is handled. I
124 # tried generalizing but the code becomes inscrutable. Until I (or someone else) has a better
125 # idea how to solve this, I think its better to have code that is readable at the expense of
126 # duplication.
127 
128 
129 class ReadablePipe(object):
130     """
131     An object-oriented wrapper for os.pipe. Clients should subclass it, implement
132     :meth:`.writeTo` to place data into the writable end of the pipe, then instantiate the class
133     as a context manager to get the writable end. See the example below.
134 
135     >>> import sys, shutil
136     >>> class MyPipe(ReadablePipe):
137     ...     def writeTo(self, writable):
138     ...         writable.write('Hello, world!\\n')
139     >>> with MyPipe() as readable:
140     ...     shutil.copyfileobj(readable, sys.stdout)
141     Hello, world!
142 
143     Each instance of this class creates a thread and invokes the :meth:`.writeTo` method in that
144     thread. The thread will be join()ed upon normal exit from the context manager, i.e. the body
145     of the `with` statement. If an exception occurs, the thread will not be joined but a
146     well-behaved :meth:`.writeTo` implementation will terminate shortly thereafter due to the
147     pipe having been closed.
148 
149     Now, exceptions in the reader thread will be reraised in the main thread:
150 
151     >>> class MyPipe(ReadablePipe):
152     ...     def writeTo(self, writable):
153     ...         raise RuntimeError('Hello, world!')
154     >>> with MyPipe() as readable:
155     ...     pass
156     Traceback (most recent call last):
157     ...
158     RuntimeError: Hello, world!
159 
160     More complicated, less illustrative tests:
161 
162     Same as above, but provving that handles are closed:
163 
164     >>> x = os.dup(0); os.close(x)
165     >>> class MyPipe(ReadablePipe):
166     ...     def writeTo(self, writable):
167     ...         raise RuntimeError('Hello, world!')
168     >>> with MyPipe() as readable:
169     ...     pass
170     Traceback (most recent call last):
171     ...
172     RuntimeError: Hello, world!
173     >>> y = os.dup(0); os.close(y); x == y
174     True
175 
176     Exceptions in the body of the with statement aren't masked, and handles are closed:
177 
178     >>> x = os.dup(0); os.close(x)
179     >>> class MyPipe(ReadablePipe):
180     ...     def writeTo(self, writable):
181     ...         pass
182     >>> with MyPipe() as readable:
183     ...     raise RuntimeError('Hello, world!')
184     Traceback (most recent call last):
185     ...
186     RuntimeError: Hello, world!
187     >>> y = os.dup(0); os.close(y); x == y
188     True
189     """
190 
191     __metaclass__ = ABCMeta
192 
193     @abstractmethod
194     def writeTo(self, writable):
195         """
196         Implement this method to read data from the pipe.
197 
198         :param file writable: the file object representing the writable end of the pipe. Do not
199         explicitly invoke the close() method of the object, that will be done automatically.
200         """
201         raise NotImplementedError()
202 
203     def _writer(self):
204         with os.fdopen(self.writable_fh, 'w') as writable:
205             # FIXME: another race here, causing a redundant attempt to close in the main thread
206             self.writable_fh = None  # signal to parent thread that we've taken over
207             self.writeTo(writable)
208 
209     def __init__(self):
210         super(ReadablePipe, self).__init__()
211         self.writable_fh = None
212         self.readable = None
213         self.thread = None
214 
215     def __enter__(self):
216         readable_fh, self.writable_fh = os.pipe()
217         self.readable = os.fdopen(readable_fh, 'r')
218         self.thread = ExceptionalThread(target=self._writer)
219         self.thread.start()
220         return self.readable
221 
222     def __exit__(self, exc_type, exc_val, exc_tb):
223         try:
224             if exc_type is None:
225                 if self.thread is not None:
226                     # reraises any exception that was raised in the thread
227                     self.thread.join()
228         finally:
229             self.readable.close()
230             # The responsibility for closing the writable end is generally that of the writer
231             # thread. To cover the small window before the writer takes over we also close it here.
232             writable_fh = self.writable_fh
233             if writable_fh is not None:
234                 # FIXME: This is still racy. The writer thread could close it now, and someone
235                 # else may immediately open a new file, reusing the file handle.
236                 os.close(writable_fh)
237
```
--- 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/src/toil/jobStores/utils.py b/src/toil/jobStores/utils.py
--- a/src/toil/jobStores/utils.py
+++ b/src/toil/jobStores/utils.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import errno
 from abc import ABCMeta
 from abc import abstractmethod
 
@@ -105,10 +106,14 @@
             self.writable.close()
             # Closeing the writable end will send EOF to the readable and cause the reader thread
             # to finish.
+            if self.thread is not None:
+                # reraises any exception that was raised in the thread
+                self.thread.join()
+        except:
             if exc_type is None:
-                if self.thread is not None:
-                    # reraises any exception that was raised in the thread
-                    self.thread.join()
+                # Only raise the child exception if there wasn't
+                # already an exception in the main thread
+                raise
         finally:
             # The responsibility for closing the readable end is generally that of the reader
             # thread. To cover the small window before the reader takes over we also close it here.
@@ -119,13 +124,6 @@
                 os.close(readable_fh)
 
 
-# FIXME: Unfortunately these two classes are almost an exact mirror image of each other.
-# Basically, read and write are swapped. The only asymmetry lies in how shutdown is handled. I
-# tried generalizing but the code becomes inscrutable. Until I (or someone else) has a better
-# idea how to solve this, I think its better to have code that is readable at the expense of
-# duplication.
-
-
 class ReadablePipe(object):
     """
     An object-oriented wrapper for os.pipe. Clients should subclass it, implement
@@ -201,10 +199,14 @@
         raise NotImplementedError()
 
     def _writer(self):
-        with os.fdopen(self.writable_fh, 'w') as writable:
-            # FIXME: another race here, causing a redundant attempt to close in the main thread
-            self.writable_fh = None  # signal to parent thread that we've taken over
-            self.writeTo(writable)
+        try:
+            with os.fdopen(self.writable_fh, 'w') as writable:
+                self.writeTo(writable)
+        except IOError as e:
+            # The other side of the pipe may have been closed by the
+            # reading thread, which is OK.
+            if e.errno != errno.EPIPE:
+                raise
 
     def __init__(self):
         super(ReadablePipe, self).__init__()
@@ -220,17 +222,16 @@
         return self.readable
 
     def __exit__(self, exc_type, exc_val, exc_tb):
+        # Close the read end of the pipe. The writing thread may
+        # still be writing to the other end, but this will wake it up
+        # if that's the case.
+        self.readable.close()
         try:
+            if self.thread is not None:
+                # reraises any exception that was raised in the thread
+                self.thread.join()
+        except:
             if exc_type is None:
-                if self.thread is not None:
-                    # reraises any exception that was raised in the thread
-                    self.thread.join()
-        finally:
-            self.readable.close()
-            # The responsibility for closing the writable end is generally that of the writer
-            # thread. To cover the small window before the writer takes over we also close it here.
-            writable_fh = self.writable_fh
-            if writable_fh is not None:
-                # FIXME: This is still racy. The writer thread could close it now, and someone
-                # else may immediately open a new file, reusing the file handle.
-                os.close(writable_fh)
+                # Only raise the child exception if there wasn't
+                # already an exception in the main thread
+                raise

{"golden_diff": "diff --git a/src/toil/jobStores/utils.py b/src/toil/jobStores/utils.py\n--- a/src/toil/jobStores/utils.py\n+++ b/src/toil/jobStores/utils.py\n@@ -1,5 +1,6 @@\n import logging\n import os\n+import errno\n from abc import ABCMeta\n from abc import abstractmethod\n \n@@ -105,10 +106,14 @@\n             self.writable.close()\n             # Closeing the writable end will send EOF to the readable and cause the reader thread\n             # to finish.\n+            if self.thread is not None:\n+                # reraises any exception that was raised in the thread\n+                self.thread.join()\n+        except:\n             if exc_type is None:\n-                if self.thread is not None:\n-                    # reraises any exception that was raised in the thread\n-                    self.thread.join()\n+                # Only raise the child exception if there wasn't\n+                # already an exception in the main thread\n+                raise\n         finally:\n             # The responsibility for closing the readable end is generally that of the reader\n             # thread. To cover the small window before the reader takes over we also close it here.\n@@ -119,13 +124,6 @@\n                 os.close(readable_fh)\n \n \n-# FIXME: Unfortunately these two classes are almost an exact mirror image of each other.\n-# Basically, read and write are swapped. The only asymmetry lies in how shutdown is handled. I\n-# tried generalizing but the code becomes inscrutable. Until I (or someone else) has a better\n-# idea how to solve this, I think its better to have code that is readable at the expense of\n-# duplication.\n-\n-\n class ReadablePipe(object):\n     \"\"\"\n     An object-oriented wrapper for os.pipe. Clients should subclass it, implement\n@@ -201,10 +199,14 @@\n         raise NotImplementedError()\n \n     def _writer(self):\n-        with os.fdopen(self.writable_fh, 'w') as writable:\n-            # FIXME: another race here, causing a redundant attempt to close in the main thread\n-            self.writable_fh = None  # signal to parent thread that we've taken over\n-            self.writeTo(writable)\n+        try:\n+            with os.fdopen(self.writable_fh, 'w') as writable:\n+                self.writeTo(writable)\n+        except IOError as e:\n+            # The other side of the pipe may have been closed by the\n+            # reading thread, which is OK.\n+            if e.errno != errno.EPIPE:\n+                raise\n \n     def __init__(self):\n         super(ReadablePipe, self).__init__()\n@@ -220,17 +222,16 @@\n         return self.readable\n \n     def __exit__(self, exc_type, exc_val, exc_tb):\n+        # Close the read end of the pipe. The writing thread may\n+        # still be writing to the other end, but this will wake it up\n+        # if that's the case.\n+        self.readable.close()\n         try:\n+            if self.thread is not None:\n+                # reraises any exception that was raised in the thread\n+                self.thread.join()\n+        except:\n             if exc_type is None:\n-                if self.thread is not None:\n-                    # reraises any exception that was raised in the thread\n-                    self.thread.join()\n-        finally:\n-            self.readable.close()\n-            # The responsibility for closing the writable end is generally that of the writer\n-            # thread. To cover the small window before the writer takes over we also close it here.\n-            writable_fh = self.writable_fh\n-            if writable_fh is not None:\n-                # FIXME: This is still racy. The writer thread could close it now, and someone\n-                # else may immediately open a new file, reusing the file handle.\n-                os.close(writable_fh)\n+                # Only raise the child exception if there wasn't\n+                # already an exception in the main thread\n+                raise\n", "issue": "readFileStream deadlocks when not read through completely\nI think I was aware of this at one point, but using readFileStream without reading all the way to EOF will totally deadlock the worker process on anything but the file jobStore. This is pretty unintuitive (surely this is one of the only use cases for streaming the file?). It may be especially confusing to new users who aren't used to toil deadlocks.\r\n\r\nUnfortunately, it's not easy to fix. The writing thread will be blocked on write, so there's no way to wake it up and tell the writing thread to quit. Python threads can't (safely) be killed from their parents. The thread *should* have dropped the GIL while blocked on the write, so we might barely get away with killing its PID. But I'd suggest adjusting the ReadableThread/WritableThread classes to use processes rather than threads. They already use pipes, so very little would need to change. There would be a bit of added overhead from the extra fork(), but surely that's worth getting rid of some of these GIL nightmares.\nreadFileStream deadlocks when not read through completely\nI think I was aware of this at one point, but using readFileStream without reading all the way to EOF will totally deadlock the worker process on anything but the file jobStore. This is pretty unintuitive (surely this is one of the only use cases for streaming the file?). It may be especially confusing to new users who aren't used to toil deadlocks.\r\n\r\nUnfortunately, it's not easy to fix. The writing thread will be blocked on write, so there's no way to wake it up and tell the writing thread to quit. Python threads can't (safely) be killed from their parents. The thread *should* have dropped the GIL while blocked on the write, so we might barely get away with killing its PID. But I'd suggest adjusting the ReadableThread/WritableThread classes to use processes rather than threads. They already use pipes, so very little would need to change. There would be a bit of added overhead from the extra fork(), but surely that's worth getting rid of some of these GIL nightmares.\n", "before_files": [{"content": "import logging\nimport os\nfrom abc import ABCMeta\nfrom abc import abstractmethod\n\nfrom bd2k.util.threading import ExceptionalThread\n\nlog = logging.getLogger(__name__)\n\nclass WritablePipe(object):\n    \"\"\"\n    An object-oriented wrapper for os.pipe. Clients should subclass it, implement\n    :meth:`.readFrom` to consume the readable end of the pipe, then instantiate the class as a\n    context manager to get the writable end. See the example below.\n\n    >>> import sys, shutil\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         shutil.copyfileobj(readable, sys.stdout)\n    >>> with MyPipe() as writable:\n    ...     writable.write('Hello, world!\\\\n')\n    Hello, world!\n\n    Each instance of this class creates a thread and invokes the readFrom method in that thread.\n    The thread will be join()ed upon normal exit from the context manager, i.e. the body of the\n    `with` statement. If an exception occurs, the thread will not be joined but a well-behaved\n    :meth:`.readFrom` implementation will terminate shortly thereafter due to the pipe having\n    been closed.\n\n    Now, exceptions in the reader thread will be reraised in the main thread:\n\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as writable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n\n    More complicated, less illustrative tests:\n\n    Same as above, but provving that handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as writable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n\n    Exceptions in the body of the with statement aren't masked, and handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         pass\n    >>> with MyPipe() as writable:\n    ...     raise RuntimeError('Hello, world!')\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n    \"\"\"\n\n    __metaclass__ = ABCMeta\n\n    @abstractmethod\n    def readFrom(self, readable):\n        \"\"\"\n        Implement this method to read data from the pipe.\n\n        :param file readable: the file object representing the readable end of the pipe. Do not\n        explicitly invoke the close() method of the object, that will be done automatically.\n        \"\"\"\n        raise NotImplementedError()\n\n    def _reader(self):\n        with os.fdopen(self.readable_fh, 'r') as readable:\n            # FIXME: another race here, causing a redundant attempt to close in the main thread\n            self.readable_fh = None  # signal to parent thread that we've taken over\n            self.readFrom(readable)\n\n    def __init__(self):\n        super(WritablePipe, self).__init__()\n        self.readable_fh = None\n        self.writable = None\n        self.thread = None\n\n    def __enter__(self):\n        self.readable_fh, writable_fh = os.pipe()\n        self.writable = os.fdopen(writable_fh, 'w')\n        self.thread = ExceptionalThread(target=self._reader)\n        self.thread.start()\n        return self.writable\n\n    def __exit__(self, exc_type, exc_val, exc_tb):\n        try:\n            self.writable.close()\n            # Closeing the writable end will send EOF to the readable and cause the reader thread\n            # to finish.\n            if exc_type is None:\n                if self.thread is not None:\n                    # reraises any exception that was raised in the thread\n                    self.thread.join()\n        finally:\n            # The responsibility for closing the readable end is generally that of the reader\n            # thread. To cover the small window before the reader takes over we also close it here.\n            readable_fh = self.readable_fh\n            if readable_fh is not None:\n                # FIXME: This is still racy. The reader thread could close it now, and someone\n                # else may immediately open a new file, reusing the file handle.\n                os.close(readable_fh)\n\n\n# FIXME: Unfortunately these two classes are almost an exact mirror image of each other.\n# Basically, read and write are swapped. The only asymmetry lies in how shutdown is handled. I\n# tried generalizing but the code becomes inscrutable. Until I (or someone else) has a better\n# idea how to solve this, I think its better to have code that is readable at the expense of\n# duplication.\n\n\nclass ReadablePipe(object):\n    \"\"\"\n    An object-oriented wrapper for os.pipe. Clients should subclass it, implement\n    :meth:`.writeTo` to place data into the writable end of the pipe, then instantiate the class\n    as a context manager to get the writable end. See the example below.\n\n    >>> import sys, shutil\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         writable.write('Hello, world!\\\\n')\n    >>> with MyPipe() as readable:\n    ...     shutil.copyfileobj(readable, sys.stdout)\n    Hello, world!\n\n    Each instance of this class creates a thread and invokes the :meth:`.writeTo` method in that\n    thread. The thread will be join()ed upon normal exit from the context manager, i.e. the body\n    of the `with` statement. If an exception occurs, the thread will not be joined but a\n    well-behaved :meth:`.writeTo` implementation will terminate shortly thereafter due to the\n    pipe having been closed.\n\n    Now, exceptions in the reader thread will be reraised in the main thread:\n\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as readable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n\n    More complicated, less illustrative tests:\n\n    Same as above, but provving that handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as readable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n\n    Exceptions in the body of the with statement aren't masked, and handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         pass\n    >>> with MyPipe() as readable:\n    ...     raise RuntimeError('Hello, world!')\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n    \"\"\"\n\n    __metaclass__ = ABCMeta\n\n    @abstractmethod\n    def writeTo(self, writable):\n        \"\"\"\n        Implement this method to read data from the pipe.\n\n        :param file writable: the file object representing the writable end of the pipe. Do not\n        explicitly invoke the close() method of the object, that will be done automatically.\n        \"\"\"\n        raise NotImplementedError()\n\n    def _writer(self):\n        with os.fdopen(self.writable_fh, 'w') as writable:\n            # FIXME: another race here, causing a redundant attempt to close in the main thread\n            self.writable_fh = None  # signal to parent thread that we've taken over\n            self.writeTo(writable)\n\n    def __init__(self):\n        super(ReadablePipe, self).__init__()\n        self.writable_fh = None\n        self.readable = None\n        self.thread = None\n\n    def __enter__(self):\n        readable_fh, self.writable_fh = os.pipe()\n        self.readable = os.fdopen(readable_fh, 'r')\n        self.thread = ExceptionalThread(target=self._writer)\n        self.thread.start()\n        return self.readable\n\n    def __exit__(self, exc_type, exc_val, exc_tb):\n        try:\n            if exc_type is None:\n                if self.thread is not None:\n                    # reraises any exception that was raised in the thread\n                    self.thread.join()\n        finally:\n            self.readable.close()\n            # The responsibility for closing the writable end is generally that of the writer\n            # thread. To cover the small window before the writer takes over we also close it here.\n            writable_fh = self.writable_fh\n            if writable_fh is not None:\n                # FIXME: This is still racy. The writer thread could close it now, and someone\n                # else may immediately open a new file, reusing the file handle.\n                os.close(writable_fh)\n", "path": "src/toil/jobStores/utils.py"}], "after_files": [{"content": "import logging\nimport os\nimport errno\nfrom abc import ABCMeta\nfrom abc import abstractmethod\n\nfrom bd2k.util.threading import ExceptionalThread\n\nlog = logging.getLogger(__name__)\n\nclass WritablePipe(object):\n    \"\"\"\n    An object-oriented wrapper for os.pipe. Clients should subclass it, implement\n    :meth:`.readFrom` to consume the readable end of the pipe, then instantiate the class as a\n    context manager to get the writable end. See the example below.\n\n    >>> import sys, shutil\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         shutil.copyfileobj(readable, sys.stdout)\n    >>> with MyPipe() as writable:\n    ...     writable.write('Hello, world!\\\\n')\n    Hello, world!\n\n    Each instance of this class creates a thread and invokes the readFrom method in that thread.\n    The thread will be join()ed upon normal exit from the context manager, i.e. the body of the\n    `with` statement. If an exception occurs, the thread will not be joined but a well-behaved\n    :meth:`.readFrom` implementation will terminate shortly thereafter due to the pipe having\n    been closed.\n\n    Now, exceptions in the reader thread will be reraised in the main thread:\n\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as writable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n\n    More complicated, less illustrative tests:\n\n    Same as above, but provving that handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as writable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n\n    Exceptions in the body of the with statement aren't masked, and handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(WritablePipe):\n    ...     def readFrom(self, readable):\n    ...         pass\n    >>> with MyPipe() as writable:\n    ...     raise RuntimeError('Hello, world!')\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n    \"\"\"\n\n    __metaclass__ = ABCMeta\n\n    @abstractmethod\n    def readFrom(self, readable):\n        \"\"\"\n        Implement this method to read data from the pipe.\n\n        :param file readable: the file object representing the readable end of the pipe. Do not\n        explicitly invoke the close() method of the object, that will be done automatically.\n        \"\"\"\n        raise NotImplementedError()\n\n    def _reader(self):\n        with os.fdopen(self.readable_fh, 'r') as readable:\n            # FIXME: another race here, causing a redundant attempt to close in the main thread\n            self.readable_fh = None  # signal to parent thread that we've taken over\n            self.readFrom(readable)\n\n    def __init__(self):\n        super(WritablePipe, self).__init__()\n        self.readable_fh = None\n        self.writable = None\n        self.thread = None\n\n    def __enter__(self):\n        self.readable_fh, writable_fh = os.pipe()\n        self.writable = os.fdopen(writable_fh, 'w')\n        self.thread = ExceptionalThread(target=self._reader)\n        self.thread.start()\n        return self.writable\n\n    def __exit__(self, exc_type, exc_val, exc_tb):\n        try:\n            self.writable.close()\n            # Closeing the writable end will send EOF to the readable and cause the reader thread\n            # to finish.\n            if self.thread is not None:\n                # reraises any exception that was raised in the thread\n                self.thread.join()\n        except:\n            if exc_type is None:\n                # Only raise the child exception if there wasn't\n                # already an exception in the main thread\n                raise\n        finally:\n            # The responsibility for closing the readable end is generally that of the reader\n            # thread. To cover the small window before the reader takes over we also close it here.\n            readable_fh = self.readable_fh\n            if readable_fh is not None:\n                # FIXME: This is still racy. The reader thread could close it now, and someone\n                # else may immediately open a new file, reusing the file handle.\n                os.close(readable_fh)\n\n\nclass ReadablePipe(object):\n    \"\"\"\n    An object-oriented wrapper for os.pipe. Clients should subclass it, implement\n    :meth:`.writeTo` to place data into the writable end of the pipe, then instantiate the class\n    as a context manager to get the writable end. See the example below.\n\n    >>> import sys, shutil\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         writable.write('Hello, world!\\\\n')\n    >>> with MyPipe() as readable:\n    ...     shutil.copyfileobj(readable, sys.stdout)\n    Hello, world!\n\n    Each instance of this class creates a thread and invokes the :meth:`.writeTo` method in that\n    thread. The thread will be join()ed upon normal exit from the context manager, i.e. the body\n    of the `with` statement. If an exception occurs, the thread will not be joined but a\n    well-behaved :meth:`.writeTo` implementation will terminate shortly thereafter due to the\n    pipe having been closed.\n\n    Now, exceptions in the reader thread will be reraised in the main thread:\n\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as readable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n\n    More complicated, less illustrative tests:\n\n    Same as above, but provving that handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         raise RuntimeError('Hello, world!')\n    >>> with MyPipe() as readable:\n    ...     pass\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n\n    Exceptions in the body of the with statement aren't masked, and handles are closed:\n\n    >>> x = os.dup(0); os.close(x)\n    >>> class MyPipe(ReadablePipe):\n    ...     def writeTo(self, writable):\n    ...         pass\n    >>> with MyPipe() as readable:\n    ...     raise RuntimeError('Hello, world!')\n    Traceback (most recent call last):\n    ...\n    RuntimeError: Hello, world!\n    >>> y = os.dup(0); os.close(y); x == y\n    True\n    \"\"\"\n\n    __metaclass__ = ABCMeta\n\n    @abstractmethod\n    def writeTo(self, writable):\n        \"\"\"\n        Implement this method to read data from the pipe.\n\n        :param file writable: the file object representing the writable end of the pipe. Do not\n        explicitly invoke the close() method of the object, that will be done automatically.\n        \"\"\"\n        raise NotImplementedError()\n\n    def _writer(self):\n        try:\n            with os.fdopen(self.writable_fh, 'w') as writable:\n                self.writeTo(writable)\n        except IOError as e:\n            # The other side of the pipe may have been closed by the\n            # reading thread, which is OK.\n            if e.errno != errno.EPIPE:\n                raise\n\n    def __init__(self):\n        super(ReadablePipe, self).__init__()\n        self.writable_fh = None\n        self.readable = None\n        self.thread = None\n\n    def __enter__(self):\n        readable_fh, self.writable_fh = os.pipe()\n        self.readable = os.fdopen(readable_fh, 'r')\n        self.thread = ExceptionalThread(target=self._writer)\n        self.thread.start()\n        return self.readable\n\n    def __exit__(self, exc_type, exc_val, exc_tb):\n        # Close the read end of the pipe. The writing thread may\n        # still be writing to the other end, but this will wake it up\n        # if that's the case.\n        self.readable.close()\n        try:\n            if self.thread is not None:\n                # reraises any exception that was raised in the thread\n                self.thread.join()\n        except:\n            if exc_type is None:\n                # Only raise the child exception if there wasn't\n                # already an exception in the main thread\n                raise\n", "path": "src/toil/jobStores/utils.py"}]}