tpl_concurrent_graph.H

# ifndef TPL_CONCURRENT_GRAPH_H
# define TPL_CONCURRENT_GRAPH_H

# include <pthread.h>

# include <autosprintf.h>

# include <gsl/gsl_rng.h>
# include <gsl/gsl_randist.h>

# include <tpl_graph.H>
# include <useMutex.H>

using namespace Aleph;


namespace Aleph 
{


 template <template <class, class> class GT, class __Node, class __Arc> 
class Concurrent_Graph;

  template <class Base_Object> 
class Lock_Object : public Base_Object 
{
public:

  pthread_mutex_t * mutex; // protege toda la data asociada al nodo

  typedef typename Base_Object::Item_Type Object_Type;

  Lock_Object(const Lock_Object & c_obj) 
    : Base_Object(c_obj) , mutex(c_obj.mutex)
  {
    I(not c_obj.delete_mutex);
  }
        
  Lock_Object(pthread_mutex_t * __mutex = NULL) 
    : Base_Object(), mutex(__mutex)
  {
    // empty
  }
        
  Lock_Object(const Object_Type & obj_info, pthread_mutex_t * __mutex = NULL) 
    : Base_Object(obj_info), mutex(__mutex)
  {
    // Empty
  }
  
  Lock_Object(Base_Object * pobj, pthread_mutex_t * __mutex = NULL) 
    : Base_Object(pobj) , mutex(__mutex)
  {
    // Empty
  }

  void set_mutex( pthread_mutex_t * ptrm)
  {
    mutex = ptrm;
  }

  void set_mutex(pthread_mutex_t & m)
  {
    mutex = &m;
  }

  class Critical_Section : public UseMutex
  {
  public:

    Critical_Section(Lock_Object * node) 
      : UseMutex(node->mutex) { /* empty */ }
  };
};


    template <template <class, class> class __GT, class __Node, class __Arc> 
class Concurrent_Graph : public __GT<Lock_Object<__Node>, Lock_Object<__Arc> >
{
public:

  typedef __GT<Lock_Object<__Node>, Lock_Object<__Arc> > GT;

  typedef typename GT::Node Node;

  typedef typename GT::Arc Arc;

protected:

  pthread_mutex_t mutex;

  size_t num_mutexes;

  DynArray<pthread_mutex_t> mutexes;

  void init_mutexes()
  {
    mutexes.reserve(num_mutexes);
    for (int i = 0; i < num_mutexes; ++i)
      init_mutex(&mutexes.access(i));
  }

  void uninit_mutexes()
  {
    for (int i = 0; i < num_mutexes; ++i)
      destroy_mutex(&mutexes.access(i));
  }

public:

  pthread_mutex_t * get_mutex(int i)
  {
    if (i >= num_mutexes)
      throw std::out_of_range(gnu::autosprintf("mutex index %d out of range", i));

    return mutexes.access(i);
  }

  pthread_mutex_t * allocate_mutex()
  {
    init_mutex(&mutexes.append());
    
    return mutexes.access(num_mutexes++);
  }
        
  typedef typename GT::Node_Type Node_Type; 

  typedef typename GT::Arc_Type Arc_Type;

  Concurrent_Graph(const size_t & n_mut = 1) : GT(), num_mutexes(n_mut)
  {
    init_mutex(mutex);  
    init_mutexes();   
  }

  Concurrent_Graph(const Concurrent_Graph & g) : GT(g), num_mutexes(g.num_mutexes)
  {
    init_mutex(mutex); 
    init_mutexes();   
  }

  void set_num_mutexes(const size_t & n)
  {
    CRITICAL_SECTION(mutex);

    if (n == 0)
      throw std::out_of_range("n is equal to zero");

    if (n < num_mutexes)
      throw std::out_of_range("n is smaller than current number of mutexes");

    mutexes.reserve(n);
    
    for (int i = num_mutexes; i < n; ++i)
      init_mutex(&mutexes.access(i));

    num_mutexes = n;
  }

  void distribute_mutexes_randomly()
  {
    gsl_rng * r = gsl_rng_alloc (gsl_rng_mt19937);
    gsl_rng_set(r, time(NULL) % gsl_rng_max(r));

    for (typename Concurrent_Graph::Node_Iterator it(*this); it.has_current();
         it.next())
      {
        int i = gsl_rng_uniform_int(r, num_mutexes); // índice del mutex al azar
        it.get_current()->set_mutex(mutexes.access(i));
      }

    for (typename Concurrent_Graph::Arc_Iterator it(*this); it.has_current();
         it.next())
      {
        int i = gsl_rng_uniform_int(r, num_mutexes); // índice del mutex al azar
        it.get_current()->set_mutex(mutexes.access(i));
      }

    gsl_rng_free(r);
  }

  void distribute_mutexes_uniformly()
  {
    {
      const size_t & num_nodes = this->get_num_nodes();
      const size_t num_nodes_by_mutex =  num_nodes / num_mutexes;

      typename GT::Node_Iterator it(*this); 
      for (int k = 0; k < num_nodes_by_mutex; ++k)
        {
          pthread_mutex_t & m = mutexes.access(k);

          for (int i = 0; it.has_current() and i < num_nodes; ++i, it.next())
            it.get_current()->set_mutex(m);
        }
    }

    {
      const size_t & num_arcs = this->get_num_arcs();
      const size_t num_arcs_by_mutex = num_arcs / num_mutexes;

      typename GT::Arc_Iterator it(*this); 
      for (int k = 0; k < num_arcs_by_mutex; ++k)
        {
          pthread_mutex_t & m = mutexes.access(k);

          for (int i = 0; it.has_current() and i < num_arcs; ++i, it.next())
            it.get_current()->set_mutex(m);
        }
    }
  }

