Skip to content
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

Jump to bottom

feature/nanocld: task manager wait_task #829

Merged
merged 2 commits into from
Feb 1, 2024
Merged

feature/nanocld: task manager wait_task #829

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8fa1831
feature/nanocld: task manager wait_task
leon3s Feb 1, 2024
40658c8
fix clippy
leon3s Feb 1, 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
82 changes: 35 additions & 47 deletions bin/nanocld/src/models/task_manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,9 @@
use std::{
sync::{Arc, Mutex},
collections::HashMap,
};
use std::{sync::Arc, collections::HashMap, time::Duration};

use ntex::{rt, web};
use futures_util::Future;
use ntex::{rt, time};
use futures_util::{Future, lock::Mutex};

use nanocl_error::io::{IoResult, IoError};
use nanocl_error::io::IoResult;

use nanocl_stubs::system::NativeEventAction;

Expand All @@ -24,6 +21,17 @@
let fut = Arc::new(Mutex::new(rt::spawn(task)));
Self { kind, fut }
}

pub async fn wait(&self) {
loop {
let fut = self.fut.lock().await;
if fut.is_finished() {
break;

Check warning on line 29 in bin/nanocld/src/models/task_manager.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/models/task_manager.rs#L29 

Added line #L29 was not covered by tests
}
drop(fut);
time::sleep(Duration::from_secs(1)).await;
}
}

Check warning on line 34 in bin/nanocld/src/models/task_manager.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/models/task_manager.rs#L34 

Added line #L34 was not covered by tests
}

#[derive(Clone, Default)]
Expand All @@ -36,54 +44,34 @@
Self::default()
}

