40 lines
7.2 KiB
PHP
40 lines
7.2 KiB
PHP
<?php
|
|
require_once __DIR__ . '/labdb.class.php';
|
|
require_once __DIR__ . '/parametercatalog.class.php';
|
|
require_once __DIR__ . '/maps.class.php';
|
|
class Job {
|
|
private static function row(array $r): array { $r['spec']=json_decode($r['spec'],true); if (isset($r['spec']['parameters']) && $r['spec']['parameters']===[]) $r['spec']['parameters']=(object)[]; return $r; }
|
|
private static function specJson(array $spec): string { if (isset($spec['parameters']) && $spec['parameters']===[]) $spec['parameters']=(object)[]; return json_encode($spec,JSON_THROW_ON_ERROR); }
|
|
public static function directory($job): string { return rtrim(lab_config()['jobs'],'/') . '/' . $job['id']; }
|
|
public static function all($owner): array { $s=LabDb::connection()->prepare('SELECT * FROM jobs WHERE owner=? ORDER BY created_at DESC LIMIT 100'); $s->execute([$owner]); return array_map([self::class,'row'],$s->fetchAll()); }
|
|
public static function get($id,$owner): ?array { $s=LabDb::connection()->prepare('SELECT * FROM jobs WHERE id=? AND owner=?'); $s->execute([$id,$owner]); $r=$s->fetch(); return $r?self::row($r):null; }
|
|
private static function queue($owner,$name,$experimentId,$spec,$kind='model'): array {
|
|
$captures=Maps::captures($spec);if(empty($captures['ok']))return ['ok'=>false,'errors'=>$captures['errors']??[]];
|
|
$id=bin2hex(random_bytes(16));$cycles=$kind==='planet'?1:(int)($spec['run']['cycles']??1);$now=gmdate('c');$db=LabDb::connection();
|
|
$db->beginTransaction();try{$q=$db->query("SELECT COUNT(*) FROM jobs WHERE status IN ('queued','running')");if((int)$q->fetchColumn()>=(int)(lab_config()['queue_limit']??100)){$db->rollBack();return ['ok'=>false,'errors'=>['queue'=>'The job queue is full.']];}
|
|
$s=$db->prepare("INSERT INTO jobs(id,experiment_id,owner,name,spec,kind,status,phase,total_cycles,created_at,updated_at) VALUES(?,?,?,?,?,?,'queued','queued',?,?,?)");$s->execute([$id,$experimentId,$owner,$name,self::specJson($spec),$kind,$cycles,$now,$now]);$insert=$db->prepare('INSERT INTO job_maps(job_id,field,version_id,digest,data) VALUES(?,?,?,?,?)');foreach($captures['data'] as $field=>$map){$insert->bindValue(1,$id);$insert->bindValue(2,$field);$insert->bindValue(3,$map['id']);$insert->bindValue(4,$map['digest']);$insert->bindValue(5,$map['data'],PDO::PARAM_LOB);$insert->execute();}$db->commit();
|
|
}catch(Throwable $e){if($db->inTransaction())$db->rollBack();return ['ok'=>false,'error'=>'Could not queue job.'];}return ['ok'=>true,'data'=>self::get($id,$owner)];
|
|
}
|
|
public static function enqueue($owner,$experiment): array {
|
|
if(!is_array($experiment)||empty($experiment['id']))return ['ok'=>false,'errors'=>['experiment'=>'A saved experiment is required.']];$s=LabDb::connection()->prepare('SELECT * FROM experiments WHERE id=? AND owner=?');$s->execute([$experiment['id'],$owner]);$e=$s->fetch();if(!$e)return ['ok'=>false,'errors'=>['experiment'=>'Experiment not found.']];
|
|
$v=ParameterCatalog::validate(json_decode($e['spec'],true));if(empty($v['ok']))return ['ok'=>false,'errors'=>$v['errors']??[]];$mapCheck=Maps::validate($v['spec']);if(empty($mapCheck['ok']))return $mapCheck;return self::queue($owner,$e['name'],$e['id'],$v['spec']);
|
|
}
|
|
public static function enqueueKind($owner,$kind,$name,$spec): array {
|
|
if($kind!=='planet')return ['ok'=>false,'errors'=>['kind'=>'Unsupported job kind.']];if(!is_string($name)||trim($name)==='')return ['ok'=>false,'errors'=>['name'=>'A name is required.']];$v=ParameterCatalog::validate($spec);if(empty($v['ok']))return ['ok'=>false,'errors'=>$v['errors']??[]];if(empty($v['spec']['surface']))return ['ok'=>false,'errors'=>['surface'=>'Planet generation parameters are required.']];$mapCheck=Maps::validate($v['spec']);if(empty($mapCheck['ok']))return $mapCheck;return self::queue($owner,trim($name),null,$v['spec'],'planet');
|
|
}
|
|
public static function cancel($id,$owner): array { $db=LabDb::connection();$s=$db->prepare("UPDATE jobs SET status=CASE WHEN status='queued' THEN 'cancelled' ELSE status END,cancel_requested=1,phase=CASE WHEN status='queued' THEN 'cancelled' ELSE phase END,finished_at=CASE WHEN status='queued' THEN ? ELSE finished_at END,updated_at=? WHERE id=? AND owner=? AND status IN ('queued','running')");$now=gmdate('c');$s->execute([$now,$now,$id,$owner]);$r=self::get($id,$owner);return $r?['ok'=>true,'data'=>$r]:['ok'=>false,'errors'=>['id'=>'Active job not found.']]; }
|
|
public static function retry($id,$owner): array {
|
|
$j=self::get($id,$owner);if(!$j||!in_array($j['status'],['failed','cancelled','interrupted'],true))return ['ok'=>false,'errors'=>['id'=>'Retry requires a stopped job.']];$newId=bin2hex(random_bytes(16));$now=gmdate('c');$db=LabDb::connection();
|
|
$db->beginTransaction();try{$q=$db->query("SELECT COUNT(*) FROM jobs WHERE status IN ('queued','running')");if((int)$q->fetchColumn()>=(int)(lab_config()['queue_limit']??100)){$db->rollBack();return ['ok'=>false,'errors'=>['queue'=>'The job queue is full.']];}$cycles=$j['kind']==='planet'?1:(int)($j['spec']['run']['cycles']??1);$s=$db->prepare("INSERT INTO jobs(id,experiment_id,owner,name,spec,kind,status,phase,total_cycles,created_at,updated_at) VALUES(?,?,?,?,?,?,'queued','queued',?,?,?)");$s->execute([$newId,$j['experiment_id'],$owner,$j['name'],self::specJson($j['spec']),$j['kind'],$cycles,$now,$now]);$copy=$db->prepare('INSERT INTO job_maps(job_id,field,version_id,digest,data) SELECT ?,field,version_id,digest,data FROM job_maps WHERE job_id=?');$copy->execute([$newId,$j['id']]);$db->commit();}catch(Throwable $e){if($db->inTransaction())$db->rollBack();return ['ok'=>false,'error'=>'Could not queue retry.'];}return ['ok'=>true,'data'=>self::get($newId,$owner)];
|
|
}
|
|
public static function mapCaptures(string $id): array { $q=LabDb::connection()->prepare('SELECT field,version_id,digest,data FROM job_maps WHERE job_id=? ORDER BY field');$q->execute([$id]);$maps=[];foreach($q->fetchAll() as $row)$maps[$row['field']]=['id'=>$row['version_id'],'digest'=>$row['digest'],'data'=>$row['data']];return $maps; }
|
|
public static function validatedMapCaptures(array $job): array {
|
|
$expected=$job['spec']['maps']??[];$captures=self::mapCaptures($job['id']);
|
|
$fields=array_keys($expected);sort($fields);
|
|
if(array_keys($captures)!==$fields)return ['ok'=>false,'error'=>'Stored job map captures do not match the job spec.'];
|
|
foreach($expected as $field=>$id)if(($captures[$field]['id']??null)!==$id||!hash_equals((string)($captures[$field]['digest']??''),hash('sha256',(string)($captures[$field]['data']??''))))return ['ok'=>false,'error'=>'Stored job map capture failed validation.'];
|
|
return ['ok'=>true,'data'=>$captures];
|
|
}
|
|
public static function log($id,$owner,$offset=0): array { $j=self::get($id,$owner);if(!$j)return ['ok'=>false,'errors'=>['id'=>'Job not found.']];$path=self::directory($j).'/worker.log';$offset=max(0,(int)$offset);if(!is_file($path))return ['ok'=>true,'data'=>['text'=>'','offset'=>0,'next_offset'=>0]];$size=filesize($path);$offset=min($offset,$size);if($offset<$size-65536)$offset=$size-65536;$f=fopen($path,'rb');fseek($f,$offset);$text=fread($f,65536);$next=ftell($f);fclose($f);return ['ok'=>true,'data'=>['text'=>$text,'offset'=>$offset,'next_offset'=>$next]]; }
|
|
}
|