tpl_agent_graph.H

# ifndef TPL_AGENT_GRAPH_H
# define TPL_AGENT_GRAPH_H

# include <sys/time.h>

# include <iostream>
# include <exception>
# include <sys/types.h>
# include <dlink.H>
# include <tpl_concurrent_graph.H>
# include <tpl_dynArray.H>
# include <tpl_arrayQueue.H>

using namespace Aleph;
using namespace std;

namespace Aleph 
{

  // Declaraciones adelantadas de clases 

template <typename Agent_Info> class Agent;     // Agente que circula por grafo


    template <typename Agent_Info> 
struct Walking_Agent
{
  enum Agent_State
  {
    Ready,                   // Listo para ejecutarse dentro de la cola ready
    Executing,               // Ejecutándose, sin estar en ninguna cola
    Suspending,              // dentro de la cola suspendidos
    Suspended,               // ejecutándose pero recibe una señal de suspender
    Deleting_From_Ready,     // ejecutándose pero recibe señal de eliminar de grafo
    Deleting_From_Suspended, // dentro de ready pero recibe señal eliminar de grafo
    Without_Queue,           // No está dentro del grafo
    Invalid                  // agente no tiene función de transición
  } ;

  enum Transition_Action 
    {
      Remain_In_Node,
      Remain_In_Arc,
      Enter_In_Node,
      Enter_In_Arc,
      Out_Of_Location, // no implementado todavía
      Leave_Node,    // no implementado todavía
      Leave_Arc,     // no implementado todavía
      Suspend,
      Dead 
    }  ;

  typedef long (*Process_From_Node) (void *  agent, 
                                     void *  graph, 
                                     void *  src, 
                                     void *& arc);
  
  typedef long (*Process_From_Arc) (void * agent, void * graph, void * arc);

  typedef long (*Process_Out_Location) (void *  agent, 
                                        void *  graph, 
                                        void *& node, 
                                        void *& arc);

  Agent_State agent_state;

  pthread_t thread_id;

  Agent_Info info;

      // enlace a lista de agentes de todo el grafo 
  Dlink agent_link_in_graph; 

      // enlace a cola de listos o suspendidos 
  Dlink schedule_link_in_graph;

      // cada nodo o arco contiene una lista de los agentes que se
      // encuentran en él. Este es el enlace 
  Dlink location_link;

  bool in_node; // true ==> agente en un nodo; false ==> agente es un arco

  void * location; // puntero a nodo o arco donde se encuentra el agente

  void * src; // si location es un arco ==> src es el nodo fuente

  void * cookie;

  Process_From_Node process_from_node; 

  Process_From_Arc  process_from_arc;

  Process_Out_Location process_out_location;

  void * get_src_node() { return src; }

  void set_agent_state()
  {
    if (process_from_node == NULL or process_from_arc)
      agent_state = Invalid;
    else
      agent_state = Without_Queue;
  }

  void set_process_node(Process_From_Node __process_from_node)
  {
    process_from_node = __process_from_node;
    set_agent_state();
  }

  void set_process_arc(Process_From_Arc __process_from_arc)
  {
    process_from_arc = __process_from_arc;
    set_agent_state();
  }

  bool is_a_valid_agent() const 
  {
    return ((process_from_node != NULL and process_from_arc != NULL) or 
            (agent_state == Invalid));
  }

  Walking_Agent() 
    : location(NULL), src(NULL), cookie(NULL), 
      process_from_node(NULL), process_from_arc(NULL)
  {
    agent_state = Invalid;
  }

  Walking_Agent(const Agent_Info &    agent_info, 
                Process_From_Node     __process_from_node     = NULL,
                Process_From_Arc      __process_from_arc      = NULL,
                Process_Out_Location  __process_out_location  = NULL) 
    : info(agent_info), location(NULL), src(NULL), cookie(NULL), 
      process_from_node(__process_from_node), 
      process_from_arc(__process_from_arc), 
      process_out_location(__process_out_location)
  {
    set_agent_state();
  }

