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 #include "sc_annotation.h"
  18
  19 #define relaxed memory_order_relaxed
  20 #define release memory_order_release
  21 #define acquire memory_order_acquire
  22 #define acq_rel memory_order_acq_rel
  23 #define seq_cst memory_order_seq_cst
  24
  25 using namespace std;
  26
  27 /**
  28         For the sake of simplicity, we do not use template but some toy structs to
  29         represent the Key and Value.
  30 */
  31 struct Key {
  32         // Probably represent the coordinate (x, y, z)
  33         int x;
  34         int y;
  35         int z;
  36
  37         int hashCode() {
  38                 return x + 31 * y + 31 * 31 * z;
  39         }
  40
  41         bool equals(Key *other) {
  42                 if (!other)
  43                         return false;
  44                 return x == other->x && y == other->y && z == other->z;
  45         }
  46
  47         Key(int x, int y, int z) :
  48                 x(x),
  49                 y(y),
  50                 z(z)
  51         {
  52
  53         }
  54 };
  55
  56 struct Value {
  57         // Probably represent the speed vector (vX, vY, vZ)
  58         int vX;
  59         int vY;
  60         int vZ;
  61
  62         Value(int vX, int vY, int vZ) :
  63                 vX(vX),
  64                 vY(vY),
  65                 vZ(vZ)
  66         {
  67
  68         }
  69
  70         bool equals(Value *other) {
  71                 if (!other)
  72                         return false;
  73                 return vX == other->vX && vY == other->vY && vZ == other->vZ;
  74         }
  75 };
  76
  77 class Entry {
  78         public:
  79         Key *key;
  80         atomic<Value*> value;
  81         int hash;
  82         atomic<Entry*> next;
  83
  84         Entry(int h, Key *k, Value *v, Entry *n) {
  85                 this->hash = h;
  86                 this->key = k;
  87                 this->value.store(v, relaxed);
  88                 this->next.store(n, relaxed);
  89         }
  90 };
  91
  92 class Segment {
  93         public:
  94         int count;
  95         mutex segMutex;
  96
  97         void lock() {
  98                 segMutex.lock();
  99         }
 100
 101         void unlock() {
 102                 segMutex.unlock();
 103         }
 104
 105         Segment() {
 106                 this->count = 0;
 107         }
 108 };
 109
 110 class HashMap {
 111         public:
 112         atomic<Entry*> *table;
 113
 114         int capacity;
 115         int size;
 116
 117         static const int CONCURRENCY_LEVEL = 16;
 118
 119         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
 120
 121         Segment *segments[CONCURRENCY_LEVEL];
 122
 123         static const int DEFAULT_INITIAL_CAPACITY = 16;
 124
 125         // Not gonna consider resize now...
 126
 127         HashMap() {
 128                 this->size = 0;
 129                 this->capacity = DEFAULT_INITIAL_CAPACITY;
 130                 this->table = new atomic<Entry*>[capacity];
 131                 for (int i = 0; i < capacity; i++) {
 132                         atomic_init(&table[i], NULL);
 133                 }
 134                 for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
 135                         segments[i] = new Segment;
 136                 }
 137         }
 138
 139         int hashKey(Key *x) {
 140                 int h = x->hashCode();
 141                 // Use logical right shift
 142                 unsigned int tmp = (unsigned int) h;
 143                 return ((h << 7) - h + (tmp >> 9) + (tmp >> 17));
 144         }
 145
 146         bool eq(Key *x, Key *y) {
 147                 return x == y || x->equals(y);
 148         }
 149
 150         Value* get(Key *key) {
 151                 MODEL_ASSERT (key);
 152                 int hash = hashKey(key);
 153
 154                 // Try first without locking...
 155                 atomic<Entry*> *tab = table;
 156                 int index = hash & (capacity - 1);
 157                 atomic<Entry*> *first = &tab[index];
 158                 Entry *e;
 159                 Value *res = NULL;
 160
 161                 // Should be a load acquire
 162                 // This load action here makes it problematic for the SC analysis, what
 163                 // we need to do is as follows: if the get() method ever acquires the
 164                 // lock, we ignore this operation for the SC analysis, and otherwise we
 165                 // take it into consideration
 166
 167                 SC_BEGIN();
 168                 Entry *firstPtr = first->load(acquire);
 169                 SC_KEEP();
 170                 SC_END();
 171
 172                 e = firstPtr;
 173                 while (e != NULL) {
 174                         if (e->hash == hash && eq(key, e->key)) {
 175                                 res = e->value.load(seq_cst);
 176                                 if (res != NULL)
 177                                         return res;
 178                                 else
 179                                         break;
 180                                 // Loading the next entry
 181                                 e = e->next.load(acquire);
 182                         }
 183                 }
 184
 185                 // Recheck under synch if key apparently not there or interference
 186                 Segment *seg = segments[hash & SEGMENT_MASK];
 187                 seg->lock(); // Critical region begins
 188                 // Not considering resize now, so ignore the reload of table...
 189
 190                 // Synchronized by locking, no need to be load acquire
 191                 Entry *newFirstPtr = first->load(relaxed);
 192                 if (e != NULL || firstPtr != newFirstPtr) {
 193                         e = newFirstPtr;
 194                         while (e != NULL) {
 195                                 if (e->hash == hash && eq(key, e->key)) {
 196                                         res = e->value.load(seq_cst);
 197                                         seg->unlock(); // Critical region ends
 198                                         return res;
 199                                 }
 200                                 // Synchronized by locking
 201                                 e = e->next.load(relaxed);
 202                         }
 203                 }
 204                 seg->unlock(); // Critical region ends
 205                 return NULL;
 206         }
 207
 208         Value* put(Key *key, Value *value) {
 209                 // Don't allow NULL key or value
 210                 MODEL_ASSERT (key && value);
 211
 212                 int hash = hashKey(key);
 213                 Segment *seg = segments[hash & SEGMENT_MASK];
 214                 atomic<Entry*> *tab;
 215
 216                 seg->lock(); // Critical region begins
 217                 tab = table;
 218                 int index = hash & (capacity - 1);
 219
 220                 atomic<Entry*> *first = &tab[index];
 221                 Entry *e;
 222                 Value *oldValue = NULL;
 223
 224                 // The written of the entry is synchronized by locking
 225                 Entry *firstPtr = first->load(relaxed);
 226                 e = firstPtr;
 227                 while (e != NULL) {
 228                         if (e->hash == hash && eq(key, e->key)) {
 229                                 // FIXME: This could be a relaxed (because locking synchronize
 230                                 // with the previous put())??
 231                                 oldValue = e->value.load(acquire);
 232                                 e->value.store(value, seq_cst);
 233                                 seg->unlock(); // Don't forget to unlock before return
 234                                 return oldValue;
 235                         }
 236                         // Synchronized by locking
 237                         e = e->next.load(relaxed);
 238                 }
 239
 240                 // Add to front of list
 241                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 242                 // Publish the newEntry to others
 243                 first->store(newEntry, release);
 244                 seg->unlock(); // Critical region ends
 245                 return NULL;
 246         }
 247
 248         Value* remove(Key *key, Value *value) {
 249                 MODEL_ASSERT (key);
 250                 int hash = hashKey(key);
 251                 Segment *seg = segments[hash & SEGMENT_MASK];
 252                 atomic<Entry*> *tab;
 253
 254                 seg->lock(); // Critical region begins
 255                 tab = table;
 256                 int index = hash & (capacity - 1);
 257
 258                 atomic<Entry*> *first = &tab[index];
 259                 Entry *e;
 260                 Value *oldValue = NULL;
 261
 262                 // The written of the entry is synchronized by locking
 263                 Entry *firstPtr = first->load(relaxed);
 264                 e = firstPtr;
 265
 266                 while (true) {
 267                         if (e != NULL) {
 268                                 seg->unlock(); // Don't forget to unlock
 269                                 return NULL;
 270                         }
 271                         if (e->hash == hash && eq(key, e->key))
 272                                 break;
 273                         // Synchronized by locking
 274                         e = e->next.load(relaxed);
 275                 }
 276
 277                 // FIXME: This could be a relaxed (because locking synchronize
 278                 // with the previous put())??
 279                 oldValue = e->value.load(acquire);
 280                 // If the value parameter is NULL, we will remove the entry anyway
 281                 if (value != NULL && value->equals(oldValue)) {
 282                         seg->unlock();
 283                         return NULL;
 284                 }
 285
 286                 // Force the get() to grab the lock and retry
 287                 e->value.store(NULL, relaxed);
 288
 289                 // The strategy to remove the entry is to keep the entries after the
 290                 // removed one and copy the ones before it
 291                 Entry *head = e->next.load(relaxed);
 292                 Entry *p;
 293                 p = first->load(relaxed);
 294                 while (p != e) {
 295                         head = new Entry(p->hash, p->key, p->value.load(relaxed), head);
 296                         p = p->next.load(relaxed);
 297                 }
 298
 299                 // Publish the new head to readers
 300                 first->store(head, release);
 301                 seg->unlock(); // Critical region ends
 302                 return oldValue;
 303         }
 304 };
 305
 306 #endif