CoolBoy/submmit
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

fix: replace duplicate submissions while refilling queues

Browse Source
This commit is contained in:
CoolBoy
2026-08-02 15:44:28 +08:00
parent 670c76fe4e
commit 0e42ef73dc
8 changed files with 309 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@@ -53,6 +53,12 @@ different candidate ordering (derived from `STRATEGY_ID`, instance ID, or
hostname), which reduces duplicate work while the platform remains the final
authority for account capacity and model/GPU uniqueness.
If the platform reports that a model/GPU is already being validated, the claim
is retained and the runner immediately draws replacement candidates from the
same scan instead of retrying the duplicate every cycle. Startup logs and the
health response expose `agent_version`; version `2026.08.02.1` or newer includes
this behavior.
## Deploy
Create a tag and submit the repository URL plus tag in "我的适配智能体".

5
main.py
View File

@@ -10,6 +10,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from modelhub_submmit_api.defaults import EMBEDDED_MODELHUB_XC_TOKEN, EMBEDDED_MODELSCOPE_TOKEN
from modelhub_submmit_api.version import AGENT_VERSION
HOST = "0.0.0.0"
@@ -74,6 +75,7 @@ def _worker_command() -> list[str]:
def _config() -> dict[str, object]:
return {
"agent_version": AGENT_VERSION,
"strategy_id_present": bool(os.getenv("STRATEGY_ID")),
"modelscope_token_present": bool(os.getenv("MODELSCOPE_API_TOKEN") or os.getenv("MODELSCOPE_TOKEN") or EMBEDDED_MODELSCOPE_TOKEN),
"modelhub_auth_present": _has_modelhub_auth(),
@@ -92,7 +94,7 @@ class Handler(BaseHTTPRequestHandler):
if worker is not None and worker.poll() is not None:
self._send_json({"status": "worker_exited", "config": _config()}, status=500)
return
self._send_json({"status": "ok"})
self._send_json({"status": "ok", "config": _config()})
return
if self.path == "/":
@@ -135,6 +137,7 @@ def _handle_signal(signum: int, _frame: object) -> None:
def main() -> int:
global config_error, worker
print(f"modelhub-submmit-agent version={AGENT_VERSION}", flush=True)
signal.signal(signal.SIGTERM, _handle_signal)
signal.signal(signal.SIGINT, _handle_signal)

13
modelhub_submmit_api/daily_runner.py
View File

@@ -198,7 +198,8 @@ def run_daily_batches(
f"[daily] wave_done name={wave.name} "
f"candidates={summary['candidateCount']} planned={summary['plannedSubmitCount']} "
f"submitted={summary['submittedCount']} skipped={summary['skippedCount']} "
f"failed={summary['failedCount']} remaining_before_run={summary['remainingDailyQuotaBeforeRun']}"
f"duplicates={summary.get('duplicateCount', 0)} failed={summary['failedCount']} "
f"remaining_before_run={summary['remainingDailyQuotaBeforeRun']}"
)
if summary.get("platformAvailableSlotsBeforeRun") == 0:
@@ -267,6 +268,14 @@ def finalize_daily_run(
stopped_reason: str,
) -> dict[str, Any]:
last_wave_summary = wave_results[-1]["summary"] if wave_results else {}
duplicate_total = sum(
int(wave_result.get("summary", {}).get("duplicateCount", 0) or 0)
for wave_result in wave_results
)
failed_total = sum(
int(wave_result.get("summary", {}).get("failedCount", 0) or 0)
for wave_result in wave_results
)
summary = {
"generatedAt": now.isoformat(),
"dryRun": bool(base_args.dry_run),
@@ -275,6 +284,8 @@ def finalize_daily_run(
"rounds": base_args.rounds,
"attemptedWaves": attempted_waves,
"submittedTotal": submitted_total,
"duplicateTotal": duplicate_total,
"failedTotal": failed_total,
"stoppedReason": stopped_reason,
"dailyRunDir": str(daily_run_dir),
"remainingDailyQuotaBeforeRun": last_wave_summary.get("remainingDailyQuotaBeforeRun"),

215
modelhub_submmit_api/main.py
View File

@@ -16,10 +16,10 @@ from history_stats import (
load_ledger,
update_history_archive,
)
from modelhub_client import ModelHubAPIError, ModelHubClient, ModelHubClientPool
from modelhub_client import ModelHubAPIError, ModelHubClient, ModelHubClientPool, is_duplicate_submission_error
from models import CandidateModel, HFModelSummary, ModelInspection
from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker
from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, diversify_candidates
from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, candidate_key, diversify_candidates
from task_registry import TASK_SPEC_BY_TYPE, all_task_types, choose_framework_for_task, choose_text_generation_framework, pipeline_tags_for_task_types, task_specs_for_model
from template_selector import TemplateSelector
@@ -319,6 +319,17 @@ def submit_candidate(
"responseData": response.get("data"),
}
except ModelHubAPIError as exc:
if is_duplicate_submission_error(exc):
print(
f"[submit] skipped repo={candidate['repoId']} gpu={candidate['targetGpu']} "
f"framework={candidate['framework']} reason=already_validating",
flush=True,
)
return {
"outcome": "duplicate",
"candidate": candidate,
"reason": str(exc),
}
print(
f"[submit] failed repo={candidate['repoId']} gpu={candidate['targetGpu']} "
f"framework={candidate['framework']} reason={exc}",
@@ -458,8 +469,11 @@ def run_submission(
"scanLimit": 0,
"scannedModels": 0,
"candidateCount": 0,
"targetSubmitCount": 0,
"maxSubmitAttempts": 0,
"plannedSubmitCount": 0,
"submittedCount": 0,
"duplicateCount": 0,
"skippedCount": 0,
"failedCount": 0,
"warnings": report.get("warnings", []),
@@ -550,7 +564,8 @@ def run_submission(
write_jsonl(run_dir / "candidates.jsonl", candidates)
submitted: list[dict[str, Any]] = []
planned_submit_count = resolve_max_submit_count(
duplicate_candidates: list[dict[str, Any]] = []
target_submit_count = resolve_max_submit_count(
args=args,
planned_count=len(candidates),
remaining_daily_quota=remaining_daily_quota,
@@ -558,94 +573,134 @@ def run_submission(
instance_id = runtime_instance_id()
diversified_candidates = diversify_candidates(candidates, instance_id=instance_id)
claim_store: SubmissionClaimStore | None = None
attempted_candidates: list[dict[str, Any]] = []
submit_workers = 1
if args.dry_run:
planned_candidates = diversified_candidates[:planned_submit_count]
attempted_candidates = diversified_candidates[:target_submit_count]
else:
claim_store = SubmissionClaimStore(
Path(getattr(args, "claims_path", DEFAULT_CLAIMS_PATH)),
owner_id=instance_id,
)
planned_candidates = claim_store.claim(diversified_candidates, limit=planned_submit_count)
submit_workers = 1
if not args.dry_run:
submit_workers = resolve_submit_concurrency(
args,
modelhub_client=modelhub_client,
planned_submit_count=len(planned_candidates),
attempted_keys: set[str] = set()
attempt_multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1))
max_submit_attempts = min(
len(diversified_candidates),
max(target_submit_count, target_submit_count * attempt_multiplier),
)
with ThreadPoolExecutor(max_workers=submit_workers) as executor:
futures = {
executor.submit(
submit_candidate,
candidate,
modelhub_client,
): index
for index, candidate in enumerate(planned_candidates)
}
ordered_results: dict[int, dict[str, Any]] = {}
for future in as_completed(futures):
index = futures[future]
try:
ordered_results[index] = future.result()
except Exception as exc:
candidate = planned_candidates[index]
ordered_results[index] = {
"outcome": "failed",
"candidate": candidate,
"reason": str(exc),
}
for index in range(len(planned_candidates)):
result = ordered_results.get(index)
if result is None:
continue
candidate = result["candidate"]
if result["outcome"] == "failed":
failed.append(
# A batch can contain candidates another machine has already submitted.
# Keep those duplicate claims and immediately draw replacements from the
# already-scanned pool until the desired number of real submissions is
# reached or account capacity is genuinely exhausted.
while len(submitted) < target_submit_count and len(attempted_candidates) < max_submit_attempts:
remaining_candidates = [
candidate
for candidate in diversified_candidates
if candidate_key(candidate) not in attempted_keys
]
desired_count = min(
target_submit_count - len(submitted),
max_submit_attempts - len(attempted_candidates),
)
batch_candidates = claim_store.claim(remaining_candidates, limit=desired_count)
if not batch_candidates:
break
attempted_candidates.extend(batch_candidates)
attempted_keys.update(candidate_key(candidate) for candidate in batch_candidates)
batch_workers = resolve_submit_concurrency(
args,
modelhub_client=modelhub_client,
planned_submit_count=len(batch_candidates),
)
submit_workers = max(submit_workers, batch_workers)
with ThreadPoolExecutor(max_workers=batch_workers) as executor:
futures = {
executor.submit(submit_candidate, candidate, modelhub_client): index
for index, candidate in enumerate(batch_candidates)
}
ordered_results: dict[int, dict[str, Any]] = {}
for future in as_completed(futures):
index = futures[future]
try:
ordered_results[index] = future.result()
except Exception as exc:
candidate = batch_candidates[index]
ordered_results[index] = {
"outcome": "failed",
"candidate": candidate,
"reason": str(exc),
}
batch_submitted_candidates: list[dict[str, Any]] = []
batch_duplicate_candidates: list[dict[str, Any]] = []
batch_failed_candidates: list[dict[str, Any]] = []
for index in range(len(batch_candidates)):
result = ordered_results.get(index)
if result is None:
continue
candidate = result["candidate"]
if result["outcome"] == "duplicate":
duplicate_candidates.append(candidate)
batch_duplicate_candidates.append(candidate)
skipped.append(
{
"repoId": candidate["repoId"],
"targetGpu": candidate["targetGpu"],
"reason": "already_validating_on_platform",
}
)
continue
if result["outcome"] == "failed":
batch_failed_candidates.append(candidate)
failed.append(
{
"repoId": candidate["repoId"],
"targetGpu": candidate["targetGpu"],
"framework": candidate["framework"],
"taskType": candidate["taskType"],
"reason": result.get("reason", "submission_failed"),
}
)
continue
batch_submitted_candidates.append(candidate)
submitted_record = {
**candidate,
"submitTime": result["submitTime"],
"taskId": result["taskId"],
"responseData": result["responseData"],
}
submitted.append(submitted_record)
append_ledger_entry(
ledger_path,
{
"repoId": candidate["repoId"],
"modelId": candidate["repoId"],
"modelAddress": candidate["modelAddress"],
"targetGpu": candidate["targetGpu"],
"framework": candidate["framework"],
"templateId": candidate["templateId"],
"taskId": result["taskId"],
"taskType": candidate["taskType"],
"reason": result.get("reason", "submission_failed"),
}
"submitTime": result["submitTime"],
},
)
outcome_tracker.record_submission(
model_id=candidate["repoId"],
target_gpu=candidate["targetGpu"],
framework=candidate["framework"],
task_type=candidate["taskType"],
task_id=result["taskId"],
submit_time=result["submitTime"],
)
continue
submitted_record = {
**candidate,
"submitTime": result["submitTime"],
"taskId": result["taskId"],
"responseData": result["responseData"],
}
submitted.append(submitted_record)
append_ledger_entry(
ledger_path,
{
"modelId": candidate["repoId"],
"modelAddress": candidate["modelAddress"],
"targetGpu": candidate["targetGpu"],
"framework": candidate["framework"],
"templateId": candidate["templateId"],
"taskId": result["taskId"],
"taskType": candidate["taskType"],
"submitTime": result["submitTime"],
},
)
outcome_tracker.record_submission(
model_id=candidate["repoId"],
target_gpu=candidate["targetGpu"],
framework=candidate["framework"],
task_type=candidate["taskType"],
task_id=result["taskId"],
submit_time=result["submitTime"],
)
claim_store.mark_submitted([*batch_submitted_candidates, *batch_duplicate_candidates])
claim_store.release(batch_failed_candidates)
if claim_store is not None:
submitted_candidates = [result["candidate"] for result in ordered_results.values() if result.get("outcome") == "submitted"]
failed_candidates = [result["candidate"] for result in ordered_results.values() if result.get("outcome") != "submitted"]
claim_store.mark_submitted(submitted_candidates)
claim_store.release(failed_candidates)
if hasattr(modelhub_client, "available_submit_slots") and modelhub_client.available_submit_slots() <= 0:
break
write_jsonl(run_dir / "submitted.jsonl", submitted)
write_jsonl(run_dir / "skipped.jsonl", skipped)
@@ -671,7 +726,13 @@ def run_submission(
"scanLimit": scan_limit,
"scannedModels": len(models),
"candidateCount": len(candidates),
"plannedSubmitCount": len(planned_candidates),
"targetSubmitCount": target_submit_count,
"maxSubmitAttempts": 0 if args.dry_run else min(
len(diversified_candidates),
max(target_submit_count, target_submit_count * max(1, int(getattr(args, "scan_multiplier", 4) or 1))),
),
"plannedSubmitCount": len(attempted_candidates),
"duplicateCount": len(duplicate_candidates),
"submittedCount": len(submitted),
"skippedCount": len(skipped),
"failedCount": len(failed),

16
modelhub_submmit_api/modelhub_client.py
View File

@@ -246,6 +246,15 @@ CAPACITY_ERROR_MARKERS = (
"active task limit",
)
DUPLICATE_SUBMISSION_MARKERS = (
"正在验证中",
"请勿重复提交",
"重复提交",
"already validating",
"already being validated",
"already in progress",
)
def is_capacity_error(error: ModelHubAPIError) -> bool:
if error.code in {409, 429}:
@@ -256,6 +265,13 @@ def is_capacity_error(error: ModelHubAPIError) -> bool:
return any(marker in message for marker in CAPACITY_ERROR_MARKERS)
def is_duplicate_submission_error(error: ModelHubAPIError) -> bool:
message = str(error).strip().lower()
if isinstance(error.payload, dict):
message = f"{message} {error.payload.get('message') or ''}".lower()
return any(marker in message for marker in DUPLICATE_SUBMISSION_MARKERS)
class ModelHubClientPool:
def __init__(
self,

14
modelhub_submmit_api/poll_runner.py
View File

@@ -17,6 +17,7 @@ from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker
from runner_common import DEFAULT_KEY_PATH, ensure_tokens
from submission_claims import DEFAULT_CLAIMS_PATH
from template_selector import TemplateSelector
from version import AGENT_VERSION
DEFAULT_POLL_RUNS_DIR = Path("poll_runs")
@@ -122,7 +123,7 @@ def run_poll_loop(
OUTCOME_SYNC_INTERVAL = 3
STATS_PRINT_INTERVAL = 10
log(f"[poll] poll_run_dir={poll_run_dir}")
log(f"[poll] version={AGENT_VERSION} poll_run_dir={poll_run_dir}")
log(
f"[poll] target={base_args.daily_target} dry_run={str(bool(base_args.dry_run)).lower()} "
f"poll_interval={base_args.poll_interval_seconds}s idle_interval={base_args.idle_interval_seconds}s"
@@ -165,6 +166,7 @@ def run_poll_loop(
log(
f"[poll] cycle_done submitted_total={cycle_summary['submittedTotal']} "
f"duplicates={cycle_summary.get('duplicateTotal', 0)} "
f"remaining_before_run={remaining_before_run if remaining_before_run is not None else 'n/a'} "
f"stop={cycle_summary['stoppedReason']}"
)
@@ -174,8 +176,14 @@ def run_poll_loop(
break
if cycle_summary["submittedTotal"] <= 0:
log(f"[poll] cycle={cycles} sleep={base_args.idle_interval_seconds}s reason=no_new_submissions")
time.sleep(base_args.idle_interval_seconds)
duplicate_total = int(cycle_summary.get("duplicateTotal", 0) or 0)
if duplicate_total > 0 and (available_slots is None or available_slots > 0):
retry_delay = max(1, int(getattr(base_args, "post_cycle_cooldown_seconds", 2) or 2))
log(f"[poll] cycle={cycles} sleep={retry_delay}s reason=duplicates_need_replacement_candidates")
time.sleep(retry_delay)
else:
log(f"[poll] cycle={cycles} sleep={base_args.idle_interval_seconds}s reason=no_new_submissions")
time.sleep(base_args.idle_interval_seconds)
continue
if cycles % OUTCOME_SYNC_INTERVAL == 0:

1
modelhub_submmit_api/version.py Normal file
View File

@@ -0,0 +1 @@
AGENT_VERSION = "2026.08.02.1"

123
tests/test_concurrency.py
View File

@@ -15,8 +15,9 @@ if str(PACKAGE_DIR) in sys.path:
sys.path.insert(0, str(PACKAGE_DIR))
from common import read_jsonl # noqa: E402
from main import make_run_dir # noqa: E402
from modelhub_client import ModelHubAPIError, ModelHubClientPool # noqa: E402
from main import build_parser, make_run_dir, run_submission, submit_candidate # noqa: E402
from modelhub_client import ModelHubAPIError, ModelHubClientPool, is_duplicate_submission_error # noqa: E402
from models import HFModelSummary, ModelInspection # noqa: E402
from outcome_tracker import OutcomeTracker # noqa: E402
from submission_claims import SubmissionClaimStore, candidate_key # noqa: E402
@@ -57,6 +58,54 @@ class FakeClient:
return None
class FakeDiscovery:
def __init__(self, count: int) -> None:
self.models = [
HFModelSummary(
repo_id=f"owner/model-{index}",
downloads=100,
last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc),
pipeline_tag="text-generation",
)
for index in range(count)
]
def list_recent_models(self, **_kwargs): # noqa: ANN003
return self.models
@staticmethod
def inspect_model(model: HFModelSummary) -> ModelInspection:
return ModelInspection(repo_id=model.repo_id, weight_files=["model.safetensors"])
class DuplicateThenSuccessClient:
def __init__(self) -> None:
self.calls = 0
self.active = 0
@staticmethod
def begin_cycle() -> None:
return None
def available_submit_slots(self) -> int:
return max(0, 2 - self.active)
@staticmethod
def list_tasks_page(**_kwargs) -> dict: # noqa: ANN003
return {"code": 0, "data": {"records": [], "pages": 0}}
@staticmethod
def processed_gpus_for_model(_model_id: str) -> set[str]:
return set()
def add_task(self, _payload: dict) -> dict:
self.calls += 1
if self.calls <= 2:
raise ModelHubAPIError("模型正在验证中，请勿重复提交")
self.active += 1
return {"code": 0, "data": {"id": f"task-{self.calls}"}}
def make_candidate(index: int) -> dict:
return {
"repoId": f"owner/model-{index}",
@@ -66,6 +115,61 @@ def make_candidate(index: int) -> dict:
class ClientPoolConcurrencyTests(unittest.TestCase):
def test_duplicate_batch_is_replaced_until_available_slots_are_filled(self) -> None:
with tempfile.TemporaryDirectory() as temporary_dir:
root = Path(temporary_dir)
args = build_parser().parse_args(
[
"--gpus",
"Iluvatar_bi-150",
"--task-types",
"text-generation",
"--limit",
"4",
"--max-scan-models",
"4",
"--skip-outcome-sync",
"--skip-history-archive",
]
)
args.runs_dir = str(root / "runs")
args.ledger_path = str(root / "ledger.jsonl")
args.outcomes_path = str(root / "outcomes.jsonl")
args.claims_path = str(root / "claims.jsonl")
args.history_archive_path = str(root / "history.jsonl")
client = DuplicateThenSuccessClient()
summary = run_submission(
args,
now=datetime(2026, 1, 1, 12, tzinfo=timezone.utc),
hf_discovery=FakeDiscovery(4), # type: ignore[arg-type]
modelhub_client=client, # type: ignore[arg-type]
)
self.assertEqual(2, summary["targetSubmitCount"])
self.assertEqual(4, summary["plannedSubmitCount"])
self.assertEqual(2, summary["duplicateCount"])
self.assertEqual(2, summary["submittedCount"])
self.assertEqual(0, client.available_submit_slots())
def test_platform_duplicate_is_classified_as_skipped_candidate(self) -> None:
error = ModelHubAPIError("模型正在验证中，请勿重复提交")
self.assertTrue(is_duplicate_submission_error(error))
class DuplicateClient:
@staticmethod
def add_task(_payload): # noqa: ANN001
raise error
candidate = {
**make_candidate(1),
"taskType": "text-generation",
"framework": "vllm",
"configParams": "framework: vllm",
}
result = submit_candidate(candidate, DuplicateClient()) # type: ignore[arg-type]
self.assertEqual("duplicate", result["outcome"])
def test_every_account_reaches_capacity_under_load(self) -> None:
clients = [FakeClient() for _ in range(12)]
pool = ModelHubClientPool(
@@ -129,6 +233,21 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
class ProcessCoordinationTests(unittest.TestCase):
def test_duplicate_claims_are_retained_so_next_batch_moves_forward(self) -> None:
with tempfile.TemporaryDirectory() as temporary_dir:
path = Path(temporary_dir) / "claims.jsonl"
candidates = [make_candidate(index) for index in range(4)]
store = SubmissionClaimStore(path, owner_id="worker-a")
duplicates = store.claim(candidates, limit=2)
store.mark_submitted(duplicates)
replacements = store.claim(candidates, limit=2)
self.assertEqual(
{candidate_key(candidate) for candidate in candidates[2:]},
{candidate_key(candidate) for candidate in replacements},
)
def test_concurrent_claim_stores_select_disjoint_candidates(self) -> None:
with tempfile.TemporaryDirectory() as temporary_dir:
path = Path(temporary_dir) / "claims.jsonl"