  Walking_Agent(Process_From_Node    __process_from_node,
                Process_From_Arc     __process_from_arc,
                Process_Out_Location __process_out_location) 
    : agent_state(Without_Queue), location(NULL), src(NULL), cookie(NULL), 
      process_from_node(__process_from_node), 
      process_from_arc(__process_from_arc),
      process_out_location(__process_out_location)
  {
    set_agent_state();
  }
    
  typedef Agent_Info Agent_Type;
 
      // funciones de conversión a Agent desde un dlink 
  LINKNAME_TO_TYPE (Walking_Agent, agent_link_in_graph);  
  LINKNAME_TO_TYPE (Walking_Agent, schedule_link_in_graph);
  LINKNAME_TO_TYPE (Walking_Agent, location_link);

  // funciones de acceso a atributos públicos de un agente

  Agent_Info & get_info() { return info; }
  
  void *& get_cookie() { return cookie; }

  bool is_in_node() const { return in_node; }

  bool is_in_arc() const { return not in_node; }
};  
  

    template <typename Agents_Node_Info> 
class Agent_Node : public Graph_Node <Agents_Node_Info>
{
public:

  Dlink agent_list; // lista de agentes que tiene el nodo

  typedef Graph_Node<Agents_Node_Info> Base_Node;

  typedef Agents_Node_Info Node_Type;

  Agent_Node() : Graph_Node<Agents_Node_Info>() { /* empty */ }

  Agent_Node(const Node_Type & info)
    : Graph_Node<Agents_Node_Info>(info) { /* empty */ }
}; 
  
  
    template <typename Agents_Arc_Info> 
class Agent_Arc : public Graph_Arc<Agents_Arc_Info>    
{
public:

  Dlink agent_list; // lista de agentes que tiene el arco

  typedef Graph_Arc<Agents_Arc_Info> Base_Arc;

  typedef Agents_Arc_Info Arc_Type;

  Agent_Arc() { /* empty */ }

  Agent_Arc(const Arc_Type & info) : Base_Arc (info) { /* empty */ }

  Agent_Arc(void * src, void * tgt) : Base_Arc (src, tgt)
  {
    // empty
  }
};
  

template <template <class, class> class  __GT, // El tipo de grafo
            class __Node,         // Definción de nodo basada Agent_Node<T>
            class __Arc,          // Definición de arco basada en Agent_Arc<T>
            class __Agent>        // Definción de agente basada en Walking_Agent<T>
class Agent_Graph : public Concurrent_Graph<__GT, __Node, __Arc>
{
  typedef Agent_Graph GT;

public:

  typedef typename GT::Node Node;

  typedef typename GT::Arc Arc;

  typedef __Agent Agent;

private:

      // Estados de ejecución del grafo
  enum Status { 
    Init,        // Estado inicial colocado por el constructor de Walking_Agent
    Running,     // Los agentes se están ejecutando
    Suspended,   // La ejecución está suspendida pero puede ser reanudada
    Stopped      // La ejecución ha sido detenida (las threads han sido
                 // destruidas)
  }; 

  size_t      num_threads;
  pthread_t * threads;     // Arreglo de threads de dimensión num_threads
  Status      status;      // Estado de ejecución del grafo

      /* protege las colas ready y suspended y en general todo el grafo */
  pthread_mutex_t  graph_mutex;

  /* 
     Esta variable de condición se utiliza para bloquear las threads en
     dos situaciones:

     1-. Tenemos num_threads para procesar una cantidad arbitraria de
         agentes. Sin embargo, en algunas circunstancias puede haber
         menos agentes que num_threads. En este caso, debemos suspender
         num_threads - num_agents threads que no se pueden procesar
         puesto que no tienen un agente para correr. Esta variable de
         condición es la que bloquea una thread que no tiene agente para
         ejecutar.

     2-. Cuando el grafo se manda a suspender
  */
  pthread_cond_t graph_cond_var;

public:

  typedef Concurrent_Graph<__GT, __Node, __Arc> Base_Graph;

  typedef typename Node::Node_Type Node_Type; 

