Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions python/sglang/srt/entrypoints/http_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -1558,6 +1558,7 @@ async def load_lora_adapter(


@app.api_route("/load_lora_adapter_from_tensors", methods=["POST"])
@auth_level(AuthLevel.ADMIN_OPTIONAL)
async def load_lora_adapter_from_tensors(
obj: Annotated[LoadLoRAAdapterFromTensorsReqInput, Body()], request: Request
):
Expand All @@ -1570,6 +1571,7 @@ async def load_lora_adapter_from_tensors(


@app.api_route("/load_lora_adapter_from_distributed", methods=["POST"])
@auth_level(AuthLevel.ADMIN_OPTIONAL)
async def load_lora_adapter_from_distributed(
obj: Annotated[LoadLoRAAdapterFromDistributedReqInput, Body()], request: Request
):
Expand Down
47 changes: 29 additions & 18 deletions python/sglang/srt/lora/lora_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@ class LoRARef(msgspec.Struct, frozen=True, array_like=True):
lora_name: Optional[str] = None
lora_path: Optional[str] = None
pinned: Optional[bool] = None
# False for adapters whose weights arrived over the wire (lora_path
# "__distributed__" / "__tensor__"): there is no disk artifact to reload
# from, so they must never be LRU-evicted nor implicitly reloaded.
# Trailing field with a default keeps the array_like wire format
# compatible with refs encoded before this field existed.
reloadable: bool = True

def __post_init__(self):
if self.lora_id is None:
Expand Down Expand Up @@ -184,24 +190,28 @@ def _lookup(name: str) -> str:
self._registry.move_to_end(name)
return lora_ref.lora_id

# Lookup and increment must be one atomic admission step under the
# writer lock: with the increment outside, an unload can slip between
# them — unregister the adapter, observe the counter at zero, delete
# the counter — leaving this request to KeyError or to run on an
# adapter that is already gone. wait_for_unload never awaits inside
# this lock, so holding it across the increment cannot deadlock.
if isinstance(lora_name, str):
async with self._registry_lock.writer_lock:
lora_id = _lookup(lora_name)

await self._counters[lora_id].increment(notify_all=False)
await self._counters[lora_id].increment(notify_all=False)
return lora_id
elif isinstance(lora_name, list):
async with self._registry_lock.writer_lock:
lora_ids = [_lookup(name) for name in lora_name]

# Increment the counters only after all IDs are looked up.
await asyncio.gather(
*[
self._counters[id].increment(notify_all=False)
for id in lora_ids
if id is not None
]
)
# Increment the counters only after all IDs are looked up.
await asyncio.gather(
*[
self._counters[id].increment(notify_all=False)
for id in lora_ids
if id is not None
]
)
return lora_ids
else:
raise TypeError("lora_name must be either a string or a list of strings.")
Expand Down Expand Up @@ -266,14 +276,15 @@ async def lru_lora_name(self, exclude_pinned=False):
If exclude_pinned is True, then return the LRU LoRA adapter that isn't pinned.
"""
async with self._registry_lock.reader_lock:
if not exclude_pinned:
return next(iter(self._registry), None)

for lora_name, lora_ref in self._registry.items():
if not lora_ref.pinned:
return lora_name
else:
return None
# Evicting a non-reloadable (wire-loaded) adapter is always
# destructive: there is no artifact to reload it from.
if not lora_ref.reloadable:
continue
if exclude_pinned and lora_ref.pinned:
continue
return lora_name
return None

def _register_adapter(self, lora_ref: LoRARef):
"""
Expand Down
11 changes: 10 additions & 1 deletion python/sglang/srt/managers/tokenizer_control_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -735,6 +735,7 @@ async def load_lora_adapter_from_tensors(
lora_name=obj.lora_name,
lora_path="__tensor__",
pinned=obj.pinned,
reloadable=False,
)
obj.lora_id = new_adapter.lora_id
result = (await self.update_lora_adapter_communicator(obj))[0]
Expand Down Expand Up @@ -811,6 +812,7 @@ async def load_lora_adapter_from_distributed(
lora_name=obj.lora_name,
lora_path="__distributed__",
pinned=obj.pinned,
reloadable=False,
),
upsert=obj.upsert,
)
Expand Down Expand Up @@ -886,7 +888,14 @@ async def unload_lora_adapter(
)

async with self.lora_update_lock:
return await self._unload_lora_adapter_locked(obj)
result = await self._unload_lora_adapter_locked(obj)
# Explicit unload is a DELETE: drop the reload-catalog entry too.
# The max_loaded_loras LRU loop calls _unload_lora_adapter_locked
# directly — an EVICT — and must keep the entry so disk-backed
# adapters can be implicitly reloaded later.
if result.success:
self.lora_ref_cache.pop(obj.lora_name, None)
return result
except ValueError as e:
return UnloadLoRAAdapterReqOutput(success=False, error_message=str(e))

Expand Down
56 changes: 49 additions & 7 deletions python/sglang/srt/managers/tokenizer_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,13 @@ class ReqState:
# request as aborted instead of sending it.
abort_before_dispatch: bool = False

# The LoRA lease for this request has already been released. Every terminal
# path funnels through TokenizerManager._finalize_lora_lease, which uses
# this flag to stay idempotent: a leaked release hangs unload's
# wait_for_zero forever, and a double release drives the ConcurrentCounter
# negative, which hangs it just the same (it waits for exactly zero).
lora_lease_released: bool = False

# For streaming output
last_output_offset: int = 0

Expand Down Expand Up @@ -1473,6 +1480,25 @@ def _coalesce_streaming_chunks(
out["meta_info"] = meta_info
return out

def _finalize_lora_lease(self, state: Optional[ReqState]) -> None:
"""Release the request's LoRA lease exactly once, however it terminates:
normal finish, scheduler abort echo (queued / tokenizer-held / disagg),
status-code abort, or a failed dispatch. A request whose LoRA was never
acquired (pre-acquire failure leaves lora_id unset) has no lease to
release."""
if state is None or state.lora_lease_released:
return
# lora_path / lora_id are declared on both input structs with default
# None (= base model / lease never acquired), so plain None checks are
# the contract. State-derived checks come first: a request without a
# lease has nothing to release, whatever the server config says.
if state.obj.lora_path is None or state.obj.lora_id is None:
return
if not self.enable_lora:
return
state.lora_lease_released = True
asyncio.create_task(self.lora_registry.release(state.obj.lora_id))

async def _handle_abort_finish_reason(
self,
out: dict,
Expand Down Expand Up @@ -1505,9 +1531,10 @@ async def _handle_abort_finish_reason(
if state.obj.rid in self.rid_to_state:
del self.rid_to_state[state.obj.rid]

# Mark ongoing LoRA request as finished.
if self.enable_lora and state.obj.lora_path:
await self.lora_registry.release(state.obj.lora_id)
# Status-code aborts also arrive through _handle_batch_output, which
# already released the lease and deleted the state — the finalizer's
# idempotency is what prevents the counter from going negative here.
self._finalize_lora_lease(state)
if not is_stream:
raise fastapi.HTTPException(
status_code=finish_reason["status_code"],
Expand Down Expand Up @@ -2220,8 +2247,7 @@ async def _handle_batch_output(
del self.rid_to_state[rid]

# Mark ongoing LoRA request as finished.
if self.enable_lora and state.obj.lora_path:
asyncio.create_task(self.lora_registry.release(state.obj.lora_id))
self._finalize_lora_lease(state)

if out_dict is not None:
state.out_list.append(out_dict)
Expand Down Expand Up @@ -2879,6 +2905,10 @@ def _handle_abort_req(self, recv_obj: AbortReq):
"output_ids": output_ids,
"meta_info": meta_info,
}
# This abort echo is the scheduler-side terminal ACK for queued,
# tokenizer-held, and disagg-retracted requests — none of them reach
# _handle_batch_output, so this is their only lease-release point.
self._finalize_lora_lease(state)
del self.rid_to_state[recv_obj.rid]

state.out_list.append(out)
Expand Down Expand Up @@ -3010,8 +3040,15 @@ async def _resolve_lora_path(self, obj: Union[GenerateReqInput, EmbeddingReqInpu
f"All loaded adapters: {self.lora_ref_cache.keys()}."
)

logger.info(f"Reloading evicted adapter: {lora_path}")
new_lora_ref = self.lora_ref_cache[lora_path]
if not new_lora_ref.reloadable:
raise ValueError(
f"LoRA adapter '{lora_path}' was loaded over the wire "
f"(lora_path={new_lora_ref.lora_path!r}) and has no disk "
"artifact to reload from; it must be re-pushed by the "
"trainer instead of implicitly reloaded."
)
logger.info(f"Reloading evicted adapter: {lora_path}")
load_result = await self.load_lora_adapter(
LoadLoRAAdapterReqInput(
lora_name=new_lora_ref.lora_name,
Expand Down Expand Up @@ -3092,7 +3129,12 @@ def _discard_pending_req_states(self, obj):
else:
rids = obj.rid
for rid in rids:
self.rid_to_state.pop(rid, None)
state = self.rid_to_state.pop(rid, None)
# A failed/partial dispatch may never produce a scheduler terminal
# for this rid, and once the state is dropped a late terminal has no
# release point either — so release here. Pre-acquire failures are
# covered by the finalizer's lora_id check.
self._finalize_lora_lease(state)

def _should_dispatch_to_encoder(
self, obj: Union[GenerateReqInput, EmbeddingReqInput]
Expand Down
Loading
Loading