Skip to content
This repository was archived by the owner on Jul 13, 2026. It is now read-only.

Set Pyworker errror for Autoscaler to handle reliability descrease only for such #39

Open
nakinnubis wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Set Pyworker errror for Autoscaler to handle reliability descrease only for such #39

Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9d75829
Enhance error handling in delete requests by distinguishing between e…
nakinnubis Dec 18, 2025
6929244
Refactor delete request
nakinnubis Dec 18, 2025
12cb625
Refactor delete request handling to simplify payload structure and im…
nakinnubis Dec 19, 2025
657c7a8
Refactor and remove deadcode
nakinnubis Dec 20, 2025
fc3eb21
Refactor delete request payload
nakinnubis Dec 20, 2025
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
24 changes: 6 additions & 18 deletions vastai/serverless/server/lib/metrics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,16 +151,13 @@ def _set_mtoken(self, mtoken: str) -> None:
#######################################Private#######################################

async def __send_delete_requests_and_reset(self):
async def post(report_addr: str, idxs: list[int], success_flag: bool) -> bool:
async def post(report_addr: str, requests: list[dict]) -> bool:
data = {
"worker_id": self.id,
"mtoken": self.mtoken,
"request_idxs": idxs,
"success": success_flag,
"requests": requests,
}
log.debug(
f"Deleting requests that {'succeeded' if success_flag else 'failed'}: {data['request_idxs']}"
)
log.debug(f"Deleting requests: {[r['request_idx'] for r in requests]}")
full_path = report_addr.rstrip("/") + "/delete_requests/"
for attempt in range(1, 4):
try:
Expand All @@ -180,28 +177,19 @@ async def post(report_addr: str, idxs: list[int], success_flag: bool) -> bool:
# Take a snapshot of what we plan to send this tick.
# New arrivals after this snapshot will remain in the queue for the next tick.
snapshot = list(self.model_metrics.requests_deleting)

@edgaratvast edgaratvast Dec 19, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

requests_payload is now just snapshot, so this is now unnecessary

@edgaratvast edgaratvast Dec 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I miswrote this: this was supposed to be "request_payload is now a snapshot", not "just snapshot, meaning that requests_payload contains all information contained in snapshot, so keeping a separate snapshot is now unnecessary.

Passing snapshot to post won't work work.

success_idxs = [r.request_idx for r in snapshot if r.success is True]
failed_idxs = [r.request_idx for r in snapshot if r.success is False]

if not success_idxs and not failed_idxs:
if not snapshot:
return # nothing to do

for report_addr in self.report_addr:
# TODO: Add a Redis subscriber queue for delete_requests
if report_addr == "https://cloud.vast.ai/api/v0":
# Patch: ignore the Redis API report_addr
continue
sent_success = True
sent_failed = True

if success_idxs:
sent_success = await post(report_addr, success_idxs, True)
if failed_idxs:
sent_failed = await post(report_addr, failed_idxs, False)

if sent_success and sent_failed:
if await post(report_addr, snapshot):
# Remove only the items we actually sent from the live queue.
sent_set = set(success_idxs) | set(failed_idxs)
sent_set = {r.request_idx for r in snapshot}
self.model_metrics.requests_deleting[:] = [
r for r in self.model_metrics.requests_deleting
if r.request_idx not in sent_set
Expand Down