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