update async

Author: Yunnglin

src/twinkle/server/model/app.py

@@ -14,7 +14,7 @@
 
 import twinkle
 from twinkle import DeviceGroup, DeviceMesh
-from twinkle.server.utils.adapter_manager import AdapterManagerMixin
+from twinkle.server.utils.lifecycle import AdapterManagerMixin
 from twinkle.server.utils.state import ServerStateProxy, get_server_state
 from twinkle.server.utils.task_queue import TaskQueueConfig, TaskQueueMixin
 from twinkle.server.utils.validation import get_token_from_request, verify_request_token
@@ -83,7 +83,7 @@ def __init__(self,
         # Initialize mixins
         self._init_task_queue(TaskQueueConfig.from_dict(queue_config))
         self._init_adapter_manager(**adapter_config)
-        self.start_adapter_countdown()
+        # Note: countdown task is started lazily in _ensure_sticky()
 
     async def _ensure_replica_registered(self):
         """Lazily register replica on first async request."""
@@ -98,6 +98,8 @@ async def _sticky_entry(self, sticky_key: str):
     async def _ensure_sticky(self):
         sticky_key = serve.get_multiplexed_model_id()
         await self._sticky_entry(sticky_key)
+        # Lazy-start countdown task on first request (requires running event loop)
+        self._ensure_countdown_started()
 
     async def _on_request_start(self, request: Request) -> str:
         await self._ensure_sticky()

src/twinkle/server/processor/app.py

@@ -20,7 +20,7 @@
 
 import twinkle
 from twinkle import DeviceGroup, DeviceMesh, get_logger
-from twinkle.server.utils.processor_manager import ProcessorManagerMixin
+from twinkle.server.utils.lifecycle import ProcessorManagerMixin
 from twinkle.server.utils.state import ServerStateProxy, get_server_state
 from twinkle.server.utils.validation import verify_request_token
 from .twinkle_handlers import _register_processor_routes
@@ -69,7 +69,7 @@ def __init__(self,
             processor_timeout=float(_cfg.get('processor_timeout', 1800.0)),
             per_token_processor_limit=int(_cfg.get('per_token_processor_limit', _env_limit)),
         )
-        self.start_processor_countdown()
+        # Note: countdown task is started lazily in _ensure_sticky()
 
     @serve.multiplexed(max_num_models_per_replica=100)
     async def _sticky_entry(self, sticky_key: str):
@@ -78,6 +78,8 @@ async def _sticky_entry(self, sticky_key: str):
     async def _ensure_sticky(self):
         sticky_key = serve.get_multiplexed_model_id()
         await self._sticky_entry(sticky_key)
+        # Lazy-start countdown task on first request (requires running event loop)
+        self._ensure_countdown_started()
 
     def _on_processor_expired(self, processor_id: str) -> None:
         """Called by the countdown thread when a processor's session expires."""

src/twinkle/server/sampler/app.py

@@ -13,7 +13,6 @@
 
 import twinkle
 from twinkle import DeviceGroup, DeviceMesh
-from twinkle.server.utils.adapter_manager import AdapterManagerMixin
 from twinkle.server.utils.state import ServerStateProxy, get_server_state
 from twinkle.server.utils.task_queue import TaskQueueConfig, TaskQueueMixin
 from twinkle.server.utils.validation import get_token_from_request, verify_request_token
@@ -25,14 +24,13 @@
 logger = get_logger()
 
 
-class SamplerManagement(TaskQueueMixin, AdapterManagerMixin):
+class SamplerManagement(TaskQueueMixin):
     """Unified sampler management service.
 
     Manages:
     - vLLM or Torch sampler initialization and lifecycle
     - Tinker inference requests (/tinker/asample) with rate limiting via TaskQueueMixin
     - Twinkle inference requests (/twinkle/*) calling sampler directly
-    - Adapter lifecycle via AdapterManagerMixin
     - Template configuration for trajectory encoding
     """
 
@@ -43,7 +41,6 @@ def __init__(self,
                  device_mesh: dict[str, Any],
                  sampler_type: str = 'vllm',
                  engine_args: dict[str, Any] | None = None,
-                 adapter_config: dict[str, Any] | None = None,
                  queue_config: dict[str, Any] | None = None,
                  **kwargs):
         self.device_group = DeviceGroup(**device_group)
@@ -82,11 +79,8 @@ def __init__(self,
         self.sampler.set_template('Template', model_id=model_id)
         self.state: ServerStateProxy = get_server_state()
 
-        # Initialize both mixins
+        # Initialize task queue mixin
         self._init_task_queue(TaskQueueConfig.from_dict(queue_config))
-        _adapter_config = adapter_config or {}
-        self._init_adapter_manager(**_adapter_config)
-        self.start_adapter_countdown()
 
     @serve.multiplexed(max_num_models_per_replica=5)
     async def _sticky_entry(self, sticky_key: str):
@@ -101,14 +95,6 @@ async def _on_request_start(self, request: Request) -> str:
         token = get_token_from_request(request)
         return token
 
-    async def _on_adapter_expired(self, adapter_name: str, token: str = None) -> None:
-        """Handle expired adapters by removing them from the sampler."""
-        try:
-            self.sampler.remove_adapter(adapter_name)
-            logger.info(f'Removed expired adapter {adapter_name}')
-        except Exception as e:
-            logger.warning(f'Failed to remove expired adapter {adapter_name}: {e}')
-
 
 def build_sampler_app(model_id: str,
                       nproc_per_node: int,
@@ -117,7 +103,6 @@ def build_sampler_app(model_id: str,
                       deploy_options: dict[str, Any],
                       sampler_type: str = 'vllm',
                       engine_args: dict[str, Any] | None = None,
-                      adapter_config: dict[str, Any] | None = None,
                       queue_config: dict[str, Any] | None = None,
                       **kwargs):
     """Build a unified sampler application for text generation inference.
@@ -133,7 +118,6 @@ def build_sampler_app(model_id: str,
         deploy_options: Ray Serve deployment options
         sampler_type: Type of sampler to use ('vllm' or 'torch')
         engine_args: Additional engine arguments for the sampler
-        adapter_config: Adapter lifecycle config (timeout, per-token limits)
         queue_config: Task queue configuration dict (rps_limit, tps_limit, etc.)
         **kwargs: Additional arguments passed to the sampler
 
@@ -161,8 +145,7 @@ def get_self() -> SamplerManagement:
     SamplerManagementWithIngress = serve.ingress(app)(SamplerManagement)
     DeploymentClass = serve.deployment(name='SamplerManagement')(SamplerManagementWithIngress)
     return DeploymentClass.options(**deploy_options).bind(model_id, nproc_per_node, device_group, device_mesh,
-                                                          sampler_type, engine_args, adapter_config, queue_config,
-                                                          **kwargs)
+                                                          sampler_type, engine_args, queue_config, **kwargs)
 
 
 build_sampler_app = wrap_builder_with_device_group_env(build_sampler_app)

src/twinkle/server/sampler/twinkle_handlers.py

@@ -154,13 +154,10 @@ def add_adapter_to_sampler(
         """Add a LoRA adapter to the sampler."""
         assert body.adapter_name, 'You need to specify a valid `adapter_name`'
         full_adapter_name = _get_twinkle_sampler_adapter_name(request, body.adapter_name)
-        from twinkle.server.utils.validation import get_token_from_request
-        token = get_token_from_request(request)
 
         from peft import LoraConfig
         config = LoraConfig(**body.config) if isinstance(body.config, dict) else body.config
 
-        self.register_adapter(full_adapter_name, token)
         self.sampler.add_adapter_to_sampler(full_adapter_name, config)
 
         return types.AddAdapterResponse(adapter_name=full_adapter_name)

src/twinkle/server/utils/__init__.py

@@ -1,8 +1,7 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
-from .adapter_manager import AdapterManagerMixin
 from .checkpoint_base import (TRAIN_RUN_INFO_FILENAME, TWINKLE_DEFAULT_SAVE_DIR, BaseCheckpointManager, BaseFileManager,
                               BaseTrainingRunManager)
 from .device_utils import auto_fill_device_group_visible_devices, wrap_builder_with_device_group_env
-from .processor_manager import ProcessorManagerMixin
+from .lifecycle import AdapterManagerMixin, ProcessorManagerMixin, SessionResourceMixin
 from .rate_limiter import RateLimiter
 from .task_queue import QueueState, TaskQueueConfig, TaskQueueMixin, TaskStatus