kaxil commented on code in PR #70690:
URL: https://github.com/apache/airflow/pull/70690#discussion_r3676511255
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -487,6 +484,46 @@ def fetch_celery_task_state(async_result: AsyncResult) ->
tuple[str, str | Excep
return async_result.task_id, ExceptionWithTraceback(e,
exception_traceback), None
+def _get_state_fetch_mp_context() -> multiprocessing.context.BaseContext:
+ """
+ Return the ``multiprocessing`` context for the bulk state-fetch pool.
+
+ ``fork`` is unsafe here because the pool is created from the
multi-threaded scheduler
+ process: a worker can inherit a mutex held by a thread that ``fork()`` did
not copy, then
+ block forever acquiring it, which stalls the scheduling loop until the
scheduler is
+ restarted. ``forkserver``/``spawn`` workers start without the parent's
locks.
+
+ Honours ``[celery] mp_start_method`` (then ``[core] mp_start_method``) so
an operator who
+ has deliberately pinned a method keeps control, and falls back to
``forkserver`` then
+ ``spawn``.
+ """
+ available = multiprocessing.get_all_start_methods()
+ configured = None
+ if AIRFLOW_V_3_3_PLUS:
+ from airflow.utils.process_utils import resolve_mp_start_method
+
+ configured = resolve_mp_start_method("celery")
+
+ if configured:
+ if configured not in available:
+ log.warning(
+ "Configured mp_start_method=%r is not available on this
platform (available: %s); "
+ "falling back to a non-fork start method for the Celery
state-fetch pool.",
+ configured,
+ available,
+ )
+ else:
+ if configured == "fork":
+ log.warning(
+ "mp_start_method is set to 'fork' for the Celery
state-fetch pool. Forking the "
+ "multi-threaded scheduler can deadlock a worker on an
inherited lock and stall "
+ "scheduling; prefer 'forkserver' or 'spawn'."
+ )
+ return multiprocessing.get_context(configured)
+
+ return multiprocessing.get_context("forkserver" if "forkserver" in
available else "spawn")
Review Comment:
The forkserver context here has no preload set, and
`_get_many_using_multiprocessing` builds a fresh pool on every `sync()`.
Children are forked from a forkserver that never imported this module, so each
pool pays a full airflow + celery import in every worker, and unlike fork it
doesn't get cheaper on later pools. Measured with a stand-in module whose
import costs 0.8s, 7 workers, 26 items, new pool per call:
```
fork: sync1=26.2ms sync2=17.1ms sync3=9.9ms
forkserver: sync1=873.3ms sync2=838.4ms sync3=841.4ms
```
Calling `multiprocessing.set_forkserver_preload([...])` first turns that
into 901ms / 19ms / 16ms. That is what `[celery] mp_forkserver_preload` and its
`[core]` fallback exist for, and `set_component_mp_start_method` already
applies them, but building the context directly here skips that. `rpc://` and
`mongodb://` result backends are neither KV nor DB, so they land in this pool
on every heartbeat, which is the same per-pool startup cost the forkserver
column in your table rejects for publishing. Setting the preload, or hoisting
the pool out of the per-sync `with`, would avoid it.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor.py:
##########
@@ -241,22 +229,13 @@ def _send_workloads(self, workload_tuples_to_send:
Sequence[WorkloadInCelery]):
def _send_workloads_to_celery(self, workload_tuples_to_send:
Sequence[WorkloadInCelery]):
from airflow.providers.celery.executors.celery_executor_utils import
send_workload_to_executor
- if len(workload_tuples_to_send) == 1 or self._sync_parallelism == 1:
- # One tuple, or max one process -> send it in the main thread.
- return list(map(send_workload_to_executor,
workload_tuples_to_send))
-
- # Use chunks instead of a work queue to reduce context switching
- # since workloads are roughly uniform in size.
- chunksize =
self._num_workloads_per_send_process(len(workload_tuples_to_send))
- num_processes = min(len(workload_tuples_to_send),
self._sync_parallelism)
-
- # Use ProcessPoolExecutor with team_name instead of workload objects
to avoid pickling issues.
- # Subprocesses reconstruct the team-specific Celery app from the team
name and existing config.
- with ProcessPoolExecutor(max_workers=num_processes) as send_pool:
- key_and_async_results = list(
- send_pool.map(send_workload_to_executor,
workload_tuples_to_send, chunksize=chunksize)
- )
- return key_and_async_results
+ # Publish in this process rather than in a worker pool. Publishing a
workload is a single
+ # short broker round-trip, so a pool costs more to start than the
sends it parallelizes at
+ # any batch size the scheduler produces ([scheduler] max_tis_per_query
defaults to 16).
+ # Forking the pool from the multi-threaded scheduler was also a
deadlock risk: a worker
+ # could inherit a mutex held by a thread fork() did not copy and block
forever acquiring
+ # it, stalling the scheduling loop with no way to recover short of a
restart.
+ return list(map(send_workload_to_executor, workload_tuples_to_send))
Review Comment:
The benchmarks all assume a healthy broker. Each send arms a `[celery]
operation_timeout` alarm (default 1.0s), so a serial batch's worst case is
`len(batch) * 1.0s` on the scheduler loop, where the pool divided it by
`sync_parallelism` (batch of 32 over 11 processes was chunksize 3, so ~3s).
`trigger_tasks` passes the executor's open slots, so batch tracks `[core]
parallelism` (default 32), and goes well past that for anyone who raised
`max_tis_per_query`. 32s without a heartbeat is over `[scheduler]
scheduler_health_check_threshold` (30), which is the 503 this PR is fixing, and
`task_publish_max_retries` (default 3) puts the timed-out keys back for the
next heartbeat to stall on again. Worth bounding total publish time per
heartbeat, or at least calling out the tradeoff?
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -487,6 +484,46 @@ def fetch_celery_task_state(async_result: AsyncResult) ->
tuple[str, str | Excep
return async_result.task_id, ExceptionWithTraceback(e,
exception_traceback), None
+def _get_state_fetch_mp_context() -> multiprocessing.context.BaseContext:
+ """
+ Return the ``multiprocessing`` context for the bulk state-fetch pool.
+
+ ``fork`` is unsafe here because the pool is created from the
multi-threaded scheduler
+ process: a worker can inherit a mutex held by a thread that ``fork()`` did
not copy, then
+ block forever acquiring it, which stalls the scheduling loop until the
scheduler is
+ restarted. ``forkserver``/``spawn`` workers start without the parent's
locks.
+
+ Honours ``[celery] mp_start_method`` (then ``[core] mp_start_method``) so
an operator who
+ has deliberately pinned a method keeps control, and falls back to
``forkserver`` then
+ ``spawn``.
+ """
+ available = multiprocessing.get_all_start_methods()
+ configured = None
+ if AIRFLOW_V_3_3_PLUS:
Review Comment:
The provider still declares `apache-airflow>=2.11.0`, so on 2.11 and 3.0 to
3.2 `configured` stays `None` and the pool is pinned to forkserver/spawn with
no way to ask for `fork` back. The `provider.yaml` change in this PR tells
operators `[celery] mp_start_method` selects this pool's start method, which
won't be true on those versions. `resolve_mp_start_method` is only
`conf.get(section, ...) or conf.get("core", ...)`, so inlining that lookup with
`fallback=None` would keep the escape hatch on every version the provider
supports.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -407,13 +408,9 @@ def send_workload_to_executor(
"""
Send workload to executor (serialized and executed as a Celery task).
- This function runs either inline in the long-lived scheduler process
(single-workload or
- sync_parallelism=1 path) or in short-lived ProcessPoolExecutor
subprocesses (multi-workload
- path). To avoid pickling issues with team-specific Celery apps, we pass
the team_name and
- create the app at call time. The cached app lives for the duration of the
caller process, so
- the main benefit is the scheduler-inline path where the cache persists
across publish cycles.
- In the ProcessPoolExecutor path, each subprocess is recreated per publish
batch and the cache
- only lasts for that single batch.
+ This runs inline in the long-lived caller process (the scheduler, when
publishing). The
Review Comment:
`_get_celery_app_for_workload`'s docstring a few functions up still has the
old framing ("may run either inline in the scheduler process or in a publisher
subprocess"). Worth updating alongside this one.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
