tpl_agent.H

/*
  This file is part of Aleph system

  Copyright (c) 2002, 2003, 2004, 2005, 2006
  UNIVERSITY LOS ANDES (ULA) Merida - REPÚBLICA BOLIVARIANA DE VENEZUELA  

  - Center of Studies in Microelectronics & Distributed Systems (CEMISID) 
  - ULA Computer Science Department
  - FUNDACITE Mérida - Ministerio de Ciencia y Tecnología

  PERMISSION TO USE, COPY, MODIFY AND DISTRIBUTE THIS SOFTWARE AND ITS
  DOCUMENTATION IS HEREBY GRANTED, PROVIDED THAT BOTH THE COPYRIGHT
  NOTICE AND THIS PERMISSION NOTICE APPEAR IN ALL COPIES OF THE
  SOFTWARE, DERIVATIVE WORKS OR MODIFIED VERSIONS, AND ANY PORTIONS
  THEREOF, AND THAT BOTH NOTICES APPEAR IN SUPPORTING DOCUMENTATION.

  Aleph is distributed in the hope that it will be useful, but WITHOUT
  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  or FITNESS FOR A PARTICULAR PURPOSE.

  UNIVERSIDAD DE LOS ANDES requests users of this software to return to 

  Leandro Leon
  CEMISID 
  Ed La Hechicera 
  3er piso, ala sur
  Facultad de Ingenieria 
  Universidad de Los Andes 
  Merida - REPÚBLICA BOLIVARIANA DE VENEZUELA    or

  lrleon@ula.ve

  any improvements or extensions that they make and grant Universidad 
  de Los Andes (ULA) the rights to redistribute these changes. 

  Aleph is (or was) granted by: 
  - Consejo de Desarrollo Cientifico, Humanistico, Tecnico de la ULA 
    (CDCHT)
  - Fundacite Mérida
*/

/*****************************************************************
                  Grafos de agentes
*****************************************************************/

/* 
   Contribuyentes:

   Aristides Castillo
   Derik Romero
   Leandro León
*/

# ifndef TPL_AGENT_GRAPH_H
# define TPL_AGENT_GRAPH_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 Node_Info> class Agent_Node; // Nodo de grafo
  

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

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

  typedef Concurrent_Node<Agents_Node_Info> Base_Node;

  typedef Agents_Node_Info Node_Type;

  Agent_Node() : Base_Node() { /* empty */ }

  Agent_Node(const Node_Type & info) : Base_Node(info) { /* empty */ }

  virtual ~Agent_Node(){ /* empty */ }
}; 
  
  
    template <typename Agents_Arc_Info> 
class Agent_Arc : public Concurrent_Arc<Agents_Arc_Info>    
{

public:

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

  typedef Concurrent_Arc<Agents_Arc_Info> Base_Arc;

  typedef Agents_Arc_Info Arc_Type;

  Agent_Arc() : Base_Arc() { /* empty */ } 

  virtual ~Agent_Arc() { /* empty */ } 
};
  

template <typename __Node,  // Definción de nodo basada Agent_Node<T>
          typename __Arc,   // Definición de arco basada en Agent_Arc<T>
       typename Agent_Info
          > 
class Agent_Graph : public Concurrent_Graph<__Node, __Arc>
{
public:

  typedef __Node Node;

  typedef __Arc Arc;

  typedef Agent_Graph<__Node, __Arc, Agent_Info> graph_type;

private:

      // Estados de ejecución del grafo
  enum Status { 
    Init,        // Estado inicial colocado por el constructor de 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)
  }; 

  const 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:

  class Agent
  {

    friend class Agent_Graph;

    enum Agent_State
      {
     Ready,
     Executing,
     Suspending,
     Suspended,
     Deleting_From_Ready,
     Deleting_From_Suspended
      };

        // protege atributos del agente 
    pthread_mutex_t mutex;

    Agent_State agent_state;

