Raise Empty when request_queue is None (#18111)

Co-authored-by: awaelchli <[email protected]>

(cherry picked from commit 2b854a8)

Author: lantiga

src/lightning/app/CHANGELOG.md

@@ -13,7 +13,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 #### Fixed
 
--
+- Fixed handling a `None` request in the file orchestration queue ([#18111](https://github.com/Lightning-AI/lightning/pull/18111))
 
 
 

src/lightning/app/storage/orchestrator.py

@@ -106,6 +106,11 @@ def run_once(self, work_name: str) -> None:
             request_queue = self.request_queues[work_name]
             try:
                 request: _PathRequest = request_queue.get(timeout=0)  # this should not block
+                # This should not happen under normal conditions, but it has occurred.
+                # For now we are tolerant with respect to requests being None in the queue
+                # and just move on.
+                if request is None:
+                    raise Empty
             except Empty:
                 pass
             else:

tests/tests_app/storage/test_orchestrator.py

@@ -47,6 +47,13 @@ def test_orchestrator():
     orchestrator.run_once("work_a")
     orchestrator.run_once("work_b")
 
+    # edge case: `None` requests on the queue get ignored
+    # TODO: Investigate how `None` values end up in the queue
+    request_queues["work_a"].put(None)
+    orchestrator.run_once("work_a")
+    orchestrator.run_once("work_b")
+    assert not request_queues["work_a"]._queue
+
     # simulate copier A confirms that the file is available on the shared volume
     response = _GetResponse(source="work_a", path="/a/b/c.txt", hash="", destination="work_b", name="")
     copy_request_queues["work_a"].get()