tags -1 +patch thanks I believe I have tracked down the problem. The changes introduced in fec9e97c51b3a8ff226a4b3b2b0563a4a680ac68 cause a file object to be manipulated on both sides of a thread boundary, which seems dangerous. The attached patch restructures things to avoid that problem, and a few other implementation issues. The commit message has more details about the rationale.
I've also attached a script I used to help reproduce the issue. It doesn't do so reliably, and it's not totally robust itself, but I found it more reliable than trying to time timeout or a ^C right. fec9d97 was meant to improve portability. It's possible that my patch has its own portability problems: it opens a FIFO in nonblocking mode, and fifo(7) makes it sound like this behavior is undefined by POSIX. We might want to check with some BSD folks to see what they do in that case. But even if the patch doesn't work as intended outside Linux, hopefully fixing that is a matter of checking for more error codes or something like that, and won't require as much restructuring as fec9d97 did. I should note that the cleanup that we do after receiving a signal is a little incomplete in the current state, too. What happens now is: when we get a signal, we raise a SystemExit exception in the main thread. That bubbles all the way back up to diffoscope.main.main, which cleans up all the temporary files and directories in a finally block, and then the Python process exits. However, subprocesses and all of our threads (which are marked as daemon threads) are still running, and have all the file descriptors open. So the files still "exist" in some sense and are taking up space, even if they're no longer addressable. We're also still spending plenty of CPU doing the real diff work. Eventually the processes will all go away because they finished their work naturally, or because they encounter an error because their working directories have disappeared out from under them. It would be nice to move to a system where, when we get a signal, we send it to all our children, then wait for them to finish before doing our own cleanup. But that's a bigger job and less urgent, so I didn't go that far here. -- Brett Smith
>From 224f61e67fa3880c6f13b6fcd580f90cd53a4522 Mon Sep 17 00:00:00 2001 From: Brett Smith <[email protected]> Date: Fri, 27 Jan 2017 22:42:23 -0500 Subject: [PATCH] diffoscope.diff: Improve FIFO writing robustness. We used to give input to diff using FIFO objects. fec9e97c51b3a8ff226a4b3b2b0563a4a680ac68 changed this to avoid relying on /dev/fd, which is a good change for portability. Unfortunately, the implementation has a few issues of its own: * It feeds diff with normal files, rather than FIFOs. This could cause trouble if diff tries to read farther than the underlying feeder processes have written. * It uses threads to write to the pseudo-FIFO, but the file object is manipulated in multiple threads. I suspect this is the primary cause of the segfaults observed in #852013. * fd_from_feeder is decorated as a context manager, but it never yields anything, which is not how it's expected to be used. I'm not sure this is causing any problems, but it makes it harder to reason about what's going on. This commit introduces a new FIFOFeeder class. It is wholly responsible for creating and feeding the FIFO, so we don't have to pass file objects across the thread boundary and risk segfaults. It uses context management to tell when the FIFO is no longer needed, so it can clean up nicely. --- diffoscope/diff.py | 103 +++++++++++++++++++++++++++-------------------------- 1 file changed, 52 insertions(+), 51 deletions(-) diff --git a/diffoscope/diff.py b/diffoscope/diff.py index 011916a..0335a26 100644 --- a/diffoscope/diff.py +++ b/diffoscope/diff.py @@ -19,6 +19,9 @@ import re import io +import os +import errno +import fcntl import hashlib import logging import threading @@ -190,54 +193,55 @@ def run_diff(fifo1, fifo2, end_nl_q1, end_nl_q2): return parser.diff -def feed(feeder, f, end_nl_q): - # work-around unified diff limitation: if there's no newlines in both - # don't make it a difference - try: - end_nl = feeder(f) - end_nl_q.put(end_nl) - finally: - f.close() - -class ExThread(threading.Thread): - """ - Inspired by https://stackoverflow.com/a/6874161 - """ - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.__status_queue = Queue() - - def run(self, *args, **kwargs): - try: - super().run(*args, **kwargs) - except Exception as ex: - #except_type, except_class, tb = sys.exc_info() - self.__status_queue.put(ex) - - self.__status_queue.put(None) +class FIFOFeeder(threading.Thread): + def __init__(self, feeder, fifo_path, end_nl_q=None, *, daemon=True): + os.mkfifo(fifo_path) + super().__init__(daemon=daemon) + self.feeder = feeder + self.fifo_path = fifo_path + self.end_nl_q = Queue() if end_nl_q is None else end_nl_q + self._exception = None + self._want_join = threading.Event() - def wait_for_exc_info(self): - return self.__status_queue.get() + def __enter__(self): + self.start() + return self - def join(self): - ex = self.wait_for_exc_info() - if ex is None: - return - raise ex + def __exit__(self, exc_type, exc_value, exc_tb): + self.join() [email protected] -def fd_from_feeder(feeder, end_nl_q, fifo): - f = open(fifo, 'wb') - t = ExThread(target=feed, args=(feeder, f, end_nl_q)) + def run(self): + try: + # Try to open the FIFO nonblocking, so we can periodically check + # if the main thread wants us to wind down. If it does, there's no + # more need for the FIFO, so stop the thread. + while True: + try: + fifo_fd = os.open(self.fifo_path, os.O_WRONLY | os.O_NONBLOCK) + except OSError as error: + if error.errno != errno.ENXIO: + raise + elif self._want_join.is_set(): + return + else: + break + + # Now clear the fd's nonblocking flag to let writes block normally. + fcntl.fcntl(fifo_fd, fcntl.F_SETFL, 0) + with open(fifo_fd, 'wb') as fifo: + # The queue works around a unified diff limitation: if there's + # no newlines in both don't make it a difference + end_nl = self.feeder(fifo) + self.end_nl_q.put(end_nl) + except Exception as error: + self._exception = error - t.daemon = True - t.start() + def join(self): + self._want_join.set() + super().join() + if self._exception is not None: + raise self._exception - try: - t.join() - finally: - f.close() def empty_file_feeder(): def feeder(f): @@ -273,15 +277,12 @@ def make_feeder_from_raw_reader(in_file, filter=lambda buf: buf): return feeder def diff(feeder1, feeder2): - end_nl_q1 = Queue() - end_nl_q2 = Queue() - with tempfile.TemporaryDirectory() as tmpdir: - fifo1 = '{}/f1'.format(tmpdir) - fifo2 = '{}/f2'.format(tmpdir) - fd_from_feeder(feeder1, end_nl_q1, fifo1) - fd_from_feeder(feeder2, end_nl_q2, fifo2) - return run_diff(fifo1, fifo2, end_nl_q1, end_nl_q2) + fifo1_path = os.path.join(tmpdir, 'fifo1') + fifo2_path = os.path.join(tmpdir, 'fifo2') + with FIFOFeeder(feeder1, fifo1_path) as fifo1, \ + FIFOFeeder(feeder2, fifo2_path) as fifo2: + return run_diff(fifo1_path, fifo2_path, fifo1.end_nl_q, fifo2.end_nl_q) def reverse_unified_diff(diff): res = [] -- 2.1.4
test.sh
Description: application/shellscript