  typedef typename Arc::Arc_Type Arc_Type;

  typedef typename Agent::Agent_Type Agent_Type;

  friend class Agent_Iterator;

private:

  Dlink  agent_list; // lista global de agentes
  size_t num_agents;
    
  Dlink ready_queue; // cola de agentes listos para procesarse
  size_t num_agents_ready; 
    
  Dlink suspended_queue; // cola de agentes suspendidos
  size_t num_agents_suspended;  

  size_t num_executing_agents;

      // bloquea start_graph() si usuario lo solicita
  pthread_cond_t block_starter_cond_var; 

      // atributos de la callback por timer
  void * (*time_callback)(void * graph);

  struct timespec time_callback_period;

  pthread_t time_callback_thread;

  pthread_cond_t time_callback_cond_var;

  typedef void * (Time_Callback)(void *);

      // inserta agente por el trasero de la cola de listos. 
      // ATENCIÓN: No toma el mutex
  void insert_agent_in_ready_queue(Agent * agent)
  {
    ready_queue.append(&agent->schedule_link_in_graph);
    num_agents_ready++;
    pthread_cond_signal(&graph_cond_var);
  }
    
      // saca de cola de listos el agente del frente. 
      // ATENCIÓN: No toma el mutex
  Agent * get_next_ready_agent() 
  {
    if (not ready_queue.is_empty())
      {
        Dlink * link = ready_queue.get_next();

        Walking_Agent<Agent_Type> * walking_agent = 
          Agent::schedule_link_in_graph_to_Walking_Agent(link);

        Agent * result = static_cast<Agent*>(walking_agent);

        ready_queue.remove_next();
        num_agents_ready--;

        return result;
      }

    return NULL;
  }
    
      // elimina el agente de la cola de listos; nótese que se trata de
      // un agente arbitrario, no del que está en el frente 
      // ATENCIÓN: No toma el mutex
  void remove_agent_from_ready_queue(Agent * agent)
  {
    num_agents_ready--;
    agent->schedule_link_in_graph.del();
  }
   
      // inserta agente por el trasero de la cola de suspendidos
      // ATENCIÓN: No toma el mutex
  void insert_agent_in_suspended_queue(Agent * agent)
  {
    suspended_queue.append(&agent->schedule_link_in_graph);
    num_agents_suspended++; 
  }

      // elimina el agente de la cola de suspendidos; nótese que se trata de
      // un agente arbitrario, no del que está en el frente 
      // ATENCIÓN: No toma el mutex
  void remove_agent_from_suspended_queue(Agent * agent)
  {
    agent->schedule_link_in_graph.del();
    num_agents_suspended--; 
  }

public:

  size_t get_num_agents() const 
  {
    CRITICAL_SECTION(graph_mutex);
    return num_agents; 
  }

  size_t get_num_threads() const
  {
    CRITICAL_SECTION(graph_mutex);
    return num_threads;
  }

private:

      // inserción genérica en nodo o arco
      template <typename Location> inline static 
  void __insert_agent(Agent * agent, Location * location)
  {
    location->agent_list.append(&agent->location_link); 
  }

  void __insert_agent_in_node(Agent * agent, Node * p)
  {
    agent->location = p;
    agent->in_node = true;
    __insert_agent(agent, p);
  }

  void __insert_agent_in_arc(Agent * agent, Arc * a)
  {
    agent->location = a;
    agent->in_node = false;
    __insert_agent(agent, a);
  }

  void __remove_agent_from_location(Agent * agent)
  {
    agent->location = NULL;
    agent->in_node = false;
    agent->location_link.del();
  }

  void __remove_agent_from_node(Agent * agent)
  {
    __remove_agent_from_location(agent);
  }

  void __remove_agent_from_arc(Agent * agent)
  {
    __remove_agent_from_location(agent);
  }

public:

  Agent * get_first_agent()
  {
    CRITICAL_SECTION(graph_mutex);

    if (get_num_agents() == 0)
      throw std::range_error("Graph has not agents ");

    return static_cast<Agent*>
      (Agent::agent_link_in_graph_to_Walking_Agent (&*agent_list.get_next()));
  } 

private:

