diff --git a/python/xoscar/backends/indigen/pool.py b/python/xoscar/backends/indigen/pool.py index 9254bf68..ac9ec9ab 100644 --- a/python/xoscar/backends/indigen/pool.py +++ b/python/xoscar/backends/indigen/pool.py @@ -501,7 +501,24 @@ 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" + # + # 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 b1d13e72..de643d81 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,138 @@ 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.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", [