ParallelHashMap.h

Go to the documentation of this file.

#ifndef __PARALLEL_HASH_MAP__
#define __PARALLEL_HASH_MAP__
#include<iostream>
#include<stdexcept>
#include<functional>
#include<omp.h>

#include "log.h"


template <class K, class V>
class FixedHashMap {
  struct node {
    node(K k_in, V v_in) : next(NULL), key(k_in), value(v_in) {}
    K key;
    V value;
    node *next;
  };

  private:
    size_t _M;      /* table size */
    size_t _N;      /* number of elements present in table */
    node ** _buckets;   /* buckets of values stored in nodes */

  public:

    FixedHashMap(size_t M = 64);
    virtual ~FixedHashMap();
    bool contains(K& key);
    V& at(K& key);
    void insert(K key, V value);
    long insert_and_get_count(K key, V value);
    size_t size();
    size_t bucket_count();
    K* keys();
    V* values();
    void clear();
    void print_buckets();
};


template <class K, class V>
class ParallelHashMap {

  /* padded pointer to hash table to avoid false sharing */
  struct paddedPointer {
    volatile long pad_L1;
    volatile long pad_L2;
    volatile long pad_L3;
    volatile long pad_L4;
    volatile long pad_L5;
    volatile long pad_L6;
    volatile long pad_L7;
    FixedHashMap<K,V> volatile* value;
    volatile long pad_R1;
    volatile long pad_R2;
    volatile long pad_R3;
    volatile long pad_R4;
    volatile long pad_R5;
    volatile long pad_R6;
    volatile long pad_R7;
    volatile long pad_R8;
  };

  private:
    FixedHashMap<K,V> *_table;
    paddedPointer *_announce;
    size_t _num_threads;
    size_t _N;
    omp_lock_t * _locks;
    size_t _num_locks;
    bool _fixed_size;
    void resize();

  public:
    ParallelHashMap(size_t M = 64, size_t L = 64);
    virtual ~ParallelHashMap();
    bool contains(K& key);
    V at(K& key);
    void update(K& key, V value);
    void insert(K key, V value);
    long insert_and_get_count(K key, V value);
    size_t size();
    size_t bucket_count();
    size_t num_locks();
    K* keys();
    V* values();
    void clear();
    void setFixedSize();
    void print_buckets();
    void realloc(size_t M);
};


template <class K, class V>
FixedHashMap<K,V>::FixedHashMap(size_t M) {

  /* ensure M is a power of 2 */
  if ((M & (M-1)) != 0) {
    /* if not, round up to nearest power of 2 */
    M--;
    for (size_t i = 1; i < 8 * sizeof(size_t); i*=2)
      M |= M >> i;
    M++;
  }

  /* allocate table */
  _M = M;
  _N = 0;
  _buckets = new node*[_M]();
}


template <class K, class V>
FixedHashMap<K,V>::~FixedHashMap() {
  /* for each bucket, scan through linked list and delete all nodes */
  for (size_t i=0; i<_M; i++) {
    node *iter_node = _buckets[i];
    while (iter_node != NULL) {
      node *next_node = iter_node->next;
      delete iter_node;
      iter_node = next_node;
    }
  }

  /* delete all buckets (now pointers to empty linked lists) */
  delete [] _buckets;
}


template <class K, class V>
bool FixedHashMap<K,V>::contains(K& key) {

  /* get hash into table assuming M is a power of 2, using fast modulus */
  size_t key_hash = std::hash<K>()(key) & (_M-1);

  /* search corresponding bucket for key */
  node *iter_node = _buckets[key_hash];
  while (iter_node != NULL) {
    if (iter_node->key == key)
      return true;
    else
      iter_node = iter_node->next;
  }
  return false;
}


