From e406f91a6e99c6d18f54dda5ea23f908ebcc515d Mon Sep 17 00:00:00 2001 From: Amir Fathi Date: Fri, 28 Aug 2026 18:40:51 +0000 Subject: [PATCH 1/2] BUG: recover_sub_pool no longer raises on concurrent create_actor recover_sub_pool iterated self._allocated_actors[address].values() directly while awaiting self.call() inside the loop. create_actor() mutates that same dict across an await too (a placeholder key set before its own call, popped after), so a create_actor() landing during recovery raised RuntimeError: dictionary changed size during iteration. Snapshot the values into a list before iterating, mirroring the same guard already used for self.sub_processes in monitor_sub_pools. Fixes #48 --- python/xoscar/backends/indigen/pool.py | 5 +- .../backends/indigen/tests/test_pool.py | 76 +++++++++++++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/python/xoscar/backends/indigen/pool.py b/python/xoscar/backends/indigen/pool.py index 9254bf68..ac34e73a 100644 --- a/python/xoscar/backends/indigen/pool.py +++ b/python/xoscar/backends/indigen/pool.py @@ -501,7 +501,10 @@ async def recover_sub_pool(self, address: str): if self._auto_recover == "actor": # need to recover all created actors - for _, message in self._allocated_actors[address].values(): + # copy to a list first: create_actor() mutates this same dict + # concurrently, and iterating it directly across the `await` below + # can raise "dictionary changed size during iteration" + for _, message in list(self._allocated_actors[address].values()): create_actor_message: CreateActorMessage = message # type: ignore await self.call(address, create_actor_message) diff --git a/python/xoscar/backends/indigen/tests/test_pool.py b/python/xoscar/backends/indigen/tests/test_pool.py index b1d13e72..fc8bd6da 100644 --- a/python/xoscar/backends/indigen/tests/test_pool.py +++ b/python/xoscar/backends/indigen/tests/test_pool.py @@ -16,11 +16,13 @@ from __future__ import annotations import asyncio +import contextlib import logging import multiprocessing import os import re import sys +import threading import time from unittest import mock @@ -59,6 +61,7 @@ ErrorMessage, HasActorMessage, MessageType, + ResultMessage, SendMessage, TellMessage, new_message_id, @@ -930,6 +933,79 @@ def on_process_recover(*_): await ctx.has_actor(actor_ref) +@pytest.mark.asyncio +async def test_recover_sub_pool_concurrent_create_actor(): + # GH-48: recover_sub_pool() iterated self._allocated_actors[address] + # directly across an await; a concurrent create_actor() mutating that + # same dict raised "dictionary changed size during iteration". + pool = object.__new__(MainActorPool) + pool.external_address = "dummy://main" + pool._config = mock.Mock(get_process_index=mock.Mock(return_value=0)) + pool.sub_processes = {} + pool._auto_recover = "actor" + pool._allocation_lock = threading.Lock() + pool.start_sub_pool = mock.AsyncMock(return_value=None) + pool.wait_sub_pools_ready = mock.AsyncMock(return_value=(["proc"], ["addr"])) + + sub_address = "dummy://sub" + existing_message = CreateActorMessage( + new_message_id(), + TestActor, + b"existing", + (), + {}, + allocate_strategy=AddressSpecified(sub_address), + ) + pool._allocated_actors = { + sub_address: {b"existing": (AddressSpecified(sub_address), existing_message)} + } + + reached_recover_call = asyncio.Event() + reached_create_call = asyncio.Event() + release_recover_call = asyncio.Event() + call_count = 0 + + async def fake_call(address, message): + nonlocal call_count + call_count += 1 + if call_count == 1: + # recover_sub_pool()'s call for the pre-existing actor + reached_recover_call.set() + await release_recover_call.wait() + else: + # the racing create_actor() call; hang so its placeholder + # entry stays in the dict while recover_sub_pool resumes + reached_create_call.set() + await asyncio.Event().wait() + return ResultMessage(new_message_id(), b"actor_ref") + + pool.call = fake_call + + recover_task = asyncio.create_task(pool.recover_sub_pool(sub_address)) + await asyncio.wait_for(reached_recover_call.wait(), timeout=5) + + new_message = CreateActorMessage( + new_message_id(), + TestActor, + b"new", + (), + {}, + allocate_strategy=AddressSpecified(sub_address), + ) + create_task_ = asyncio.create_task(pool.create_actor(new_message)) + await asyncio.wait_for(reached_create_call.wait(), timeout=5) + # create_actor() has already inserted its placeholder into the same + # dict recover_sub_pool() is mid-iteration over + assert len(pool._allocated_actors[sub_address]) == 2 + + release_recover_call.set() + await asyncio.wait_for(recover_task, timeout=5) + + create_task_.cancel() + with contextlib.suppress(asyncio.CancelledError): + await create_task_ + + @pytest.mark.parametrize( "exception_config", [ From c8bf23836a69aa8f85eeece77c1218595a2bd2af Mon Sep 17 00:00:00 2001 From: Amir Fathi Date: Sun, 30 Aug 2026 06:31:30 +0000 Subject: [PATCH 2/2] BUG: skip in-flight placeholder during recover_sub_pool replay create_actor() inserts a placeholder under key None before awaiting the sub pool create and only pops it once that call returns. If the sub pool dies mid create, the placeholder is left in the dict holding the caller's original, unfinished message. recover_sub_pool()'s replay loop was iterating that entry too, resending an unfinished, from_main=False CreateActorMessage straight to the sub pool and duplicating the pending create. Skip entries keyed by None when replaying, and add a test that pins this behavior. --- python/xoscar/backends/indigen/pool.py | 16 ++++- .../backends/indigen/tests/test_pool.py | 59 +++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/python/xoscar/backends/indigen/pool.py b/python/xoscar/backends/indigen/pool.py index ac34e73a..ac9ec9ab 100644 --- a/python/xoscar/backends/indigen/pool.py +++ b/python/xoscar/backends/indigen/pool.py @@ -504,7 +504,21 @@ async def recover_sub_pool(self, address: str): # copy to a list first: create_actor() mutates this same dict # concurrently, and iterating it directly across the `await` below # can raise "dictionary changed size during iteration" - for _, message in list(self._allocated_actors[address].values()): + # + # skip in-flight placeholder entries (keyed by None): + # MainActorPoolBase.create_actor() inserts one before awaiting + # the sub pool create and only pops it once that call returns, + # so a placeholder can still be sitting in the dict here if the + # sub pool died mid create. Replaying it would resend the + # caller's original, unfinished CreateActorMessage + # (from_main=False) straight to the sub pool, which duplicates + # the still-pending create and can raise ActorAlreadyExist or + # register the actor a second time via notify_main_pool_to_create. + for actor_ref, (_, message) in list( + self._allocated_actors[address].items() + ): + if actor_ref is None: + continue create_actor_message: CreateActorMessage = message # type: ignore await self.call(address, create_actor_message) diff --git a/python/xoscar/backends/indigen/tests/test_pool.py b/python/xoscar/backends/indigen/tests/test_pool.py index fc8bd6da..de643d81 100644 --- a/python/xoscar/backends/indigen/tests/test_pool.py +++ b/python/xoscar/backends/indigen/tests/test_pool.py @@ -1006,6 +1006,65 @@ async def fake_call(address, message): await create_task_ +@pytest.mark.asyncio +async def test_recover_sub_pool_skips_in_flight_placeholder(): + # GH-48 review follow-up: create_actor() inserts a placeholder under + # key None before awaiting the sub pool create and only removes it + # once that call returns; if the sub pool dies mid create, the + # placeholder is left behind holding the caller's original, + # unfinished CreateActorMessage. recover_sub_pool() must skip that + # placeholder rather than replay it, since replaying resends an + # unfinished create straight to the sub pool. + pool = object.__new__(MainActorPool) + pool.external_address = "dummy://main" + pool._config = mock.Mock(get_process_index=mock.Mock(return_value=0)) + pool.sub_processes = {} + pool._auto_recover = "actor" + pool._allocation_lock = threading.Lock() + pool.start_sub_pool = mock.AsyncMock(return_value=None) + pool.wait_sub_pools_ready = mock.AsyncMock(return_value=(["proc"], ["addr"])) + + sub_address = "dummy://sub" + existing_message = CreateActorMessage( + new_message_id(), + TestActor, + b"existing", + (), + {}, + allocate_strategy=AddressSpecified(sub_address), + ) + # a still in-flight create_actor() call left its placeholder (key + # None) behind because the sub pool died before it could be popped + placeholder_message = CreateActorMessage( + new_message_id(), + TestActor, + b"in-flight", + (), + {}, + allocate_strategy=AddressSpecified(sub_address), + ) + pool._allocated_actors = { + sub_address: { + b"existing": (AddressSpecified(sub_address), existing_message), + None: (AddressSpecified(sub_address), placeholder_message), + } + } + + replayed_actor_ids = [] + + async def fake_call(address, message): + replayed_actor_ids.append(message.actor_id) + return ResultMessage(new_message_id(), b"actor_ref") + + pool.call = fake_call + + await pool.recover_sub_pool(sub_address) + + # only the real, completed entry is replayed; the placeholder is + # skipped, never resent to the sub pool + assert replayed_actor_ids == [b"existing"] + + @pytest.mark.parametrize( "exception_config", [