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
  69 class Entry {
  70         public:
  71         Key *key;
  72         atomic<Value*> value;
  73         int hash;
  74         atomic<Entry*> next;
  75
  76         Entry(int h, Key *k, Value *v, Entry *n) {
  77                 this->hash = h;
  78                 this->key = k;
  79                 this->value.store(v, relaxed);
  80                 this->next.store(n, relaxed);
  81         }
  82 };
  83
  84 class Segment {
  85         public:
  86         int count;
  87         mutex segMutex;
  88
  89         void lock() {
  90                 segMutex.lock();
  91         }
  92
  93         void unlock() {
  94                 segMutex.unlock();
  95         }
  96
  97         Segment() {
  98                 this->count = 0;
  99         }
 100 };
 101
 102 class HashMap {
 103         public:
 104         atomic<Entry*> *table;
 105
 106         int capacity;
 107         int size;
 108
 109         static const int CONCURRENCY_LEVEL = 16;
 110
 111         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
 112
 113         Segment *segments[CONCURRENCY_LEVEL];
 114
 115         static const int DEFAULT_INITIAL_CAPACITY = 16;
 116
 117         // Not gonna consider resize now...
 118
 119         HashMap() {
 120                 this->size = 0;
 121                 this->capacity = DEFAULT_INITIAL_CAPACITY;
 122                 this->table = new atomic<Entry*>[capacity];
 123                 for (int i = 0; i < capacity; i++) {
 124                         atomic_init(&table[i], NULL);
 125                 }
 126                 for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
 127                         segments[i] = new Segment;
 128                 }
 129         }
 130
 131         int hashKey(Key *x) {
 132                 int h = x->hashCode();
 133                 // Use logical right shift
 134                 unsigned int tmp = (unsigned int) h;
 135                 return ((h << 7) - h + (tmp >> 9) + (tmp >> 17));
 136         }
 137
 138         bool eq(Key *x, Key *y) {
 139                 return x == y || x->equals(y);
 140         }
 141
 142         Value* get(Key *key) {
 143                 int hash = hashKey(key);        // throws null pointer exception if key null
 144
 145                 // Try first without locking...
 146                 atomic<Entry*> *tab = table;
 147                 int index = hash & (capacity - 1);
 148                 atomic<Entry*> *first = &tab[index];
 149                 Entry *e;
 150                 Value *res = NULL;
 151
 152                 // Should be a load acquire
 153                 Entry *firstPtr = first->load(acquire);
 154                 e = firstPtr;
 155                 while (e != NULL) {
 156                         if (e->hash == hash && eq(key, e->key)) {
 157                                 res = e->value.load(seq_cst);
 158                                 if (res != NULL)
 159                                         return res;
 160                                 else
 161                                         break;
 162                                 // Loading the next entry
 163                                 e = e->next.load(acquire);
 164                         }
 165                 }
 166
 167                 // Recheck under synch if key apparently not there or interference
 168                 Segment *seg = segments[hash & SEGMENT_MASK];
 169                 seg->lock(); // Critical region begins
 170                 // Not considering resize now, so ignore the reload of table...
 171
 172                 // Synchronized by locking, no need to be load acquire
 173                 Entry *newFirstPtr = first->load(relaxed);
 174                 if (e != NULL || firstPtr != newFirstPtr) {
 175                         e = newFirstPtr;
 176                         while (e != NULL) {
 177                                 if (e->hash == hash && eq(key, e->key)) {
 178                                         res = e->value.load(seq_cst);
 179                                         return res;
 180                                 }
 181                                 // Synchronized by locking
 182                                 e = e->next.load(relaxed);
 183                         }
 184                 }
 185                 seg->unlock(); // Critical region ends
 186                 return NULL;
 187         }
 188
 189         Value* put(Key *key, Value *value) {
 190                 // Don't allow NULL value
 191                 //ASSERT (value != NULL);
 192
 193                 int hash = hashKey(key);
 194                 Segment *seg = segments[hash & SEGMENT_MASK];
 195                 atomic<Entry*> *tab;
 196
 197                 seg->lock(); // Critical region begins
 198                 tab = table;
 199                 int index = hash & (capacity - 1);
 200
 201                 atomic<Entry*> *first = &tab[index];
 202                 Entry *e;
 203                 Value *res = NULL;
 204
 205                 // The written of the entry is synchronized by locking
 206                 Entry *firstPtr = first->load(relaxed);
 207                 e = firstPtr;
 208                 while (e != NULL) {
 209                         if (e->hash == hash && eq(key, e->key)) {
 210                                 // FIXME: This could be an acquire??
 211                                 res = e->value.load(acquire);
 212                                 e->value.store(value, seq_cst);
 213                                 seg->unlock(); // Don't forget to unlock before return
 214                                 return res;
 215                         }
 216                 }
 217
 218                 // Add to front of list
 219                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 220                 // Publish the newEntry to others
 221                 first->store(newEntry, release);
 222                 seg->unlock(); // Critical region ends
 223                 return NULL;
 224         }
 225 };
 226
 227 #endif