    pthread_t thread_id;

    Agent_Info info;

        // Enlace a lista de agentes de todo el grafo. Protegido por
        // mutex del grafo.
    Dlink agent_link_in_graph; 

        // Enlace a cola de listos o suspendidos. Protegido por mutex
        // del grafo.
    Dlink schedule_link_in_graph;

        // Cada nodo o arco contiene una lista de los agentes que se
        // encuentran en él. Este es el enlace. Aquí la protección es
        // delicada cuando hay cambio de localidad porque involucra
        // tomar el mutex de la localidad, eliminar el agente y soltar
        // el mutex. Luego, con la localidad destino, tomar el mutex,
        // insertar el agente y soltar.
    Dlink location_link;

    bool in_node; // true si agente se encuentra en nodo

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

    void * cookie;

  public:

    typedef void (*Transit) (Agent_Graph * /* grafo  */ , 
                    Agent * /* agente */ , 
                    void * /* cookie */);
    
    static void transit_fct(Agent_Graph *, Agent *, void*)
    {
      // empty
    }
    

  private:

    Transit transit; 

  public:

    Agent() : location(NULL), cookie(NULL) 
    {
      init_mutex(&mutex);
    }

    Agent(const Agent_Info & agent_info, Transit __transit) 
      : info(agent_info), location(NULL), cookie(NULL), transit(__transit)
    {
      init_mutex(&mutex);
    }
    
    virtual ~Agent() 
    {
      destroy_mutex(&mutex);
    }

   typedef Agent_Info Agent_Type;

  private:

    // ejecuta función transito y retorna true si el agente debe ser
    // post-procesado en cuyo caso (ATENCIÓN) el mutex queda tomado
    bool execute(Agent_Graph * graph)
    { 
      if (transit == NULL)
     throw std::domain_error("Transit function was not specified");

      {
     CRITICAL_SECTION(mutex);

     I(agent_state == Ready);

     thread_id   = pthread_self();
     agent_state = Executing;
      }

      transit(graph, this, cookie); 

      {
     CRITICAL_SECTION(mutex);

     if (agent_state == Executing)
       {
         agent_state = Ready; /* sólo cambiamos estado si éste no ha sido
                        cambiado por otra thread */

         return false;
       }
      
     critical_section.disallow_unlock();

     return true;
      }
    }
 
        // funciones de conversión a Agent desde un dlink 
    LINKNAME_TO_TYPE (Agent, agent_link_in_graph);  
    LINKNAME_TO_TYPE (Agent, schedule_link_in_graph);
    LINKNAME_TO_TYPE (Agent, location_link);

  public:

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

    Agent_Info & get_info() 
    {
      CRITICAL_SECTION(mutex);

      return info;
    }
  
    void *& get_cookie()
    {
      CRITICAL_SECTION(mutex);

      return cookie;
    }

    bool is_in_node()
    {
      CRITICAL_SECTION(mutex);

      return in_node;
    }

    struct Critical_Section : public UseMutex
    {
      Critical_Section(Agent * agent) : UseMutex(agent->mutex) { /* empty */ }
    };
  };  

  class Node_to_Node_Agent : public Agent
  {

    friend class Agent_Graph;

  public:
    
    typedef Arc * (*Leave_Node_Fct)(Agent_Graph*, Node*, Node_to_Node_Agent*);

    typedef bool (*Enter_Node_Fct)(Agent_Graph*, Node*, Node_to_Node_Agent*);

  private:

    Leave_Node_Fct leave_node_fct;

    Enter_Node_Fct enter_node_fct;

