diff --git a/luigi/contrib/external_program.py b/luigi/contrib/external_program.py index bd60ee24c8..a1c84aed5f 100644 --- a/luigi/contrib/external_program.py +++ b/luigi/contrib/external_program.py @@ -38,7 +38,6 @@ import tempfile from contextlib import contextmanager from multiprocessing import Process -from time import sleep import luigi from luigi.parameter import ParameterVisibility @@ -189,16 +188,15 @@ def _track_url_by_pattern(): If tmp_stdout is passed, also appends lines to this file. """ pattern = re.compile(self.tracking_url_pattern) - for new_line in iter(pipe_to_read.readline, ""): - if new_line: - if file_to_write: - file_to_write.write(new_line) - match = re.search(pattern, new_line.decode("utf-8")) - if match: - self.set_tracking_url(self.build_tracking_url(match.group(1))) - else: - file_to_write.flush() - sleep(time_to_sleep) + # PIPE is binary, so EOF is b""; the old "" sentinel never stopped the iterator. + for new_line in iter(pipe_to_read.readline, b""): + if file_to_write: + file_to_write.write(new_line) + match = re.search(pattern, new_line.decode("utf-8")) + if match: + self.set_tracking_url(self.build_tracking_url(match.group(1))) + if file_to_write: + file_to_write.flush() track_proc = Process(target=_track_url_by_pattern) try: diff --git a/test/contrib/external_program_test.py b/test/contrib/external_program_test.py index 086fcca1ee..8ed9363a97 100644 --- a/test/contrib/external_program_test.py +++ b/test/contrib/external_program_test.py @@ -20,7 +20,7 @@ import tempfile from functools import partial from io import BytesIO -from multiprocessing import Value +from multiprocessing import Process, Value from subprocess import Popen import mock @@ -198,6 +198,24 @@ def fake_set_tracking_url(val, url): task.run() self.assertEqual(test_val.value, 1) + def test_tracking_process_exits_cleanly_when_capture_output_disabled(self): + # When output is not captured, there is no file to tee into. EOF used to call + # flush() on that None handle and crash the tracker (issue #3131). + procs = [] + + class RecordingProcess(Process): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + procs.append(self) + + task = TestEchoTask(capture_output=False, stream_for_searching_tracking_url="stdout", tracking_url_pattern=r"Hello, (.*)!") + with mock.patch("luigi.contrib.external_program.Process", RecordingProcess): + with mock.patch.object(task, "set_tracking_url"): + task.run() + + self.assertEqual(len(procs), 1) + self.assertEqual(procs[0].exitcode, 0) + def test_tracking_url_pattern_works_with_capture_output_enabled(self): test_val = Value("i", 0)