mirror of
https://github.com/Niko1221/Strata.git
synced 2026-11-07 06:15:58 -10:00
serve: never send the all-cards split engine a GEN key its parser refuses: pen_from is dropped for it (its penalty window is the last penalty_last_n tokens of the whole context, so a continued prompt gives the same window), logprobs and logit_bias get a 400 naming --split-engine off; a test pins the set of keys the server can send against sfnx.cpp's parse_gen
This commit is contained in:
1 parent
079d7a6244
commit
6174eed4ac
3 files changed
+126
-2
No files matched your search
@@ -106,6 +106,8 @@ def validate_request(req, engine, vocab_size):
|
||||
"""Fail closed for old engines and unsupported backends (continuous batching carries a slot's bias since the slot-penalty change)."""
|
||||
normalized = normalize(req.get("logit_bias"), vocab_size)
|
||||
if normalized:
|
||||
if getattr(engine, "split_engine", False):
|
||||
raise ValueError("logit_bias: not supported by the all-cards split engine; start with --split-engine off")
|
||||
if getattr(engine, "info", {}).get("logit_bias") != 1:
|
||||
raise ValueError("logit_bias: this engine does not support token biases; use a supporting CUDA/HIP build")
|
||||
return normalized
|
||||
+26
-2
@@ -1090,6 +1090,8 @@ class StrataEngine:
|
||||
retry_args: list[str] | None = None):
|
||||
# retry_args: the arguments to start again with, once, when this start ends before the engine is ready (an Auto start that picked DFlash2: mtp_retry_args)
|
||||
self.spawn = (exe, list(args), cwd, log, env, False, cpus) # to start it again after it died (issue #27)
|
||||
# the all-cards split engine (sFNX, strata-split-N[-arch]): its GEN parser accepts fewer keys than Strata's engine (tests/split/test_split_gen_keys.py pins them)
|
||||
self.split_engine = os.path.basename(str(exe)).startswith("strata-split")
|
||||
self.session_save_reclaim = '--session-save-reclaim' in args
|
||||
paths = {k: v for k, v in zip(args, args[1:]) if k in ("--native", "--pack")}
|
||||
self.model_path = paths.get("--native") or paths.get("--pack", "pack/full")
|
||||
@@ -1533,6 +1535,14 @@ class StrataEngine:
|
||||
if time.time() > deadline:
|
||||
raise EngineDied("the engine did not answer the VRAM command")
|
||||
|
||||
def pen_from_key(self, orig_len: int, prompt_len: int) -> str:
|
||||
"""` pen_from=N` for a leg that continues a request (prompt + what it generated): where the reply began, Strata's penalty window start. The all-cards split engine
|
||||
(sFNX) counts its penalty window over the last penalty_last_n tokens of the whole context, so the continued prompt gives it the same window and the engine refuses the
|
||||
key (GEN: unknown key pen_from): it is not sent."""
|
||||
if self.split_engine or prompt_len <= orig_len:
|
||||
return ""
|
||||
return f" pen_from={orig_len - 1}"
|
||||
|
||||
@staticmethod
|
||||
def sampling_keys(sampling: dict) -> str:
|
||||
keys = ""
|
||||
@@ -2082,7 +2092,7 @@ class StrataEngine:
|
||||
with self.slot_cv:
|
||||
alone = not any(self.slot_busy) and self.waiting == 0
|
||||
if alone and left > 1:
|
||||
pf = f" pen_from={orig_len - 1}" if len(prompt) > orig_len else ""
|
||||
pf = self.pen_from_key(orig_len, len(prompt))
|
||||
head = f"GENI {left}{keys}{pf} {embeddings}" if embeddings else f"GEN {left}{keys}{pf}"
|
||||
self._last_done = None # (#879: this request's DONE decides below)
|
||||
deadline = getattr(cancel, "_strata_queue_deadline", None)
|
||||
@@ -2211,7 +2221,7 @@ class StrataEngine:
|
||||
holding, born = True, self.gen # (the engine it now has the control lines of)
|
||||
while not self.slot_q[slot].empty():
|
||||
self.slot_q[slot].get_nowait()
|
||||
pf = f" pen_from={orig_len - 1}" if len(prompt) > orig_len else ""
|
||||
pf = self.pen_from_key(orig_len, len(prompt))
|
||||
head = f"BGENI {slot} {left}{keys}{pf} {embeddings}" if embeddings else f"BGEN {slot} {left}{keys}{pf}"
|
||||
live = {"slot": slot, "request_id": getattr(cancel, "_strata_request_id", None),
|
||||
"state": "reading", "prompt_tokens": len(prompt), "prompt_read": 0, "generated": 0,
|
||||
@@ -2477,6 +2487,7 @@ class StrataEngine:
|
||||
if not self.alive():
|
||||
raise EngineDied("the engine is unavailable; this request was not sent")
|
||||
bias_api.validate_request(sampling or {}, self, None)
|
||||
refuse_for_split(self, sampling or {})
|
||||
if getattr(self, "batch", 0) and not self.multi:
|
||||
yield from self.generate_batched(ids, max_new, sampling, cancel, embeddings)
|
||||
return
|
||||
@@ -4324,6 +4335,18 @@ class Detokenizer:
|
||||
OPENAI_MAX_STOP = 4 # OpenAI's limit on `stop`
|
||||
|
||||
|
||||
#: What a request may ask for that the all-cards split engine cannot do (its GEN parser has no such keys): answered with a 400 before anything is sent, not with an error in
|
||||
#: the middle of a reply. logit_bias is refused the same way by logit_bias.validate_request (the split engine does not report logit_bias=1).
|
||||
SPLIT_REFUSED_FEATURES = ("logprobs", "logit_bias")
|
||||
|
||||
|
||||
def refuse_for_split(engine, req: dict) -> None:
|
||||
if not getattr(engine, "split_engine", False):
|
||||
return
|
||||
if top_logprobs(req) is not None:
|
||||
raise ValueError("logprobs: not supported by the all-cards split engine; start with --split-engine off")
|
||||
|
||||
|
||||
def top_logprobs(req: dict):
|
||||
"""The K for the engine's logprobs=K, or None when the request did not ask.
|
||||
|
||||
@@ -8238,6 +8261,7 @@ def make_handler(svc: Service):
|
||||
raise ValueError(f"allowed_token_ids: {bad[0]} is outside the vocabulary "
|
||||
f"(0..{n_vocab - 1})")
|
||||
bias_api.validate_request(req, svc.engine, len(getattr(svc.tok, "tokens", ())) or None)
|
||||
refuse_for_split(svc.engine, req)
|
||||
use_mcp = req.get("strata_mcp") is True and svc.mcp is not None # the web app's opt-in (serve/mcp.py)
|
||||
own = {t.get("name") for t in tools or [] if isinstance(t, dict)} # #592: a second line of defence
|
||||
if use_mcp:
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""The GEN keys serve/server.py can send an engine, against the keys the all-cards split engine's parser (src/split/tools/sfnx.cpp, parse_gen) accepts.
|
||||
|
||||
A key the split engine does not know ends the request with "GEN: unknown key ..." (an HTTP 503 in the middle of a reply: pen_from was found that way on the AMD package
|
||||
gate). Every key the server can send is classified here; a new one fails this test until it is sent to the split engine safely, dropped for it, or refused with a 400."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
# keys the split engine's parse_gen accepts (some only note that they change nothing: pin, pcie_frac, spec_min_p, pipe, aux_cpus, cvec)
|
||||
ACCEPTED = {"temperature", "top_p", "top_k", "min_p", "penalty_repeat", "penalty_freq", "penalty_present", "penalty_last_n", "seed", "keep", "ckpt", "pin",
|
||||
"pcie_frac", "spec_min_p", "pipe", "aux_cpus", "cvec", "img_keys"}
|
||||
# keys the server does not send to a split engine: pen_from is dropped (StrataEngine.pen_from_key), logprobs / allowed / logit_bias are refused with a 400 first
|
||||
# (refuse_for_split, logit_bias.validate_request)
|
||||
DROPPED = {"pen_from"}
|
||||
REFUSED = {"logprobs", "allowed", "logit_bias"}
|
||||
# keys only other engines are sent: the DeepSeek engine's restore points (STRATA_DS4_RESTORE_POINTS), the prefill-only request of serve/split.py, rid for several writers
|
||||
OTHER_ENGINES = {"marks", "tail", "shared", "prefill", "rid"}
|
||||
|
||||
|
||||
def server_keys() -> set[str]:
|
||||
text = (ROOT / "serve" / "server.py").read_text(encoding="utf-8")
|
||||
keys = set(re.findall(r'keys \+= f?" (\w+)=', text))
|
||||
keys |= set(re.findall(r'keys \+= " (\w+)="', text))
|
||||
for tup in re.findall(r'for k in \(((?:"\w+",? ?)+)\):\n\s+v = tune\.get\(k\)', text):
|
||||
keys |= set(re.findall(r'"(\w+)"', tup))
|
||||
keys |= set(re.findall(r'f" (pen_from)=', text))
|
||||
keys |= set(re.findall(r'f" (rid)=', text))
|
||||
keys |= set(re.findall(r'" (prefill)=1', text))
|
||||
keys |= set(re.findall(r'(marks|tail|shared)=', text.split("def restore_keys", 1)[1].split("def projection_key", 1)[0]))
|
||||
keys |= set(re.findall(r'f" (cvec)=', text))
|
||||
keys |= set(re.findall(r'"(logit_bias)="', (ROOT / "serve" / "logit_bias.py").read_text(encoding="utf-8")))
|
||||
keys |= set(re.findall(r'" (\w+)=" \+', (ROOT / "serve" / "logit_bias.py").read_text(encoding="utf-8")))
|
||||
keys |= set(re.findall(r'keys \+= f" (\w+)=', text))
|
||||
return keys
|
||||
|
||||
|
||||
def parser_keys() -> set[str]:
|
||||
text = (ROOT / "src" / "split" / "tools" / "sfnx.cpp").read_text(encoding="utf-8")
|
||||
body = text.split("static GenReq parse_gen", 1)[-1] if "static GenReq parse_gen" in text else text
|
||||
return set(re.findall(r'k == "(\w+)"', body))
|
||||
|
||||
|
||||
class SplitGenKeys(unittest.TestCase):
|
||||
def test_every_key_the_server_can_send_is_classified(self):
|
||||
sent = server_keys()
|
||||
known = ACCEPTED | DROPPED | REFUSED | OTHER_ENGINES
|
||||
self.assertEqual(sorted(sent - known), [], "a GEN key the server sends that nobody classified for the split engine")
|
||||
for must in ("pen_from", "logprobs", "allowed", "temperature", "penalty_last_n", "seed"):
|
||||
self.assertIn(must, sent, f"{must}: the parse of serve/server.py no longer finds it (update this test)")
|
||||
|
||||
def test_the_keys_sent_to_a_split_engine_are_accepted_by_its_parser(self):
|
||||
parser = parser_keys()
|
||||
self.assertEqual(sorted(ACCEPTED - parser), [], "a key the split engine is sent but its parse_gen does not know")
|
||||
|
||||
def test_pen_from_is_not_sent_to_a_split_engine_and_is_to_the_others(self):
|
||||
from serve import server
|
||||
|
||||
split = object.__new__(server.StrataEngine)
|
||||
split.split_engine = True
|
||||
other = object.__new__(server.StrataEngine)
|
||||
other.split_engine = False
|
||||
self.assertEqual(split.pen_from_key(100, 130), "")
|
||||
self.assertEqual(other.pen_from_key(100, 130), " pen_from=99")
|
||||
self.assertEqual(other.pen_from_key(100, 100), "")
|
||||
|
||||
def test_a_split_engine_name_is_found_by_its_executable(self):
|
||||
from serve import server
|
||||
|
||||
for exe, want in (("/x/strata-split-4-amd", True), ("/x/strata-split-2-sm86", True), ("C:/x/strata-split-2", True), ("/x/strata", False),
|
||||
("/x/strata-dsv41", False)):
|
||||
e = object.__new__(server.StrataEngine)
|
||||
e.split_engine = __import__("os").path.basename(exe).startswith("strata-split")
|
||||
self.assertEqual(e.split_engine, want, exe)
|
||||
|
||||
def test_logprobs_and_logit_bias_are_refused_with_a_400_for_a_split_engine(self):
|
||||
from serve import logit_bias, server
|
||||
|
||||
e = object.__new__(server.StrataEngine)
|
||||
e.split_engine = True
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
server.refuse_for_split(e, {"logprobs": True, "top_logprobs": 3})
|
||||
self.assertIn("--split-engine off", str(cm.exception))
|
||||
server.refuse_for_split(e, {"temperature": 0.7}) # an ordinary request goes on
|
||||
with self.assertRaises(ValueError) as cm:
|
||||
logit_bias.validate_request({"logit_bias": {"5": 1.0}}, e, 1000)
|
||||
self.assertIn("--split-engine off", str(cm.exception))
|
||||
other = object.__new__(server.StrataEngine)
|
||||
other.split_engine = False
|
||||
server.refuse_for_split(other, {"logprobs": True, "top_logprobs": 3}) # other engines unchanged
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user