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         bool equals(Value *other) {
  69                 if (!other)
  70                         return false;
  71                 return vX == other->vX && vY == other->vY && vZ == other->vZ;
  72         }
  73 };
  74
  75 class Entry {
  76         public:
  77         Key *key;
  78         atomic<Value*> value;
  79         int hash;
  80         atomic<Entry*> next;
  81
  82         Entry(int h, Key *k, Value *v, Entry *n) {
  83                 this->hash = h;
  84                 this->key = k;
  85                 this->value.store(v, relaxed);
  86                 this->next.store(n, relaxed);
  87         }
  88 };
  89
  90 class Segment {
  91         public:
  92         int count;
  93         mutex segMutex;
  94
  95         void lock() {
  96                 segMutex.lock();
  97         }
  98
  99         void unlock() {
 100                 segMutex.unlock();
 101         }
 102
 103         Segment() {
 104                 this->count = 0;
 105         }
 106 };
 107
 108 class HashMap {
 109         public:
 110         atomic<Entry*> *table;
 111
 112         int capacity;
 113         int size;
 114
 115         static const int CONCURRENCY_LEVEL = 16;
 116
 117         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
 118
 119         Segment *segments[CONCURRENCY_LEVEL];
 120
 121         static const int DEFAULT_INITIAL_CAPACITY = 16;
 122
 123         // Not gonna consider resize now...
 124
 125         HashMap() {
 126                 this->size = 0;
 127                 this->capacity = DEFAULT_INITIAL_CAPACITY;
 128                 this->table = new atomic<Entry*>[capacity];
 129                 for (int i = 0; i < capacity; i++) {
 130                         atomic_init(&table[i], NULL);
 131                 }
 132                 for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
 133                         segments[i] = new Segment;
 134                 }
 135         }
 136
 137         int hashKey(Key *x) {
 138                 int h = x->hashCode();
 139                 // Use logical right shift
 140                 unsigned int tmp = (unsigned int) h;
 141                 return ((h << 7) - h + (tmp >> 9) + (tmp >> 17));
 142         }
 143
 144         bool eq(Key *x, Key *y) {
 145                 return x == y || x->equals(y);
 146         }
 147
 148         Value* get(Key *key) {
 149                 MODEL_ASSERT (key);
 150                 int hash = hashKey(key);
 151
 152                 // Try first without locking...
 153                 atomic<Entry*> *tab = table;
 154                 int index = hash & (capacity - 1);
 155                 atomic<Entry*> *first = &tab[index];
 156                 Entry *e;
 157                 Value *res = NULL;
 158
 159                 // Should be a load acquire
 160                 Entry *firstPtr = first->load(acquire);
 161                 e = firstPtr;
 162                 while (e != NULL) {
 163                         if (e->hash == hash && eq(key, e->key)) {
 164                                 res = e->value.load(seq_cst);
 165                                 if (res != NULL)
 166                                         return res;
 167                                 else
 168                                         break;
 169                                 // Loading the next entry
 170                                 e = e->next.load(acquire);
 171                         }
 172                 }
 173
 174                 // Recheck under synch if key apparently not there or interference
 175                 Segment *seg = segments[hash & SEGMENT_MASK];
 176                 seg->lock(); // Critical region begins
 177                 // Not considering resize now, so ignore the reload of table...
 178
 179                 // Synchronized by locking, no need to be load acquire
 180                 Entry *newFirstPtr = first->load(relaxed);
 181                 if (e != NULL || firstPtr != newFirstPtr) {
 182                         e = newFirstPtr;
 183                         while (e != NULL) {
 184                                 if (e->hash == hash && eq(key, e->key)) {
 185                                         res = e->value.load(seq_cst);
 186                                         return res;
 187                                 }
 188                                 // Synchronized by locking
 189                                 e = e->next.load(relaxed);
 190                         }
 191                 }
 192                 seg->unlock(); // Critical region ends
 193                 return NULL;
 194         }
 195
 196         Value* put(Key *key, Value *value) {
 197                 // Don't allow NULL key or value
 198                 MODEL_ASSERT (key && value);
 199
 200                 int hash = hashKey(key);
 201                 Segment *seg = segments[hash & SEGMENT_MASK];
 202                 atomic<Entry*> *tab;
 203
 204                 seg->lock(); // Critical region begins
 205                 tab = table;
 206                 int index = hash & (capacity - 1);
 207
 208                 atomic<Entry*> *first = &tab[index];
 209                 Entry *e;
 210                 Value *oldValue = NULL;
 211
 212                 // The written of the entry is synchronized by locking
 213                 Entry *firstPtr = first->load(relaxed);
 214                 e = firstPtr;
 215                 while (e != NULL) {
 216                         if (e->hash == hash && eq(key, e->key)) {
 217                                 // FIXME: This could be a relaxed (because locking synchronize
 218                                 // with the previous put())??
 219                                 oldValue = e->value.load(acquire);
 220                                 e->value.store(value, seq_cst);
 221                                 seg->unlock(); // Don't forget to unlock before return
 222                                 return oldValue;
 223                         }
 224                         // Synchronized by locking
 225                         e = e->next.load(relaxed);
 226                 }
 227
 228                 // Add to front of list
 229                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 230                 // Publish the newEntry to others
 231                 first->store(newEntry, release);
 232                 seg->unlock(); // Critical region ends
 233                 return NULL;
 234         }
 235
 236         Value* remove(Key *key, Value *value) {
 237                 MODEL_ASSERT (key);
 238                 int hash = hashKey(key);
 239                 Segment *seg = segments[hash & SEGMENT_MASK];
 240                 atomic<Entry*> *tab;
 241
 242                 seg->lock(); // Critical region begins
 243                 tab = table;
 244                 int index = hash & (capacity - 1);
 245
 246                 atomic<Entry*> *first = &tab[index];
 247                 Entry *e;
 248                 Value *oldValue = NULL;
 249
 250                 // The written of the entry is synchronized by locking
 251                 Entry *firstPtr = first->load(relaxed);
 252                 e = firstPtr;
 253
 254                 while (true) {
 255                         if (e != NULL) {
 256                                 seg->unlock(); // Don't forget to unlock
 257                                 return NULL;
 258                         }
 259                         if (e->hash == hash && eq(key, e->key))
 260                                 break;
 261                         // Synchronized by locking
 262                         e = e->next.load(relaxed);
 263                 }
 264
 265                 // FIXME: This could be a relaxed (because locking synchronize
 266                 // with the previous put())??
 267                 oldValue = e->value.load(acquire);
 268                 // If the value parameter is NULL, we will remove the entry anyway
 269                 if (value != NULL && value->equals(oldValue))
 270                         return NULL;
 271
 272                 // Force the get() to grab the lock and retry
 273                 e->value.store(NULL, relaxed);
 274
 275                 // The strategy to remove the entry is to keep the entries after the
 276                 // removed one and copy the ones before it
 277                 Entry *head = e->next.load(relaxed);
 278                 Entry *p;
 279                 p = first->load(relaxed);
 280                 while (p != e) {
 281                         head = new Entry(p->hash, p->key, p->value.load(relaxed), head);
 282                         p = p->next.load(relaxed);
 283                 }
 284
 285                 // Publish the new head to readers
 286                 first->store(head, release);
 287                 seg->unlock(); // Critical region ends
 288                 return oldValue;
 289         }
 290 };
 291
 292 #endif