Avoid scheduler/parser manager deadlock by using non-blocking IO

There have been long standing issues where the scheduler would "stop
responding" that we haven't been able to track down.

Someone was able to catch the scheduler in this state in 2.0.1 and
inspect it with py-spy (thanks, MatthewRBruce!)

The stack traces (slightly shortened) were:

```
Process 6: /usr/local/bin/python /usr/local/bin/airflow scheduler
Python v3.8.7 (/usr/local/bin/python3.8)
Thread 0x7FF5C09C8740 (active): "MainThread"
    _send (multiprocessing/connection.py:368)
    _send_bytes (multiprocessing/connection.py:411)
    send (multiprocessing/connection.py:206)
    send_callback_to_execute (airflow/utils/dag_processing.py:283)
    _send_dag_callbacks_to_processor (airflow/jobs/scheduler_job.py:1795)
    _schedule_dag_run (airflow/jobs/scheduler_job.py:1762)

Process 77: airflow scheduler -- DagFileProcessorManager
Python v3.8.7 (/usr/local/bin/python3.8)
Thread 0x7FF5C09C8740 (active): "MainThread"
    _send (multiprocessing/connection.py:368)
    _send_bytes (multiprocessing/connection.py:405)
    send (multiprocessing/connection.py:206)
    _run_parsing_loop (airflow/utils/dag_processing.py:698)
    start (airflow/utils/dag_processing.py:596)
```

What this shows is that both processes are stuck trying to send data to
each other, but neither can proceed as both buffers are full, but since
both are trying to send, neither side is going to read and make more
space in the buffer. A classic deadlock!

The fix for this is two fold:

1) Enable non-blocking IO on the DagFileProcessorManager side.

   The only thing the Manager sends back up the pipe is (now, as of 2.0)
   the DagParsingStat object, and the scheduler will happily continue
   without receiving these, so in the case of a blocking error, it is
   simply better to ignore the error, continue the loop and try sending
   one again later.

2) Reduce the size of DagParsingStat

   In the case of a large number of dag files we included the path for
   each and every one (in full) in _each_ parsing stat. Not only did the
   scheduler do nothing with this field, meaning it was larger than it
   needed to be, by making it such a large object, it increases the
   likely hood of hitting this send-buffer-full deadlock case!

Author: ashb

airflow/utils/dag_processing.py

@@ -141,7 +141,7 @@ def waitable_handle(self):
 class DagParsingStat(NamedTuple):
     """Information on processing progress"""
 
-    file_paths: List[str]
+    num_file_paths: int
     done: bool
     all_files_processed: bool
 
