concurrent-hashmap/hashmap.h

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