pub async fn add_task(&self, key: &str, task: ObjTask) -> IoResult<()> {
pub async fn add_task(&self, key: &str, task: ObjTask) {
let key = key.to_owned();
let tasks = Arc::clone(&self.tasks);
web::block(move || {
let mut tasks = tasks.lock()?;
log::debug!("Adding task: {key} {}", task.kind);
tasks.insert(key.clone(), task.clone());
Ok::<_, IoError>(())
})
.await?;
Ok(())
let mut tasks = self.tasks.lock().await;
log::debug!("Adding task: {key} {}", task.kind);
tasks.insert(key.clone(), task.clone());
}

pub async fn remove_task(&self, key: &str) -> IoResult<()> {
pub async fn remove_task(&self, key: &str) {
let key = key.to_owned();
let tasks = Arc::clone(&self.tasks);
web::block(move || {
let mut tasks = tasks.lock().map_err(|err| {
IoError::interrupted("Task", err.to_string().as_str())
})?;
let task = tasks.get(&key);
if let Some(task) = task {
log::debug!("Removing task: {key} {}", task.kind);
task.fut.lock()?.abort();
}
tasks.remove(&key);
Ok::<_, IoError>(())
})
.await?;
Ok(())
let mut tasks = self.tasks.lock().await;
let task = tasks.get(&key);
if let Some(task) = task {
log::debug!("Removing task: {key} {}", task.kind);
task.fut.lock().await.abort();

Check warning on line 60 in bin/nanocld/src/models/task_manager.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/models/task_manager.rs#L59-L60 

Added lines #L59 - L60 were not covered by tests
}
tasks.remove(&key);
}

pub async fn get_task(&self, key: &str) -> Option<ObjTask> {
let key = key.to_owned();
let tasks = Arc::clone(&self.tasks);
let res = web::block(move || {
let tasks = tasks.lock().map_err(|err| {
IoError::interrupted("Task", err.to_string().as_str())
})?;
Ok::<_, IoError>(tasks.get(&key).cloned())
})
.await;
match res {
Ok(res) => res,
Err(err) => {
log::error!("Failed to get task: {}", err);
None
}
let tasks = self.tasks.lock().await;
tasks.get(&key).cloned()
}

pub async fn wait_task(&self, key: &str) {
if let Some(task) = self.get_task(key).await {
task.wait().await;
}
self.remove_task(key).await;
}
}
35 changes: 10 additions & 25 deletions bin/nanocld/src/subsystem/event.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,10 +89,7 @@
let task_key = format!("{}@{key}", actor.kind);
let cargo = CargoDb::transform_read_by_pk(&key, &state.pool).await?;
let state_ptr = state.clone();
let curr_task = state.task_manager.get_task(&task_key).await;
if curr_task.is_some() {
state.task_manager.remove_task(&task_key).await?;
}
state.task_manager.wait_task(&task_key).await;
let task = ObjTask::new(action, async move {
let mut processes =
ProcessDb::read_by_kind_key(&cargo.spec.cargo_key, &state_ptr.pool)
Expand Down Expand Up @@ -128,17 +125,14 @@
state_ptr.emit_normal_native_action(&cargo, NativeEventAction::Start);
Ok::<_, IoError>(())
});
state.task_manager.add_task(&task_key, task).await?;
state.task_manager.add_task(&task_key, task).await;
}
EventActorKind::Vm => {}
EventActorKind::Job => {
let task_key = format!("{}@{key}", actor.kind);
let job = JobDb::read_by_pk(&key, &state.pool).await?.try_to_spec()?;
let state_ptr = state.clone();
let curr_task = state.task_manager.get_task(&task_key).await;
if curr_task.is_some() {
state.task_manager.remove_task(&task_key).await?;
}
state.task_manager.wait_task(&task_key).await;
let task = ObjTask::new(action, async move {
for mut container in job.containers {
let job_name = job.name.clone();
Expand Down Expand Up @@ -178,7 +172,7 @@
}
Ok::<_, IoError>(())
});
state.task_manager.add_task(&task_key, task).await?;
state.task_manager.add_task(&task_key, task).await;
}
_ => {}
}
Expand All @@ -200,10 +194,7 @@
EventActorKind::Cargo => {
log::debug!("handling delete event for cargo {key}");
let task_key = format!("{}@{key}", &actor.kind);
let curr_task = state.task_manager.get_task(&task_key).await;
if curr_task.is_some() {
state.task_manager.remove_task(&task_key).await?;
}
state.task_manager.wait_task(&task_key).await;
let state_ptr = state.clone();
let task = ObjTask::new(action, async move {
let processes =
Expand All @@ -226,16 +217,13 @@
state_ptr.emit_normal_native_action(&cargo, NativeEventAction::Delete);
Ok::<_, IoError>(())
});
state.task_manager.add_task(&task_key, task).await?;
state.task_manager.add_task(&task_key, task).await;
}
EventActorKind::Vm => {}
EventActorKind::Job => {
let job = JobDb::read_by_pk(&key, &state.pool).await?.try_to_spec()?;
let task_key = format!("{}@{key}", &actor.kind);
let curr_task = state.task_manager.get_task(&task_key).await;
if curr_task.is_some() {
state.task_manager.remove_task(&task_key).await?;
}
state.task_manager.wait_task(&task_key).await;

Check warning on line 226 in bin/nanocld/src/subsystem/event.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/subsystem/event.rs#L226 

Added line #L226 was not covered by tests
let state_ptr = state.clone();
let task = ObjTask::new(action, async move {
let processes =
Expand All @@ -259,7 +247,7 @@
state_ptr.emit_normal_native_action(&job, NativeEventAction::Delete);
Ok::<_, IoError>(())
});
state.task_manager.add_task(&task_key, task).await?;
state.task_manager.add_task(&task_key, task).await;

Check warning on line 250 in bin/nanocld/src/subsystem/event.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/subsystem/event.rs#L250 

Added line #L250 was not covered by tests
}
_ => {}
};
Expand All @@ -280,10 +268,7 @@
match actor.kind {
EventActorKind::Cargo => {
let task_key = format!("{}@{key}", &actor.kind);
let curr_task = state.task_manager.get_task(&task_key).await;
if curr_task.is_some() {
state.task_manager.remove_task(&task_key).await?;
}
state.task_manager.wait_task(&task_key).await;
let state_ptr = state.clone();
let task = ObjTask::new(action, async move {
let cargo =
Expand Down Expand Up @@ -333,7 +318,7 @@
}
Ok::<_, IoError>(())
});
state.task_manager.add_task(&task_key, task).await?;
state.task_manager.add_task(&task_key, task).await;

Check warning on line 321 in bin/nanocld/src/subsystem/event.rs

View check run for this annotation

Codecov / codecov/patch

bin/nanocld/src/subsystem/event.rs#L321 

Added line #L321 was not covered by tests
}
EventActorKind::Vm => {}
EventActorKind::Job => {}
Expand Down
Loading