template <class K, class V>
V& FixedHashMap<K,V>::at(K& key) {

  /* get hash into table assuming M is a power of 2, using fast modulus */
  size_t key_hash = std::hash<K>()(key) & (_M-1);

  /* search bucket for key and return the corresponding value if found */
  node *iter_node = _buckets[key_hash];
  while (iter_node != NULL) {
    if (iter_node->key == key)
      return iter_node->value;
    else
      iter_node = iter_node->next;
  }

  /* after the bucket has been completely searched without finding the key,
     print an error message */
  log_printf(ERROR, "Key not present in map");
}


template <class K, class V>
void FixedHashMap<K,V>::insert(K key, V value) {

  /* get hash into table using fast modulus */
  size_t key_hash = std::hash<K>()(key) & (_M-1);

  /* check to see if key already exists in map */
  if (contains(key))
    return;

  /* create new node */
  node *new_node = new node(key, value);

  /* find where to place element in linked list */
  node **iter_node = &_buckets[key_hash];
  while (*iter_node != NULL)
    iter_node = &(*iter_node)->next;

  /* place element in linked list */
  *iter_node = new_node;

  /* increment counter */
#pragma omp atomic
  _N++;
}


template <class K, class V>
long FixedHashMap<K,V>::insert_and_get_count(K key, V value) {

  /* get hash into table using fast modulus */
  size_t key_hash = std::hash<K>()(key) & (_M-1);

  /* check to see if key already exists in map */
  if (contains(key))
    return -1;

  /* create new node */
  node *new_node = new node(key, value);

  /* find where to place element in linked list */
  node **iter_node = &_buckets[key_hash];
  while (*iter_node != NULL)
    iter_node = &(*iter_node)->next;

  /* place element in linked list */
  *iter_node = new_node;

  /* increment counter and return number */
  size_t N;
#pragma omp atomic capture
    N = _N++;

  return (long) N;
}


template <class K, class V>
size_t FixedHashMap<K,V>::size() {
  return _N;
}


template <class K, class V>
size_t FixedHashMap<K,V>::bucket_count() {
  return _M;
}


template <class K, class V>
K* FixedHashMap<K,V>::keys() {

  /* allocate array of keys */
  K *key_list = new K[_N];

  /* fill array with keys */
  size_t ind = 0;
  for (size_t i=0; i<_M; i++) {
    node *iter_node = _buckets[i];
    while (iter_node != NULL) {
      key_list[ind] = iter_node->key;
      iter_node = iter_node->next;
      ind++;
    }
  }
  return key_list;
}


template <class K, class V>
V* FixedHashMap<K,V>::values() {

  /* allocate array of values */
  V *values = new V[_N];

  /* fill array with values */
  size_t ind = 0;
  for (size_t i=0; i<_M; i++) {
    node *iter_node = _buckets[i];
    while (iter_node != NULL) {
      values[ind] = iter_node->value;
      iter_node = iter_node->next;
      ind++;
    }
  }
  return values;
}


template <class K, class V>
void FixedHashMap<K,V>::clear() {

  /* for each bucket, scan through linked list and delete all nodes */
  for (size_t i=0; i<_M; i++) {
    node *iter_node = _buckets[i];
    while (iter_node != NULL) {
      node *next_node = iter_node->next;
      delete iter_node;
      iter_node = next_node;
    }
  }

  /* reset each bucket to null */
  for (size_t i=0; i<_M; i++)
    _buckets[i] = NULL;

  /* reset the number of entries to zero */
  _N = 0;
}


template <class K, class V>
void FixedHashMap<K,V>::print_buckets() {
  log_printf(NORMAL, "Printing all buckets in the hash map...");
  for (size_t i=0; i<_M; i++) {
    if (_buckets[i] == NULL)
      log_printf(NORMAL, "Bucket %d -> NULL", i);
    else
      log_printf(NORMAL, "Bucket %d -> %p", i, _buckets[i]);
  }
}


