mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-10-26 06:04:11 +00:00
fix(core): finish application cleanup before propagating cancellation
This commit is contained in:
parent
638c81e88e
commit
eb5c2c9899
2 changed files with 63 additions and 1 deletions
|
|
@ -227,15 +227,21 @@ class Application(BaseComponent):
|
|||
|
||||
async def _close_started_components(self) -> None:
|
||||
"""Close resources while the caller serializes component mutations."""
|
||||
cancelled_error: asyncio.CancelledError | None = None
|
||||
for c in reversed(self._started_components):
|
||||
try:
|
||||
await c.close()
|
||||
except asyncio.CancelledError as exc:
|
||||
if cancelled_error is None:
|
||||
cancelled_error = exc
|
||||
except Exception as e:
|
||||
self.logger.exception(f"Failed to close {component_type_name(c.component_type)}:{c.name}: {e}")
|
||||
self._started_components.clear()
|
||||
if self.context.thread_pool is not None:
|
||||
self.context.thread_pool.shutdown(wait=True)
|
||||
self.context.thread_pool = None
|
||||
if cancelled_error is not None:
|
||||
raise cancelled_error
|
||||
|
||||
async def update_component(self, component_enum: ComponentType, name: str, /, **kwargs) -> BaseComponent:
|
||||
"""Update an existing component by type/name; never creates missing components."""
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ import pytest
|
|||
from reme.enumeration import ComponentEnum
|
||||
from reme.application import Application
|
||||
from reme.components.base_component import BaseComponent
|
||||
from reme.components.component_registry import ComponentRegistry
|
||||
from reme.components.component_registry import ComponentRegistry, R
|
||||
from reme.components.job.background_job import BackgroundJob
|
||||
from reme.components.job.base_job import BaseJob
|
||||
from reme.components.job.cron_job import CronJob
|
||||
|
|
@ -489,3 +489,59 @@ def test_application_start_failure_propagates_and_closes_started_components():
|
|||
assert not app._started_components
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_cancelled_application_close_finishes_remaining_cleanup(tmp_path):
|
||||
"""Cancellation still closes other components and shuts down the owned pool."""
|
||||
|
||||
async def run():
|
||||
closing = asyncio.Event()
|
||||
closed = []
|
||||
|
||||
class ShutdownComponent(BaseComponent):
|
||||
component_type = ComponentEnum.FILE_CHUNKER
|
||||
|
||||
async def _close(self):
|
||||
try:
|
||||
if self.name == "b":
|
||||
closing.set()
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
closed.append(self.name)
|
||||
|
||||
with R.preserve(allow_mutation=True):
|
||||
R.register(ShutdownComponent, "test_shutdown")
|
||||
app = Application(
|
||||
workspace_dir=str(tmp_path),
|
||||
enable_logo=False,
|
||||
log_to_console=False,
|
||||
log_to_file=False,
|
||||
thread_pool_max_workers=1,
|
||||
service={"backend": "cli"},
|
||||
components={"file_chunker": {name: {"backend": "test_shutdown"} for name in ("a", "b")}},
|
||||
)
|
||||
|
||||
await app.start()
|
||||
pool = app.context.thread_pool
|
||||
assert pool is not None
|
||||
close_task = asyncio.create_task(app.close())
|
||||
try:
|
||||
await asyncio.wait_for(closing.wait(), timeout=5)
|
||||
close_task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await close_task
|
||||
|
||||
assert closed == ["b", "a"]
|
||||
assert all(not component.is_started for component in app.context.components["file_chunker"].values())
|
||||
assert not app.is_started
|
||||
assert app.context.thread_pool is None
|
||||
with pytest.raises(RuntimeError, match="cannot schedule new futures after shutdown"):
|
||||
pool.submit(lambda: None)
|
||||
finally:
|
||||
close_task.cancel()
|
||||
await asyncio.gather(close_task, return_exceptions=True)
|
||||
for component in app.context.components["file_chunker"].values():
|
||||
await component.close()
|
||||
pool.shutdown(wait=True)
|
||||
|
||||
asyncio.run(run())
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue