Dag processor cannot run in daemon mode

[yzhsieh]

Apache Airflow version

3.0.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

When starting dag-processor using command: airflow dag-processor --daemon, it will end with following error:

OSError: [Errno 22] Invalid argument

Logs:

2025-05-01 00:50:40,229 INFO - Starting the Dag Processor Job
2025-05-01 00:50:40,229 INFO - Processing files using up to 2 processes at a time 
2025-05-01 00:50:40,230 INFO - Process each file at most once every 30 seconds
2025-05-01 00:50:40,231 INFO - DAG bundles loaded: dags-folder, example_dags
2025-05-01 00:50:40,235 INFO - DAG bundles loaded: dags-folder, example_dags
2025-05-01 00:50:40,235 INFO - Checking for new files in bundle dags-folder every 300 seconds
2025-05-01 00:50:40,236 INFO - Checking for new files in bundle example_dags every 300 seconds
2025-05-01 00:50:40,241 INFO - Refreshing bundle dags-folder
2025-05-01 00:50:40,246 INFO - Searching for files in dags-folder at /home/yzhsieh/airflow/dags
2025-05-01 00:50:40,246 INFO - Found 0 files for bundle dags-folder
2025-05-01 00:50:40,258 INFO - Refreshing bundle example_dags
2025-05-01 00:50:40,263 INFO - Searching for files in example_dags at /home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/example_dags
2025-05-01 00:50:40,273 INFO - Found 58 files for bundle example_dags
2025-05-01 00:50:40,296 ERROR - Exception when executing DagProcessorJob
Traceback (most recent call last):
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute
    self.processor.run()
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 262, in run
    return self._run_parsing_loop()
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 347, in _run_parsing_loop
    self._start_new_processes()
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 894, in _start_new_processes
    processor = self._create_process(file)
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 876, in _create_process
    return DagFileProcessorProcess.start(
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/dag_processing/processor.py", line 245, in start
    proc: Self = super().start(target=target, **kwargs)
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 487, in start
    proc._register_pipe_readers(
  File "/home/yzhsieh/tmp/airflow_venv/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 506, in _register_pipe_readers
    self.selector.register(
  File "/usr/lib/python3.10/selectors.py", line 360, in register
    self._selector.register(key.fd, poller_events)
OSError: [Errno 22] Invalid argument

What you think should happen instead?

No response

How to reproduce

1. Create an python venv and install airflow via Pypi
2. Use airflow standalone to setup a simple Airflow instance
3. Issue airflow dag-processor --daemon, then we can find error in airflow-dag-processor.log

In step 3, if I use airflow dag-processor instead, then it will work without any error

Operating System

Ubuntu 22.04.5 LTS

Versions of Apache Airflow Providers

(Not install any extra provider)

Deployment

Virtualenv installation

Deployment details

Python 3.10.12

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[yaobaishijie]

I also encountered the same problem, as shown in the picture
2025-05-01 20:49:51,117 INFO - Starting the Dag Processor Job 2025-05-01 20:49:51,117 INFO - Processing files using up to 2 processes at a time 2025-05-01 20:49:51,118 INFO - Process each file at most once every 30 seconds 2025-05-01 20:49:51,120 INFO - DAG bundles loaded: dags-folder 2025-05-01 20:49:51,124 INFO - DAG bundles loaded: dags-folder 2025-05-01 20:49:51,124 INFO - Checking for new files in bundle dags-folder every 300 seconds 2025-05-01 20:49:51,128 INFO - Refreshing bundle dags-folder 2025-05-01 20:49:51,130 INFO - Searching for files in dags-folder at /root/airflow/dags 2025-05-01 20:49:51,133 INFO - Found 15 files for bundle dags-folder 2025-05-01 20:49:51,152 ERROR - Exception when executing DagProcessorJob Traceback (most recent call last): File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute self.processor.run() File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 262, in run return self._run_parsing_loop() File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 347, in _run_parsing_loop self._start_new_processes() File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 894, in _start_new_processes processor = self._create_process(file) File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 876, in _create_process return DagFileProcessorProcess.start( File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/dag_processing/processor.py", line 245, in start proc: Self = super().start(target=target, **kwargs) File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/sdk/execution_time/supervisor.py", line 487, in start proc._register_pipe_readers( File "/root/miniconda3/envs/airflow/lib/python3.9/site-packages/airflow/sdk/execution_time/supervisor.py", line 506, in _register_pipe_readers self.selector.register( File "/root/miniconda3/envs/airflow/lib/python3.9/selectors.py", line 360, in register self._selector.register(key.fd, poller_events) OSError: [Errno 22] Invalid argument

[vatsrahul1001]

I tried this and getting below error

2025-05-02 11:07:55,432 ERROR - Exception when executing DagProcessorJob
Traceback (most recent call last):
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute
    self.processor.run()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 262, in run
    return self._run_parsing_loop()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 347, in _run_parsing_loop
    self._start_new_processes()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 894, in _start_new_processes
    processor = self._create_process(file)
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 876, in _create_process
    return DagFileProcessorProcess.start(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/processor.py", line 245, in start
    proc: Self = super().start(target=target, **kwargs)
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 487, in start
    proc._register_pipe_readers(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 506, in _register_pipe_readers
    self.selector.register(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/lib/python3.10/selectors.py", line 523, in register
    self._selector.control([kev], 0, 0)
OSError: [Errno 9] Bad file descriptor

[joyceguouk]

I have the same issue. cannot run airflow dag-processor in daemon mode

[amoghrajesh]

@yzhsieh are you still seeing this issue? I think it is due to macos having some issues with forking + multithreading.

[ravi-simtel]

when in DAEMON mode, it seems to be not passing the argument, hence the error
2025-05-11 09:26:50,356 ERROR - Exception when executing DagProcessorJob Traceback (most recent call last): File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute self.processor.run() File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 262, in run return self._run_parsing_loop() File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 347, in _run_parsing_loop self._start_new_processes() File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 894, in _start_new_processes processor = self._create_process(file) File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/dag_processing/manager.py", line 876, in _create_process return DagFileProcessorProcess.start( File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/dag_processing/processor.py", line 245, in start proc: Self = super().start(target=target, **kwargs) File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/sdk/execution_time/supervisor.py", line 487, in start proc._register_pipe_readers( File "/home/ec2-user/airflow/lib/python3.9/site-packages/airflow/sdk/execution_time/supervisor.py", line 506, in _register_pipe_readers self.selector.register( File "/usr/lib64/python3.9/selectors.py", line 360, in register self._selector.register(key.fd, poller_events) OSError: [Errno 22] Invalid argument

You can use this for now,

nohup airflow dag-processor > nohup.out &

[phanikumv]

@uranusjr would you like to take a look into this?

[Lee-W]

I tried this and getting below error

2025-05-02 11:07:55,432 ERROR - Exception when executing DagProcessorJob
Traceback (most recent call last):
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/jobs/dag_processor_job_runner.py", line 61, in _execute
    self.processor.run()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 262, in run
    return self._run_parsing_loop()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 347, in _run_parsing_loop
    self._start_new_processes()
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 894, in _start_new_processes
    processor = self._create_process(file)
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 876, in _create_process
    return DagFileProcessorProcess.start(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/dag_processing/processor.py", line 245, in start
    proc: Self = super().start(target=target, **kwargs)
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 487, in start
    proc._register_pipe_readers(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/envs/airflow-python310-env/lib/python3.10/site-packages/airflow/sdk/execution_time/supervisor.py", line 506, in _register_pipe_readers
    self.selector.register(
  File "/Users/rahulvats/.pyenv/versions/3.10.12/lib/python3.10/selectors.py", line 523, in register
    self._selector.control([kev], 0, 0)
OSError: [Errno 9] Bad file descriptor

I'm now able to repro this one

[Lee-W]

I suspect it might relate to using daemon with

airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py

Line 455 in f360bc8

pid = os.fork()

, but I'm somewhat stuck. It would be much appreciated if @uranusjr had some suggestions on this one. Thanks!