    static void transit_fct(Agent_Graph * g, Agent * __agent, void*)
    {
      Node_to_Node_Agent * agent = static_cast<Node_to_Node_Agent*>(__agent);

      Arc * arc       = NULL;
      Node * tgt_node = NULL;

      I(agent->in_node);

          // Determinar nodo de residencia actual
      Node * curr_node = static_cast<Node*>(agent->location);

          // Entrar a sección crítica curr_node
      CRITICAL_SECTION(curr_node->mutex); 

      if (agent->leave_node_fct != NULL)
     // Modificar nodo actual y decidir arco.
     arc = agent->leave_node_fct(g, curr_node, agent);

# ifdef nada
      if (arc == NULL)
     {
       g->remove_agent(agent); // eliminar agente del grafo
       
       return;
     }
# endif

      g->leave_agent_from_node(agent); // sacar agente del nodo

      tgt_node = g->get_connected_node(arc, curr_node);

      int lock_status = 0;

      do 
     lock_status = pthread_mutex_trylock(&tgt_node->mutex);
      while (lock_status == EBUSY);

      if (lock_status != 0)
     throw std::domain_error("Mutex cannot be locked");

      // En este punto los mutexes de los dos nodos están tomados

      if (not agent->enter_node_fct(g, tgt_node, agent))
     {
       pthread_mutex_unlock(&tgt_node->mutex);

       g->remove_agent(agent);
       
       return;
     }

      g->enter_agent_in_node(agent, tgt_node);

      pthread_mutex_unlock(&tgt_node->mutex);
  }

  public:

    Node_to_Node_Agent() 
    {
      leave_node_fct = NULL;
      enter_node_fct = NULL;
    }

    Node_to_Node_Agent (const Agent_Info & agent_info,
               Leave_Node_Fct __leave_node_fct, 
               Enter_Node_Fct __enter_node_fct)
      : Agent(agent_info, &transit_fct)
    {
      leave_node_fct = __leave_node_fct;
      enter_node_fct = __enter_node_fct;
    }
  };
  

  class Node_to_Arc_Agent : public Agent
  {

  };
  

  typedef Concurrent_Graph<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;

private:

  Dlink  agent_list; // lista 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;  

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

