concurrent-hashmap/hashmap.h

   1 #ifndef _HASHMAP_H
   2 #define _HASHMAP_H
   3
   4 #include <iostream>
   5 #include <atomic>
   6 #include "stdio.h"
   7 //#include <common.h>
   8 #ifdef STANDALONE
   9 #include <assert.h>
  10 #define MODEL_ASSERT assert
  11 #else
  12 #include <model-assert.h>
  13 #endif
  14 #include <stdlib.h>
  15 #include <mutex>
  16
  17 #define relaxed memory_order_relaxed
  18 #define release memory_order_release
  19 #define acquire memory_order_acquire
  20 #define acq_rel memory_order_acq_rel
  21 #define seq_cst memory_order_seq_cst
  22
  23 using namespace std;
  24
  25 /**
  26         For the sake of simplicity, we do not use template but some toy structs to
  27         represent the Key and Value.
  28 */
  29 struct Key {
  30         // Probably represent the coordinate (x, y, z)
  31         int x;
  32         int y;
  33         int z;
  34
  35         int hashCode() {
  36                 return x + 31 * y + 31 * 31 * z;
  37         }
  38
  39         bool equals(Key *other) {
  40                 if (!other)
  41                         return false;
  42                 return x == other->x && y == other->y && z == other->z;
  43         }
  44
  45         Key(int x, int y, int z) :
  46                 x(x),
  47                 y(y),
  48                 z(z)
  49         {
  50
  51         }
  52 };
  53
  54 struct Value {
  55         // Probably represent the speed vector (vX, vY, vZ)
  56         int vX;
  57         int vY;
  58         int vZ;
  59
  60         Value(int vX, int vY, int vZ) :
  61                 vX(vX),
  62                 vY(vY),
  63                 vZ(vZ)
  64         {
  65
  66         }
  67 };
  68
  69 class Entry {
  70         public:
  71         Key *key;
  72         atomic<Value*> value;
  73         int hash;
  74         atomic<Entry*> next;
  75
  76         Entry(int h, Key *k, Value *v, Entry *n) {
  77                 this->hash = h;
  78                 this->key = k;
  79                 this->value.store(v, relaxed);
  80                 this->next.store(n, relaxed);
  81         }
  82 };
  83
  84 class Segment {
  85         public:
  86         int count;
  87         mutex segMutex;
  88
  89         void lock() {
  90                 segMutex.lock();
  91         }
  92
  93         void unlock() {
  94                 segMutex.unlock();
  95         }
  96
  97         Segment() {
  98                 this->count = 0;
  99         }
 100 };
 101
 102 class HashMap {
 103         public:
 104         atomic<Entry*> *table;
 105
 106         int capacity;
 107         int size;
 108
 109         static const int CONCURRENCY_LEVEL = 16;
 110
 111         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
 112
 113         Segment segments[CONCURRENCY_LEVEL];
 114
 115         static const int DEFAULT_INITIAL_CAPACITY = 16;
 116
 117         // Not gonna consider resize now...
 118
 119         HashMap() {
 120                 this->size = 0;
 121                 this->capacity = DEFAULT_INITIAL_CAPACITY;
 122                 this->table = new atomic<Entry*>[capacity];
 123                 for (int i = 0; i < capacity; i++) {
 124                         atomic_init(&table[i], NULL);
 125                 }
 126         }
 127
 128         int hashKey(Key *x) {
 129                 int h = x->hashCode();
 130                 // Use logical right shift
 131                 unsigned int tmp = (unsigned int) h;
 132                 return ((h << 7) - h + (tmp >> 9) + (tmp >> 17));
 133         }
 134
 135         bool eq(Key *x, Key *y) {
 136                 return x == y || x->equals(y);
 137         }
 138
 139         Value* get(Key *key) {
 140                 int hash = hashKey(key);        // throws null pointer exception if key null
 141
 142                 // Try first without locking...
 143                 atomic<Entry*> *tab = table;
 144                 int index = hash & (capacity - 1);
 145                 atomic<Entry*> *first = &tab[index];
 146                 Entry *e;
 147                 Value *res = NULL;
 148
 149                 // Should be a load acquire
 150                 Entry *firstPtr = first->load(acquire);
 151                 e = firstPtr;
 152                 while (e != NULL) {
 153                         if (e->hash == hash && eq(key, e->key)) {
 154                                 res = e->value.load(seq_cst);
 155                                 if (res != NULL)
 156                                         return res;
 157                                 else
 158                                         break;
 159                                 // Loading the next entry
 160                                 e = e->next.load(acquire);
 161                         }
 162                 }
 163
 164                 // Recheck under synch if key apparently not there or interference
 165                 Segment seg = segments[hash & SEGMENT_MASK];
 166                 seg.lock(); // Critical region begins
 167                 // Not considering resize now, so ignore the reload of table...
 168                 atomic<Entry*> *newFirst = &tab[index];
 169
 170                 // Synchronized by locking, no need to be load acquire
 171                 Entry *newFirstPtr = newFirst->load(relaxed);
 172                 if (e != NULL || firstPtr != newFirstPtr) {
 173                         e = newFirstPtr;
 174                         while (e != NULL) {
 175                                 if (e->hash == hash && eq(key, e->key)) {
 176                                         res = e->value.load(seq_cst);
 177                                         return res;
 178                                 }
 179                                 // Synchronized by locking
 180                                 e = e->next.load(relaxed);
 181                         }
 182                 }
 183                 seg.unlock(); // Critical region ends
 184                 return NULL;
 185         }
 186
 187         Value* put(Key *key, Value *value) {
 188                 // Don't allow NULL value
 189                 //ASSERT (value != NULL);
 190
 191                 int hash = hashKey(key);
 192                 Segment seg = segments[hash & SEGMENT_MASK];
 193                 atomic<Entry*> *tab;
 194
 195                 seg.lock(); // Critical region begins
 196                 tab = table;
 197                 int index = hash & (capacity - 1);
 198
 199                 atomic<Entry*> *first = &tab[index];
 200                 Entry *e;
 201                 Value *res = NULL;
 202
 203                 // The written of the entry is synchronized by locking
 204                 Entry *firstPtr = first->load(relaxed);
 205                 e = firstPtr;
 206                 while (e != NULL) {
 207                         if (e->hash == hash && eq(key, e->key)) {
 208                                 // FIXME: This could be an acquire??
 209                                 res = e->value.load(acquire);
 210                                 e->value.store(value, seq_cst);
 211                                 seg.unlock(); // Don't forget to unlock before return
 212                                 return res;
 213                         }
 214                 }
 215
 216                 // Add to front of list
 217                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 218                 // Publish the newEntry to others
 219                 tab[index].store(newEntry, release);
 220                 seg.unlock(); // Critical region ends
 221                 return NULL;
 222         }
 223 };
 224
 225 #endif