[BugFix] Preserve tokenizer worker fanout when skip_tokenizer_init is enabled (#30682)
Co-authored-by: hjzhang <zhanghjzzz@qq.com> Co-authored-by: hnyls2002 <lsyincs@gmail.com>
This commit is contained in:
co-authored by
hjzhang
hnyls2002
parent
4d60c4540c
commit
db7e6807de
@@ -53,7 +53,8 @@ class SchedulerIpcChannels:
|
|||||||
context, zmq.PUSH, port_args.tokenizer_ipc_name, False
|
context, zmq.PUSH, port_args.tokenizer_ipc_name, False
|
||||||
)
|
)
|
||||||
if skip_tokenizer_init:
|
if skip_tokenizer_init:
|
||||||
# Directly send to the TokenizerManager
|
# No decode work: send outputs straight to the tokenizer side
|
||||||
|
# (MultiTokenizerRouter fans out when tokenizer_worker_num > 1).
|
||||||
send_to_detokenizer_raw = get_zmq_socket(
|
send_to_detokenizer_raw = get_zmq_socket(
|
||||||
context, zmq.PUSH, port_args.tokenizer_ipc_name, False
|
context, zmq.PUSH, port_args.tokenizer_ipc_name, False
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -601,6 +601,8 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
|
|||||||
(FreezeGCReq, lambda x: None),
|
(FreezeGCReq, lambda x: None),
|
||||||
# For handling case when scheduler skips detokenizer and forwards back to the tokenizer manager, we ignore it.
|
# For handling case when scheduler skips detokenizer and forwards back to the tokenizer manager, we ignore it.
|
||||||
(HealthCheckOutput, lambda x: None),
|
(HealthCheckOutput, lambda x: None),
|
||||||
|
# Same skip-detokenizer forwarding case as above.
|
||||||
|
(ConfigureLoggingReq, lambda x: None),
|
||||||
(ActiveRanksOutput, self.update_active_ranks),
|
(ActiveRanksOutput, self.update_active_ranks),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -6089,16 +6089,12 @@ class ServerArgs:
|
|||||||
)
|
)
|
||||||
|
|
||||||
if self.skip_tokenizer_init:
|
if self.skip_tokenizer_init:
|
||||||
if self.tokenizer_worker_num != 1:
|
# Tokenizer workers still serve HTTP / state / output work, so
|
||||||
logger.warning(
|
# their fanout is preserved; detokenizer workers only decode.
|
||||||
"skip_tokenizer_init=True disables tokenizer workers; forcing tokenizer_worker_num=1 "
|
|
||||||
f"(requested {self.tokenizer_worker_num})."
|
|
||||||
)
|
|
||||||
self.tokenizer_worker_num = 1
|
|
||||||
if self.detokenizer_worker_num != 1:
|
if self.detokenizer_worker_num != 1:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"skip_tokenizer_init=True disables detokenizer workers; forcing detokenizer_worker_num=1 "
|
"skip_tokenizer_init=True leaves no decode work for detokenizer workers; "
|
||||||
f"(requested {self.detokenizer_worker_num})."
|
f"forcing detokenizer_worker_num=1 (requested {self.detokenizer_worker_num})."
|
||||||
)
|
)
|
||||||
self.detokenizer_worker_num = 1
|
self.detokenizer_worker_num = 1
|
||||||
|
|
||||||
|
|||||||
@@ -36,7 +36,12 @@ class TestSkipTokenizerInit(CustomTestCase):
|
|||||||
cls.model,
|
cls.model,
|
||||||
cls.base_url,
|
cls.base_url,
|
||||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||||
other_args=["--skip-tokenizer-init", "--incremental-streaming-output"],
|
other_args=[
|
||||||
|
"--skip-tokenizer-init",
|
||||||
|
"--incremental-streaming-output",
|
||||||
|
"--tokenizer-worker-num",
|
||||||
|
"4",
|
||||||
|
],
|
||||||
)
|
)
|
||||||
cls.eos_token_id = [119690]
|
cls.eos_token_id = [119690]
|
||||||
cls.tokenizer = AutoTokenizer.from_pretrained(
|
cls.tokenizer = AutoTokenizer.from_pretrained(
|
||||||
@@ -216,7 +221,11 @@ class TestSkipTokenizerInitVLM(TestSkipTokenizerInit):
|
|||||||
cls.model,
|
cls.model,
|
||||||
cls.base_url,
|
cls.base_url,
|
||||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||||
other_args=["--skip-tokenizer-init"],
|
other_args=[
|
||||||
|
"--skip-tokenizer-init",
|
||||||
|
"--tokenizer-worker-num",
|
||||||
|
"4",
|
||||||
|
],
|
||||||
)
|
)
|
||||||
cls.eos_token_id = [cls.tokenizer.eos_token_id]
|
cls.eos_token_id = [cls.tokenizer.eos_token_id]
|
||||||
|
|
||||||
|
|||||||
@@ -232,6 +232,22 @@ class TestLoadBalanceMethod(unittest.TestCase):
|
|||||||
self.assertEqual(server_args.disaggregation_transfer_backend, "mooncake")
|
self.assertEqual(server_args.disaggregation_transfer_backend, "mooncake")
|
||||||
|
|
||||||
|
|
||||||
|
class TestSkipTokenizerInit(unittest.TestCase):
|
||||||
|
def test_skip_tokenizer_worker_counts(self):
|
||||||
|
server_args = ServerArgs(
|
||||||
|
model_path="dummy",
|
||||||
|
skip_tokenizer_init=True,
|
||||||
|
tokenizer_worker_num=4,
|
||||||
|
detokenizer_worker_num=3,
|
||||||
|
)
|
||||||
|
|
||||||
|
server_args._handle_tokenizer_batching()
|
||||||
|
|
||||||
|
# Tokenizer fanout preserved; detokenizer coerced to 1 (no decode work).
|
||||||
|
self.assertEqual(server_args.tokenizer_worker_num, 4)
|
||||||
|
self.assertEqual(server_args.detokenizer_worker_num, 1)
|
||||||
|
|
||||||
|
|
||||||
class TestHiSparseDsaBackendPolicy(unittest.TestCase):
|
class TestHiSparseDsaBackendPolicy(unittest.TestCase):
|
||||||
# The backend selection moved to the resolution pipeline; these policy
|
# The backend selection moved to the resolution pipeline; these policy
|
||||||
# tests drive the pass through its read-only view.
|
# tests drive the pass through its read-only view.
|
||||||
|
|||||||
Reference in New Issue
Block a user