      // asume que mutex del grafo ya está tomados.
      // ATENCIÓN: INVOCANTE DEBE SOLTAR EL MUTEX Y LUEGO HACER DELETE 
  void __remove_agent_from_graph(Agent * agent)
  { 
    if (not agent->location_link.is_empty()) 
      agent->location_link.del();

    agent->agent_link_in_graph.del();
    num_agents--;
        
    agent->schedule_link_in_graph.del();

    switch (agent->agent_state)
      {
      case Agent::Deleting_From_Ready:
        num_agents_ready--; break;

      case Agent::Deleting_From_Suspended:
        num_agents_suspended--; break;

      case Agent::Without_Queue:
      case Agent::Invalid:              break; 

      default: EXIT("Invalid call (state %ld)", agent->agent_state);
      }
  }

public:

  void remove_agent(Agent * agent)
  { 
    CRITICAL_SECTION(graph_mutex);

    switch (agent->agent_state)
      {
      case Agent::Ready:
        agent->agent_state = Agent::Deleting_From_Ready;
        break; // eliminará agente después del switch

      case Agent::Suspended:
        agent->agent_state = Agent::Deleting_From_Suspended;
        break; // eliminará agente después del switch

      case Agent::Without_Queue:
      case Agent::Invalid:              break; 

      case Agent::Executing:
      case Agent::Suspending:
             // Eliminación será realizada por thread que está
             // ejecutando el agente
        agent->agent_state = Agent::Deleting_From_Ready; 
        return;

      case Agent::Deleting_From_Ready:
      case Agent::Deleting_From_Suspended:
        throw std::domain_error("Agent is already deleting");

      default: EXIT("Invalid agent state %ld", agent->agent_state);
      }

    __remove_agent_from_graph(agent);
    
    delete agent;
  }

private:

  void create_thread(pthread_t & thread, void *( *start_routine)(void*))
  {
    int result = pthread_create(&thread, NULL, start_routine, this); 

    if (result != 0)
      {
        if (result == EAGAIN)
          throw std::bad_alloc();
        else
          throw std::exception();
      }
  }

      // retorna el índice dentro del arreglo threads[] de thread
  int search_thread(const pthread_t & thread)
  {
    return sequential_search(threads, thread, 0, num_threads - 1);
  }

  void cancel_threads()
  {
    if (threads == NULL) // ¿fueron las threads creadas?
      return; // NO

    for (int i = 0; i < num_threads; i++) 
      {
        int st = pthread_cancel(threads[i]);

        if (st != 0)
          ERROR("Error %d in pthread_join", st);
      }
  }

  void wait_for_threads_termination()
  {
    if (threads == NULL)
      return;

    for (int i = 0; i < num_threads; i++) 
      {
        int st = pthread_join(threads[i] , NULL);

        if (st != 0)
          ERROR("Error %d in pthread_join", st);
      }
  }

public:
   
  void start_graph(const bool block_caller = true) 
  {
    CRITICAL_SECTION(graph_mutex);

    if (status == Running)
      throw std::domain_error("Graph is already running");
    
    status = Running;

    threads = new pthread_t[num_threads];

        // crear las threads y colocarlas a ejecutar inmediatamente
    for (int i = 0; i < num_threads; i++) 
      create_thread(threads[i], agent_handler);

    if (time_callback != NULL)
      create_thread(time_callback_thread, time_callback_handler); 

    if (block_caller)
      pthread_cond_wait(&block_starter_cond_var, &graph_mutex);
  }

private:

  void __stop_graph()
  {
    if (threads == NULL)
      return;

    if (status == Stopped)
      throw std::domain_error("Graph is already stopped");

    if (status == Suspended)
      throw std::domain_error("Graph is suspended");

    status = Stopped;

    pthread_cond_broadcast(&graph_cond_var);
    pthread_cond_signal(&time_callback_cond_var);
  }

public:

  void stop_graph()
  {
    {
      CRITICAL_SECTION(graph_mutex);
      __stop_graph();
    }

    wait_for_threads_termination();
  }

  void suspend_graph()
  {
    CRITICAL_SECTION(graph_mutex);
    status = Suspended;
  }

  void resume_graph()
  {
    {
      CRITICAL_SECTION(graph_mutex);

      if (status != Suspended)
        throw std::domain_error("Graph has not been previously suspended");

      status = Running;
    }
    
    pthread_cond_broadcast(&graph_cond_var);

    if (time_callback)
      pthread_cond_signal(&time_callback_cond_var);
  }

  void clear_agent_list()
  {
    if (status == Running)
      suspend_graph();

    for (Dlink::Iterator it(&agent_list); it.has_current(); /* empty */)
      {
        Walking_Agent<Agent_Type> * walking_agent = 
          Agent::agent_link_in_graph_to_Walking_Agent(it.get_current());

        Agent * agent = static_cast<Agent*>(walking_agent);
          
        it.next();
        remove_agent(agent);
      }
   }

  const long & get_num_agents_ready() const 
  {
    CRITICAL_SECTION(graph_mutex);

    return num_agents_ready; 
  }

  const long & get_num_agents_suspended() const 
  {
    CRITICAL_SECTION(graph_mutex);

    return num_agents_suspended;
  }

private:

  void __suspend_agent_in_graph(Agent * agent)
  {
    remove_agent_from_ready_queue(agent);
    insert_agent_in_suspended_queue(agent);    
  }

public:
    
  void suspend_agent(Agent * agent)
  {
    CRITICAL_SECTION(graph_mutex);

        // validar que el agente se suspenda en un estado correcto
    switch (agent->agent_state)
      {
      case Agent::Suspended:
        throw std::domain_error("Agent is already suspended");

      case Agent::Suspending:
        throw std::domain_error("Agent is already suspending");

      case Agent::Executing:
            // suspención será realizada por run() luego de ejecución 
        agent->agent_state = Agent::Suspending; 
        return;

      case Agent::Ready: 
        agent->agent_state = Agent::Suspended;
        break;

      default: 
        EXIT("Invalid agent state %ld", agent->agent_state);
      }
    __suspend_agent_in_graph(agent);
  }

private:

  void __resume_agent_in_graph(Agent * agent) 
  {
    remove_agent_from_suspended_queue(agent);
    insert_agent_in_ready_queue(agent);
  }

public:

  void resume_agent(Agent * agent) 
  {
    CRITICAL_SECTION(graph_mutex);

    switch (agent->agent_state)
      {
      case Agent::Ready: 
      case Agent::Executing:
        throw throw std::domain_error("Agent is not suspended");

      case Agent::Suspending:
      case Agent::Suspended:
        agent->agent_state = Agent::Ready;
        break;

      default:
        Exit("Invalid agent state %ld", agent->agent_state);
      }
  
    __resume_agent_in_graph(agent);
  }

private:

  void init()
  {
    init_mutex(graph_mutex);
    pthread_cond_init(&graph_cond_var, NULL);
    pthread_cond_init(&block_starter_cond_var, NULL); 
  }

public:

  Agent_Graph(const size_t & __num_threads = 3, const size_t & num_mutexes = 4) 
    : Base_Graph(num_mutexes),
      num_threads(__num_threads), threads(NULL), status(Init),
      num_agents(0), num_agents_ready(0), num_agents_suspended(0), 
      num_executing_agents(0), time_callback(NULL)
   {
     init();
   }

  void set_num_threads(const size_t & __num_threads)
  {
    if (threads != NULL)
      throw std::domain_error("The graph has already threads");

    num_threads = __num_threads;
  }

  virtual ~Agent_Graph() 
  {
    {
      CRITICAL_SECTION(graph_mutex);

      if (threads != NULL)
        if (status == Running or status == Suspended or num_agents_ready == 0)
          {
            status = Stopped;
            pthread_cond_broadcast(&graph_cond_var);
          }
    }

    cancel_time_callback();

    wait_for_threads_termination();

    clear_agent_list();
    destroy_mutex(graph_mutex);
    pthread_cond_destroy(&graph_cond_var);
    pthread_cond_destroy(&block_starter_cond_var);

    delete [] threads; 
  } 

