agent_tk/decision/

default.rs

//! This module defines a default decision module.

use async_broadcast::Sender;
use futures::join;

use crate::prelude::*;
use definitions::task as definitions_task;
use delegation::Task as DelTask;

//   ____        _   _
//  / __ \      | | (_)
//  | |  | |_ __ | |_ _  ___  _ __  ___
//  | |  | | '_ \| __| |/ _ \| '_ \/ __|
//  | |__| | |_) | |_| | (_) | | | \__ \
//   \____/| .__/ \__|_|\___/|_| |_|___/
//         | |
//         |_|

/// Options for the default decision module
pub struct Options {}

impl Default for Options
{
  fn default() -> Self
  {
    Self::new()
  }
}

impl Options
{
  /// Create options for the default decision module
  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,
}

/// This structure implements a decision module
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;
}