@@ -515,6 +515,15 @@ def __init__(
         self._async_mode = async_mode
         self._parsing_start_time: Optional[int] = None
 
+        # Set the signal conn in to non-blocking mode, so that attempting to
+        # send when the buffer is full errors, rather than hangs for-ever
+        # attempting to send (this is to avoid deadlocks!)
+        #
+        # Don't do this in sync_mode, as we _need_ the DagParsingStat sent to
+        # continue the scheduler
+        if self._async_mode:
+            os.set_blocking(self._signal_conn.fileno(), False)
+
         self._parallelism = conf.getint('scheduler', 'parsing_processes')
         if 'sqlite' in conf.get('core', 'sql_alchemy_conn') and self._parallelism > 1:
             self.log.warning(
@@ -623,6 +632,7 @@ def _run_parsing_loop(self):
             ready = multiprocessing.connection.wait(self.waitables.keys(), timeout=poll_time)
             if self._signal_conn in ready:
                 agent_signal = self._signal_conn.recv()
+
                 self.log.debug("Received %s signal from DagFileProcessorAgent", agent_signal)
                 if agent_signal == DagParsingSignal.TERMINATE_MANAGER:
                     self.terminate()
@@ -696,11 +706,20 @@ def _run_parsing_loop(self):
             max_runs_reached = self.max_runs_reached()
 
             dag_parsing_stat = DagParsingStat(
-                self._file_paths,
+                len(self._file_paths),
                 max_runs_reached,
                 all_files_processed,
             )
-            self._signal_conn.send(dag_parsing_stat)
+            try:
+                self._signal_conn.send(dag_parsing_stat)
+            except BlockingIOError:
+                # Try again next time around the loop!
+
+                # It is better to fail, than it is deadlock. This should
+                # "almost never happen" since the DagParsingStat object is
+                # small, and in async mode this stat is not actually _required_
+                # for normal operation (It only drives "max runs")
+                self.log.debug("BlockingIOError recived trying to send DagParsingStat, ignoring")
 
             if max_runs_reached:
                 self.log.info(

pylintrc-tests

@@ -443,6 +443,7 @@ good-names=e,
            i,
            j,
            k,
+           n,
            v, # Commonly used when iterating dict.items()
            _,
            ti,  # Commonly used in Airflow as shorthand for taskinstance
@@ -454,7 +455,8 @@ good-names=e,
            cm,  # Commonly used as shorthand for context manager
            ds,  # Used in Airflow templates
            ts,  # Used in Airflow templates
-           id   # Commonly used as shorthand for identifier
+           id,  # Commonly used as shorthand for identifier
+           fd,  # aka "file-descriptor" -- common in socket code
 
 # Include a hint for the correct naming format with invalid-name.
 include-naming-hint=no

tests/utils/test_dag_processing.py

@@ -36,7 +36,7 @@
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.taskinstance import SimpleTaskInstance
 from airflow.utils import timezone
-from airflow.utils.callback_requests import TaskCallbackRequest
+from airflow.utils.callback_requests import CallbackRequest, TaskCallbackRequest
 from airflow.utils.dag_processing import (
     DagFileProcessorAgent,
     DagFileProcessorManager,
@@ -521,17 +521,97 @@ def test_dag_with_system_exit(self):
 
         manager._run_parsing_loop()
 
+        result = None
         while parent_pipe.poll(timeout=None):
             result = parent_pipe.recv()
             if isinstance(result, DagParsingStat) and result.done:
                 break
 
         # Three files in folder should be processed
-        assert len(result.file_paths) == 3
+        assert result.num_file_paths == 3
 
         with create_session() as session:
             assert session.query(DagModel).get(dag_id) is not None
 
+    @conf_vars({('core', 'load_examples'): 'False'})
+    @pytest.mark.backend("mysql", "postgres")
+    def test_pipe_full_deadlock(self):
+        import threading
+
+        dag_filepath = TEST_DAG_FOLDER / "test_scheduler_dags.py"
+
+        child_pipe, parent_pipe = multiprocessing.Pipe()
+
+        import socket
+
+        # Shrink the buffers to exacerbate the problem!
+        for fd in (parent_pipe.fileno(),):
+            sock = socket.socket(fileno=fd)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
+            sock.detach()
+
+        exit_event = threading.Event()
+
+        # To test this behaviour we need something that continually fills the
+        # parent pipe's bufffer (and keeps it full).
+        def keep_pipe_full(pipe, exit_event):
+            import logging
+
+            n = 0
+            while True:
+                if exit_event.is_set():
+                    break
+
+                req = CallbackRequest(str(dag_filepath))
+                try:
+                    logging.debug("Sending CallbackRequests %d", n + 1)
+                    pipe.send(req)
+                except TypeError:
+                    # This is actually the error you get when the parent pipe
+                    # is closed! Nicely handled, eh?
+                    break
+                except OSError:
+                    break
+                n += 1
+                logging.debug("   Sent %d CallbackRequests", n)
+
+        thread = threading.Thread(target=keep_pipe_full, args=(parent_pipe, exit_event))
+
+        fake_processors = []
+
+        def fake_processor_factory(*args, **kwargs):
+            nonlocal fake_processors
+            processor = FakeDagFileProcessorRunner._fake_dag_processor_factory(*args, **kwargs)
+            fake_processors.append(processor)
+            return processor
+
+        manager = DagFileProcessorManager(
+            dag_directory=dag_filepath,
+            dag_ids=[],
+            # A reasonable large number to ensure that we trigger the deadlock
+            max_runs=100,
+            processor_factory=fake_processor_factory,
+            processor_timeout=timedelta(seconds=5),
+            signal_conn=child_pipe,
+            pickle_dags=False,
+            async_mode=True,
+        )
+
+        try:
+            thread.start()
+
+            # If this completes without hanging, then the test is good!
+            manager._run_parsing_loop()
+            exit_event.set()
+        finally:
+            import logging
+
+            logging.info("Closing pipes")
+            parent_pipe.close()
+            child_pipe.close()
+            thread.join(timeout=1.0)
+
 
 class TestDagFileProcessorAgent(unittest.TestCase):
     def setUp(self):