  virtual ~Concurrent_Graph() 
  {
    uninit_mutexes();
    destroy_mutex(mutex);
  }

  class Node_Iterator  
  {
    Concurrent_Graph & cg;

    typename GT::Node_Iterator base_iterator;

  public:

    typedef typename GT::Node_Iterator::Item_Type Item_Type;
    
    Node_Iterator() : base_iterator() 
    { 
      // empty
    }

    Node_Iterator(Concurrent_Graph & _g) : cg(_g), base_iterator(_g)
    {
      // empty
    }

    Node_Iterator(const Node_Iterator & it) 
      : cg(it.cg), base_iterator(it.base_iterator)
    {
      // empty
    }

    Node_Iterator & operator = (const Node_Iterator & it) 
    {
      CRITICAL_SECTION(cg.mutex);

      if (&it == this)
        return *this;

      base_iterator = it.base_iterator;
      
      return *this;
    }
    
    Node * get_current_node() 
    {
      CRITICAL_SECTION(cg.mutex);

      Node * node = base_iterator.get_current_node();
        
      return node;
    }

    Node * get_current() { return get_current_node(); }  
        
    bool has_current() 
    {
      CRITICAL_SECTION(cg.mutex);
      return base_iterator.has_current();
    }
        
    void next() 
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.next();
    }

    void prev() 
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.prev();
    }

    void reset_first()
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.reset_first();
    }

    void reset_last()
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.reset_last();
    }
  };

  class Arc_Iterator 
  {
    Concurrent_Graph & cg;
        
    typename GT::Arc_Iterator base_iterator;

  public:

    typedef typename GT::Node_Iterator::Item_Type Item_Type;
    
    Arc_Iterator() : base_iterator() 
    { 
      // empty
    }

    Arc_Iterator(Concurrent_Graph & _g) : cg(_g), base_iterator(_g)
    {
      // empty
    }

    Arc_Iterator(const Arc_Iterator & it) 
      : cg(it.cg), base_iterator(it.base_iterator)
    {
      // empty
    }
        
    Arc_Iterator & operator = (const Arc_Iterator & it) 
    {
      CRITICAL_SECTION(cg.mutex);

      base_iterator = it.base_iterator;

      return *this;
    }

    Arc * get_current_arc() 
    { 
      CRITICAL_SECTION(cg.mutex);

      Arc * arc = base_iterator.get_current_arc();

      return arc;
    }

    Arc * get_current() { return get_current_arc(); }

    bool has_current() 
    {
      CRITICAL_SECTION(cg.mutex);
      return base_iterator.has_current();
    }
        
    void next() 
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.next();
    }

    void prev() 
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.prev();
    }

    void reset_first()
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.reset_first();
    }

    void reset_last()
    {
      CRITICAL_SECTION(cg.mutex);
      base_iterator.reset_last();
    }
  };

  size_t get_num_nodes()  
  {
    CRITICAL_SECTION(mutex);
    return GT::get_num_nodes(); 
  }

      // Hace visible get_num_arcs(Node*) heredada de List_Graph
  using List_Graph<Node, Arc>::get_num_arcs;
        
  size_t get_num_arcs() 
  {
    CRITICAL_SECTION(mutex); 
    return GT::get_num_arcs(); 
  }

  size_t get_num_mutexes()
  {
    CRITICAL_SECTION(mutex); 
    return num_mutexes;
  }
 
  Node * search_node(const Node_Type & node_info)
  {
    CRITICAL_SECTION(mutex);
    return GT::search_node(node_info);
  } 

  Node * insert_node(Node * node)
  {
    CRITICAL_SECTION(mutex);
    return GT::insert_node(node);
  }
        
  Node * insert_node(const Node_Type & node_info)
  {
    return insert_node( new Node(node_info) );
  }

  Node * get_first_node() 
  {
    CRITICAL_SECTION(mutex);
    return GT::get_first_node();
  }
        
  Arc * get_first_arc() 
  {
    CRITICAL_SECTION(mutex);
    return GT::get_first_arc(); 
  }
        
  void verify_graphs(Concurrent_Graph& g) 
  {
    CRITICAL_SECTION(mutex); 
    GT::verify_graphs(g);               
  }
        
  void remove_node(Node * node)
  {
    CRITICAL_SECTION(mutex);
    GT::remove_node(node);              
  }
  
  template <class Compare> void sort_arcs()
  {
    CRITICAL_SECTION(mutex);
    GT::template sort_arcs<Compare> ();
  }

  Arc * insert_arc(Node *           src_node, 
                   Node *           tgt_node, 
                   const Arc_Type & arc_info) 
  {
    CRITICAL_SECTION(mutex);
    return GT::insert_arc(src_node, tgt_node, arc_info);        
  }

  Arc * insert_arc(Node *           src_node, 
                   Node *           tgt_node) 
  {
    CRITICAL_SECTION(mutex);
    return GT::insert_arc(src_node, tgt_node);          
  }

  void remove_arc(Arc * arc)
  {
    CRITICAL_SECTION(mutex);
    GT::remove_arc(arc);
  }
    
  Arc * search_arc(Node * src_node, Node * tgt_node) 
  {
    CRITICAL_SECTION(mutex);
    return GT::search_arc(src_node, tgt_node);          
  }
    
  Arc * search_arc(const Arc_Type & arc_info)
  {
    CRITICAL_SECTION(mutex);
    return GT::search_arc(arc_info);            
  }
  
  bool arc_belong_to_graph(Arc * arc) 
  {
    CRITICAL_SECTION(mutex);
    return GT::arc_belong_to_graph(arc);
  }
};

}

# endif // TPL_CONCURRENT_GRAPH_H