diff --git a/engine/crates/ava1-ctest/csrc/test_shim.c b/engine/crates/ava1-ctest/csrc/test_shim.c index 4fe392b5..c6abe93d 100644 --- a/engine/crates/ava1-ctest/csrc/test_shim.c +++ b/engine/crates/ava1-ctest/csrc/test_shim.c @@ -2469,6 +2469,92 @@ int ava1_test_reap_rules(const char *jobs_dir) { return rc; } +/* #475: an upload that ended on a session still running leaves the table on its own. Marks a listed + * job as a concluded upload (what ava1_apply_fail does once its JobDone is out), ended `ago_ms` ago. */ +static void conclude(uint8_t tag, uint64_t ago_ms) { + uint8_t id[16]; + ava1_job_t *j; + memset(id, tag, 16); + if (!(j = ava1_job_find(id))) return; + j->kind = AVA1_JOB_UPLOAD; + __atomic_store_n(&j->finished, 1, __ATOMIC_RELEASE); + __atomic_store_n(&j->concluded_ms, (ava1_mono_ms() - ago_ms) | 1u, __ATOMIC_RELEASE); + ava1_job_put(j); +} + +/* 0, or the step that failed. A full table of uploads attached to one live session: the ended ones + * make room for the next JobOpen (the one that ended longest ago and nobody holds goes first), the + * ones still running or still settling never do, and the reaper lets an ended one go 10 s after its + * end although its session never let go of it. */ +int ava1_test_ended_transfers(const char *jobs_dir) { + static const uint8_t SID[16] = { 0xe1 }; + ava1_data_cfg_t dc; + ava1_job_t *j, *held = NULL; + uint8_t id[16]; + int rc = 0, t; + memset(&dc, 0, sizeof dc); + snprintf(dc.jobs_dir, sizeof dc.jobs_dir, "%s", jobs_dir); + if (ava1_data_start(&dc) != 0) return -100; + /* 32 running uploads on one session: the table is full and nothing may be evicted */ + for (t = 0; t < AVA1_MAX_JOBS && !rc; t++) { + if (!(j = mk((uint8_t)(0x40 + t), SID))) rc = -1; + else { + j->kind = AVA1_JOB_UPLOAD; + ava1_job_put(j); + } + } + if (!rc && (j = mk(0x70, SID)) != NULL) { + ava1_job_put(j); + rc = -2; /* a full table of running jobs took another */ + } + if (!rc) { + /* 0x40 ended longest ago but a reader holds it; 0x41 ended next; 0x42 ended but still settles */ + conclude(0x40, 5000); + conclude(0x41, 4000); + conclude(0x42, 6000); + memset(id, 0x42, 16); + if ((j = ava1_job_find(id)) != NULL) { + __atomic_store_n(&j->unswept_n, 1, __ATOMIC_RELAXED); + ava1_job_put(j); + } + memset(id, 0x40, 16); + held = ava1_job_find(id); + if (!(j = mk(0x71, SID))) rc = -3; /* 0x41 makes room */ + else ava1_job_put(j); + if (!rc && listed(0x41)) rc = -4; + if (!rc && (!listed(0x40) || !listed(0x42))) rc = -5; + if (held) ava1_job_put(held); + if (!rc && !(j = mk(0x72, SID))) rc = -6; /* now 0x40 makes room */ + else if (!rc) ava1_job_put(j); + if (!rc && listed(0x40)) rc = -7; + if (!rc && (j = mk(0x73, SID)) != NULL) { /* only the settling 0x42 has ended: still full */ + ava1_job_put(j); + rc = -8; + } + } + if (!rc) { + /* the reaper: an ended upload still attached goes 10 s after its end, not before; a running one stays */ + conclude(0x43, 0); + ava1_job_reap(ava1_mono_ms() + 5000u); + if (!listed(0x43)) rc = -9; + ava1_job_reap(ava1_mono_ms() + 11000u); + if (!rc && listed(0x43)) rc = -10; + if (!rc && (!listed(0x44) || !listed(0x42))) rc = -11; + /* a copy is read by job id after it ends: never taken this way */ + memset(id, 0x45, 16); + if (!rc && (j = ava1_job_find(id)) != NULL) { + j->kind = AVA1_JOB_COPY; + __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE); + ava1_job_put(j); + ava1_job_reap(ava1_mono_ms() + 11000u); + if (!listed(0x45)) rc = -12; + } + } + ava1_job_free_all(); + ava1_data_stop(); + return rc; +} + /* N3: a job a cancel unlisted, still referenced by the caller and one other holder. */ int ava1_test_retire_unlisted(void) { ava1_data_cfg_t dc; diff --git a/engine/crates/ava1-ctest/src/lib.rs b/engine/crates/ava1-ctest/src/lib.rs index 0edca133..c2d0bdf1 100644 --- a/engine/crates/ava1-ctest/src/lib.rs +++ b/engine/crates/ava1-ctest/src/lib.rs @@ -460,6 +460,7 @@ pub mod ffi { pub fn ava1_test_data_delays(open_ms: u32, map_ms: u32); pub fn ava1_test_job_attached(id: *const u8) -> c_int; pub fn ava1_test_reap_rules(jobs_dir: *const c_char) -> c_int; + pub fn ava1_test_ended_transfers(jobs_dir: *const c_char) -> c_int; pub fn ava1_test_retire_unlisted() -> c_int; pub fn ava1_test_job_counts(id: *const u8, out: *mut u64) -> c_int; pub fn ava1_test_data_knob(name: *const c_char, v: u32) -> c_int; @@ -1071,6 +1072,14 @@ pub fn c_reap_rules(jobs: &Path) -> i32 { unsafe { ffi::ava1_test_reap_rules(j.as_ptr()) } } +/// Ended uploads on a live session leave the job table (#475), checked in C on a private data +/// layer (0 = all hold, else the step that failed). Holds the C server lock. +pub fn c_ended_transfers(jobs: &Path) -> i32 { + let _lock = C_SERVER.lock().unwrap_or_else(|e| e.into_inner()); + let j = CString::new(jobs.to_str().unwrap()).unwrap(); + unsafe { ffi::ava1_test_ended_transfers(j.as_ptr()) } +} + extern "C" { fn ava1_test_firmware(kernel_version: *const c_char, out: *mut c_char, cap: usize); } diff --git a/engine/crates/ava1-ctest/tests/many_jobs_one_session.rs b/engine/crates/ava1-ctest/tests/many_jobs_one_session.rs new file mode 100644 index 00000000..67fcc4a0 --- /dev/null +++ b/engine/crates/ava1-ctest/tests/many_jobs_one_session.rs @@ -0,0 +1,99 @@ +#![cfg(unix)] +//! Issue #475: the Files screen uploads a pick of many files one job per file, all on the +//! engine's one shared session. The console's job table holds 32 jobs, and an upload that ended +//! stayed listed until its session ended, so the 33rd file was answered BUSY for 12 retries and +//! failed ("the console stayed busy"). Ended uploads must make room on a session that runs on. +mod common; + +use std::sync::atomic::{AtomicBool, AtomicU64}; +use std::sync::Arc; + +use ava1::keys::Identity; +use ava1::peers::PeerStore; +use ava1_ctest::{c_job_count, CServer, LogOpts}; +use common::{dir, SECRET}; +use ps5upload_ava1::upload::upload_file_in; +use ps5upload_ava1::Pool; +use ps5upload_core::transfer::TransferConfig; + +fn cfg(addr: &str) -> TransferConfig { + let mut c = TransferConfig::new(addr); + c.progress_bytes = Some(Arc::new(AtomicU64::new(0))); + c.progress_files = Some(Arc::new(AtomicU64::new(0))); + c.progress_files_finalized = Some(Arc::new(AtomicU64::new(0))); + c.progress_bytes_finalized = Some(Arc::new(AtomicU64::new(0))); + c.cancel = Some(Arc::new(AtomicBool::new(false))); + c +} + +#[tokio::test(flavor = "multi_thread")] +async fn forty_single_file_jobs_in_a_row_on_one_session_all_land() { + let t = dir("many-jobs"); + let ava = t.join("ava"); + std::fs::create_dir_all(&ava).unwrap(); + let me = Identity::load_or_create(&ava.join("identity")).unwrap(); + PeerStore::load(&t.join("srv-peers")) + .unwrap() + .add(me.public(), "engine") + .unwrap(); + PeerStore::load(&ava.join("peers")) + .unwrap() + .add(Identity::from_secret(SECRET).public(), "C receiver") + .unwrap(); + let srv = CServer::start_data_opts( + SECRET, + &t.join("srv-peers"), + &t.join("jobs"), + 200, + 2000, + 2000, + 0, + 0, + LogOpts::ON, + ); + let src = t.join("src"); + let dest = t.join("dest"); + std::fs::create_dir_all(&src).unwrap(); + std::fs::create_dir_all(&dest).unwrap(); + let pool = Arc::new(Pool::new(ava.clone()).with_addr(srv.addr())); + let mut slowest = std::time::Duration::ZERO; + for n in 0..40u32 { + let f = src.join(format!("f{n}.nca")); + std::fs::write(&f, vec![n as u8; 1000 + n as usize * 37]).unwrap(); + let mut id = [0x47u8; 16]; + id[..4].copy_from_slice(&n.to_le_bytes()); + let (p, c, d, s) = ( + pool.clone(), + cfg(&srv.addr()), + dest.join(format!("f{n}.nca")).to_str().unwrap().to_string(), + f.clone(), + ); + let started = std::time::Instant::now(); + tokio::task::spawn_blocking(move || upload_file_in(&p, &c, id, &d, &s)) + .await + .unwrap() + .unwrap_or_else(|e| panic!("file {n}: {e:#} (jobs listed: {})", c_job_count())); + slowest = slowest.max(started.elapsed()); + assert!(c_job_count() <= 32); + } + for n in 0..40u32 { + assert_eq!( + std::fs::read(dest.join(format!("f{n}.nca"))).unwrap(), + std::fs::read(src.join(format!("f{n}.nca"))).unwrap() + ); + } + // A BUSY answer costs the sender at least one backoff (250 ms and doubling); a job that + // waited for a slot shows up as a slow file. + assert!( + slowest < std::time::Duration::from_secs(5), + "a file waited for a table slot: slowest {slowest:?}" + ); + // The session is still up (the pool holds it): the ended jobs leave anyway. + ava1_ctest::c_reap_far(); + assert_eq!( + c_job_count(), + 0, + "ended jobs stayed listed on a live session" + ); + drop(pool); +} diff --git a/engine/crates/ava1-ctest/tests/wire_upload.rs b/engine/crates/ava1-ctest/tests/wire_upload.rs index 692386c7..20def40f 100644 --- a/engine/crates/ava1-ctest/tests/wire_upload.rs +++ b/engine/crates/ava1-ctest/tests/wire_upload.rs @@ -640,6 +640,15 @@ fn the_reaper_takes_parked_jobs_only_and_never_one_just_attached() { assert_eq!(ava1_ctest::c_reap_rules(&d.join("jobs")), 0); } +#[test] +fn ended_uploads_leave_the_table_while_their_session_runs_on() { + // #475, in C: a session runs job after job, so the uploads that ended on it make room for + // the next JobOpen (the one that ended longest ago and nobody holds goes first) and leave + // 10 s after their end; running or still-settling jobs never do, nor a copy. + let d = dir("wire-ended"); + assert_eq!(ava1_ctest::c_ended_transfers(&d.join("jobs")), 0); +} + #[tokio::test(flavor = "multi_thread")] async fn the_server_restarts_on_the_same_port() { let d = dir("wire-restart"); diff --git a/payload/ava1/ava1_apply.c b/payload/ava1/ava1_apply.c index 04750a39..c1b1bf5d 100644 --- a/payload/ava1/ava1_apply.c +++ b/payload/ava1/ava1_apply.c @@ -524,6 +524,7 @@ void ava1_apply_fail(ava1_job_t *j, uint16_t status, const char *what, int err, } pthread_mutex_unlock(&j->mu); emit_done(j); + __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE); /* the table may now let it go (ava1_job.c) */ } void ava1_apply_done_again(ava1_job_t *j) { diff --git a/payload/ava1/ava1_job.c b/payload/ava1/ava1_job.c index 029f5bed..f3670bca 100644 --- a/payload/ava1/ava1_job.c +++ b/payload/ava1/ava1_job.c @@ -71,8 +71,36 @@ ava1_job_t *ava1_job_find(const uint8_t id[16]) { return j; } +/* An upload or download that ended and has nothing left to settle: what it still holds (its threads, its + * slot) serves only a late Status or a reopen, and a reopen replays its journal (an upload) or starts over (a + * download). A copy, an operation and a send to another console keep their own rules: their status is read + * by job id after they end. `concluded_ms` (set after `finished`, once the job's JobDone is out: its last commit + * and journal append are behind it) and `unswept_n` are read without j->mu, as in the reaper. */ +static int transfer_ended(const ava1_job_t *j) { + return (j->kind == AVA1_JOB_UPLOAD || j->kind == AVA1_JOB_DOWNLOAD) && !j->c2c && + __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) && !__atomic_load_n(&j->unswept_n, __ATOMIC_RELAXED); +} + +/* Caller holds T.mu, and the table is full: the ended transfer nobody else holds (only the table's + * reference) that ended longest ago, or -1. A session runs job after job (the Files screen sends a pick of + * files one job each, #475), so its ended jobs must make room for the next one rather than answer BUSY. */ +static int evictable_locked(void) { + int i, best = -1; + for (i = 0; i < AVA1_MAX_JOBS; i++) { + ava1_job_t *j = T.jobs[i]; + if (!j || j->refs != 1 || !transfer_ended(j)) continue; + if (best < 0 || __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) < + __atomic_load_n(&T.jobs[best]->concluded_ms, __ATOMIC_ACQUIRE)) + best = i; + } + return best; +} + +static void unlist(ava1_job_t *j); + static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const uint8_t *sid) { ava1_job_t *j = calloc(1, sizeof *j); + ava1_job_t *evicted = NULL; int i, slot = -1; if (!j) return NULL; memcpy(j->id, id, 16); @@ -100,6 +128,11 @@ static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const u } if (!T.jobs[i] && slot == -1) slot = i; } + if (slot == -1 && (i = evictable_locked()) >= 0) { + evicted = T.jobs[i]; + unlist(evicted); /* its id stays retiring until the put below has destroyed it */ + slot = i; + } if (slot >= 0) for (i = 0; i < (int)T.retiring_n; i++) if (memcmp(T.retiring[i], id, 16) == 0) { @@ -112,6 +145,7 @@ static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const u if (slot >= 0 && T.retiring_n >= AVA1_MAX_JOBS) slot = -2; if (slot >= 0) T.jobs[slot] = j; pthread_mutex_unlock(&T.mu); + if (evicted) ava1_job_put(evicted); /* the table's reference, the last one: joins its idle threads */ if (slot < 0) { pthread_mutex_destroy(&j->mu); pthread_mutex_destroy(&j->jnl_mu); @@ -357,7 +391,18 @@ void ava1_job_reap(uint64_t now_ms) { * j->mu: that lock can be held across disk I/O, and waiting for it under T.mu would * stall every reader thread's job lookup. It only ever goes 0 -> 1, so a stale read * reaps later. */ - if (!j || j->attached || j->parked_at_ms == 0) continue; + if (!j) continue; + if (j->attached) { + /* An upload or download ends on the session that ran it, and that session goes on to the + * next job: the ended one leaves the table 10 s after it ended, as if its session had let go + * of it (SPEC.md §11.7), not when the session ends (#475). */ + if (transfer_ended(j) && now_ms > __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) + done_age) { + gone[n++] = j; + unlist(j); + } + continue; + } + if (j->parked_at_ms == 0) continue; fin = __atomic_load_n(&j->finished, __ATOMIC_ACQUIRE); /* Files still settling (durable-by-log) keep their job alive for the full park age: it owns the * sweep, and the engine polls its Status for `unswept`. */ diff --git a/payload/ava1/ava1_job.h b/payload/ava1/ava1_job.h index 9275dffe..c7739116 100644 --- a/payload/ava1/ava1_job.h +++ b/payload/ava1/ava1_job.h @@ -92,6 +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; + uint64_t concluded_ms; /* an upload or download: when its end went out (our JobDone) or came in (the peer's), + * nothing of it in flight after; 0 before. The table lets it go from then (#475) */ 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 */ diff --git a/payload/ava1/ava1_send.c b/payload/ava1/ava1_send.c index 122ab7a4..6433a568 100644 --- a/payload/ava1/ava1_send.c +++ b/payload/ava1/ava1_send.c @@ -734,6 +734,7 @@ static int on_frame(ava1_job_t *j, uint8_t type, const uint8_t *body, size_t len } s->stop = 1; j->finished = 1; + __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE); /* the peer ended it: the table may let it go */ pthread_cond_broadcast(&j->cv); pthread_mutex_unlock(&j->mu); return 0; diff --git a/protocol/ava1/SPEC.md b/protocol/ava1/SPEC.md index cfd0a224..34e68411 100644 --- a/protocol/ava1/SPEC.md +++ b/protocol/ava1/SPEC.md @@ -720,7 +720,10 @@ or `job.copy` whose root is, or lies inside or around, the root of another job t (and, for a move, its source) is answered `ERR_BUSY`; parked jobs count, because they can resume. When a session ends its jobs are parked, not ended: they stay listed, detached, for 10 minutes and then leave the table; their journal stays on disk (§14.3), so a later `JobOpen` resumes them. -A finished upload or download leaves the table 10 s after its session lets go of it; a finished local job +A finished upload or download leaves the table 10 s after it ended (its `JobDone` went out or came in), whether +or not its session goes on: a session runs job after job, so the jobs it ended must not hold the table until it +ends. A `JobOpen` that finds the table full takes the slot of the ended upload or download (with nothing left +to settle) that ended longest ago instead of answering `ERR_BUSY` (a reopen of an upload replays its journal; a download starts over). A finished local job (`job.copy`) stays listed for the full park age so `job.status` can still answer. Either peer ends a job with `JobCancel{job_id, reason}`, where `reason` is the `ERR_*` code the ender wants reported (`ERR_CANCELLED` for a user's cancel, `ERR_IO` or `ERR_VERIFY` when a sender's source fails); the