Handle queue errors in streaming dataset reader

[awaelchli]

What does this PR do?

Suppresses sporadic errors when querying the queue:

Error 1:

Exception in thread Thread-5:
 Traceback (most recent call last):
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
     self.run()
 Exception in thread Thread-5:
 Traceback (most recent call last):
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
     self.run()
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 136, in run
     self._maybe_delete_chunks()
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 99, in _maybe_delete_chunks
     chunk_index = _get_from_queue(self._to_delete_queue)
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 309, in _get_from_queue
     raise e
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 301, in _get_from_queue
     return queue.get(timeout=0.1)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/queues.py", line 117, in get
     res = self._recv_bytes()
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 216, in recv_bytes
     buf = self._recv_bytes(maxlength)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 414, in _recv_bytes
     buf = self._recv(4)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 379, in _recv
     chunk = read(handle, remaining)
 OSError: [Errno 9] Bad file descriptor

Error 2:

Traceback (most recent call last):
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/threading.py", line 1016, in _bootstrap_inner
     self.run()
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 134, in run
     self._maybe_delete_chunks()
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 89, in _maybe_delete_chunks
     chunk_index = _get_from_queue(self._to_delete_queue, timeout=None if reached_pre_download else _DEFAULT_TIMEOUT)
   File "/teamspace/studios/this_studio/lightning/src/lightning/data/streaming/reader.py", line 298, in _get_from_queue
     return queue.get(timeout=timeout)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/queues.py", line 103, in get
     res = self._recv_bytes()
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 216, in recv_bytes
     buf = self._recv_bytes(maxlength)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 414, in _recv_bytes
     buf = self._recv(4)
   File "/home/zeus/miniconda3/envs/cloudspace/lib/python3.10/multiprocessing/connection.py", line 383, in _recv
     raise EOFError
 EOFError

Any clue why we get this error in the first place?

---

📚 Documentation preview 📚: https://pytorch-lightning--19167.org.readthedocs.build/en/19167/

cc @Borda

[github-actions]

⚡ Required checks status: All passing 🟢

Groups summary

🟢 lightning_data: CPU workflow

Check ID	Status
data-cpu (macOS-11, lightning, 3.10, 2.1)	success	✅
data-cpu (ubuntu-20.04, lightning, 3.10, 2.1)	success	✅
data-cpu (windows-2022, lightning, 3.10, 2.1)	success	✅

These checks are required after the changes to src/lightning/data/streaming/reader.py.

🟢 mypy

Check ID	Status
mypy	success	✅

These checks are required after the changes to src/lightning/data/streaming/reader.py.

🟢 install

Check ID	Status
install-pkg (ubuntu-22.04, app, 3.8)	success	✅
install-pkg (ubuntu-22.04, app, 3.11)	success	✅
install-pkg (ubuntu-22.04, fabric, 3.8)	success	✅
install-pkg (ubuntu-22.04, fabric, 3.11)	success	✅
install-pkg (ubuntu-22.04, pytorch, 3.8)	success	✅
install-pkg (ubuntu-22.04, pytorch, 3.11)	success	✅
install-pkg (ubuntu-22.04, lightning, 3.8)	success	✅
install-pkg (ubuntu-22.04, lightning, 3.11)	success	✅
install-pkg (ubuntu-22.04, notset, 3.8)	success	✅
install-pkg (ubuntu-22.04, notset, 3.11)	success	✅
install-pkg (macOS-12, app, 3.8)	success	✅
install-pkg (macOS-12, app, 3.11)	success	✅
install-pkg (macOS-12, fabric, 3.8)	success	✅
install-pkg (macOS-12, fabric, 3.11)	success	✅
install-pkg (macOS-12, pytorch, 3.8)	success	✅
install-pkg (macOS-12, pytorch, 3.11)	success	✅
install-pkg (macOS-12, lightning, 3.8)	success	✅
install-pkg (macOS-12, lightning, 3.11)	success	✅
install-pkg (macOS-12, notset, 3.8)	success	✅
install-pkg (macOS-12, notset, 3.11)	success	✅
install-pkg (windows-2022, app, 3.8)	success	✅
install-pkg (windows-2022, app, 3.11)	success	✅
install-pkg (windows-2022, fabric, 3.8)	success	✅
install-pkg (windows-2022, fabric, 3.11)	success	✅
install-pkg (windows-2022, pytorch, 3.8)	success	✅
install-pkg (windows-2022, pytorch, 3.11)	success	✅
install-pkg (windows-2022, lightning, 3.8)	success	✅
install-pkg (windows-2022, lightning, 3.11)	success	✅
install-pkg (windows-2022, notset, 3.8)	success	✅
install-pkg (windows-2022, notset, 3.11)	success	✅

These checks are required after the changes to src/lightning/data/streaming/reader.py.

---

Thank you for your contribution! 💜

Note
This comment is automatically generated and updates for 60 minutes every 180 seconds. If you have any other questions, contact carmocca for help.

[tchaton]

LGTM !