Don't try to close or join mp manager queues (#18866)

Multiprocessing Manager queues don't have a close() or join_thread() method, and the Manager will clean it up appropriately after we empty it. This prevents an infinite loop when an AttributeError exception fires for Manager AutoProxy queue objects.
This commit is contained in:
Josh Hawkins 2025-06-24 16:19:09 -05:00 committed by GitHub
parent 47a0097e95
commit 32f82a947d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -5,7 +5,7 @@ import copy
import datetime
import logging
import math
import multiprocessing as mp
import multiprocessing.queues
import queue
import re
import shlex
@ -338,16 +338,23 @@ def clear_and_unlink(file: Path, missing_ok: bool = True) -> None:
file.unlink(missing_ok=missing_ok)
def empty_and_close_queue(q: mp.Queue):
def empty_and_close_queue(q):
while True:
try:
try:
q.get(block=True, timeout=0.5)
except (queue.Empty, EOFError):
q.close()
q.join_thread()
return
except AttributeError:
q.get(block=True, timeout=0.5)
except (queue.Empty, EOFError):
break
except Exception as e:
logger.debug(f"Error while emptying queue: {e}")
break
# close the queue if it is a multiprocessing queue
# manager proxy queues do not have close or join_thread method
if isinstance(q, multiprocessing.queues.Queue):
try:
q.close()
q.join_thread()
except Exception:
pass