Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix processor cleanup on DagFileProcessorManager #22685

Merged
merged 16 commits into from
Apr 6, 2022
Merged

Fix processor cleanup on DagFileProcessorManager #22685

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
fa7712a
Fix processor cleanup
pcolladosoto Apr 1, 2022
48f0243
Reap the zombie when killing the processor
pcolladosoto Apr 1, 2022
27502be
Reap resulting zombies in a safe way
pcolladosoto Apr 2, 2022
60f20b8
Explain why we are actively waiting
pcolladosoto Apr 2, 2022
2b3eccc
Fix small typo and triling whitespace
pcolladosoto Apr 2, 2022
4a74193
Fix call to `poll()`
pcolladosoto Apr 3, 2022
85f5440
Fix processor cleanup
pcolladosoto Apr 1, 2022
9232c2b
Reap the zombie when killing the processor
pcolladosoto Apr 1, 2022
1f392d6
Reap resulting zombies in a safe way
pcolladosoto Apr 2, 2022
454c383
Explain why we are actively waiting
pcolladosoto Apr 2, 2022
3534263
Fix small typo and triling whitespace
pcolladosoto Apr 2, 2022
ab6555e
Fix call to `poll()`
pcolladosoto Apr 3, 2022
e3379b1
Merge branch 'main' of origin
pcolladosoto Apr 5, 2022
c6ee385
Prevent static check from failing
pcolladosoto Apr 5, 2022
7f651b6
Fix test for `_kill_timed_out_processors()`
pcolladosoto Apr 5, 2022
4fc2df3
Fix `black` checks and `mock` decorators
pcolladosoto Apr 6, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions airflow/dag_processing/manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1058,6 +1058,7 @@ def prepare_file_path_queue(self):
def _kill_timed_out_processors(self):
"""Kill any file processors that timeout to defend against process hangs."""
now = timezone.utcnow()
processors_to_remove = []
for file_path, processor in self._processors.items():
duration = now - processor.start_time
if duration > self._processor_timeout:
Expand All @@ -1073,6 +1074,14 @@ def _kill_timed_out_processors(self):
Stats.incr('dag_file_processor_timeouts')
processor.kill()

# Clean up processor references
potiuk marked this conversation as resolved.
Show resolved Hide resolved
self.waitables.pop(processor.waitable_handle)
processors_to_remove.append(file_path)

# Clean up `self._processors` after iterating over it
for proc in processors_to_remove:
self._processors.pop(proc)
malthe marked this conversation as resolved.
Show resolved Hide resolved

def max_runs_reached(self):
""":return: whether all file paths have been processed max_runs times"""
if self._max_runs == -1: # Unlimited runs.
Expand Down
3 changes: 3 additions & 0 deletions airflow/dag_processing/processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,9 @@ def _kill_process(self) -> None:
if self._process.is_alive() and self._process.pid:
self.log.warning("Killing DAGFileProcessorProcess (PID=%d)", self._process.pid)
os.kill(self._process.pid, signal.SIGKILL)

# Reap the spawned zombie
pcolladosoto marked this conversation as resolved.
Show resolved Hide resolved
os.waitpid(self._process.pid, 0)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Indeed there might be a short time between even SIGKILL gets processed so waiting here makes sense.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know if there's anything here that could be racy, but there are some details here: https://bugs.python.org/issue42558.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ough. Very interesting . So it seems that this one (or similar) is safer (ad waitpid might crash for Python 3.9+) :)

Do I read it right ?

while self._process.poll() is None:
    sleep(0.001)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes that is how I read it too.

@pcolladosoto you wanna make that change? I don't think I had anything else that needed changing.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @malthe! I just pushed the changes: I decided to import the time module so that I could call time.sleep() within the while loop. Hope I made the right choice... You can check out the changes on 27502be. Thanks a lot for the input! This is something that I'm sure will come in handy for future projects 😸

if self._parent_channel:
self._parent_channel.close()

Expand Down