  Agent_Graph(const Agent_Graph & g)
    : Base_Graph(g),
      num_threads(g.num_threads), threads(NULL), status(Init),
      num_agents(0), num_agents_ready(), num_agents_suspended(0), 
      num_executing_agents(0), time_callback(g.time_callback)
   {
     init(); 
   }

private:

      template <typename Location> Agent * 
  insert_agent_in_location(Agent *    agent,
                           Location * location, 
                           const bool suspended = false)
  {
    if (not (agent->agent_state == Agent::Invalid))
      agent->agent_state = suspended ? Agent::Suspended : Agent::Ready;

    CRITICAL_SECTION(graph_mutex);

    if (not agent->location_link.is_empty())
      throw std::domain_error("agent is already inside the graph");

    { CRITICAL_SECTION(location->mutex);
      __insert_agent_in_node(agent, location);
    }

    agent_list.append(&agent->agent_link_in_graph);
    num_agents++;

    if (not suspended)
      insert_agent_in_ready_queue(agent);
    else
      insert_agent_in_suspended_queue(agent);

    return agent;
  }

public:

  Agent * insert_agent_in_node(Agent * agent, Node * p, 
                               const bool suspended = false)
  {
    return insert_agent_in_location(agent, p, suspended);
  }

  Agent * insert_agent_in_arc(Agent * agent, Arc * a, 
                              const bool suspended = false)
  {
    return insert_agent_in_location(agent, a, suspended);
  }

private:

      template <typename Location> Agent * 
  create_agent_in_location(const Agent_Type &   agent_data, 
                           Location *           location, 
                           long (*process_node) (void *  agent, 
                                                 void *  graph, 
                                                 void *  src, 
                                                 void *& arc) = NULL,
                           long (*process_arc) (void * agent, 
                                                void * graph, 
                                                void * arc) = NULL,
                           const bool          suspended = false)
  {
        /* puesto que esta función crea el agente y determina su dirección,
           no es necesario proteger sus atributos con el mutex */
    Agent * agent = new Agent(process_node, process_arc);

    agent->info    = agent_data;

    if (agent->is_a_valid_agent())
       return insert_agent_in_location(agent, location, suspended);
    else
      return insert_agent_in_location(agent, location, true); // suspendido

    return agent;
  }

public:

      Agent * 
  create_agent_in_node(const Agent_Type & agent_data, 
                       Node *               node, 
                       long (*process_node) (void *  agent, 
                                             void *  graph, 
                                             void *  src, 
                                              void *& arc) = NULL,
                       long (*process_arc) (void * agent, 
                                            void * graph, 
                                            void * arc) = NULL,
                       const bool          suspended = false)
  {
    return create_agent_in_location(agent_data, node, 
                                    process_node, process_arc, suspended);
  }
    
      Agent * 
  create_agent_in_arc(const Agent_Type &   agent_data, 
                      Arc *                arc, 
                      long (*process_node) (void *  agent, 
                                            void *  graph, 
                                            void *  src, 
                                            void *& arc) = NULL,
                      long (*process_arc) (void * agent, 
                                           void * graph, 
                                           void * arc) = NULL,
                      const bool &       suspended = false)
  {
    return create_agent_in_location(agent_data, arc, 
                                    process_node, process_arc, suspended);
  }
    
      template <class Equal>
  Agent * search_agent(const Agent_Type & agent_data)
  {
    CRITICAL_SECTION(graph_mutex);

    for (Dlink::Iterator it(agent_list); it.has_current(); it.next()) 
      {
        Agent * agent = static_cast<Agent*>
          (Agent::agent_link_in_graph_to_Walking_Agent(it.get_current()));

        if (Equal () (agent->get_info(), agent_data)) 
          return agent;
      }
      
    return NULL;
  }
   
