bingshuo/guanghu-ice-heart
Watch
1
0
Fork
You've already forked guanghu-ice-heart
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

feat(pool): grant deterministic worker leases

Browse source
This commit is contained in:
冰朔 2026-09-11 18:33:44 +08:00
commit 9c31e7f90c
1 changed files with 11 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

13
server-tools/persona-compute-pool/entry_service.py
View file

@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Loopback-first compute-pool entry with authenticated worker heartbeats.""" """Loopback-first compute-pool entry with authenticated worker heartbeats."""
from __future__ import annotations from __future__ import annotations
import json, os, secrets, ssl, threading import json, os, secrets, ssl, threading, time, uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
@ -50,7 +50,16 @@ class Handler(BaseHTTPRequestHandler):
if node != peer: self.send_json(403,{'error':'WORKER_CERT_ID_MISMATCH'}); return if node != peer: self.send_json(403,{'error':'WORKER_CERT_ID_MISMATCH'}); return
state['workers'][node]={**payload,'last_seen':payload.get('observed_at'),'authority_granted':False}; state['registered_workers']=len(state['workers']); write_state(state); self.send_json(200,{'outcome':'HEARTBEAT_ACCEPTED','node_id':node,'authority_granted':False}); return state['workers'][node]={**payload,'last_seen':payload.get('observed_at'),'authority_granted':False}; state['registered_workers']=len(state['workers']); write_state(state); self.send_json(200,{'outcome':'HEARTBEAT_ACCEPTED','node_id':node,'authority_granted':False}); return
if self.path=='/v1/lease/request': if self.path=='/v1/lease/request':
self.send_json(200,{'outcome':'LEASE_QUEUED','state':'PENDING_DETERMINISTIC_SELECTION','authority_granted':False}); return min_cpu=int(payload.get('min_cpu_cores',1)); min_mem=int(payload.get('min_memory_mb',256)); candidates=[]
for worker in state['workers'].values():
metrics=worker.get('metrics') or {}
if worker.get('state')!='ACTIVE_WORKER': continue
if int(metrics.get('cpu_cores') or 0)<min_cpu: continue
if int(metrics.get('memory_available_bytes') or 0)<min_mem*1024*1024: continue
candidates.append(worker)
if not candidates: self.send_json(409,{'outcome':'NO_ELIGIBLE_WORKER','state':'LEASE_NOT_GRANTED','authority_granted':False}); return
candidates.sort(key=lambda worker:(int((worker.get('metrics') or {}).get('cpu_cores') or 0),int((worker.get('metrics') or {}).get('memory_available_bytes') or 0)))
node=candidates[0]['node_id']; lease_id=f'LEASE-{uuid.uuid4()}'; lease={'lease_id':lease_id,'node_id':node,'task_id':payload.get('task_id','UNNAMED_TASK'),'created_at':time.time(),'expires_at':time.time()+900,'state':'GRANTED','authority_granted':False}; state['leases'][lease_id]=lease; write_state(state); self.send_json(200,{'outcome':'LEASE_GRANTED','state':'GRANTED','lease':lease,'authority_granted':False}); return
self.send_json(404,{'error':'NOT_FOUND'}) self.send_json(404,{'error':'NOT_FOUND'})
def log_message(self,*args): pass def log_message(self,*args): pass