tpl_sim_agent_graph.H

#ifndef TPL_SIM_AGENT_GRAPH_H
#define TPL_SIM_AGENT_GRAPH_H

#include <iostream>
#include <cstdlib>
#include <sys/types.h>
#include <dlink.H>
#include <tpl_concurrent_graph.H>
#include <tpl_dynBinHeap.H>
#include <useMutex.H>

using namespace Aleph;
using namespace std;

namespace Aleph {

#define STOPPED 0  
#define RUNNING 1
#define SUSPENDED 2
  
  template <typename Agent_Info> class Sim_Agent;
  
  template <typename Node_Info> class Sim_Agent_Node;  
  
  template <typename Arc_Info> class Sim_Agent_Arc;
  
      template <typename Arc_Agents, typename Node_Agents, typename Agent> 
  class Sim_Agent_Graph;

  struct Sim_Event
  {
    long actual_time;

    bool to_be_destroyed;

    Sim_Event() : actual_time(0) { }

    Sim_Event(const long & _curr_time) : actual_time(_curr_time) 
    { 
      to_be_destroyed = false;
    }

    virtual ~Sim_Event() { }

    bool operator < (const Sim_Event & event) const
    {
      return actual_time > event.actual_time;    
    }

    virtual void execute_event(void * graph)
    {
      graph = NULL;

      throw 
        domain_error("Debe sobreescribir este método en una clase derivada");
    }
  };

  struct Cmp_Sim_Event
  {
    bool operator () (Sim_Event * op1, Sim_Event * op2)
    {
      return op1->actual_time < op2->actual_time;
    }
  };

  template <typename Agent_Info> struct Sim_Agent : public Sim_Event   
  {
    typedef void (*Transit) (void *, void *);

    Agent_Info info;
    Dlink agent_link;
    Dlink schedule_link;
    Dlink location_link;
    bool in_node;
    void * location;
    void * cookie;
    Transit transit;
    
    typedef Agent_Info Agent_Type;

    Sim_Agent(Agent_Info agent_info, Transit _transit, long actual_time) 
      : Sim_Event(actual_time), info (agent_info), transit(_transit) 
    {
      // empty
    }

    ~Sim_Agent() { /* empty */ }

    LINKNAME_TO_TYPE (Sim_Agent, agent_link);  
    LINKNAME_TO_TYPE (Sim_Agent, schedule_link);
    LINKNAME_TO_TYPE (Sim_Agent, location_link);

    Agent_Info & get_info() 
    {
      return info;
    }

    virtual void execute_event (void * graph)
    {
      if (transit != NULL)
      {
        transit(graph, this);
      }
    }
  };  

  template <typename Agents_Node_Info> 
  struct Sim_Agent_Node :  public Concurrent_Node <Agents_Node_Info>
  {
    Dlink agent_list;
    typedef Concurrent_Node<Agents_Node_Info> Parent_Node;
    typedef Agents_Node_Info Node_Type;
    Sim_Agent_Node() : Parent_Node() { }
    Sim_Agent_Node(const Node_Type & info) : Parent_Node(info) {}
    ~Sim_Agent_Node(){ }
  }; 
  

  template <typename Agents_Arc_Info> 
  struct Sim_Agent_Arc : public Concurrent_Arc<Agents_Arc_Info>    {
    Dlink agent_list;
    typedef Concurrent_Arc<Agents_Arc_Info> Parent_Arc;
    typedef Agents_Arc_Info Arc_Type;
    Sim_Agent_Arc() : Parent_Arc() { } 
    virtual ~Sim_Agent_Arc(){ } 
  };
  
      template <typename Node,typename Arc, typename Agent > 
  struct Sim_Agent_Graph : public Concurrent_Graph<Node, Arc>
  {
   public:
    typedef Concurrent_Graph< Node, Arc > Parent_Graph;
    typedef Node Node_Class;
    typedef Arc Arc_Class;
    typedef typename Node::Node_Type Node_Type; 
    typedef typename Arc::Arc_Type Arc_Type;
    typedef typename Agent::Agent_Type Agent_Type;

   private:
    size_t thread_count;
    pthread_t * threads;
    int status ;
    pthread_mutex_t sch_mutex;
    pthread_mutex_t suspend_mutex;
    pthread_cond_t sch_cond;
    pthread_cond_t suspend_cond;
    long actual_time;
    long callback_rate;
    typedef void (* Update_Callback) (Sim_Agent_Graph *, Agent *, void *);
    Update_Callback update_callback; 

    Dlink agent_list; // lista de agentes
    size_t num_agents;
   
    DynBinHeap<Sim_Event *,Cmp_Sim_Event > events_heap;
  
  public:

    const size_t & get_num_agents() const {return num_agents;}
   
    Node * get_agent_node_location (Agent * agent) 
    {
      if (agent->in_node) 
      {
        return static_cast<Node*>(agent->location);
      }
      return NULL;
    }

    void set_agent_node_location (Agent * agent, Node * node)
    {
      agent->location_link.del();
      agent->in_node = true;
      agent->location = node;
      node->agent_list.append(&agent->location_link);
    }

    Arc * get_agent_arc_location (Agent * agent)
    {
      if (!agent->in_node) 
      {
        return static_cast<Arc*>(agent->location);
      }
      return NULL;
    }

    void set_agent_arc_location(Agent * agent, Arc * arc)
    {
      agent->location_link.del();
      agent->in_node = false;
      agent->location = arc;
      arc->agent_list.append(&agent->location_link);
    }

