diff --git a/engine/crates/ava1-ctest/src/lib.rs b/engine/crates/ava1-ctest/src/lib.rs index 3800285e..0edca133 100644 --- a/engine/crates/ava1-ctest/src/lib.rs +++ b/engine/crates/ava1-ctest/src/lib.rs @@ -2507,6 +2507,42 @@ extern "C" { fn ava1_jobs_list_all(me: *const u8, out: *mut u8, cap: usize, out_len: *mut usize) -> c_int; fn ava1_jobs_owner_is(id: *const u8, me: *const u8) -> c_int; fn ava1_job_cancel_any(id: *const u8) -> c_int; + fn ava1_jobs_release_seen(me: *const u8); + fn ava1_server_set_rpc_peer(peer: *const u8); + fn ava1_data_rpc( + method: u16, + body: *const u8, + len: u32, + out: *mut u8, + cap: usize, + out_len: *mut usize, + ) -> c_int; +} + +/// `ava1_jobs_release_seen` (the console page read /api/jobs): `me`'s finished operations and +/// copies keep only the grace. +pub fn c_jobs_release_seen(me: &[u8; 32]) { + unsafe { ava1_jobs_release_seen(me.as_ptr()) } +} + +/// A data-layer RPC (`job.copy`, `job.run`, ...) as `me`, the way the console page's routes call +/// it (`ava1_server_set_rpc_peer` on the calling thread, then the data plane): (status, reply). +pub fn c_data_rpc_as(me: &[u8; 32], method: u16, body: &[u8]) -> (i32, Vec) { + let mut out = vec![0u8; 64 * 1024]; + let mut n = 0usize; + let rc = unsafe { + ava1_server_set_rpc_peer(me.as_ptr()); + ava1_data_rpc( + method, + body.as_ptr(), + body.len() as u32, + out.as_mut_ptr(), + out.len(), + &mut n, + ) + }; + out.truncate(n); + (rc, out) } /// `ava1_jobs_list_all` (the console page's job list): every job the data layer holds, of diff --git a/engine/crates/ava1-ctest/tests/jobs_any_owner.rs b/engine/crates/ava1-ctest/tests/jobs_any_owner.rs index d88ee82c..b4c89608 100644 --- a/engine/crates/ava1-ctest/tests/jobs_any_owner.rs +++ b/engine/crates/ava1-ctest/tests/jobs_any_owner.rs @@ -184,3 +184,167 @@ async fn job_list_rpc_unchanged_for_pcs() { assert_eq!(all, vec![ja, jb]); drop(t.srv); } + +const FSCK_NOW: &[u8] = br#"{"device":"/dev/md1","loops":0}"#; + +fn fsck_body(job: [u8; 16]) -> Vec { + JobRun { + job_id: job, + op: gen::JOB_OP_FSCK, + args: FSCK_NOW.to_vec(), + } + .to_bytes() + .unwrap() +} + +/// `job.run` of a quick fsck as the console page (the page's routes call the data layer with its +/// identity set on the thread): the status. +fn run_as_console(job: [u8; 16]) -> i32 { + c_data_rpc_as(&CONSOLE, gen::METHOD_JOB_RUN, &fsck_body(job)).0 +} + +/// `job.copy` as the console page. +fn copy_as_console(job: [u8; 16], src: &std::path::Path, dest: &std::path::Path) -> i32 { + let body = JobCopy { + job_id: job, + src: src.to_str().unwrap().into(), + dest: dest.to_str().unwrap().into(), + flags: 0, + }; + c_data_rpc_as(&CONSOLE, gen::METHOD_JOB_COPY, &body.to_bytes().unwrap()).0 +} + +/// Waits until every job in `ids` is listed and finished. +async fn all_finished(ids: &[[u8; 16]]) { + for _ in 0..400 { + let all = list_all(&CONSOLE); + if ids + .iter() + .all(|id| all.iter().any(|e| e.job_id == *id && e.state != 0)) + { + return; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + panic!("jobs did not finish"); +} + +/// Final review: the 32-slot job table is shared by every owner. Finished jobs hold their slot +/// until they are released; the console page lists all 32 and, once it has read them finished, +/// releases only its own, which frees room for new jobs. +#[tokio::test(flavor = "multi_thread")] +async fn table_holds_32_jobs_and_seen_console_jobs_make_room() { + let t = two("any-cap32", 0).await; + let mut pc = Vec::new(); + let mut mine = Vec::new(); + for n in 0..32u8 { + let id = [0x40 + n; 16]; + // At most eight operations run at once: a quick one is retried while the others end. + let mut st = gen::ERR_BUSY as i32; + for _ in 0..200 { + st = if n % 2 == 0 { + run_fsck_now(&t.a, id).await as i32 + } else { + run_as_console(id) + }; + if st != gen::ERR_BUSY as i32 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(st, gen::STATUS_OK as i32, "job {n}"); + if n % 2 == 0 { + pc.push(id) + } else { + mine.push(id) + } + } + let every: Vec<_> = pc.iter().chain(mine.iter()).copied().collect(); + all_finished(&every).await; + // list_all fits all 32 in one reply, whoever owns them. + let all = list_all(&CONSOLE); + assert_eq!(all.len(), 32); + for id in &mine { + assert_eq!(c_jobs_owner_is(id, &CONSOLE), 1); + } + // Full, and nothing was delivered: a 33rd job is refused, for the page as for a PC. + assert_eq!(run_as_console([0x70; 16]), gen::ERR_BUSY as i32); + assert_eq!(run_fsck_now(&t.b, [0x71; 16]).await, gen::ERR_BUSY); + // The page read them finished: its own are released (oldest first when room is needed) ... + c_jobs_release_seen(&CONSOLE); + assert_eq!(list_all(&CONSOLE).len(), 32, "the grace keeps them listed"); + assert_eq!(run_as_console([0x70; 16]), gen::STATUS_OK as i32); + assert_eq!(run_fsck_now(&t.b, [0x71; 16]).await, gen::STATUS_OK); + // ... and a PC's job never is. + let all = list_all(&CONSOLE); + assert_eq!(all.len(), 32); + for id in &pc { + assert!( + all.iter().any(|e| e.job_id == *id), + "a PC's job stays listed" + ); + } + drop(t.srv); +} + +/// A finished copy or operation the console page started is released a grace after the page first +/// read it finished (not the 10-minute park age); a paired PC's finished jobs stay listed. +#[tokio::test(flavor = "multi_thread")] +async fn seen_console_jobs_leave_after_the_grace_pc_jobs_stay() { + let t = two("any-seen", 0).await; + write_tree(&t.d.join("usb/g"), 5, |_| 1024); + let (my_op, my_copy, pc_op, pc_copy) = ([0x81; 16], [0x82; 16], [0x83; 16], [0x84; 16]); + assert_eq!(run_as_console(my_op), gen::STATUS_OK as i32); + assert_eq!( + copy_as_console(my_copy, &t.d.join("usb/g"), &t.d.join("data/mine")), + gen::STATUS_OK as i32 + ); + assert_eq!(run_fsck_now(&t.a, pc_op).await, gen::STATUS_OK); + assert_eq!( + copy(&t.a, pc_copy, &t.d.join("usb/g"), &t.d.join("data/pc")).await, + gen::STATUS_OK + ); + all_finished(&[my_op, my_copy, pc_op, pc_copy]).await; + assert_eq!(c_jobs_owner_is(&my_copy, &CONSOLE), 1); + c_jobs_release_seen(&CONSOLE); + let started = std::time::Instant::now(); + let listed = |id: [u8; 16]| list_all(&CONSOLE).iter().any(|e| e.job_id == id); + assert!( + listed(my_op) && listed(my_copy), + "still listed during the grace" + ); + // The page reads /api/jobs about once a second. + let mut gone = false; + while started.elapsed() < Duration::from_secs(20) { + c_jobs_release_seen(&CONSOLE); + if !listed(my_op) && !listed(my_copy) { + gone = true; + break; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + assert!( + gone, + "the page's finished jobs are released after the grace" + ); + assert!( + started.elapsed() >= Duration::from_secs(9), + "not before the grace: {:?}", + started.elapsed() + ); + assert!(listed(pc_op), "a PC's finished operation stays"); + assert!(listed(pc_copy), "a PC's finished copy stays"); + // The PC still reads its own results. + assert_eq!( + code(&t.a, gen::METHOD_JOB_STATUS, pc_copy).await, + gen::STATUS_OK + ); + drop(t.srv); +} + +async fn run_fsck_now(s: &Session, job: [u8; 16]) -> u16 { + s.rpc(gen::METHOD_JOB_RUN, &fsck_body(job)) + .await + .unwrap() + .status +} diff --git a/payload/ava1/ava1_job.c b/payload/ava1/ava1_job.c index bf4ed3c3..029f5bed 100644 --- a/payload/ava1/ava1_job.c +++ b/payload/ava1/ava1_job.c @@ -377,7 +377,9 @@ void ava1_job_reap(uint64_t now_ms) { if (j->kind == AVA1_JOB_OPKIND ? (fin && now_ms - j->parked_at_ms > ava1_op_keep_ms(__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE))) : j->kind == AVA1_JOB_COPY || j->c2c - ? (fin && now_ms - j->parked_at_ms > age) + /* a copy the console page has seen finished (ava1_jobs_release_seen) keeps only the grace */ + ? (fin && now_ms - j->parked_at_ms > + (__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE) ? ava1_op_keep_ms(1) : age)) : (now_ms - j->parked_at_ms > age || (fin && now_ms - j->parked_at_ms > done_age))) { gone[n++] = j; unlist(j); diff --git a/payload/ava1/ava1_job.h b/payload/ava1/ava1_job.h index 7f461944..9275dffe 100644 --- a/payload/ava1/ava1_job.h +++ b/payload/ava1/ava1_job.h @@ -92,7 +92,8 @@ struct ava1_job { int attached, refs, stopping, finished, prepared; int discard_parts; /* a cancelled local copy: destroy removes the .ava-part files this job itself was writing (final review #9) */ uint64_t parked_at_ms; - int op_delivered; /* an operation job: its terminal status was first delivered at parked_at_ms (the grace starts) */ + int op_delivered; /* an operation job, or a console-page copy (ava1_jobs_release_seen): its terminal status was + * first delivered at parked_at_ms (the grace starts) */ int c2c; /* a send to another console (SPEC.md ยง18): kept the full park age once ended */ uint8_t c2c_by[32]; /* ... and the device that asked for it, which may read its status or cancel it */ uint8_t kind, policy; diff --git a/payload/ava1/ava1_op.c b/payload/ava1/ava1_op.c index c61e9468..9d5be044 100644 --- a/payload/ava1/ava1_op.c +++ b/payload/ava1/ava1_op.c @@ -460,6 +460,45 @@ int ava1_jobs_list_all(const uint8_t me[32], uint8_t *out, size_t cap, size_t *o return list_jobs(NULL, out, cap, out_len); } +typedef struct { + const uint8_t *me; + uint8_t id[AVA1_MAX_JOBS][16]; + unsigned n; +} seen_t; + +static void seen_one(ava1_job_t *j, void *ctx) { + seen_t *s = ctx; + if (s->n >= AVA1_MAX_JOBS || memcmp(j->owner, s->me, 32) != 0) return; + if (j->kind != AVA1_JOB_OPKIND && j->kind != AVA1_JOB_COPY) return; + if (!__atomic_load_n(&j->finished, __ATOMIC_ACQUIRE)) return; + /* the listing's rule (list_one): a move is running until its source is gone */ + if (j->kind == AVA1_JOB_COPY && __atomic_load_n(&j->copy_move, __ATOMIC_ACQUIRE) && + !__atomic_load_n(&j->copy_delete_done, __ATOMIC_ACQUIRE)) + return; + memcpy(s->id[s->n++], j->id, 16); +} + +void ava1_jobs_release_seen(const uint8_t me[32]) { + seen_t *s = calloc(1, sizeof *s); + unsigned i; + if (!s) return; + s->me = me; + ava1_job_foreach(seen_one, s); /* under the table lock: only collects ids */ + for (i = 0; i < s->n; i++) { + ava1_job_t *j = ava1_job_find(s->id[i]); + if (!j) continue; + if (j->kind == AVA1_JOB_OPKIND) { + ava1_op_status_delivered(j, 1); + } else if (!__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE)) { + /* a copy: the reaper releases it once the grace from now has passed */ + __atomic_store_n(&j->parked_at_ms, ava1_mono_ms(), __ATOMIC_RELEASE); + __atomic_store_n(&j->op_delivered, 1, __ATOMIC_RELEASE); + } + ava1_job_put(j); + } + free(s); +} + int ava1_jobs_owner_is(const uint8_t id[16], const uint8_t me[32]) { ava1_job_t *j = ava1_job_find(id); int mine; diff --git a/payload/ava1/ava1_op.h b/payload/ava1/ava1_op.h index 441550a6..7a4c6f8b 100644 --- a/payload/ava1/ava1_op.h +++ b/payload/ava1/ava1_op.h @@ -57,6 +57,15 @@ int ava1_op_list_rpc(const uint8_t owner[32], uint8_t *out, size_t cap, size_t * int ava1_jobs_list_all(const uint8_t me[32], uint8_t *out, size_t cap, size_t *out_len); /* 1 when job `id` is listed and `me` owns it, else 0 (an absent job included). */ int ava1_jobs_owner_is(const uint8_t id[16], const uint8_t me[32]); +/* The console page has read the job list: every FINISHED operation or copy that `me` owns is + * treated as delivered, exactly as a JOB_STATUS reply delivers an operation's end + * (ava1_op_status_delivered): it stays listed for the grace (AVA1_OP_GRACE_MS, capped by the park + * age) after the first such read, so other readers still see how it ended, and is then released + * (an operation also by the next read after the grace, and at once when the table is full). Without + * this a page-started copy stayed listed the full park age (10 min) and 32 of them filled the table + * every owner shares. A move still deleting its source is not finished. Jobs of any other owner (a + * paired PC's) are never touched. Never call it with the table lock held. */ +void ava1_jobs_release_seen(const uint8_t me[32]); /* Encodes the Status of an operation job (job.status / job.run reply). */ int ava1_op_encode_status(ava1_job_t *j, uint8_t *out, size_t cap, size_t *out_len); /* job.cancel on an operation job: raises the flag and returns (never waits); the worker stops at