      // 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 * remove_next_ready_agent() 
  {
    if (not ready_queue.is_empty())
      {
     Dlink * link = ready_queue.get_next();

     Agent * agent = Agent::schedule_link_in_graph_to_Agent(link);

     Agent * result = static_cast<Agent*>(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:


  const size_t & get_num_agents() const 
  {
    CRITICAL_SECTION(graph_mutex);

    return num_agents; 
  }

  const size_t & get_num_threads() const
  {
    return num_threads;
  }

  Node * get_agent_node_location(Agent * agent) 
  {
    CRITICAL_SECTION(agent->mutex);

    I(not agent->location_link.is_empty());

    if (agent->in_node) 
      return static_cast<Node*>(agent->location);

    return NULL;
  }

private:

      // inserción genérica en nodo o arco
      template <typename Location> inline static 
  void insert_agent(Agent * agent, Location * location)
  {
    agent->location = location;

    location->agent_list.append(&agent->location_link); 
  }

  
      template <typename Location> inline static 
  void remove_agent_from_location(Agent * agent, Location * location)
  {
    agent->location_link->del(); 
  }

public:

  void leave_agent_from_node(Agent * agent)
  {
    I(agent->in_node);

    agent->location_link.del();
  }

  void enter_agent_in_node(Agent * agent, Node * node)
  {
    agent->in_node  = true;
    agent->location = node;

    insert_agent(agent, node);
  }

  Arc * get_agent_arc_location(Agent * agent)
  {
    CRITICAL_SECTION(agent->mutex);

    if (not agent->in_node) 
      return static_cast<Arc*>(agent->location);

    return NULL;
  }

      // cambia el agente a un nuevo nodo. Se asume que el mutex del nodo o
      // arco donde se encuentra el agente ya está tomado
  void set_agent_arc_location(Agent * agent, Arc * arc)
  {
    if (agent->agent_state == Agent::Suspended)
      throw std::domain_error("cannot set location of suspended agent");

    agent->location_link.del();

    agent->in_node = false;
    agent->location = arc;

    insert_agent(agent, arc); 
  }

  void enter_agent_in_arc(Agent *       agent, 
                 Arc *         arc, 
                 const bool &  with_mutex = false)
  {
    if (with_mutex)
      pthread_mutex_lock(&agent->mutex);

    {
      CRITICAL_SECTION(arc->mutex);

      agent->in_node  = false;
      agent->location = arc;

      insert_agent(agent, arc);
    }

    if (with_mutex)
      pthread_mutex_unlock(&agent->mutex);
  }

  void change_agent_arc_location(Agent * agent, Arc * arc)
  {
    CRITICAL_SECTION(agent->mutex);

    set_agent_arc_location(agent, arc);
  }

  bool & is_agent_in_node(Agent * agent) 
  {
    CRITICAL_SECTION(agent->mutex);

    return agent->in_node;
  }

  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_Agent (&*agent_list.get_next()));
  } 

private:

      // asume que mutex del grafo y del agente ya están tomados.
      // ATENCIÓN: INVOCANTE DEBE SOLTAR EL MUTEX Y LUEGO HACER DELETE 
  void __remove_agent_in_graph(Agent * agent)
  { 
    if (not agent->location_link.is_empty()) 
      {
     if (agent->in_node)
       pthread_mutex_lock(&static_cast<Node*>(agent->location)->mutex);
     else
       pthread_mutex_lock(&static_cast<Arc*>(agent->location)->mutex);

     agent->location_link.del();

     if (agent->in_node)
       pthread_mutex_unlock(&static_cast<Node*>(agent->location)->mutex);
     else
       pthread_mutex_unlock(&static_cast<Arc*>(agent->location)->mutex);
      }

    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;

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

public:

  void remove_agent(Agent * agent)
  { 
    {
      CRITICAL_SECTION(agent->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::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);
     }

      {
     CRITICAL_SECTION(graph_mutex);

     __remove_agent_in_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 wait_for_threads_termination()
  {
    for (int i = 0; i < num_threads; i++) 
      {
     int st = pthread_join(threads[i] , NULL);

     if (st != 0)
       EXIT("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;

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

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

  void stop_graph()
  {
    {
      CRITICAL_SECTION(graph_mutex);

      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);
    }

    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);
  }

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

    for (Dlink::Iterator it(&agent_list); it.has_current(); /* empty */)
      {
     Agent * agent = Agent::agent_link_in_graph_to_Agent(it.get_current());

        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)
  {
    {     // validar que el agente se suspenda en un estado correcto
      CRITICAL_SECTION(agent->mutex);

      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);
     }
    }
    
    CRITICAL_SECTION(graph_mutex);

    __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(agent->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);
     }
    }

    CRITICAL_SECTION(graph_mutex);

    __resume_agent_in_graph(agent);
  }

  typedef void * (Callback)(void *);

private:

  void * (*callback)(void *);

  int callback_freq;

  pthread_t callback_thread;

  void init()
  {
    threads = new pthread_t[num_threads];

    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 = 1) 
    : num_threads(num_threads), status(Init),
      num_agents(0), num_agents_ready(0), num_agents_suspended(0), 
      callback(NULL)
   {
     init();
   }

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

      if (status == Running)
     {
       status = Stopped;

       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), status(Init),
      num_agents(0), num_agents_ready(), num_agents_suspended(0), 
      callback(g.callback)
   {
     init(); 
   }

  Agent_Graph & operator = (const Agent_Graph & g)
  {
    if (this == &g)
      return *this;

    clear_agent_list();

    copy_graph(*this, const_cast<Agent_Graph&>(g));
  }

      template <typename Location> 
  Agent * insert_agent_in_location(Agent *      agent,
                       Location *   location, 
                       const bool & suspended = false)
  {
        // puesto que agent debe ser "nuevo", sus datos no están manejados
        // por otra thread y por lo tanto no tienen necesidad de protegerse
    agent->agent_state = suspended ? Agent::Suspended : Agent::Ready;
      
    agent_list.append(&agent->agent_link_in_graph);

    insert_agent(agent, location);

    {
      CRITICAL_SECTION(graph_mutex);

      num_agents++;

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

    return agent;
  }

      template <typename Location> Agent * 
  create_agent_in_location(const Agent_Info &      agent_data, 
                  typename Agent::Transit fnct,
                  Location *              location, 
                  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;

    agent->info    = agent_data;
    agent->transit = fnct;

    return insert_agent_in_location(agent, location, suspended);
  }

  Agent * create_agent_in_node(const Agent_Info &      agent_data, 
                      typename Agent::Transit fnct,
                      Node *                  node, 
                      const bool &            suspended = false)
  {
    return create_agent_in_location(agent_data, fnct, node, suspended);
  }
    
  Agent * create_agent_in_arc(const Agent_Info &      agent_data, 
                     typename Agent::Transit fnct,
                     Arc *                   arc,
                     const bool &            suspended = false)
  {
    return create_agent_in_location(agent_data, fnct, arc, suspended);
  }

      template <class Agent_Class>
  Agent_Class * insert_agent_in_node(Agent_Class * agent, 
                         Node *        node, 
                         const bool &  suspended = false)
  {
    Agent * ret_val = insert_agent_in_location(agent, node, suspended);

    return static_cast<Agent_Class*>(ret_val);
  }
    
    

  // TODO: inserción sin función de transición

      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_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 * callback_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:

         sleep(graph->callback_freq); // esperar 

         (*graph->callback)(cookie);  // invocar callback

         break; // case Running:

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

public:

  void set_callback(Callback __callback, const int & freq = 2000)
  {
    CRITICAL_SECTION(graph_mutex);

    I(__callback != NULL);

    callback      = __callback;
    callback_freq = freq;

    create_thread(callback_thread, callback_handler); 
  }

  void cancel_callback()
  {
    CRITICAL_SECTION(graph_mutex);

    if (callback == NULL)
      throw std::domain_error("callback has not been set");

    pthread_cancel(callback_thread);
  }

private:

  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->remove_next_ready_agent();
       
           if (ready_agent == NULL) 
          { // si no hay agente se señaliza a start_graph() y
            // se suspende thread  
            pthread_cond_signal(&graph->block_starter_cond_var);

            pthread_cond_wait(&graph->graph_cond_var, 
                        &graph->graph_mutex);
          }
           else
          {      // libera mutex grafo antes ejecutar agente
            critical_section.unlock(); 

            if (not ready_agent->execute(graph)) // ejecuta agente
              {
                critical_section.lock(); // retoma el mutex del grafo
            
                graph->insert_agent_in_ready_queue(ready_agent);
              }
            else
              {
                critical_section.lock(); // retoma el mutex del grafo

                // Revisar el estado del agente y post-procesarlo
                // mutex del agente sigue tomado pues se leerá estado
                switch (ready_agent->agent_state) 
               {
               case Agent::Suspending:
                 // libera mutex de agente pues ya leimos estado
                 pthread_mutex_unlock(&ready_agent->mutex);

                 graph->__suspend_agent_in_graph(ready_agent);
               
                 break;
               
               case Agent::Deleting_From_Ready:
               case Agent::Deleting_From_Suspended:

                 pthread_mutex_unlock(&ready_agent->mutex);

                 graph->__remove_agent_in_graph(ready_agent);

                 delete ready_agent;

                 break;

               default: EXIT("Invalid agent state %ld in run()",
                          ready_agent->agent_state);
               }
              } // end else of if (not ready_agent->execute(graph))
          }
           break; // case Running:
         }
     default: EXIT("Invalid execution state %ld", graph->status);
     }
      } // end while (true)
  }
};  // end Agent_Graph 


}

# endif 

---