fix(hil-serl): drain queue on get_last_item_from_queue (#1524)

* fix(hil-serl): drain queue on get_last_item_from_queue

* parametrize queue tests

* revert changes for Darwin

* revert parametrize queue tests

* add test_get_last_item_multiple_items_with_torch_queue

* update test_get_last_item_multiple_items_with_torch_queue

* update test_get_last_item_multiple_items_with_torch_queue
This commit is contained in:
Juan Pizarro
2025-07-17 17:01:48 +02:00
committed by GitHub
parent 378e1f0338
commit 7e9f955b40
2 changed files with 33 additions and 4 deletions

View File

@@ -14,6 +14,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import platform
from contextlib import suppress
from queue import Empty
from typing import Any
@@ -30,10 +32,21 @@ def get_last_item_from_queue(queue: Queue, block=True, timeout: float = 0.1) ->
item = None
# Drain queue and keep only the most recent parameters
try:
while True:
if platform.system() == "Darwin":
# On Mac, avoid using `qsize` due to unreliable implementation.
# There is a comment on `qsize` code in the Python source:
# Raises NotImplementedError on Mac OSX because of broken sem_getvalue()
try:
while True:
item = queue.get_nowait()
except Empty:
pass
return item
# Details about using qsize in https://github.com/huggingface/lerobot/issues/1523
while queue.qsize() > 0:
with suppress(Empty):
item = queue.get_nowait()
except Empty:
pass
return item

View File

@@ -18,6 +18,8 @@ import threading
import time
from queue import Queue
from torch.multiprocessing import Queue as TorchMPQueue
from lerobot.utils.queue import get_last_item_from_queue
@@ -46,6 +48,20 @@ def test_get_last_item_multiple_items():
assert queue.empty()
def test_get_last_item_multiple_items_with_torch_queue():
"""Test getting the last item when queue has multiple items."""
queue = TorchMPQueue()
items = ["first", "second", "third", "fourth", "last"]
for item in items:
queue.put(item)
result = get_last_item_from_queue(queue)
assert result == "last"
assert queue.empty()
def test_get_last_item_different_types():
"""Test with different data types in the queue."""
queue = Queue()