use super::*;
use saddle_core::ComponentLifecycle;
use saddle_runtime::startup_assembly::StartupDbPoolFactory;
use saddle_runtime::{profusegw::*, request_task::reserved::*};
use std::{
alloc::Layout,
future::Future,
pin::pin,
sync::{
Arc,
atomic::{AtomicUsize, Ordering},
},
time::Duration,
};
static POLLS: AtomicUsize = AtomicUsize::new(0);
static DROPS: AtomicUsize = AtomicUsize::new(0);
static ENTERED: tokio::sync::Notify = tokio::sync::Notify::const_new();
struct Watch(Cell<u64>);
impl Drop for Watch {
fn drop(&mut self) {
DROPS.fetch_add(1, Ordering::SeqCst);
}
}
saddle::transaction_logic! {
fn wait_logic(tx: Accounts; watch:Watch) -> (u64,()) {
POLLS.fetch_add(1,Ordering::SeqCst);
let row=tx.read(ops::ReadParams{id:1}).await
.map_err(|e|saddle::database::TransactionAbort::Technical(e.into_failure()))?;
watch.0.set(row.unwrap().value);
ENTERED.notify_one();
std::future::pending::<()>().await;
Ok(watch.0.get())
}
}
saddle::transaction_logic! {
fn large_logic(_tx: Accounts; watch:Watch) -> (u64,()) {
POLLS.fetch_add(1,Ordering::SeqCst);
let bytes=[7_u8;65536];
tokio::task::yield_now().await;
std::hint::black_box(bytes);
Ok(watch.0.get())
}
}
type Cancel = tokio::sync::futures::OwnedNotified;
struct State {
cap: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
mode: u8,
cancel: Option<Cancel>,
serial: Option<ProfuseGwSerialScope<Cancel>>,
terminal: Option<ProfuseGwPostDatabaseRequestTerminal>,
watch: Option<Watch>,
}
type Owner = saddle_runtime::profusegw::ReservedDispatchOwner<State>;
fn prepared<R, B: __saddle_cf::Logic<R>>(
cap: &Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
body: B,
) -> __saddle_cf::Prepared<R, B> {
__saddle_cf::Prepared {
body,
target: saddle::transaction_construction::target(cap.clone()),
route: std::marker::PhantomData,
}
}
async fn body(
owner: &mut Owner,
context: ReservedTaskContext,
) -> Result<(), ReservedRequestFailure> {
owner.1.serial = Some(
owner
.0
.take()
.unwrap()
.into_parameter_scope(owner.1.cancel.take().unwrap()),
);
let isolation = saddle::database::TransactionIsolation::RepeatableRead;
let result = match owner.1.mode {
0 => {
saddle::transaction_construction::submit(
&owner.1.cap,
&mut owner.1.serial,
&context,
isolation,
prepared(&owner.1.cap, module_logic(1, Cell::new(0))),
)
.await
}
1 => {
saddle::transaction_construction::submit(
&owner.1.cap,
&mut owner.1.serial,
&context,
isolation,
prepared(&owner.1.cap, route_logic(Cell::new(0))),
)
.await
}
_ => {
saddle::transaction_construction::submit(
&owner.1.cap,
&mut owner.1.serial,
&context,
isolation,
prepared(&owner.1.cap, wait_logic(Watch(Cell::new(0)))),
)
.await
}
};
let complete = match result {
Ok(c) => c,
Err(e) => {
let suspended = owner
.1
.cap
.finish_reserved_without_connection(owner.1.serial.take().unwrap(), e);
let (_, terminal) = suspended.into_response_parts().ok().unwrap();
owner.1.terminal = Some(terminal);
panic!("unexpected outer preparation refusal");
}
};
let (result, terminal) = complete.suspended.into_response_parts().ok().unwrap();
owner.1.terminal = Some(terminal); use saddle_db::internal::{
ScopeDatabaseError as E, ScopeTransactionAbort as A, ScopeTransactionOutcome as O,
};
match (owner.1.mode, result.outcome) {
(0, O::Committed(20)) | (1, O::Committed(11)) => (),
(2, O::Rejected(A::Technical(E::Resource))) => (),
(3, O::Rejected(A::Technical(E::Cancelled))) => (),
(_, other) => panic!("unexpected outcome {other:?}"),
}
if let Some(original) = complete.unretained {
return Err(original);
}
Ok(())
}
fn factory(owner: &mut Owner, context: ReservedTaskContext) -> ReservedBorrowedFuture<'_, ()> {
Box::pin(body(owner, context))
}
async fn oversized_body(
owner: &mut Owner,
context: ReservedTaskContext,
) -> Result<(), ReservedRequestFailure> {
POLLS.fetch_add(1, Ordering::SeqCst);
owner.1.serial = Some(
owner
.0
.take()
.unwrap()
.into_parameter_scope(owner.1.cancel.take().unwrap()),
);
let _completion = saddle::transaction_construction::submit(
&owner.1.cap,
&mut owner.1.serial,
&context,
saddle::database::TransactionIsolation::RepeatableRead,
prepared(&owner.1.cap, large_logic(owner.1.watch.take().unwrap())),
)
.await;
panic!("oversized body must not be polled")
}
fn oversized_factory(
owner: &mut Owner,
context: ReservedTaskContext,
) -> ReservedBorrowedFuture<'_, ()> {
Box::pin(oversized_body(owner, context))
}
fn layout<I, O>(_: impl FnOnce(I) -> O) -> Layout {
Layout::new::<O>()
}
fn budget() -> saddle_admission::DeploymentResourceBudget {
let pending = saddle_admission::freeze_deployment_resource_budget(
1, 32, 5000, 1, 1, 1_000_000, 1_000_000, 1_000_000, 1_000_000,
)
.unwrap();
let (app, listener) = saddle_core::BootstrapRendezvousIssuer::issue()
.freeze_application(saddle_core::GeneratedApplicationFreezeSource::new(
"app",
b"descriptor",
&["route"],
))
.unwrap();
let listener = listener
.freeze_listener(saddle_core::ListenerStartupFreezeSource::new(
"app",
"127.0.0.1:8000".parse().unwrap(),
"127.0.0.1:9000".parse().unwrap(),
Duration::from_millis(5000),
))
.ok()
.unwrap();
let (whole, receipt) = saddle_core::pair_bootstrap_rendezvous(app, listener)
.ok()
.unwrap();
saddle_admission::bind_deployment_resource_budget_bootstrap(pending, whole, receipt)
.ok()
.unwrap()
}
pub fn run() {
let mut coordinator = pin!(coordinate_profusegw_app_run(budget(), |process| {
std::future::ready(run_profusegw_owned_application(
process,
|lease| async move {
let config = saddle_db::DatabaseConfig::new(
std::env::var("SADDLE_TEST_DATABASE_URL").unwrap(),
)
.name_mapping_directory(std::env::var("SADDLE_SESSION_MAPPING_DIR").unwrap())
.register_owned_query::<ops::Read>()
.register_owned_write::<ops::Write>();
let observer = saddle_observability::Observer::with_writer(
Default::default(),
std::io::sink(),
)
.unwrap();
let pool =
saddle_db::internal::StartupManagedDatabaseFactory::new(Some(config), observer)
.construct(saddle_admission::DbCreditProfile {
connections: 1,
operations: 1,
})
.await
.unwrap();
pool.start().await.unwrap();
let cap = Arc::new(
pool.issue_process_capability(lease.take_database_startup_half().unwrap())
.ok()
.unwrap(),
);
for mode in 0..4 {
let notify = Arc::new(tokio::sync::Notify::new());
let polls = POLLS.load(Ordering::SeqCst);
let drops = DROPS.load(Ordering::SeqCst);
if mode == 2 {
let actual = layout(|(o, c): (&'static mut Owner, ReservedTaskContext)| {
oversized_body(o, c)
});
let admission = lease.try_reserved_dispatch(
saddle_core::ContextLabel::checked("cf-app").unwrap(),
None,
actual,
&[],
State {
cap: cap.clone(),
mode,
cancel: Some(notify.clone().notified_owned()),
serial: None,
terminal: None,
watch: Some(Watch(Cell::new(42))),
},
oversized_factory,
);
let ReservedDispatchOutcome::Rejected {
input,
make: _,
reason,
} = admission
else {
panic!("oversized concrete construction unexpectedly admitted")
};
match reason {
ReservedDispatchRejection::Admitted {
admission: _,
failure: saddle_runtime::profusegw::ReservedDispatchConstructionFailure::Storage(
saddle_admission::AdmissionError::BudgetExceeded {
requested,
available,
}
| saddle_admission::AdmissionError::FrameworkReserveExceeded {
requested,
available,
}
| saddle_admission::AdmissionError::TaskReserveExceeded {
requested,
available,
},
) } => {
assert!(requested > available);
println!(
"CF permit refusal requested={requested} available={available}"
);
}
_ => panic!("refusal was not a concrete storage shortage"),
}
assert_eq!(input.watch.as_ref().unwrap().0.get(), 42);
assert_eq!(POLLS.load(Ordering::SeqCst), polls);
assert_eq!(DROPS.load(Ordering::SeqCst), drops);
drop(input);
assert_eq!(DROPS.load(Ordering::SeqCst), drops + 1);
println!(
"CF_CASE_PASS mode=2 preallocation_rejected original_input no_poll body_bytes={}",
actual.size()
);
continue;
}
let actual =
layout(|(o, c): (&'static mut Owner, ReservedTaskContext)| body(o, c));
println!(
"CF outer body={} owner={} mode={mode}",
actual.size(),
std::mem::size_of::<Owner>()
);
let admission = lease.try_reserved_dispatch(
saddle_core::ContextLabel::checked("cf-app").unwrap(),
None,
actual,
&[],
State {
cap: cap.clone(),
mode,
cancel: Some(notify.clone().notified_owned()),
serial: None,
terminal: None,
watch: None,
},
factory,
);
let ReservedDispatchOutcome::Ready {
root,
future,
mut ticket,
..
} = admission
else {
panic!("actual CF owner/body not within original profile")
};
let held = root
.view(saddle_core::request_context::RequestViewPhase::Reading)
.unwrap();
let mut tasks = tokio::task::JoinSet::new();
let task = tasks.spawn(future);
ticket.bind(task.id()).unwrap();
if mode == 3 {
ENTERED.notified().await;
notify.notify_one();
}
let mut joined = ticket
.complete(tasks.join_next().await.unwrap())
.ok()
.unwrap();
let terminal = joined.owner_mut().1.terminal.take().unwrap();
finish_profusegw_after_database(terminal);
let recovered = joined.recover(Default::default()).ok().unwrap();
assert!(matches!(recovered.result, Some(Ok(()))));
drop((root, recovered));
if mode == 2 {
assert_eq!(POLLS.load(Ordering::SeqCst), polls);
assert_eq!(DROPS.load(Ordering::SeqCst), drops + 1);
}
if mode == 3 {
assert_eq!(POLLS.load(Ordering::SeqCst), polls + 1);
assert_eq!(DROPS.load(Ordering::SeqCst), drops + 1);
}
assert!(matches!(
lease.try_admit(),
ProfuseGwCoordinatorAdmissionOutcome::CapacityRejected(_)
));
drop(held);
let ProfuseGwCoordinatorAdmissionOutcome::Ready(owner, _) = lease.try_admit()
else {
panic!("last reference did not release original slot")
};
assert!(finish_profusegw_without_database(owner, ()).is_ok());
println!("CF_CASE_PASS mode={mode} matching_join last_reference_refund");
}
drop(cap);
pool.shutdown().await.unwrap();
drop(pool);
assert!(
std::process::Command::new("kill")
.args(["-TERM", &std::process::id().to_string()])
.status()
.unwrap()
.success()
);
Ok(saddle_runtime::Application::new())
},
))
}));
let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
assert!(matches!(
coordinator.as_mut().poll(&mut cx),
std::task::Poll::Ready(Ok(()))
));
}