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