benchmark/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 <stdlib.h>
   8 #include <mutex>
   9 #include "common.h"
  10
  11 #define relaxed memory_order_relaxed
  12 #define release memory_order_release
  13 #define acquire memory_order_acquire
  14 #define acq_rel memory_order_acq_rel
  15 #define seq_cst memory_order_seq_cst
  16
  17 using namespace std;
  18
  19 /**
  20         For the sake of simplicity, we map Integer -> Integer.
  21 */
  22
  23
  24
  25 class Entry {
  26         public:
  27         int key;
  28         atomic_int value;
  29         int hash;
  30         atomic<Entry*> next;
  31
  32         Entry(int h, int k, int v, Entry *n) {
  33                 this->hash = h;
  34                 this->key = k;
  35                 this->value.store(v, relaxed);
  36                 this->next.store(n, relaxed);
  37         }
  38 };
  39
  40 class Segment {
  41         public:
  42         int count;
  43         mutex segMutex;
  44
  45         void lock() {
  46                 segMutex.lock();
  47         }
  48
  49         void unlock() {
  50                 segMutex.unlock();
  51         }
  52
  53         Segment() {
  54                 this->count = 0;
  55         }
  56 };
  57
  58 /**
  59     @Begin
  60         @Class_begin
  61         @End*/
  62 class HashMap {
  63         public:
  64         atomic<Entry*> *table;
  65
  66         int capacity;
  67         int size;
  68
  69         static const int CONCURRENCY_LEVEL = 4;
  70
  71         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
  72
  73         Segment *segments[CONCURRENCY_LEVEL];
  74
  75         static const int DEFAULT_INITIAL_CAPACITY = 16;
  76
  77         // Not gonna consider resize now...
  78
  79         HashMap() {
  80                 this->size = 0;
  81                 this->capacity = DEFAULT_INITIAL_CAPACITY;
  82                 this->table = new atomic<Entry*>[capacity];
  83                 for (int i = 0; i < capacity; i++) {
  84                         atomic_init(&table[i], NULL);
  85                 }
  86                 for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
  87                         segments[i] = new Segment;
  88                 }
  89         }
  90
  91         int hashKey(int key) {
  92                 return key;
  93         }
  94
  95         int get(int key) {
  96                 ASSERT (key);
  97                 int hash = hashKey(key);
  98
  99                 // Try first without locking...
 100                 atomic<Entry*> *tab = table;
 101                 int index = hash & (capacity - 1);
 102                 atomic<Entry*> *first = &tab[index];
 103                 Entry *e;
 104                 int res = 0;
 105
 106                 // Should be a load acquire
 107                 // This load action here makes it problematic for the SC analysis, what
 108                 // we need to do is as follows: if the get() method ever acquires the
 109                 // lock, we ignore this operation for the SC analysis, and otherwise we
 110                 // take it into consideration
 111
 112                 Entry *firstPtr = first->load(acquire);
 113
 114                 e = firstPtr;
 115                 while (e != NULL) {
 116                         if (key, e->key) {
 117                                 res = e->value.load(seq_cst);
 118                                 if (res != 0)
 119                                         return res;
 120                                 else
 121                                         break;
 122                         }
 123                         // Loading the next entry, this can be relaxed because the
 124                         // synchronization has been established by the load of head
 125                         e = e->next.load(relaxed);
 126                 }
 127
 128                 // Recheck under synch if key apparently not there or interference
 129                 Segment *seg = segments[hash & SEGMENT_MASK];
 130                 seg->lock(); // Critical region begins
 131                 // Not considering resize now, so ignore the reload of table...
 132
 133                 // Synchronized by locking, no need to be load acquire
 134                 Entry *newFirstPtr = first->load(relaxed);
 135                 if (e != NULL || firstPtr != newFirstPtr) {
 136                         e = newFirstPtr;
 137                         while (e != NULL) {
 138                                 if (key == e->key) {
 139                                         // Protected by lock, no need to be SC
 140                                         res = e->value.load(relaxed);
 141                                         seg->unlock(); // Critical region ends
 142                                         return res;
 143                                 }
 144                                 // Synchronized by locking
 145                                 e = e->next.load(relaxed);
 146                         }
 147                 }
 148                 seg->unlock(); // Critical region ends
 149                 return 0;
 150         }
 151
 152         int put(int key, int value) {
 153                 ASSERT (key && value);
 154                 int hash = hashKey(key);
 155                 Segment *seg = segments[hash & SEGMENT_MASK];
 156                 atomic<Entry*> *tab;
 157
 158                 seg->lock(); // Critical region begins
 159                 tab = table;
 160                 int index = hash & (capacity - 1);
 161
 162                 atomic<Entry*> *first = &tab[index];
 163                 Entry *e;
 164                 int oldValue = 0;
 165
 166                 // The written of the entry is synchronized by locking
 167                 Entry *firstPtr = first->load(relaxed);
 168                 e = firstPtr;
 169                 while (e != NULL) {
 170                         if (key == e->key) {
 171                                 // This could be a relaxed (because locking synchronize
 172                                 // with the previous put()),  no need to be acquire
 173                                 oldValue = e->value.load(relaxed);
 174                                 e->value.store(value, seq_cst);
 175                                 seg->unlock(); // Don't forget to unlock before return
 176                                 return oldValue;
 177                         }
 178                         // Synchronized by locking
 179                         e = e->next.load(relaxed);
 180                 }
 181
 182                 // Add to front of list
 183                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 184                 // Publish the newEntry to others
 185                 first->store(newEntry, release);
 186                 seg->unlock(); // Critical region ends
 187                 return 0;
 188         }
 189 };
 190
 191 #endif