  Agent * search_agent(const Agent_Type & agent)
  {
    return search_agent<Aleph::equal_to<Agent_Type> >(agent);
  }

  int get_status()
  {
    CRITICAL_SECTION(graph_mutex);
    return status;
  }

private:

  static void * time_callback_handler(void * cookie)
  {
    Agent_Graph * graph = static_cast<Agent_Graph*>(cookie);

    timeval now;
    timespec timeout;

    time_t sec = graph->time_callback_period.tv_sec;
    long  nsec = graph->time_callback_period.tv_nsec;
 
    while (true)
      {
        CRITICAL_SECTION(graph->graph_mutex);

        switch (graph->status)
          {
          case Stopped: return NULL; // terminar thread

          case Suspended: // suspender la thread con variable condición
            pthread_cond_wait(&graph->time_callback_cond_var, 
                              &graph->graph_mutex);
            break;
            
          case Running:
            gettimeofday(&now, NULL);
            timeout.tv_sec  = now.tv_sec + sec;
            timeout.tv_nsec = now.tv_usec * 1000 + nsec;
            if (timeout.tv_nsec >= 1000000000)
              {
                timeout.tv_sec++;
                timeout.tv_nsec %= 1000000000;
              }
            
            if (pthread_cond_timedwait(&graph->time_callback_cond_var, 
                                       &graph->graph_mutex, 
                                       &timeout) == ETIMEDOUT)
              if (graph->status == Running)
                (*graph->time_callback)(graph);  // invocar time_callback

            break; // case Running:

          default: EXIT("Invalid execution state %ld", graph->status);
          }
      }
  }


public:

  void set_time_callback(Time_Callback __time_callback, 
                    const int  & sec     = 10,
                    const long & nanosec = 0)
  {
    CRITICAL_SECTION(graph_mutex);

    I(__time_callback != NULL);

    time_callback                = __time_callback;
    time_callback_period.tv_sec  = sec;
    time_callback_period.tv_nsec = nanosec;

    pthread_cond_init(&time_callback_cond_var, NULL);
  }

  void cancel_time_callback()
  {
    if (time_callback == NULL)
      return;

    CRITICAL_SECTION(graph_mutex);
    pthread_cancel(time_callback_thread);
    time_callback = NULL;
    pthread_cond_destroy(&time_callback_cond_var);
  }

private:

  long execute_from_node(Agent * agent, Node * src)
  {
    if (agent->process_from_node == NULL)
      throw std::domain_error("There is no function for arc processing");

    I(agent->is_in_node());

    void * a  = NULL;
    long ret   = -1;

    { CRITICAL_SECTION(src->mutex);
      ret = agent->process_from_node(agent, this, src, a);
    }

    switch (ret)
    {
    case Agent::Remain_In_Node:
    case Agent::Suspend:
    case Agent::Dead:
      break;

    case Agent::Enter_In_Arc:
      {
        if (a == NULL)
          ERROR("there is no arc to enter");
        agent->src = src;

        { CRITICAL_SECTION(src->mutex);
          __remove_agent_from_node(agent);
        }

        Arc * arc = (Arc*) a;

        { CRITICAL_SECTION(arc->mutex);
          __insert_agent_in_arc(agent, arc);
        }
        break;
      }
    case Agent::Enter_In_Node:
      ERROR("Invalid action: Enter_In_Node transition from a node");

    case Agent::Remain_In_Arc:
      ERROR("Invalid action: Remain_In_Node transition from a node");

    default: ERROR("Panic: invalid transtion action");
    }

    return ret;
  }

