Skip to content

Commit

Permalink
Fix
Browse files Browse the repository at this point in the history
  • Loading branch information
@gurinderu
gurinderu committed Apr 14, 2023
1 parent 797cb01 commit 55e41cc
Show file tree
Hide file tree
Showing 7 changed files with 25 additions and 25 deletions.
2 changes: 1 addition & 1 deletion aquamarine/src/particle_data_store.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ impl DataStore for ParticleDataStore {
}

fn store_data(&mut self, data: &[u8], particle_id: &str, current_peer_id: &str) -> Result<()> {
log::debug!(target: "particle_reap", "Stroring data for particle {}", particle_id);
log::debug!(target: "particle_reap", "Storing data for particle {}", particle_id);
let data_path = self.data_file(particle_id, current_peer_id);
std::fs::write(&data_path, data).map_err(|err| StoreData(err, data_path))?;

Expand Down
15 changes: 0 additions & 15 deletions aquamarine/src/plumber.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -188,17 +188,6 @@ impl<RT: AquaRuntime, F: ParticleFunctionStatic> Plumber<RT, F> {
// Remove expired actors
if let Some((vm_id, mut vm)) = self.vm_pool.get_vm() {
let now = now_ms();
for ((particle_id, peer_id), actor) in self.actors.iter() {
log::debug!(
target: "particle_reap",
"Current actors particle_id={:#?}, peer_id={:#?}, actor_deadline={:#?}, actor_mailbox_size={:#?}, actor_is_executing={:#?}",
particle_id,
peer_id,
actor.deadline,
actor.mailbox.len(),
actor.is_executing()
);
}

self.actors.retain(|(particle_id, peer_id), actor| {
// if actor hasn't yet expired or is still executing, keep it
Expand All @@ -208,10 +197,6 @@ impl<RT: AquaRuntime, F: ParticleFunctionStatic> Plumber<RT, F> {
return true; // keep actor
}

log::debug!(
target: "particle_reap",
"Reaping particle's actor particle_id={particle_id}, peer_id={peer_id})"
);
// cleanup files and dirs after particle processing (vault & prev_data)
// TODO: do not pass vm https://github.com/fluencelabs/fluence/issues/1216
if let Err(err) = actor.cleanup(particle_id, &peer_id.to_string(), &mut vm) {
Expand Down
3 changes: 1 addition & 2 deletions particle-builtins/src/builtins.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -846,8 +846,7 @@ where
}

fn call_service(&self, function_args: Args, particle: ParticleParams) -> FunctionOutcome {
log::debug!(target: "particle_reap", "Before call_service call_service {}", particle.id);
self.services.call_service(function_args, particle)
self.services.call_service(function_args, particle, true)
}

fn get_interface(&self, args: Args, params: ParticleParams) -> Result<JValue, JError> {
Expand Down
18 changes: 11 additions & 7 deletions particle-services/src/app_services.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ use fluence_app_service::{
AppService, AppServiceError, CallParameters, MarineError, SecurityTetraplet, ServiceInterface,
};
use humantime_serde::re::humantime::format_duration as pretty;
use log::log;
use parking_lot::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value as JValue};
Expand Down Expand Up @@ -409,7 +408,12 @@ impl ParticleAppServices {
Ok(())
}

pub fn call_service(&self, function_args: Args, particle: ParticleParams) -> FunctionOutcome {
pub fn call_service(
&self,
function_args: Args,
particle: ParticleParams,
create_vault: bool,
) -> FunctionOutcome {
let call_time_start = Instant::now();
let services = self.services.read();
let aliases = self.aliases.read();
Expand Down Expand Up @@ -448,8 +452,9 @@ impl ParticleAppServices {
let service_type = self.get_service_type(service, &worker_id);

// TODO: move particle vault creation to aquamarine::particle_functions
log::debug!(target: "particle_reap", "Creating vault for particle {} {} {} {:?}", particle.id, service_id, service.blueprint_id, service_type);
self.create_vault(&particle.id)?;
if create_vault {
self.create_vault(&particle.id)?;
}
let params = CallParameters {
host_id: worker_id.to_string(),
particle_id: particle.id,
Expand Down Expand Up @@ -537,6 +542,7 @@ impl ParticleAppServices {
particle_id: Option<String>,
init_peer_id: PeerId,
particle_ttl: Duration,
create_vault: bool,
) -> FunctionOutcome {
let args = Args {
service_id: service_id.to_string(),
Expand All @@ -555,8 +561,7 @@ impl ParticleAppServices {
signature: vec![],
};

log::debug!(target: "particle_reap", "Before call_service app_services {}", particle.id);
self.call_service(args, particle)
self.call_service(args, particle, create_vault)
}

fn add_alias_inner(
Expand Down Expand Up @@ -892,7 +897,6 @@ impl ParticleAppServices {
}

fn create_vault(&self, particle_id: &str) -> Result<(), VaultError> {
log::debug!(target: "particle_reap", "Creating particle vault {}", particle_id);
self.vault.create(particle_id)
}

Expand Down
4 changes: 4 additions & 0 deletions sorcerer/src/script_executor.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ impl Sorcerer {
None,
worker_id,
self.spell_script_particle_ttl,
false,
);

let res = process_func_outcome::<U32Value>(func_outcome, &spell_id, "get_u32");
Expand Down Expand Up @@ -65,6 +66,7 @@ impl Sorcerer {
None,
worker_id,
self.spell_script_particle_ttl,
false,
);

process_func_outcome::<UnitValue>(func_outcome, &spell_id, "set_u32").map(drop)
Expand All @@ -79,6 +81,7 @@ impl Sorcerer {
None,
worker_id,
self.spell_script_particle_ttl,
false,
);

Ok(process_func_outcome::<ScriptValue>(
Expand Down Expand Up @@ -138,6 +141,7 @@ impl Sorcerer {
None,
worker_id,
self.spell_script_particle_ttl,
false,
);

process_func_outcome::<UnitValue>(func_outcome, &event.spell_id, "list_push_string")
Expand Down
1 change: 1 addition & 0 deletions sorcerer/src/sorcerer.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@ impl Sorcerer {
None,
spell_owner,
self.spell_script_particle_ttl,
false,
),
spell_id,
"get_trigger_config",
Expand Down
7 changes: 7 additions & 0 deletions sorcerer/src/spells.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,7 @@ pub(crate) async fn spell_install(
None,
worker_id,
Duration::from_millis(params.ttl as u64),
false,
),
&spell_id,
"set_script_source_to_file",
Expand All @@ -112,6 +113,7 @@ pub(crate) async fn spell_install(
None,
worker_id,
Duration::from_millis(params.ttl as u64),
false,
),
&spell_id,
"set_json_fields",
Expand All @@ -127,6 +129,7 @@ pub(crate) async fn spell_install(
None,
worker_id,
Duration::from_millis(params.ttl as u64),
false,
),
&spell_id,
"set_trigger_config",
Expand Down Expand Up @@ -249,6 +252,7 @@ pub(crate) async fn spell_update_config(
None,
worker_id,
Duration::from_millis(params.ttl as u64),
false,
),
&spell_id,
"set_trigger_config",
Expand Down Expand Up @@ -298,6 +302,7 @@ pub(crate) fn get_spell_arg(
Some(params.id.clone()),
params.init_peer_id,
Duration::from_millis(params.ttl as u64),
true, //TODO: check it
),
spell_id,
"get_string",
Expand All @@ -324,6 +329,7 @@ pub(crate) fn store_error(
Some(params.id.clone()),
params.init_peer_id,
Duration::from_millis(params.ttl as u64),
false,
),
spell_id,
"store_error",
Expand Down Expand Up @@ -353,6 +359,7 @@ pub(crate) fn store_response(
Some(params.id.clone()),
params.init_peer_id,
Duration::from_millis(params.ttl as u64),
false,
),
spell_id,
"set_json_fields",
Expand Down

0 comments on commit 55e41cc

Please sign in to comment.