    bool is_agent_in_node (Agent * agent)
    {
      return agent->in_node;
    }

    Agent * get_first_agent ()
    {
      if(get_num_agents() == 0)
              throw std::range_error("Graph has not agents ");
      return Agent::agent_link_to_Sim_Agent(&*agent_list.get_next());
    }
    
    void remove_agent (Agent * agent)
    { 
            agent->agent_link.del();
      agent->location_link.del();
      num_agents--;
            delete agent;
    }
   
    void clear_agent_list()
    {
      Agent * agent = NULL;
      for (Dlink::Iterator it(&agent_list); it.has_current(); )
      {
       agent = Agent::agent_link_to_Sim_Agent(it.get_current());
        it.next();
        remove_agent(agent);
      }
    }

    Sim_Agent_Graph(const size_t & num_threads) : thread_count(num_threads) 
    {
      threads = (pthread_t*)malloc(num_threads * sizeof(pthread_t));
      init_mutex(sch_mutex);
      init_mutex(suspend_mutex);
      pthread_cond_init(&suspend_cond, NULL);
      pthread_cond_init(&sch_cond, NULL);
      status =RUNNING; 
      num_agents = 0;
    }

    virtual ~Sim_Agent_Graph()
    {
      clear_agent_list();
      destroy_mutex(sch_mutex);
      destroy_mutex(suspend_mutex);
      pthread_cond_destroy(&suspend_cond);
      pthread_cond_destroy(&sch_cond);
        free(threads); 
    } 
    
    Agent * create_agent_in_node(Agent_Type agent_info, void (*funcptr) (void *, void *), Node * node, long activation_time)
    {
      Agent * agent = new Agent ( agent_info, funcptr, activation_time );
      set_agent_node_location(agent, node);
      agent_list.append(&agent->agent_link);
      num_agents++;
      schedule_event(agent);
      return agent;
    }
    
    Agent * create_agent_in_arc(Agent_Type agent_info, void (*funcptr) (void *, void *), Arc * arc, long activation_time)
    {
      Agent * agent = new Agent( agent_info, funcptr, activation_time );
      set_agent_arc_location(agent, arc);
      agent_list.append(&agent->agent_link);
      num_agents++;
      schedule_event(agent);
      return agent;
    }
    
    template <class Equal>
    Agent * search_agent(const Agent_Type & agent)
    {
      Agent * result = NULL;
      for (Dlink::Iterator it(agent_list); it.has_current(); it.next()) 
      {
        result = Agent::agent_link_to_Sim_Agent(it.get_current());
        if (Equal() (result->get_info(), agent)) 
        {
          return result;
        }
      }
      return NULL;
    }
   
    Agent * search_agent(const Agent_Type & agent)
    {
       return search_agent<Aleph::equal_to<Agent_Type> >(agent);
    }

    void start_graph() 
    {
      for (unsigned int i = 0; i < thread_count; i++) 
            {
              int result = pthread_create(&threads[i] , NULL, Sim_Agent_Graph::run, this);
              if (result == 0) 
                    {
                      cout<<" Thread "<<i<<endl; 
                    }
            }
    }

    void stop_graph()
    {
      if (get_status() == SUSPENDED)
      {
        resume_graph();
      }
      set_status(STOPPED);
      for (unsigned int i = 0; i < thread_count; i++) 
            {
              pthread_join(threads[i] , NULL);
      }
    }
    
    int get_status()
    {
      CRITICAL_SECTION(sch_mutex);
      return status;
    }

    void set_status(int value)
    {
      CRITICAL_SECTION(sch_mutex);
      status = value;
    }

    void suspend_graph()
    {
      set_status(SUSPENDED);
    }

    void resume_graph()
    {
      set_status(RUNNING);
      pthread_cond_broadcast(&suspend_cond);

    }

    void cancel_event(Sim_Event * event)
    {
    }

    void schedule_event (Sim_Event * event)
    {
      events_heap.insert(event);
    }

    void set_update_callback(Update_Callback fnct)
    {
      update_callback = fnct;
    }

    long get_actual_time()
    {
      return actual_time;
    }

    void set_callback_rate(long rate)
    {
      callback_rate = rate;
    }

    long get_callback_rate()
    {
      return callback_rate();
    }
    
    static void * run(void * cookie) 
    {
      Sim_Agent_Graph * graph = static_cast<Sim_Agent_Graph*>(cookie);
      // Lógica de ejecución de eventos
      while (graph->get_status() != STOPPED)
      {
        while (graph->get_status() == SUSPENDED)
        {
          pthread_mutex_lock(&graph->suspend_mutex);
          pthread_cond_wait(&graph->suspend_cond, &graph->suspend_mutex);
          pthread_mutex_unlock(&graph->suspend_mutex);
        }
        
        if (graph->events_heap.is_empty())
        {
          break;
        }

        Sim_Event * next_event = graph->events_heap.getMin();

        graph->actual_time = next_event->actual_time;
        next_event->execute_event(graph);
       
        //TODO Si ha pasado tiempo > callback_rate, ejecutar update_callback
        //TODO Incluir evento en el heap
       
        if (next_event->to_be_destroyed)
        {
          delete next_event;
        }
      }
      return NULL;
    }

    
    void run_static()
    {
      while (!events_heap.is_empty())
      {
        Sim_Event * next_event = events_heap.getMin();

        actual_time = next_event->actual_time;
        next_event->execute_event(this);
        if (next_event->to_be_destroyed)
        {
          delete next_event;
        }
      }
    }
    
  };
}
#endif