  long execute_from_arc(Agent * agent, Arc * a) 
  {
    if (agent->process_from_arc == NULL)
      throw std::domain_error("There is no function for arc processing");

    long ret = -1;
    { CRITICAL_SECTION(a->mutex);
      ret = agent->process_from_arc(agent, this, a);
    }

    switch (ret)
    {
    case Agent::Remain_In_Arc:
    case Agent::Suspend:
    case Agent::Dead:
      break;

    case Agent::Enter_In_Node:
      {
        Node * tgt = NULL;
        Node * src = (Node *) agent->src;

        { CRITICAL_SECTION(a->mutex);
          tgt = get_connected_node(a, src);
          __remove_agent_from_arc(agent);
        }

        { CRITICAL_SECTION(tgt->mutex);
          __insert_agent_in_node(agent, tgt); 
        }

        agent->src = NULL;

        break;
      }
    case Agent::Enter_In_Arc:
      ERROR("Invalid action: Enter_In_Arc transition from a arc");

    case Agent::Remain_In_Node:
      ERROR("Invalid action: Remain_In_Arc transition from a arc");

    default: ERROR("Panic: invalid transtion action");
    }

    return ret;
  }

  long execute_out_location(Agent * agent) 
  {
    if (agent->process_out_location == NULL)
      throw std::domain_error("There is no function for arc processing");
    
    I(agent->location_link.is_empty());

    
  }

  long execute_agent(Agent * agent)
  {
    // if (agent->location_link.is_empty())
    // TODO: una nueva función si el agente no está en localidad
    if (agent->is_in_node())
      {
        Node * p = (Node*)agent->location;
        return execute_from_node(agent, p);
      }

    Arc * a = (Arc*)agent->location;
    return execute_from_arc(agent, a);
  }

  static void * agent_handler(void * cookie) 
  {
    Agent_Graph * graph = static_cast<Agent_Graph*>(cookie);

    while (true)
      {
        CRITICAL_SECTION(graph->graph_mutex);

        switch (graph->status)
          {
          case Stopped: return NULL; // terminar thread

          case Suspended: // suspender la thread con variable condición
            pthread_cond_wait(&graph->graph_cond_var, &graph->graph_mutex);
            break;
            
          case Running:
            {
              Agent * ready_agent = graph->get_next_ready_agent();

              if (ready_agent == NULL) 
                {
                  if (graph->num_executing_agents == 0 and 
                      graph->num_agents_ready == 0)
                        // despertar a thread start_graph() (si fue suspendida)
                    pthread_cond_signal(&graph->block_starter_cond_var);

                      // si no hay agente se señaliza start_graph() y se
                      // suspende esta thread    
                  pthread_cond_wait(&graph->graph_cond_var, &graph->graph_mutex);
                }
              else
                {  graph->num_executing_agents++;

                      // libera mutex grafo antes ejecutar agente 
                  critical_section.unlock(); 
                  long action = graph->execute_agent(ready_agent);
                  critical_section.lock(); // retoma el mutex del grafo

                  graph->num_executing_agents--;

                  switch (action)
                    {
                    case Agent::Remain_In_Node:
                    case Agent::Remain_In_Arc:
                    case Agent::Enter_In_Node:
                    case Agent::Enter_In_Arc:
                      graph->insert_agent_in_ready_queue(ready_agent);
                      break;
                      
                    case Agent::Suspend:
                      graph->__suspend_agent_in_graph(ready_agent);
                      break;
                      
                    case Agent::Dead:
                      graph->__remove_agent_from_graph(ready_agent);
                      delete ready_agent;
                      break;
                    } // end switch (action)
                }
              break; // case Running:
            }
          default: EXIT("Invalid execution state %ld", graph->status);
          }
      } // end while (true)
  }

  void verify_graph_conditions_of_modification()
  {
    if (status == Running)
      throw std::domain_error("graph is running");

    if (status == Init)
      throw std::domain_error("graph is Init state. Use without mutex");
  }

public:

  class Agent_Iterator : public Dlink::Iterator
  {
    Agent_Graph * g;

  public:

    Agent_Iterator(Agent_Graph & __g) : Dlink::Iterator(&(__g.agent_list)), g(&__g)
    {
      // empty
    }

    Agent * get_current()
    {
      Agent * curr_agent = static_cast<Agent*>
        (Agent::agent_link_in_graph_to_Walking_Agent(Dlink::Iterator::get_current()));

      return curr_agent;
    }
  }; 


};  // end Agent_Graph 


} // namespace Aleph
# endif