Skip to content

Fix wait_for_ops logic #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Handy-caT merged 3 commits into from
Jul 4, 2025
Merged

Fix wait_for_ops logic #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f83f237
wait correction
Handy-caT Jul 4, 2025
c093278
add recheck for long waits just in case
Handy-caT Jul 4, 2025
7da3542
clippy
Handy-caT Jul 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ members = ["codegen", "examples", "performance_measurement", "performance_measur

[package]
name = "worktable"
version = "0.6.7"
version = "0.6.8"
edition = "2024"
authors = ["Handy-caT"]
license = "MIT"
Expand Down
84 changes: 55 additions & 29 deletions src/persistence/task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,19 @@
use data_bucket::page::PageId;
use std::collections::HashSet;
use std::fmt::Debug;
use std::sync::atomic::{AtomicBool, AtomicU16, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use worktable_codegen::worktable;

use crate::persistence::operation::{
BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId, PosByOpIdQuery,
};
use crate::persistence::PersistenceEngineOps;
use crate::prelude::*;
use crate::util::OptimizedVec;

use std::collections::HashSet;
use std::fmt::Debug;
use std::sync::atomic::{AtomicU16, Ordering};
use std::sync::Arc;

use data_bucket::page::PageId;
use tokio::sync::Notify;
use worktable_codegen::worktable;

worktable! (
name: QueueInner,
columns: {
Expand All @@ -34,7 +34,7 @@ const MAX_PAGE_AMOUNT: usize = 16;

pub struct QueueAnalyzer<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> {
operations: OptimizedVec<Operation<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>>,
queue_inner_wt: QueueInnerWorkTable,
queue_inner_wt: Arc<QueueInnerWorkTable>,
}

impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>
Expand All @@ -44,10 +44,10 @@ where
PrimaryKey: Debug,
SecondaryKeys: Debug,
{
pub fn new() -> Self {
pub fn new(queue_inner_wt: Arc<QueueInnerWorkTable>) -> Self {
Self {
operations: OptimizedVec::with_capacity(256),
queue_inner_wt: QueueInnerWorkTable::default(),
queue_inner_wt,
}
}

Expand Down Expand Up @@ -291,6 +291,8 @@ pub struct PersistenceTask<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> {
#[allow(dead_code)]
engine_task_handle: tokio::task::AbortHandle,
queue: Arc<Queue<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>>,
analyzer_inner_wt: Arc<QueueInnerWorkTable>,
analyzer_in_progress: Arc<AtomicBool>,
progress_notify: Arc<Notify>,
}

Expand All @@ -313,19 +315,23 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>

let engine_queue = queue.clone();
let engine_progress_notify = progress_notify.clone();
let analyzer_inner_wt: Arc<QueueInnerWorkTable> = Default::default();
let mut analyzer = QueueAnalyzer::new(analyzer_inner_wt.clone());
let analyzer_in_progress = Arc::new(AtomicBool::new(true));
let task_analyzer_in_progress = analyzer_in_progress.clone();

let task = async move {
let mut analyzer = QueueAnalyzer::new();
loop {
let op = if let Some(next_op) = engine_queue.immediate_pop() {
Some(next_op)
} else if analyzer.len() == 0 {
engine_progress_notify.notify_waiters();
task_analyzer_in_progress.store(false, Ordering::Release);
let res = Some(engine_queue.pop().await);
task_analyzer_in_progress.store(true, Ordering::Release);
res
} else {
// println!("Queue is {:?}", analyzer.len());
if analyzer.len() == 0 {
engine_progress_notify.notify_waiters();
Some(engine_queue.pop().await)
} else {
None
}
None
};
if let Some(op) = op {
if let Err(err) = analyzer.push(op) {
Expand All @@ -342,11 +348,6 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>
tracing::warn!("Error collecting batch operation: {}", e);
} else {
let batch_op = batch_op.unwrap();
// println!(
// "Batch len is {}, queue len is {}",
// batch_op.ops.len(),
// analyzer.len()
// );
let res = engine.apply_batch_operation(batch_op).await;
if let Err(e) = res {
tracing::warn!(
Expand All @@ -362,15 +363,40 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>
Self {
queue,
engine_task_handle,
analyzer_inner_wt,
analyzer_in_progress,
progress_notify,
}
}

fn check_wait_triggers(&self) -> bool {
if self.queue.len() != 0 {
return false;
}
if self.analyzer_inner_wt.count() != 0 {
return false;
}
if self.analyzer_in_progress.load(Ordering::Acquire) {
return false;
}
true
}

pub async fn wait_for_ops(&self) {
let count = self.queue.len();
if count != 0 {
tracing::info!("Waiting for {} operations", count);
self.progress_notify.notified().await
while !self.check_wait_triggers() {
let queue_count = self.queue.len();
let analyzer_count = self.analyzer_inner_wt.count();
let count = queue_count + analyzer_count;
if count == 0 {
tracing::info!("Waiting for last operation");
} else {
tracing::info!("Waiting for {} operations", count);
}

tokio::select! {
_ = self.progress_notify.notified() => {},
_ = tokio::time::sleep(Duration::from_secs(1)) => {}
}
}
}
}
170 changes: 170 additions & 0 deletions tests/persistence/sync/string_re_read.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,13 @@ fn test_key_delete_scenario() {
})
}

// #[test]
// fn test_key_delete_scenario_multiple() {
// for _ in 0..100 {
// test_key_delete_by_unique()
// }
// }

#[test]
fn test_key_delete() {
let config = PersistenceConfig::new("tests/data/key/delete", "tests/data/key/delete");
Expand Down Expand Up @@ -263,6 +270,169 @@ fn test_key_delete() {
})
}

#[test]
fn test_key_delete_all() {
let config = PersistenceConfig::new("tests/data/key/delete_all", "tests/data/key/delete_all");

let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_io()
.enable_time()
.build()
.unwrap();

runtime.block_on(async {
remove_dir_if_exists("tests/data/key/delete_all".to_string()).await;

let (pk0, pk1) = {
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
let pk0 = table
.insert(StringReReadRow {
first: "first".to_string(),
id: table.get_next_pk().into(),
third: "third".to_string(),
second: "second".to_string(),
last: "_________________________last_____________________".to_string(),
})
.unwrap();
let pk1 = table
.insert(StringReReadRow {
first: "first".to_string(),
id: table.get_next_pk().into(),
third: "third_again".to_string(),
second: "second_again".to_string(),
last: "_________________________last_____________________".to_string(),
})
.unwrap();

table.wait_for_ops().await;
(pk0, pk1)
};
{
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
table.delete(pk0.clone()).await.unwrap();
table.delete(pk1.clone()).await.unwrap();

table.wait_for_ops().await
}
{
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
assert_eq!(table.select_all().execute().unwrap().len(), 0);

assert!(table.select(pk0).is_none());
assert!(table.select(pk1).is_none());
assert_eq!(
table
.select_by_first("first".to_string())
.execute()
.unwrap()
.len(),
0
);
assert!(table.select_by_second("second_again".to_string()).is_none());
assert!(table.select_by_second("second".to_string()).is_none())
}
})
}

#[test]
fn test_key_delete_all_and_insert() {
let config = PersistenceConfig::new(
"tests/data/key/delete_all_and_insert",
"tests/data/key/delete_all_and_insert",
);

let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_io()
.enable_time()
.build()
.unwrap();

runtime.block_on(async {
remove_dir_if_exists("tests/data/key/delete_all_and_insert".to_string()).await;

let (pk0, pk1) = {
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
let pk0 = table
.insert(StringReReadRow {
first: "first".to_string(),
id: table.get_next_pk().into(),
third: "third".to_string(),
second: "second".to_string(),
last: "_________________________last_____________________".to_string(),
})
.unwrap();
let pk1 = table
.insert(StringReReadRow {
first: "first".to_string(),
id: table.get_next_pk().into(),
third: "third_again".to_string(),
second: "second_again".to_string(),
last: "_________________________last_____________________".to_string(),
})
.unwrap();

table.wait_for_ops().await;
(pk0, pk1)
};
{
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
table.delete(pk0.clone()).await.unwrap();
table.delete(pk1.clone()).await.unwrap();

table.wait_for_ops().await
}
let pk = {
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();
assert_eq!(table.select_all().execute().unwrap().len(), 0);

let pk = table
.insert(StringReReadRow {
first: "first".to_string(),
id: table.get_next_pk().into(),
third: "third_again".to_string(),
second: "second".to_string(),
last: "_________________________last_____________________".to_string(),
})
.unwrap();

table.wait_for_ops().await;
pk
};
{
let table = StringReReadWorkTable::load_from_file(config.clone())
.await
.unwrap();

assert_eq!(table.select_all().execute().unwrap().len(), 1);

assert!(table.select(pk).is_some());
assert_eq!(
table
.select_by_first("first".to_string())
.execute()
.unwrap()
.len(),
1
);
assert!(table.select_by_second("second".to_string()).is_some())
}
})
}

#[test]
fn test_key_delete_by_unique() {
let config = PersistenceConfig::new(
Expand Down