Skip to content

switched to tokio tasks #862

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
merged 17 commits into from
Aug 10, 2024
Merged

switched to tokio tasks #862

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
c6e416f
switched to tokio tasks
vishalkrishnads Jul 29, 2024
8ca8ac0
Merge branch 'main' into issue-652
vishalkrishnads Jul 29, 2024
f15b137
corrected intendation
vishalkrishnads Jul 29, 2024
dba4bbc
fixed intendation
vishalkrishnads Jul 29, 2024
306dd17
corrected intendation
vishalkrishnads Jul 29, 2024
591487c
restructured `run_local_sync()`
vishalkrishnads Aug 3, 2024
79c2965
corrected time sync issue
vishalkrishnads Aug 3, 2024
1929d05
Merge branch 'main' into issue-652
nitisht Aug 3, 2024
66dbbc8
pass server integrity test
vishalkrishnads Aug 3, 2024
93152d8
switched back to `AsyncScheduler`
vishalkrishnads Aug 4, 2024
0f8d4ad
Merge branch 'main' into issue-652
vishalkrishnads Aug 4, 2024
ac49651
corrected intendation
vishalkrishnads Aug 4, 2024
cd90c17
changed hot tier manager execution order
vishalkrishnads Aug 4, 2024
e9ade78
Revert to `0f8d4ad`
vishalkrishnads Aug 5, 2024
7d90b22
fix linter
nitisht Aug 5, 2024
dc00962
Merge pull request #1 from nitisht/issue-652
vishalkrishnads Aug 5, 2024
6912a47
Merge branch 'main' into issue-652
nitisht Aug 5, 2024
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
19 changes: 13 additions & 6 deletions server/src/handlers/http/modal/ingest_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,9 +339,10 @@ impl IngestServer {

migration::run_migration(&CONFIG).await?;

let (localsync_handler, mut localsync_outbox, localsync_inbox) = sync::run_local_sync();
let (localsync_handler, mut localsync_outbox, localsync_inbox) =
sync::run_local_sync().await;
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
sync::object_store_sync();
sync::object_store_sync().await;

tokio::spawn(airplane::server());

Expand All @@ -354,8 +355,12 @@ impl IngestServer {
// actix server finished .. stop other threads and stop the server
remote_sync_inbox.send(()).unwrap_or(());
localsync_inbox.send(()).unwrap_or(());
localsync_handler.join().unwrap_or(());
remote_sync_handler.join().unwrap_or(());
if let Err(e) = localsync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
return e
},
_ = &mut localsync_outbox => {
Expand All @@ -365,8 +370,10 @@ impl IngestServer {
},
_ = &mut remote_sync_outbox => {
// remote_sync failed, this is recoverable by just starting remote_sync thread again
remote_sync_handler.join().unwrap_or(());
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync();
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
}

};
Expand Down
24 changes: 15 additions & 9 deletions server/src/handlers/http/modal/query_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,8 +105,8 @@ impl ParseableServer for QueryServer {
fn validate(&self) -> anyhow::Result<()> {
if CONFIG.get_storage_mode_string() == "Local drive" {
return Err(anyhow::anyhow!(
"Query Server cannot be started in local storage mode. Please start the server in a supported storage mode.",
));
"Query Server cannot be started in local storage mode. Please start the server in a supported storage mode.",
));
}

Ok(())
Expand Down Expand Up @@ -188,13 +188,13 @@ impl QueryServer {
if matches!(init_cluster_metrics_schedular(), Ok(())) {
log::info!("Cluster metrics scheduler started successfully");
}

if let Some(hot_tier_manager) = HotTierManager::global() {
hot_tier_manager.download_from_s3()?;
};
let (localsync_handler, mut localsync_outbox, localsync_inbox) = sync::run_local_sync();
let (localsync_handler, mut localsync_outbox, localsync_inbox) =
sync::run_local_sync().await;
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
sync::object_store_sync();
sync::object_store_sync().await;

tokio::spawn(airplane::server());
let app = self.start(prometheus, CONFIG.parseable.openid.clone());
Expand All @@ -206,8 +206,12 @@ impl QueryServer {
// actix server finished .. stop other threads and stop the server
remote_sync_inbox.send(()).unwrap_or(());
localsync_inbox.send(()).unwrap_or(());
localsync_handler.join().unwrap_or(());
remote_sync_handler.join().unwrap_or(());
if let Err(e) = localsync_handler.await {
log::error!("Error joining localsync_handler: {:?}", e);
}
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
return e
},
_ = &mut localsync_outbox => {
Expand All @@ -217,8 +221,10 @@ impl QueryServer {
},
_ = &mut remote_sync_outbox => {
// remote_sync failed, this is recoverable by just starting remote_sync thread again
remote_sync_handler.join().unwrap_or(());
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync();
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
}

};
Expand Down
19 changes: 13 additions & 6 deletions server/src/handlers/http/modal/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -533,9 +533,10 @@ impl Server {

storage::retention::load_retention_from_global();

let (localsync_handler, mut localsync_outbox, localsync_inbox) = sync::run_local_sync();
let (localsync_handler, mut localsync_outbox, localsync_inbox) =
sync::run_local_sync().await;
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
sync::object_store_sync();
sync::object_store_sync().await;

if CONFIG.parseable.send_analytics {
analytics::init_analytics_scheduler()?;
Expand All @@ -553,8 +554,12 @@ impl Server {
// actix server finished .. stop other threads and stop the server
remote_sync_inbox.send(()).unwrap_or(());
localsync_inbox.send(()).unwrap_or(());
localsync_handler.join().unwrap_or(());
remote_sync_handler.join().unwrap_or(());
if let Err(e) = localsync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
return e
},
_ = &mut localsync_outbox => {
Expand All @@ -564,8 +569,10 @@ impl Server {
},
_ = &mut remote_sync_outbox => {
// remote_sync failed, this is recoverable by just starting remote_sync thread again
remote_sync_handler.join().unwrap_or(());
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync();
if let Err(e) = remote_sync_handler.await {
log::error!("Error joining remote_sync_handler: {:?}", e);
}
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
}
};
}
Expand Down
55 changes: 19 additions & 36 deletions server/src/storage/retention.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,36 +19,28 @@
use std::hash::Hash;
use std::num::NonZeroU32;
use std::sync::Mutex;
use std::thread;
use std::time::Duration;

use clokwerk::AsyncScheduler;
use clokwerk::Job;
use clokwerk::TimeUnits;
use derive_more::Display;
use once_cell::sync::Lazy;
use tokio::task::JoinHandle;

use crate::metadata::STREAM_INFO;

type SchedulerHandle = thread::JoinHandle<()>;
type SchedulerHandle = JoinHandle<()>;

static SCHEDULER_HANDLER: Lazy<Mutex<Option<SchedulerHandle>>> = Lazy::new(|| Mutex::new(None));

fn async_runtime() -> tokio::runtime::Runtime {
tokio::runtime::Builder::new_current_thread()
.thread_name("retention-task-thread")
.enable_all()
.build()
.unwrap()
}

pub fn load_retention_from_global() {
log::info!("loading retention for all streams");
init_scheduler();
}

pub fn init_scheduler() {
log::info!("Setting up schedular");
log::info!("Setting up scheduler");
let mut scheduler = AsyncScheduler::new();
let func = move || async {
//get retention every day at 12 am
Expand All @@ -57,22 +49,17 @@ pub fn init_scheduler() {

match retention {
Ok(config) => {
if config.is_none() {
continue;
}
for Task { action, days, .. } in config.unwrap().tasks.into_iter() {
match action {
Action::Delete => {
let stream = stream.to_string();
thread::spawn(move || {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
// Run the asynchronous delete action
if let Some(config) = config {
for Task { action, days, .. } in config.tasks.into_iter() {
match action {
Action::Delete => {
let stream = stream.to_string();
tokio::spawn(async move {
action::delete(stream.clone(), u32::from(days)).await;
});
});
}
};
}
};
}
}
}
Err(err) => {
Expand All @@ -83,21 +70,17 @@ pub fn init_scheduler() {
};

// Execute once on startup
thread::spawn(move || {
let rt = async_runtime();
rt.block_on(func());
tokio::spawn(async move {
func().await;
});

scheduler.every(1.day()).at("00:00").run(func);

let scheduler_handler = thread::spawn(|| {
let rt = async_runtime();
rt.block_on(async move {
loop {
tokio::time::sleep(Duration::from_secs(10)).await;
scheduler.run_pending().await;
}
});
let scheduler_handler = tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(10)).await;
scheduler.run_pending().await;
}
});

*SCHEDULER_HANDLER.lock().unwrap() = Some(scheduler_handler);
Expand Down
Loading
Loading