aljaz/turso
1
0
Fork 0
mirror of https://github.com/aljazceru/turso.git synced 2026-01-21 00:54:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

run simulation on the total list of interactions to avoid recomputing whole vec of interactions on each loop + change state management in preparation for multiple connections change

Browse Source
This commit is contained in:
pedrocarlo
2025-09-15 19:23:00 -03:00
parent faf38fe196
commit b98c22a90e
5 changed files with 135 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
simulator/generation/plan.rs
View File

@@ -99,12 +99,22 @@ impl InteractionPlan {
let _ = plan.split_off(j);
plan
}
pub fn interactions_list(&self) -> impl Iterator<Item = Interaction> {
self.plan
.clone()
.into_iter()
.flat_map(|interactions| interactions.interactions().into_iter())
}
}
pub(crate) struct InteractionPlanState {
pub(crate) stack: Vec<ResultSet>,
pub(crate) interaction_pointer: usize,
pub(crate) secondary_pointer: usize,
pub struct InteractionPlanState {
pub interaction_pointer: usize,
}
#[derive(Debug, Default)]
pub struct ConnectionState {
pub stack: Vec<ResultSet>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]

120
simulator/main.rs
View File

@@ -8,7 +8,7 @@ use rand::prelude::*;
use runner::bugbase::{Bug, BugBase, LoadedBug};
use runner::cli::{SimulatorCLI, SimulatorCommand};
use runner::env::SimulatorEnv;
use runner::execution::{Execution, ExecutionHistory, ExecutionResult, execute_plans};
use runner::execution::{Execution, ExecutionHistory, ExecutionResult, execute_interactions};
use runner::{differential, watch};
use std::any::Any;
use std::backtrace::Backtrace;
@@ -21,6 +21,7 @@ use tracing_subscriber::field::MakeExt;
use tracing_subscriber::fmt::format;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use crate::generation::plan::ConnectionState;
use crate::profiles::Profile;
use crate::runner::doublecheck;
use crate::runner::env::{Paths, SimulationPhase, SimulationType};
@@ -217,7 +218,7 @@ fn run_simulator(
mut bugbase: Option<&mut BugBase>,
cli_opts: &SimulatorCLI,
env: SimulatorEnv,
plans: Vec<InteractionPlan>,
plan: InteractionPlan,
) -> anyhow::Result<()> {
std::panic::set_hook(Box::new(move |info| {
tracing::error!("panic occurred");
@@ -239,7 +240,7 @@ fn run_simulator(
let env = Arc::new(Mutex::new(env));
let result = SandboxedResult::from(
std::panic::catch_unwind(|| {
run_simulation(env.clone(), &mut plans.clone(), last_execution.clone())
run_simulation(env.clone(), &mut plan.clone(), last_execution.clone())
}),
last_execution.clone(),
);
@@ -284,34 +285,23 @@ fn run_simulator(
tracing::info!("shrinking is disabled, skipping shrinking");
if let Some(bugbase) = bugbase.as_deref_mut() {
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
return Err(anyhow!("failed with error: '{}'", error));
}
tracing::info!("Starting to shrink");
let (shrunk_plans, shrunk) = if !cli_opts.disable_heuristic_shrinking {
let shrunk_plans = plans
.iter()
.map(|plan| {
let shrunk = plan.shrink_interaction_plan(last_execution);
tracing::info!("{}", shrunk.stats());
shrunk
})
.collect::<Vec<_>>();
let (shrunk_plan, shrunk) = if !cli_opts.disable_heuristic_shrinking {
let shrunk_plan = plan.shrink_interaction_plan(last_execution);
tracing::info!("{}", shrunk_plan.stats());
// Write the shrunk plan to a file
let shrunk_plan_path = env
.paths
.plan(&SimulationType::Default, &SimulationPhase::Shrink);
let mut f = std::fs::File::create(&shrunk_plan_path).unwrap();
tracing::trace!("writing shrunk plan to {}", shrunk_plan_path.display());
f.write_all(shrunk_plans[0].to_string().as_bytes()).unwrap();
f.write_all(shrunk_plan.to_string().as_bytes()).unwrap();
let last_execution = Arc::new(Mutex::new(*last_execution));
let env = env.clone_at_phase(SimulationPhase::Shrink);
@@ -320,15 +310,15 @@ fn run_simulator(
std::panic::catch_unwind(|| {
run_simulation(
env.clone(),
&mut shrunk_plans.clone(),
&mut shrunk_plan.clone(),
last_execution.clone(),
)
}),
last_execution,
);
(shrunk_plans, shrunk)
(shrunk_plan, shrunk)
} else {
(plans.to_vec(), result.clone())
(plan, result.clone())
};
match (&shrunk, &result) {
@@ -344,16 +334,11 @@ fn run_simulator(
tracing::trace!(
"adding bug to bugbase, seed: {}, plan: {}, error: {}",
env.opts.seed,
plans[0].plan.len(),
plan.plan.len(),
error
);
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
@@ -369,30 +354,26 @@ fn run_simulator(
let env = env.clone_at_phase(SimulationPhase::Shrink);
let env = Arc::new(Mutex::new(env));
let final_plans = if cli_opts.enable_brute_force_shrinking {
let brute_shrunk_plans = shrunk_plans
.iter()
.map(|plan| {
plan.brute_shrink_interaction_plan(&shrunk, env.clone())
})
.collect::<Vec<_>>();
let final_plan = if cli_opts.enable_brute_force_shrinking {
let brute_shrunk_plan =
shrunk_plan.brute_shrink_interaction_plan(&shrunk, env.clone());
tracing::info!("Brute force shrinking completed");
brute_shrunk_plans
brute_shrunk_plan
} else {
shrunk_plans
shrunk_plan
};
tracing::info!(
"shrinking succeeded, reduced the plan from {} to {}",
plans[0].plan.len(),
final_plans[0].plan.len()
plan.plan.len(),
final_plan.plan.len()
);
// Save the shrunk database
if let Some(bugbase) = bugbase.as_deref_mut() {
bugbase.make_shrunk(
seed,
cli_opts,
final_plans[0].clone(),
final_plan.clone(),
Some(e1.clone()),
)?;
}
@@ -410,12 +391,7 @@ fn run_simulator(
);
if let Some(bugbase) = bugbase {
bugbase
.add_bug(
env.opts.seed,
plans[0].clone(),
Some(error.clone()),
cli_opts,
)
.add_bug(env.opts.seed, plan.clone(), Some(error.clone()), cli_opts)
.unwrap();
}
Err(anyhow!("failed with error: '{}'", error))
@@ -482,7 +458,7 @@ fn setup_simulation(
bugbase: Option<&mut BugBase>,
cli_opts: &SimulatorCLI,
profile: &Profile,
) -> (u64, SimulatorEnv, Vec<InteractionPlan>) {
) -> (u64, SimulatorEnv, InteractionPlan) {
if let Some(seed) = &cli_opts.load {
let seed = seed.parse::<u64>().expect("seed should be a number");
let bugbase = bugbase.expect("BugBase must be enabled to load a bug");
@@ -521,8 +497,7 @@ fn setup_simulation(
serde_json::to_string_pretty(&plan).unwrap(),
)
.unwrap();
let plans = vec![plan];
(seed, env, plans)
(seed, env, plan)
} else {
let seed = cli_opts.seed.unwrap_or_else(|| {
let mut rng = rand::rng();
@@ -549,12 +524,9 @@ fn setup_simulation(
tracing::info!("Generating database interaction plan...");
let plans = (1..=env.opts.max_connections)
.map(|_| InteractionPlan::generate_plan(&mut env.rng.clone(), &mut env))
.collect::<Vec<_>>();
let plan = InteractionPlan::generate_plan(&mut env.rng.clone(), &mut env);
// todo: for now, we only use 1 connection, so it's safe to use the first plan.
let plan = &plans[0];
tracing::info!("{}", plan.stats());
std::fs::write(env.get_plan_path(), plan.to_string()).unwrap();
std::fs::write(
@@ -563,13 +535,13 @@ fn setup_simulation(
)
.unwrap();
(seed, env, plans)
(seed, env, plan)
}
}
fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: &mut InteractionPlan,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let simulation_type = {
@@ -580,14 +552,14 @@ fn run_simulation(
};
match simulation_type {
SimulationType::Default => run_simulation_default(env, plans, last_execution),
SimulationType::Default => run_simulation_default(env, plan, last_execution),
SimulationType::Differential => {
let limbo_env = {
let env = env.lock().unwrap();
env.clone_as(SimulationType::Default)
};
let limbo_env = Arc::new(Mutex::new(limbo_env));
differential::run_simulation(limbo_env, env, plans, last_execution)
differential::run_simulation(limbo_env, env, plan, last_execution)
}
SimulationType::Doublecheck => {
let limbo_env = {
@@ -595,28 +567,40 @@ fn run_simulation(
env.clone_as(SimulationType::Default)
};
let limbo_env = Arc::new(Mutex::new(limbo_env));
doublecheck::run_simulation(limbo_env, env, plans, last_execution)
doublecheck::run_simulation(limbo_env, env, plan, last_execution)
}
}
}
fn run_simulation_default(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: &mut InteractionPlan,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
tracing::info!("Executing database interaction plan...");
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
let num_conns = {
let env = env.lock().unwrap();
env.connections.len()
};
let mut conn_states = (0..num_conns)
.map(|_| ConnectionState::default())
.collect::<Vec<_>>();
let mut result = execute_plans(env.clone(), plans, &mut states, last_execution);
let mut state = InteractionPlanState {
interaction_pointer: 0,
};
let interactions = plan.interactions_list().collect::<Vec<_>>();
let mut result = execute_interactions(
env.clone(),
interactions,
&mut state,
&mut conn_states,
last_execution,
);
let env = env.lock().unwrap();
env.io.print_stats();

8
simulator/runner/differential.rs
View File

@@ -4,9 +4,13 @@ use sql_generation::{generation::pick_index, model::table::SimValue};
use turso_core::Value;
use crate::{
InteractionPlan,
generation::{
plan::{Interaction, InteractionPlanState, InteractionType, ResultSet}, Shadow as _
}, model::Query, runner::execution::ExecutionContinuation, InteractionPlan
Shadow as _,
plan::{Interaction, InteractionPlanState, InteractionType, ResultSet},
},
model::Query,
runner::execution::ExecutionContinuation,
};
use super::{

41
simulator/runner/doublecheck.rs
View File

@@ -18,35 +18,29 @@ use super::{
pub(crate) fn run_simulation(
env: Arc<Mutex<SimulatorEnv>>,
doublecheck_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
plan: &mut InteractionPlan,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
tracing::info!("Executing database interaction plan...");
let mut states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
.collect::<Vec<_>>();
let mut state = InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
};
let mut doublecheck_states = plans
.iter()
.map(|_| InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
})
.collect::<Vec<_>>();
let mut doublecheck_state = InteractionPlanState {
stack: vec![],
interaction_pointer: 0,
secondary_pointer: 0,
};
let mut result = execute_plans(
env.clone(),
doublecheck_env.clone(),
plans,
&mut states,
&mut doublecheck_states,
plan,
&mut state,
&mut doublecheck_state,
last_execution,
);
@@ -85,9 +79,9 @@ pub(crate) fn run_simulation(
pub(crate) fn execute_plans(
env: Arc<Mutex<SimulatorEnv>>,
doublecheck_env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
doublecheck_states: &mut [InteractionPlanState],
plans: &mut InteractionPlan,
states: &mut InteractionPlanState,
doublecheck_states: &mut InteractionPlanState,
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
@@ -99,7 +93,6 @@ pub(crate) fn execute_plans(
for _tick in 0..env.opts.ticks {
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
history.history.push(Execution::new(
connection_index,

116
simulator/runner/execution.rs
View File

@@ -1,65 +1,59 @@
use std::sync::{Arc, Mutex};
use sql_generation::generation::pick_index;
use tracing::instrument;
use turso_core::{Connection, LimboError, Result, StepResult};
use crate::generation::{
Shadow as _,
plan::{Interaction, InteractionPlan, InteractionPlanState, InteractionType, ResultSet},
plan::{ConnectionState, Interaction, InteractionPlanState, InteractionType, ResultSet},
};
use super::env::{SimConnection, SimulatorEnv};
#[derive(Debug, Clone, Copy)]
pub(crate) struct Execution {
pub(crate) connection_index: usize,
pub(crate) interaction_index: usize,
pub(crate) secondary_index: usize,
pub struct Execution {
pub connection_index: usize,
pub interaction_index: usize,
}
impl Execution {
pub(crate) fn new(
connection_index: usize,
interaction_index: usize,
secondary_index: usize,
) -> Self {
pub fn new(connection_index: usize, interaction_index: usize) -> Self {
Self {
connection_index,
interaction_index,
secondary_index,
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct ExecutionHistory {
pub(crate) history: Vec<Execution>,
pub struct ExecutionHistory {
pub history: Vec<Execution>,
}
impl ExecutionHistory {
pub(crate) fn new() -> Self {
pub fn new() -> Self {
Self {
history: Vec::new(),
}
}
}
pub(crate) struct ExecutionResult {
pub(crate) history: ExecutionHistory,
pub(crate) error: Option<LimboError>,
pub struct ExecutionResult {
pub history: ExecutionHistory,
pub error: Option<LimboError>,
}
impl ExecutionResult {
pub(crate) fn new(history: ExecutionHistory, error: Option<LimboError>) -> Self {
pub fn new(history: ExecutionHistory, error: Option<LimboError>) -> Self {
Self { history, error }
}
}
pub(crate) fn execute_plans(
pub(crate) fn execute_interactions(
env: Arc<Mutex<SimulatorEnv>>,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
interactions: Vec<Interaction>,
state: &mut InteractionPlanState,
conn_states: &mut [ConnectionState],
last_execution: Arc<Mutex<Execution>>,
) -> ExecutionResult {
let mut history = ExecutionHistory::new();
@@ -71,21 +65,24 @@ pub(crate) fn execute_plans(
for _tick in 0..env.opts.ticks {
tracing::trace!("Executing tick {}", _tick);
// Pick the connection to interact with
let connection_index = pick_index(env.connections.len(), &mut env.rng);
let state = &mut states[connection_index];
history.history.push(Execution::new(
connection_index,
state.interaction_pointer,
state.secondary_pointer,
));
if state.interaction_pointer >= interactions.len() {
break;
}
let interaction = &interactions[state.interaction_pointer];
let connection_index = interaction.connection_index;
let conn_state = &mut conn_states[connection_index];
history
.history
.push(Execution::new(connection_index, state.interaction_pointer));
let mut last_execution = last_execution.lock().unwrap();
last_execution.connection_index = connection_index;
last_execution.interaction_index = state.interaction_pointer;
last_execution.secondary_index = state.secondary_pointer;
// Execute the interaction for the selected connection
match execute_plan(&mut env, connection_index, plans, states) {
match execute_plan(&mut env, interaction, conn_state, state) {
Ok(_) => {}
Err(err) => {
return ExecutionResult::new(history, Some(err));
@@ -107,48 +104,26 @@ pub(crate) fn execute_plans(
fn execute_plan(
env: &mut SimulatorEnv,
connection_index: usize,
plans: &mut [InteractionPlan],
states: &mut [InteractionPlanState],
interaction: &Interaction,
conn_state: &mut ConnectionState,
state: &mut InteractionPlanState,
) -> Result<()> {
let connection = &env.connections[connection_index];
let plan = &mut plans[connection_index];
let state = &mut states[connection_index];
if state.interaction_pointer >= plan.plan.len() {
return Ok(());
}
let interaction = &plan.plan[state.interaction_pointer].interactions()[state.secondary_pointer];
let connection_index = interaction.connection_index;
let connection = &mut env.connections[connection_index];
if let SimConnection::Disconnected = connection {
tracing::debug!("connecting {}", connection_index);
env.connections[connection_index] = SimConnection::LimboConnection(
*connection = SimConnection::LimboConnection(
env.db.as_ref().expect("db to be Some").connect().unwrap(),
);
} else {
tracing::debug!("connection {} already connected", connection_index);
match execute_interaction(env, connection_index, interaction, &mut state.stack) {
match execute_interaction(env, interaction, &mut conn_state.stack) {
Ok(next_execution) => {
tracing::debug!("connection {} processed", connection_index);
// Move to the next interaction or property
match next_execution {
ExecutionContinuation::NextInteraction => {
if state.secondary_pointer + 1
>= plan.plan[state.interaction_pointer].interactions().len()
{
// If we have reached the end of the interactions for this property, move to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
} else {
// Otherwise, move to the next interaction
state.secondary_pointer += 1;
}
}
ExecutionContinuation::NextProperty => {
// Skip to the next property
state.interaction_pointer += 1;
state.secondary_pointer = 0;
}
}
}
@@ -170,22 +145,22 @@ fn execute_plan(
pub(crate) enum ExecutionContinuation {
/// Default continuation, execute the next interaction.
NextInteraction,
/// Typically used in the case of preconditions failures, skip to the next property.
NextProperty,
// /// Typically used in the case of preconditions failures, skip to the next property.
// NextProperty,
}
#[instrument(skip(env, interaction, stack), fields(seed = %env.opts.seed, interaction = %interaction))]
pub(crate) fn execute_interaction(
env: &mut SimulatorEnv,
connection_index: usize,
interaction: &Interaction,
stack: &mut Vec<ResultSet>,
) -> Result<ExecutionContinuation> {
// Leave this empty info! here to print the span of the execution
let connection = &mut env.connections[interaction.connection_index];
tracing::info!("");
match &interaction.interaction {
InteractionType::Query(_) => {
let conn = match &mut env.connections[connection_index] {
let conn = match connection {
SimConnection::LimboConnection(conn) => conn,
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
@@ -199,7 +174,7 @@ pub(crate) fn execute_interaction(
limbo_integrity_check(conn)?;
}
InteractionType::FsyncQuery(query) => {
let conn = match &env.connections[connection_index] {
let conn = match &connection {
SimConnection::LimboConnection(conn) => conn.clone(),
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),
@@ -216,7 +191,7 @@ pub(crate) fn execute_interaction(
InteractionType::Query(query.clone()),
);
execute_interaction(env, connection_index, &query_interaction, stack)?;
execute_interaction(env, &query_interaction, stack)?;
}
InteractionType::Assertion(_) => {
interaction.execute_assertion(stack, env)?;
@@ -228,14 +203,15 @@ pub(crate) fn execute_interaction(
if assumption_result.is_err() {
tracing::warn!("assumption failed: {:?}", assumption_result);
return Ok(ExecutionContinuation::NextProperty);
todo!("remove assumptions");
// return Ok(ExecutionContinuation::NextProperty);
}
}
InteractionType::Fault(_) => {
interaction.execute_fault(env, connection_index)?;
interaction.execute_fault(env, interaction.connection_index)?;
}
InteractionType::FaultyQuery(_) => {
let conn = match &env.connections[connection_index] {
let conn = match &connection {
SimConnection::LimboConnection(conn) => conn.clone(),
SimConnection::SQLiteConnection(_) => unreachable!(),
SimConnection::Disconnected => unreachable!(),