template <class K, class V>
ParallelHashMap<K,V>::ParallelHashMap(size_t M, size_t L) {

  /* allocate table */
  _table = new FixedHashMap<K,V>(M);
  _fixed_size = false;

  /* get number of threads and create concurrency structures */
  _num_threads = 1;
  _num_threads = omp_get_max_threads();
  _num_locks = L;
  _locks = new omp_lock_t[_num_locks];
  for (size_t i=0; i<_num_locks; i++)
    omp_init_lock(&_locks[i]);

  _announce = new paddedPointer[_num_threads];
  for (size_t t=0; t<_num_threads; t++)
    _announce[t].value = NULL;
}


template <class K, class V>
ParallelHashMap<K,V>::~ParallelHashMap() {
  delete _table;
  delete [] _locks;
  delete [] _announce;
}


template <class K, class V>
bool ParallelHashMap<K,V>::contains(K& key) {

  /* get thread ID */
  size_t tid = 0;
  tid = omp_get_thread_num();

  /* get pointer to table, announce it will be searched,
     and ensure consistency */
  FixedHashMap<K,V> *table_ptr;
  do {
    table_ptr = _table;
    _announce[tid].value = table_ptr;
#pragma omp flush
  } while (table_ptr != _table);

  /* see if current table contains the thread */
  bool present = table_ptr->contains(key);

  /* reset table announcement to not searching */
  _announce[tid].value = NULL;

  return present;
}


template <class K, class V>
V ParallelHashMap<K,V>::at(K& key) {

  /* If the size is fixed, simply return the value from the fixed hash map */
  if (_fixed_size)
    return _table->at(key);

  /* get thread ID */
  size_t tid = 0;
  tid = omp_get_thread_num();

  /* get pointer to table, announce it will be searched */
  FixedHashMap<K,V> *table_ptr;
  do {
    table_ptr = _table;
    _announce[tid].value = table_ptr;
#pragma omp flush
  } while (table_ptr != _table);

  /* get value associated with the key in the underlying table */
  V value = table_ptr->at(key);

  /* reset table announcement to not searching */
  _announce[tid].value = NULL;

  return value;
}


template <class K, class V>
void ParallelHashMap<K,V>::insert(K key, V value) {
  /* check if resize needed */
  if (2*_table->size() > _table->bucket_count())
    resize();

  /* check to see if key is already contained in the table */
  if (contains(key))
    return;

  /* get lock hash */
  size_t lock_hash = (std::hash<K>()(key) & (_table->bucket_count() - 1))
    % _num_locks;

  /* acquire lock */
  omp_set_lock(&_locks[lock_hash]);

  /* insert value */
  _table->insert(key, value);

  /* release lock */
  omp_unset_lock(&_locks[lock_hash]);
}


template <class K, class V>
void ParallelHashMap<K,V>::update(K& key, V value) {

  /* get lock hash */
  size_t lock_hash = (std::hash<K>()(key) & (_table->bucket_count() - 1))
    % _num_locks;

  /* acquire lock */
  omp_set_lock(&_locks[lock_hash]);

  /* insert value */
  _table->at(key) = value;

  /* release lock */
  omp_unset_lock(&_locks[lock_hash]);
}


template <class K, class V>
long ParallelHashMap<K,V>::insert_and_get_count(K key, V value) {

  /* check if resize needed */
  if (2*_table->size() > _table->bucket_count())
    resize();

  /* check to see if key is already contained in the table */
  if (contains(key))
    return -1;

  /* get lock hash */
  size_t lock_hash = (std::hash<K>()(key) & (_table->bucket_count() - 1))
    % _num_locks;

  /* acquire lock */
  omp_set_lock(&_locks[lock_hash]);

  /* insert value */
  long N =_table->insert_and_get_count(key, value);

  /* release lock */
  omp_unset_lock(&_locks[lock_hash]);

  return N;
}


