use async_broadcast::Sender;
use futures::join;
use crate::prelude::*;
use definitions::task as definitions_task;
use delegation::Task as DelTask;
pub struct Options {}
impl Default for Options
{
fn default() -> Self
{
Self::new()
}
}
impl Options
{
pub fn new() -> Self
{
Self {}
}
}
module::create_module_private_interface!(
ModulePrivateInterface,
decision::InputMessage,
decision::OutputMessage
);
struct EndOfQueue
{
total_cost: f32,
final_states: crate::states::States,
queue_length: usize,
}
pub struct Module {}
impl Module
{
async fn execute_tst_task(
uuid: &uuid::Uuid,
tst: &definitions::tst::Node,
requester_uri: &String,
agent_data: &agent::AgentData,
end_of_queue: &ccutils::sync::ArcMutex<EndOfQueue>,
output_sender: &Sender<decision::OutputMessage>,
_execution_input_sender: &Sender<execution::InputMessage>,
) -> Result<()>
{
agent_data.knowledge_base.insert(
"tasks",
uuid.to_hex(),
&definitions_task::Task::from_tst(tst.clone()),
)?;
let queue_length = end_of_queue.lock()?.queue_length;
if 10 * queue_length > 8 * consts::TASK_QUEUE_LENGTH
{
return Ok(());
}
let states = end_of_queue.lock()?.final_states.clone();
let sr = crate::simulation::tst::simulate_execution(
states,
agent_data.capabilities.clone(),
tst.clone(),
)
.await;
let total_cost = end_of_queue.lock()?.total_cost;
match sr
{
Ok(sr) =>
{
agent_data
.knowledge_base
.insert("task_simulation_result", uuid.to_hex(), &sr)?;
ccutils::log::log_error!(
output_sender
.broadcast(decision::OutputMessage::CFPProposal {
proposal: delegation::Proposal {
agent_uri: agent_data.agent_uri.to_owned(),
requester_uri: requester_uri.to_owned(),
cost: total_cost + sr.get_estimated_cost(),
signature: Default::default(),
task_uuid: uuid.to_owned(),
},
})
.await,
"While sending CFP Proposal"
);
Ok(())
}
Err(e) => match e
{
Error::ExecutionFailed(ef) => match *ef
{
Error::UnknownCapability(_) => Ok(()),
e => Err(e),
},
e => Err(e),
},
}
}
async fn handle_input_message(
msg: decision::InputMessage,
agent_data: &agent::AgentData,
end_of_queue: &ccutils::sync::ArcMutex<EndOfQueue>,
output_sender: &Sender<decision::OutputMessage>,
execution_input_sender: &Sender<execution::InputMessage>,
) -> Result<()>
{
match msg
{
decision::InputMessage::DecideCFPAcceptance { cfp } =>
{
let task = definitions_task::Task::from_description(&cfp.task_type, &cfp.task_description)?;
match task.get_container()
{
definitions_task::TaskContainer::Tst(tst) =>
{
Self::execute_tst_task(
&task.task_id(),
tst,
&cfp.requester_uri,
agent_data,
end_of_queue,
output_sender,
execution_input_sender,
)
.await?;
}
definitions_task::TaskContainer::Goal(goal) =>
{
let tst = conversion::goal::ToTst::convert(goal)?;
Self::execute_tst_task(
&task.task_id(),
&tst,
&cfp.requester_uri,
agent_data,
end_of_queue,
output_sender,
execution_input_sender,
)
.await?;
}
}
}
decision::InputMessage::QueueExecution {
uuid,
requester_uri,
} =>
{
let accept = 10 * end_of_queue.lock()?.queue_length < 8 * consts::TASK_QUEUE_LENGTH;
if accept
{
ccutils::log::log_error!(
execution_input_sender
.broadcast(execution::InputMessage::QueueExecution { uuid })
.await,
"Sending queue execution"
);
}
ccutils::log::log_error!(
output_sender
.broadcast(decision::OutputMessage::QueueExecutionResult {
uuid,
accepted: accept,
requester_uri,
})
.await,
"sneding queued execution result"
);
}
decision::InputMessage::CancelExecution { uuid } =>
{
ccutils::log::log_error!(
execution_input_sender
.broadcast(execution::InputMessage::CancelExecution {
uuid: uuid.to_owned(),
})
.await,
"sending execution cancelled"
);
}
}
Ok(())
}
}
impl decision::Module for Module
{
type Options = Options;
async fn start(
agent_data: agent::AgentData,
module_interfaces: (decision::ModuleInterface, ModulePrivateInterface),
execution_interface: execution::ModuleInterface,
_: Options,
)
{
let (_, module_private_interface) = module_interfaces;
let mut input_receiver = module_private_interface.input_receiver.activate();
let output_sender = module_private_interface.output_sender;
let end_of_queue: ccutils::sync::ArcMutex<_> = EndOfQueue {
total_cost: 0.0,
final_states: agent_data.states.to_owned_states().unwrap(),
queue_length: 0,
}
.into();
let execution_input_sender = execution_interface.input_sender();
let fut_ir = {
let end_of_queue = end_of_queue.clone();
async move {
loop
{
let msg = input_receiver.recv().await;
if let Ok(msg) = msg
{
if let Err(e) = Self::handle_input_message(
msg,
&agent_data,
&end_of_queue,
&output_sender,
&execution_input_sender,
)
.await
{
log::error!(
"An error occured for agent {} when handling decision message: {}",
agent_data.agent_uri,
e
);
}
}
else
{
log::info!("Decision loop for {:?} is closing", agent_data.agent_uri);
return;
}
}
}
};
let mut execution_output_receiver = execution_interface.output_receiver();
let fut_exec_or = async move {
loop
{
let msg = execution_output_receiver.recv().await;
if let Ok(msg) = msg
{
match msg
{
execution::OutputMessage::CurrentEstimatedIdlingState {
current_cost,
final_states,
queue_length,
} =>
{
let mut end_of_queue = end_of_queue.lock().unwrap();
end_of_queue.total_cost = current_cost;
end_of_queue.final_states = final_states;
end_of_queue.queue_length = queue_length;
}
}
}
else
{
return;
}
}
};
join!(fut_ir, fut_exec_or);
}
}
impl module::Module for Module
{
type InputMessage = decision::InputMessage;
type OutputMessage = decision::OutputMessage;
type ModulePrivateInterface = ModulePrivateInterface;
}