You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/05/26 16:38:28 UTC
[GitHub] [airflow] ashb opened a new pull request, #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
ashb opened a new pull request, #23944:
URL: https://github.com/apache/airflow/pull/23944
Something changed between Python 3.7 and 3.10 meaning that a limited
parallelism LocalExecutor scheduler now doesn't shutdown cleanly on
receiving a signal.
On closer investigation it appears that much of the code we are doing to
manage this is a poor imitation of the built in
concurrent.futures.ProcessPoolExecutor, so re-write it to use just that.
I have also removed the Unlimited/Limited versions and simplified the
code massively by having just a single version and "unlimited" is now
implemented as having a limit of 2 billion concurrent processes. Should
be enough for anyone
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
uranusjr commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139258100
Tests are not happy 😢
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on a diff in pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r883957884
##########
airflow/executors/local_executor.py:
##########
@@ -201,161 +124,27 @@ class LocalExecutor(BaseExecutor):
:param parallelism: how many parallel processes are run in the executor
"""
+ futures_executor: Optional[concurrent.futures.Executor]
+ futures: Dict[concurrent.futures.Future, TaskInstanceKey]
+
def __init__(self, parallelism: int = PARALLELISM):
super().__init__(parallelism=parallelism)
if self.parallelism < 0:
raise AirflowException("parallelism must be bigger than or equal to 0")
- self.manager: Optional[SyncManager] = None
- self.result_queue: Optional['Queue[TaskInstanceStateType]'] = None
- self.workers: List[QueuedLocalWorker] = []
- self.workers_used: int = 0
- self.workers_active: int = 0
- self.impl: Optional[
- Union['LocalExecutor.UnlimitedParallelism', 'LocalExecutor.LimitedParallelism']
- ] = None
-
- class UnlimitedParallelism:
- """
- Implements LocalExecutor with unlimited parallelism, starting one process
- per each command to execute.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
-
- def start(self) -> None:
- """Starts the executor."""
- self.executor.workers_used = 0
- self.executor.workers_active = 0
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: Name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- local_worker = LocalWorker(self.executor.result_queue, key=key, command=command)
- self.executor.workers_used += 1
- self.executor.workers_active += 1
- local_worker.start()
-
- def sync(self) -> None:
- """Sync will get called periodically by the heartbeat method."""
- if not self.executor.result_queue:
- raise AirflowException("Executor should be started first")
- while not self.executor.result_queue.empty():
- results = self.executor.result_queue.get()
- self.executor.change_state(*results)
- self.executor.workers_active -= 1
-
- def end(self) -> None:
- """
- This method is called when the caller is done submitting job and
- wants to wait synchronously for the job submitted previously to be
- all done.
- """
- while self.executor.workers_active > 0:
- self.executor.sync()
-
- class LimitedParallelism:
- """
- Implements LocalExecutor with limited parallelism using a task queue to
- coordinate work distribution.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
- self.queue: Optional['Queue[ExecutorWorkType]'] = None
-
- def start(self) -> None:
- """Starts limited parallelism implementation."""
- if not self.executor.manager:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue = self.executor.manager.Queue()
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.executor.workers = [
- QueuedLocalWorker(self.queue, self.executor.result_queue)
- for _ in range(self.executor.parallelism)
- ]
-
- self.executor.workers_used = len(self.executor.workers)
-
- for worker in self.executor.workers:
- worker.start()
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue.put((key, command))
-
- def sync(self):
- """Sync will get called periodically by the heartbeat method."""
- while True:
- try:
- results = self.executor.result_queue.get_nowait()
- try:
- self.executor.change_state(*results)
- finally:
- self.executor.result_queue.task_done()
- except Empty:
- break
-
- def end(self):
- """Ends the executor. Sends the poison pill to all workers."""
- for _ in self.executor.workers:
- self.queue.put((None, None))
-
- # Wait for commands to finish
- self.queue.join()
- self.executor.sync()
+ self.futures = {}
def start(self) -> None:
"""Starts the executor"""
old_proctitle = getproctitle()
setproctitle("airflow executor -- LocalExecutor")
- self.manager = Manager()
setproctitle(old_proctitle)
- self.result_queue = self.manager.Queue()
- self.workers = []
- self.workers_used = 0
- self.workers_active = 0
- self.impl = (
- LocalExecutor.UnlimitedParallelism(self)
- if self.parallelism == 0
- else LocalExecutor.LimitedParallelism(self)
- )
+ # This isn't _truely_ unlimited, but it should be good enough!
+ size = 2**30 if self.parallelism == 0 else self.parallelism
Review Comment:
Interestingly there is no promises from the unlimited: https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#parallelism
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850602344
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
Review Comment:
In 2f7d6c53d2 I've switched it to use a synchronized integer via `multiprocessing.Value` that we can pass to the worker process, so they can decrement the count as soon as it reads the message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1193426028
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on a diff in pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r883957884
##########
airflow/executors/local_executor.py:
##########
@@ -201,161 +124,27 @@ class LocalExecutor(BaseExecutor):
:param parallelism: how many parallel processes are run in the executor
"""
+ futures_executor: Optional[concurrent.futures.Executor]
+ futures: Dict[concurrent.futures.Future, TaskInstanceKey]
+
def __init__(self, parallelism: int = PARALLELISM):
super().__init__(parallelism=parallelism)
if self.parallelism < 0:
raise AirflowException("parallelism must be bigger than or equal to 0")
- self.manager: Optional[SyncManager] = None
- self.result_queue: Optional['Queue[TaskInstanceStateType]'] = None
- self.workers: List[QueuedLocalWorker] = []
- self.workers_used: int = 0
- self.workers_active: int = 0
- self.impl: Optional[
- Union['LocalExecutor.UnlimitedParallelism', 'LocalExecutor.LimitedParallelism']
- ] = None
-
- class UnlimitedParallelism:
- """
- Implements LocalExecutor with unlimited parallelism, starting one process
- per each command to execute.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
-
- def start(self) -> None:
- """Starts the executor."""
- self.executor.workers_used = 0
- self.executor.workers_active = 0
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: Name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- local_worker = LocalWorker(self.executor.result_queue, key=key, command=command)
- self.executor.workers_used += 1
- self.executor.workers_active += 1
- local_worker.start()
-
- def sync(self) -> None:
- """Sync will get called periodically by the heartbeat method."""
- if not self.executor.result_queue:
- raise AirflowException("Executor should be started first")
- while not self.executor.result_queue.empty():
- results = self.executor.result_queue.get()
- self.executor.change_state(*results)
- self.executor.workers_active -= 1
-
- def end(self) -> None:
- """
- This method is called when the caller is done submitting job and
- wants to wait synchronously for the job submitted previously to be
- all done.
- """
- while self.executor.workers_active > 0:
- self.executor.sync()
-
- class LimitedParallelism:
- """
- Implements LocalExecutor with limited parallelism using a task queue to
- coordinate work distribution.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
- self.queue: Optional['Queue[ExecutorWorkType]'] = None
-
- def start(self) -> None:
- """Starts limited parallelism implementation."""
- if not self.executor.manager:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue = self.executor.manager.Queue()
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.executor.workers = [
- QueuedLocalWorker(self.queue, self.executor.result_queue)
- for _ in range(self.executor.parallelism)
- ]
-
- self.executor.workers_used = len(self.executor.workers)
-
- for worker in self.executor.workers:
- worker.start()
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue.put((key, command))
-
- def sync(self):
- """Sync will get called periodically by the heartbeat method."""
- while True:
- try:
- results = self.executor.result_queue.get_nowait()
- try:
- self.executor.change_state(*results)
- finally:
- self.executor.result_queue.task_done()
- except Empty:
- break
-
- def end(self):
- """Ends the executor. Sends the poison pill to all workers."""
- for _ in self.executor.workers:
- self.queue.put((None, None))
-
- # Wait for commands to finish
- self.queue.join()
- self.executor.sync()
+ self.futures = {}
def start(self) -> None:
"""Starts the executor"""
old_proctitle = getproctitle()
setproctitle("airflow executor -- LocalExecutor")
- self.manager = Manager()
setproctitle(old_proctitle)
- self.result_queue = self.manager.Queue()
- self.workers = []
- self.workers_used = 0
- self.workers_active = 0
- self.impl = (
- LocalExecutor.UnlimitedParallelism(self)
- if self.parallelism == 0
- else LocalExecutor.LimitedParallelism(self)
- )
+ # This isn't _truely_ unlimited, but it should be good enough!
+ size = 2**30 if self.parallelism == 0 else self.parallelism
Review Comment:
Interestingly there is no promises for the unlimited: https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#parallelism
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850499129
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
+ if exc:
+ # TODO: This needs a better stacktrace, it appears from here
+ if hasattr(exc, "add_note"):
+ exc.add_note("(This stacktrace is incorrect -- the exception came from a subprocess)")
Review Comment:
Planning to leave this -- it happens so rarely (as the `_execute_work` fns have try/except) so this is just an final line of defense to show something when the task goes wrong.
(The only way I was able to trigger this was a syntax error when developing it)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850495715
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
Review Comment:
We call `_check_workers` directly from in `execute_async`, so we'll spawn workers directly after sending the message on the queue. I'll leave a comment to that effect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850509324
##########
airflow/executors/local_executor.py:
##########
@@ -25,192 +25,139 @@
from __future__ import annotations
-import contextlib
import logging
+import multiprocessing
import os
import subprocess
-from abc import abstractmethod
-from multiprocessing import Manager, Process
-from queue import Empty
+from multiprocessing import Queue, SimpleQueue
from typing import TYPE_CHECKING, Any, Optional, Tuple
-from setproctitle import getproctitle, setproctitle
-
from airflow import settings
-from airflow.exceptions import AirflowException
from airflow.executors.base_executor import PARALLELISM, BaseExecutor
-from airflow.traces.tracer import Trace, add_span
+from airflow.traces.tracer import add_span
from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
-from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
- from multiprocessing.managers import SyncManager
- from queue import Queue
-
from airflow.executors.base_executor import CommandType
- from airflow.models.taskinstance import TaskInstanceStateType
from airflow.models.taskinstancekey import TaskInstanceKey
# This is a work to be executed by a worker.
# It can Key and Command - but it can also be None, None which is actually a
# "Poison Pill" - worker seeing Poison Pill should take the pill and ... die instantly.
- ExecutorWorkType = Tuple[Optional[TaskInstanceKey], Optional[CommandType]]
+ ExecutorWorkType = Optional[Tuple[TaskInstanceKey, CommandType]]
+ TaskInstanceStateType = Tuple[TaskInstanceKey, TaskInstanceState, Optional[Exception]]
-class LocalWorkerBase(Process, LoggingMixin):
- """
- LocalWorkerBase implementation to run airflow commands.
+def _run_worker(logger_name: str, input: SimpleQueue[ExecutorWorkType], output: Queue[TaskInstanceStateType]):
+ import signal
- Executes the given command and puts the result into a result queue when done, terminating execution.
+ from setproctitle import setproctitle
Review Comment:
I'm not sure this actually makes any difference now I think about it, as this is in the mp process, but it's about to fork a new process anyway. I'll move it back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139463868
> Error: Process completed with exit code 137.
Damn, thats OOM.
I wonder why this is using so much more memory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139504410
Okay, another re-think needed.
This seems to work fine on 3.10, but on 3.7 it ends up spawning thousands of processes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850494139
##########
airflow/executors/local_executor.py:
##########
@@ -25,192 +25,139 @@
from __future__ import annotations
-import contextlib
import logging
+import multiprocessing
import os
import subprocess
-from abc import abstractmethod
-from multiprocessing import Manager, Process
-from queue import Empty
+from multiprocessing import Queue, SimpleQueue
from typing import TYPE_CHECKING, Any, Optional, Tuple
-from setproctitle import getproctitle, setproctitle
-
from airflow import settings
-from airflow.exceptions import AirflowException
from airflow.executors.base_executor import PARALLELISM, BaseExecutor
-from airflow.traces.tracer import Trace, add_span
+from airflow.traces.tracer import add_span
from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
-from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
- from multiprocessing.managers import SyncManager
- from queue import Queue
-
from airflow.executors.base_executor import CommandType
- from airflow.models.taskinstance import TaskInstanceStateType
from airflow.models.taskinstancekey import TaskInstanceKey
# This is a work to be executed by a worker.
# It can Key and Command - but it can also be None, None which is actually a
# "Poison Pill" - worker seeing Poison Pill should take the pill and ... die instantly.
- ExecutorWorkType = Tuple[Optional[TaskInstanceKey], Optional[CommandType]]
+ ExecutorWorkType = Optional[Tuple[TaskInstanceKey, CommandType]]
+ TaskInstanceStateType = Tuple[TaskInstanceKey, TaskInstanceState, Optional[Exception]]
-class LocalWorkerBase(Process, LoggingMixin):
- """
- LocalWorkerBase implementation to run airflow commands.
+def _run_worker(logger_name: str, input: SimpleQueue[ExecutorWorkType], output: Queue[TaskInstanceStateType]):
+ import signal
- Executes the given command and puts the result into a result queue when done, terminating execution.
+ from setproctitle import setproctitle
Review Comment:
on OSX importing secproctitle module does some stuff with threads.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850542416
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
Review Comment:
I think this is fine actually -- the only way we can decrement is if the work finishes to completion. If the process exits abnormally it by definition couldn't have sent the completion message.
So lets say that it picks up a message of the `activity_queue` and then dies.
We'll never get a completion message on `result_queue`, so `_outstanding_messages` will be off by 1.
Let me see if I can use a JoinableQueue (which has `task_done()` that we might be able to use instead. Maybe. It depends if we can get at the count without forcing a join.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Re-write LocalExecutor workers to cleanly shutdown on 3.10 [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-2488456363
This branch name was "rewrite-local-exexc-concurrentfutures" (sic) as when originally opened in 2022 for 3.10 that was the plan.
However since then 3.12 has come out and it now starts issuing warnings when Fork and threads are used, and concurrent.futures uses a thread internally, so a different approach was used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139423139
Looking. I only checked the local executor tests before pushing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139505574
Oh right, prior to Python 3.9 ProcessPoolExecutor *precreates* the specified number of worker processes.
So essentially this was a fork-bomb on Py 3.7 and 3.8.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-1139096778
This could either be in a 2.3.3 or 2.4.0 -- it's not critical and I haven't tested it all that thoroughly to make sure it doesn't have edge cases
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r884612358
##########
airflow/executors/local_executor.py:
##########
@@ -201,161 +124,27 @@ class LocalExecutor(BaseExecutor):
:param parallelism: how many parallel processes are run in the executor
"""
+ futures_executor: Optional[concurrent.futures.Executor]
+ futures: Dict[concurrent.futures.Future, TaskInstanceKey]
+
def __init__(self, parallelism: int = PARALLELISM):
super().__init__(parallelism=parallelism)
if self.parallelism < 0:
raise AirflowException("parallelism must be bigger than or equal to 0")
- self.manager: Optional[SyncManager] = None
- self.result_queue: Optional['Queue[TaskInstanceStateType]'] = None
- self.workers: List[QueuedLocalWorker] = []
- self.workers_used: int = 0
- self.workers_active: int = 0
- self.impl: Optional[
- Union['LocalExecutor.UnlimitedParallelism', 'LocalExecutor.LimitedParallelism']
- ] = None
-
- class UnlimitedParallelism:
- """
- Implements LocalExecutor with unlimited parallelism, starting one process
- per each command to execute.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
-
- def start(self) -> None:
- """Starts the executor."""
- self.executor.workers_used = 0
- self.executor.workers_active = 0
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: Name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- local_worker = LocalWorker(self.executor.result_queue, key=key, command=command)
- self.executor.workers_used += 1
- self.executor.workers_active += 1
- local_worker.start()
-
- def sync(self) -> None:
- """Sync will get called periodically by the heartbeat method."""
- if not self.executor.result_queue:
- raise AirflowException("Executor should be started first")
- while not self.executor.result_queue.empty():
- results = self.executor.result_queue.get()
- self.executor.change_state(*results)
- self.executor.workers_active -= 1
-
- def end(self) -> None:
- """
- This method is called when the caller is done submitting job and
- wants to wait synchronously for the job submitted previously to be
- all done.
- """
- while self.executor.workers_active > 0:
- self.executor.sync()
-
- class LimitedParallelism:
- """
- Implements LocalExecutor with limited parallelism using a task queue to
- coordinate work distribution.
-
- :param executor: the executor instance to implement.
- """
-
- def __init__(self, executor: 'LocalExecutor'):
- self.executor: 'LocalExecutor' = executor
- self.queue: Optional['Queue[ExecutorWorkType]'] = None
-
- def start(self) -> None:
- """Starts limited parallelism implementation."""
- if not self.executor.manager:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue = self.executor.manager.Queue()
- if not self.executor.result_queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.executor.workers = [
- QueuedLocalWorker(self.queue, self.executor.result_queue)
- for _ in range(self.executor.parallelism)
- ]
-
- self.executor.workers_used = len(self.executor.workers)
-
- for worker in self.executor.workers:
- worker.start()
-
- def execute_async(
- self,
- key: TaskInstanceKey,
- command: CommandType,
- queue: Optional[str] = None,
- executor_config: Optional[Any] = None,
- ) -> None:
- """
- Executes task asynchronously.
-
- :param key: the key to identify the task instance
- :param command: the command to execute
- :param queue: name of the queue
- :param executor_config: configuration for the executor
- """
- if not self.queue:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.queue.put((key, command))
-
- def sync(self):
- """Sync will get called periodically by the heartbeat method."""
- while True:
- try:
- results = self.executor.result_queue.get_nowait()
- try:
- self.executor.change_state(*results)
- finally:
- self.executor.result_queue.task_done()
- except Empty:
- break
-
- def end(self):
- """Ends the executor. Sends the poison pill to all workers."""
- for _ in self.executor.workers:
- self.queue.put((None, None))
-
- # Wait for commands to finish
- self.queue.join()
- self.executor.sync()
+ self.futures = {}
def start(self) -> None:
"""Starts the executor"""
old_proctitle = getproctitle()
setproctitle("airflow executor -- LocalExecutor")
- self.manager = Manager()
setproctitle(old_proctitle)
- self.result_queue = self.manager.Queue()
- self.workers = []
- self.workers_used = 0
- self.workers_active = 0
- self.impl = (
- LocalExecutor.UnlimitedParallelism(self)
- if self.parallelism == 0
- else LocalExecutor.LimitedParallelism(self)
- )
+ # This isn't _truely_ unlimited, but it should be good enough!
+ size = 2**30 if self.parallelism == 0 else self.parallelism
Review Comment:
Huh. TIL
Sadly I think that although this isn't documented it's still "relied upon" behaviour so we can't just remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850603063
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
Review Comment:
Updated the comment. I don't think this is a big enough thing to create a GH issue for tbh
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] closed pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
URL: https://github.com/apache/airflow/pull/23944
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on a diff in pull request #23944: Re-write LocalExecutor workers to cleanly shutdown on 3.10
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r883944216
##########
airflow/executors/local_executor.py:
##########
@@ -365,34 +154,42 @@ def execute_async(
executor_config: Optional[Any] = None,
) -> None:
"""Execute asynchronously."""
- if not self.impl:
+ if not self.futures_executor:
raise AirflowException(NOT_STARTED_MESSAGE)
self.validate_command(command)
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ result = self.futures_executor.submit(_execute_work, key=key, command=command)
+ self.futures[result] = key
- def sync(self) -> None:
+ def sync(self, timeout=0) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if not self.impl:
- raise AirflowException(NOT_STARTED_MESSAGE)
- self.impl.sync()
+ try:
+ done, _ = concurrent.futures.wait(self.futures, timeout=timeout)
+ for future in done:
+ key = self.futures.pop(future)
+ exc = future.exception()
+ result: Any
+ if exc:
+ result = (key, State.FAILED, exc)
+ else:
+ result = future.result()
+ self.change_state(*result)
+ except concurrent.futures.TimeoutError:
+ pass
def end(self) -> None:
- """
- Ends the executor.
- :return:
- """
- if not self.impl:
- raise AirflowException(NOT_STARTED_MESSAGE)
- if not self.manager:
+ """Ends the executor."""
+ if not self.futures_executor:
raise AirflowException(NOT_STARTED_MESSAGE)
self.log.info(
"Shutting down LocalExecutor"
"; waiting for running tasks to finish. Signal again if you don't want to wait."
)
- self.impl.end()
- self.manager.shutdown()
+ self.futures_executor.shutdown()
+ self.sync(timeout=None)
def terminate(self):
"""Terminate the executor is not doing anything."""
+ if self.futures_executor:
+ self.futures_executor.shutdown(wait=False)
Review Comment:
Should we set `cancel_futures=True` in either of the shutdown calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #23944:
URL: https://github.com/apache/airflow/pull/23944#issuecomment-2488683965
Tests currently failing cos I moved setup out of start() into init(), but ExecutorLoader caches instances. I'll move it back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850467773
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
Review Comment:
Worth creating a GH issue for it if someone wants to take care of it
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
Review Comment:
Is there ever a risk of this becoming inconsistent due to exceptions or unexpected shutdowns?
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
Review Comment:
If we have 2 workers and `_outstanding_messages` are 4, this is just going to create 1 worker. So we rely on multiple `_check_workers` invocations in sync/heartbeat. Worth adding a comment here or somewhere
##########
airflow/executors/local_executor.py:
##########
@@ -25,192 +25,139 @@
from __future__ import annotations
-import contextlib
import logging
+import multiprocessing
import os
import subprocess
-from abc import abstractmethod
-from multiprocessing import Manager, Process
-from queue import Empty
+from multiprocessing import Queue, SimpleQueue
from typing import TYPE_CHECKING, Any, Optional, Tuple
-from setproctitle import getproctitle, setproctitle
-
from airflow import settings
-from airflow.exceptions import AirflowException
from airflow.executors.base_executor import PARALLELISM, BaseExecutor
-from airflow.traces.tracer import Trace, add_span
+from airflow.traces.tracer import add_span
from airflow.utils.dag_parsing_context import _airflow_parsing_context_manager
-from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
- from multiprocessing.managers import SyncManager
- from queue import Queue
-
from airflow.executors.base_executor import CommandType
- from airflow.models.taskinstance import TaskInstanceStateType
from airflow.models.taskinstancekey import TaskInstanceKey
# This is a work to be executed by a worker.
# It can Key and Command - but it can also be None, None which is actually a
# "Poison Pill" - worker seeing Poison Pill should take the pill and ... die instantly.
- ExecutorWorkType = Tuple[Optional[TaskInstanceKey], Optional[CommandType]]
+ ExecutorWorkType = Optional[Tuple[TaskInstanceKey, CommandType]]
+ TaskInstanceStateType = Tuple[TaskInstanceKey, TaskInstanceState, Optional[Exception]]
-class LocalWorkerBase(Process, LoggingMixin):
- """
- LocalWorkerBase implementation to run airflow commands.
+def _run_worker(logger_name: str, input: SimpleQueue[ExecutorWorkType], output: Queue[TaskInstanceStateType]):
+ import signal
- Executes the given command and puts the result into a result queue when done, terminating execution.
+ from setproctitle import setproctitle
Review Comment:
This is not at the top for some reason, right? I remember discussing but can't remember
##########
tests/executors/test_local_executor.py:
##########
@@ -44,85 +53,72 @@ def test_serve_logs_default_value(self):
assert LocalExecutor.serve_logs
@mock.patch("airflow.executors.local_executor.subprocess.check_call")
- def execution_parallelism_subprocess(self, mock_check_call, parallelism=0):
- success_command = ["airflow", "tasks", "run", "true", "some_parameter", "2020-10-07"]
- fail_command = ["airflow", "tasks", "run", "false", "task_id", "2020-10-07"]
+ @mock.patch("airflow.cli.commands.task_command.task_run")
+ def _test_execute(self, mock_run, mock_check_call, parallelism=1):
+ success_command = ["airflow", "tasks", "run", "success", "some_parameter", "2020-10-07"]
+ fail_command = ["airflow", "tasks", "run", "failure", "task_id", "2020-10-07"]
+ # We just mock both styles here, only one will be hit though
def fake_execute_command(command, close_fds=True):
if command != success_command:
raise subprocess.CalledProcessError(returncode=1, cmd=command)
else:
return 0
- mock_check_call.side_effect = fake_execute_command
-
- self._test_execute(parallelism, success_command, fail_command)
-
- @mock.patch("airflow.cli.commands.task_command.task_run")
- def execution_parallelism_fork(self, mock_run, parallelism=0):
- success_command = ["airflow", "tasks", "run", "success", "some_parameter", "2020-10-07"]
- fail_command = ["airflow", "tasks", "run", "failure", "some_parameter", "2020-10-07"]
-
def fake_task_run(args):
+ print(repr(args))
Review Comment:
```suggestion
```
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
+ if exc:
+ # TODO: This needs a better stacktrace, it appears from here
+ if hasattr(exc, "add_note"):
+ exc.add_note("(This stacktrace is incorrect -- the exception came from a subprocess)")
Review Comment:
Planning to fix it here or separate PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #23944:
URL: https://github.com/apache/airflow/pull/23944#discussion_r1850496343
##########
airflow/executors/local_executor.py:
##########
@@ -403,32 +202,91 @@ def execute_async(
executor_config: Any | None = None,
) -> None:
"""Execute asynchronously."""
- if TYPE_CHECKING:
- assert self.impl
-
self.validate_airflow_tasks_run_command(command)
+ self.activity_queue.put((key, command))
+ self._outstanding_messages += 1
+ self._check_workers(can_start=True)
+
+ def _check_workers(self, can_start: bool = True):
+ # Reap any dead workers
+ to_remove = set()
+ for pid, proc in self.workers.items():
+ if not proc.is_alive():
+ to_remove.add(pid)
+ proc.close()
+
+ if to_remove:
+ self.workers = {pid: proc for pid, proc in self.workers.items() if pid not in to_remove}
+
+ # If we're using spawn in multiprocessing (default on macos now) to start tasks, this can get called a
+ # via sync() a few times before the spawned process actually starts picking up messages. Try not to
+ # create too much
+
+ if self._outstanding_messages <= 0 or self.activity_queue.empty():
+ # Nothing to do, should we shut down idle workers?
+ return
- self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
+ need_more_workers = len(self.workers) < self._outstanding_messages
+ if need_more_workers and (self.parallelism == 0 or len(self.workers) < self.parallelism):
+ self._spawn_worker()
+
+ def _spawn_worker(self):
+ p = multiprocessing.Process(
+ target=_run_worker,
+ kwargs={
+ "logger_name": self.log.name,
+ "input": self.activity_queue,
+ "output": self.result_queue,
+ },
+ )
+ p.start()
+ if TYPE_CHECKING:
+ assert p.pid # Since we've called start
+ self.workers[p.pid] = p
def sync(self) -> None:
"""Sync will get called periodically by the heartbeat method."""
- if TYPE_CHECKING:
- assert self.impl
+ self._read_results()
+ self._check_workers()
- self.impl.sync()
+ def _read_results(self):
+ while not self.result_queue.empty():
+ key, state, exc = self.result_queue.get()
+ self._outstanding_messages = self._outstanding_messages - 1
+
Review Comment:
Good question, let me have a think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Rewite LocalExecutor to be simpler, and to shutdown cleanly on Python 3.10+ [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #23944:
URL: https://github.com/apache/airflow/pull/23944
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org