template <class K, class V>
void ParallelHashMap<K,V>::resize() {

  /* do not resize if fixed size */
  if (_fixed_size)
    return;

  /* acquire all locks in order */
  for (size_t i=0; i<_num_locks; i++)
    omp_set_lock(&_locks[i]);

  /* recheck if resize needed */
  if (2*_table->size() < _table->bucket_count()) {
    /* release locks */
    for (size_t i=0; i<_num_locks; i++)
      omp_unset_lock(&_locks[i]);

    return;
  }

  /* allocate new hash map of double the size */
  FixedHashMap<K,V> *new_map =
    new FixedHashMap<K,V>(2*_table->bucket_count());

  /* get keys, values, and number of elements */
  K *key_list = _table->keys();
  V *value_list = _table->values();

  /* insert key/value pairs into new hash map */
  for (size_t i=0; i<_table->size(); i++)
    new_map->insert(key_list[i], value_list[i]);

  /* save pointer of old table */
  FixedHashMap<K,V> *old_table = _table;

  /* reassign pointer */
  _table = new_map;
#pragma omp flush

  /* release all locks */
  for (size_t i=0; i<_num_locks; i++)
    omp_unset_lock(&_locks[i]);

  /* delete key and value list */
  delete [] key_list;
  delete [] value_list;

  /* wait for all threads to stop reading from the old table */
  for (size_t i=0; i<_num_threads; i++) {
    while (_announce[i].value == old_table) {
#pragma omp flush
      continue;
    }
  }

  /* free memory associated with old table */
  delete old_table;
}


template <class K, class V>
size_t ParallelHashMap<K,V>::size() {
  return _table->size();
}


template <class K, class V>
size_t ParallelHashMap<K,V>::bucket_count() {
  return _table->bucket_count();
}


template <class K, class V>
size_t ParallelHashMap<K,V>::num_locks() {
  return _num_locks;
}


template <class K, class V>
K* ParallelHashMap<K,V>::keys() {

  /* get thread ID */
  size_t tid = 0;
  tid = omp_get_thread_num();

  /* get pointer to table, announce it will be searched */
  FixedHashMap<K,V> *table_ptr;
  do {
    table_ptr = _table;
    _announce[tid].value = table_ptr;
#pragma omp flush
  } while (table_ptr != _table);

  /* get key list */
  K* key_list = table_ptr->keys();

  /* reset table announcement to not searching */
  _announce[tid].value = NULL;

  return key_list;
}


template <class K, class V>
V* ParallelHashMap<K,V>::values() {

  /* get thread ID */
  size_t tid = 0;
  tid = omp_get_thread_num();

  /* get pointer to table, announce it will be searched */
  FixedHashMap<K,V> *table_ptr;
  do {
    table_ptr = _table;
    _announce[tid].value = table_ptr;
#pragma omp flush
  } while (table_ptr != _table);

  /* get value list */
  V* value_list = table_ptr->values();

  /* reset table announcement to not searching */
  _announce[tid].value = NULL;

  return value_list;
}


template <class K, class V>
void ParallelHashMap<K,V>::setFixedSize() {
  _fixed_size = true;
}


template <class K, class V>
void ParallelHashMap<K,V>::clear() {

  /* acquire all locks in order */
  for (size_t i=0; i<_num_locks; i++)
    omp_set_lock(&_locks[i]);

  /* clear underlying fixed table */
  _table->clear();

  /* release all locks in order */
  for (size_t i=0; i<_num_locks; i++)
    omp_unset_lock(&_locks[i]);
}


template <class K, class V>
void ParallelHashMap<K,V>::print_buckets() {

  /* get thread ID */
  size_t tid = 0;
  tid = omp_get_thread_num();

  /* get pointer to table, announce it will be searched */
  FixedHashMap<K,V> *table_ptr;
  do {
    table_ptr = _table;
    _announce[tid].value = table_ptr;
#pragma omp flush
  } while (table_ptr != _table);

  /* print buckets */
  table_ptr->print_buckets();

  /* reset table announcement to not searching */
  _announce[tid].value = NULL;
}

template <class K, class V>
void ParallelHashMap<K,V>::realloc(size_t M) {

  /* delete old table */
  delete _table;

  /* allocate new table */
  _table = new FixedHashMap<K,V>(M);
}
#endif