[Bugfix] Add missing auto_create_handle_loop to communicator methods (#19610)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
0b3ddbcf10
commit
98224de29b
@@ -344,10 +344,12 @@ class TokenizerCommunicatorMixin:
|
|||||||
)
|
)
|
||||||
|
|
||||||
async def flush_cache(self: TokenizerManager) -> FlushCacheReqOutput:
|
async def flush_cache(self: TokenizerManager) -> FlushCacheReqOutput:
|
||||||
|
self.auto_create_handle_loop()
|
||||||
return (await self.flush_cache_communicator(FlushCacheReqInput()))[0]
|
return (await self.flush_cache_communicator(FlushCacheReqInput()))[0]
|
||||||
|
|
||||||
async def clear_hicache_storage(self: TokenizerManager) -> ClearHiCacheReqOutput:
|
async def clear_hicache_storage(self: TokenizerManager) -> ClearHiCacheReqOutput:
|
||||||
"""Clear the hierarchical cache storage."""
|
"""Clear the hierarchical cache storage."""
|
||||||
|
self.auto_create_handle_loop()
|
||||||
# Delegate to the scheduler to handle HiCacheStorage clearing
|
# Delegate to the scheduler to handle HiCacheStorage clearing
|
||||||
return (await self.clear_hicache_storage_communicator(ClearHiCacheReqInput()))[
|
return (await self.clear_hicache_storage_communicator(ClearHiCacheReqInput()))[
|
||||||
0
|
0
|
||||||
@@ -361,6 +363,7 @@ class TokenizerCommunicatorMixin:
|
|||||||
hicache_write_policy: Optional[str] = None,
|
hicache_write_policy: Optional[str] = None,
|
||||||
) -> AttachHiCacheStorageReqOutput:
|
) -> AttachHiCacheStorageReqOutput:
|
||||||
"""Attach (enable) HiCache storage backend at runtime."""
|
"""Attach (enable) HiCache storage backend at runtime."""
|
||||||
|
self.auto_create_handle_loop()
|
||||||
results = await self.attach_hicache_storage_communicator(
|
results = await self.attach_hicache_storage_communicator(
|
||||||
AttachHiCacheStorageReqInput(
|
AttachHiCacheStorageReqInput(
|
||||||
hicache_storage_backend=hicache_storage_backend,
|
hicache_storage_backend=hicache_storage_backend,
|
||||||
@@ -392,6 +395,7 @@ class TokenizerCommunicatorMixin:
|
|||||||
self: TokenizerManager,
|
self: TokenizerManager,
|
||||||
) -> DetachHiCacheStorageReqOutput:
|
) -> DetachHiCacheStorageReqOutput:
|
||||||
"""Detach (disable) HiCache storage backend at runtime."""
|
"""Detach (disable) HiCache storage backend at runtime."""
|
||||||
|
self.auto_create_handle_loop()
|
||||||
results = await self.detach_hicache_storage_communicator(
|
results = await self.detach_hicache_storage_communicator(
|
||||||
DetachHiCacheStorageReqInput()
|
DetachHiCacheStorageReqInput()
|
||||||
)
|
)
|
||||||
@@ -855,6 +859,7 @@ class TokenizerCommunicatorMixin:
|
|||||||
await self.slow_down_communicator(obj)
|
await self.slow_down_communicator(obj)
|
||||||
|
|
||||||
async def get_internal_state(self: TokenizerManager) -> List[Dict[Any, Any]]:
|
async def get_internal_state(self: TokenizerManager) -> List[Dict[Any, Any]]:
|
||||||
|
self.auto_create_handle_loop()
|
||||||
req = GetInternalStateReq()
|
req = GetInternalStateReq()
|
||||||
responses: List[GetInternalStateReqOutput] = (
|
responses: List[GetInternalStateReqOutput] = (
|
||||||
await self.get_internal_state_communicator(req)
|
await self.get_internal_state_communicator(req)
|
||||||
@@ -865,6 +870,7 @@ class TokenizerCommunicatorMixin:
|
|||||||
async def set_internal_state(
|
async def set_internal_state(
|
||||||
self: TokenizerManager, obj: SetInternalStateReq
|
self: TokenizerManager, obj: SetInternalStateReq
|
||||||
) -> List[bool]:
|
) -> List[bool]:
|
||||||
|
self.auto_create_handle_loop()
|
||||||
responses: List[SetInternalStateReqOutput] = (
|
responses: List[SetInternalStateReqOutput] = (
|
||||||
await self.set_internal_state_communicator(obj)
|
await self.set_internal_state_communicator(obj)
|
||||||
)
|
)
|
||||||
@@ -873,9 +879,11 @@ class TokenizerCommunicatorMixin:
|
|||||||
async def dumper_control(
|
async def dumper_control(
|
||||||
self: TokenizerManager, obj: DumperControlReqInput
|
self: TokenizerManager, obj: DumperControlReqInput
|
||||||
) -> List[DumperControlReqOutput]:
|
) -> List[DumperControlReqOutput]:
|
||||||
|
self.auto_create_handle_loop()
|
||||||
return await self.dumper_control_communicator(obj)
|
return await self.dumper_control_communicator(obj)
|
||||||
|
|
||||||
async def get_load(self: TokenizerManager) -> List[GetLoadReqOutput]:
|
async def get_load(self: TokenizerManager) -> List[GetLoadReqOutput]:
|
||||||
|
self.auto_create_handle_loop()
|
||||||
req = GetLoadReqInput()
|
req = GetLoadReqInput()
|
||||||
return await self.get_load_communicator(req)
|
return await self.get_load_communicator(req)
|
||||||
|
|
||||||
@@ -894,6 +902,7 @@ class TokenizerCommunicatorMixin:
|
|||||||
Returns:
|
Returns:
|
||||||
List of GetLoadsReqOutput, one per scheduler (filtered by dp_rank if specified)
|
List of GetLoadsReqOutput, one per scheduler (filtered by dp_rank if specified)
|
||||||
"""
|
"""
|
||||||
|
self.auto_create_handle_loop()
|
||||||
req = GetLoadsReqInput(
|
req = GetLoadsReqInput(
|
||||||
include=include if include else ["all"],
|
include=include if include else ["all"],
|
||||||
dp_rank=dp_rank,
|
dp_rank=dp_rank,
|
||||||
|
|||||||
Reference in New Issue
Block a user