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
  10 #include <spec_lib.h>
  11 #include <cdsannotate.h>
  12 #include <specannotation.h>
  13 #include <model_memory.h>
  14 #include "common.h"
  15
  16
  17 #include "common.h"
  18
  19 #define relaxed memory_order_relaxed
  20 #define release memory_order_release
  21 #define acquire memory_order_acquire
  22 #define acq_rel memory_order_acq_rel
  23 #define seq_cst memory_order_seq_cst
  24
  25 using namespace std;
  26
  27 /**
  28         For the sake of simplicity, we map Integer -> Integer.
  29 */
  30
  31 class Entry {
  32         public:
  33         int key;
  34         atomic_int value;
  35         int hash;
  36         atomic<Entry*> next;
  37
  38         Entry(int h, int k, int v, Entry *n) {
  39                 this->hash = h;
  40                 this->key = k;
  41                 this->value.store(v, relaxed);
  42                 this->next.store(n, relaxed);
  43         }
  44 };
  45
  46 class Segment {
  47         public:
  48         int count;
  49         mutex segMutex;
  50
  51         void lock() {
  52                 segMutex.lock();
  53         }
  54
  55         void unlock() {
  56                 segMutex.unlock();
  57         }
  58
  59         Segment() {
  60                 this->count = 0;
  61         }
  62 };
  63
  64 /**
  65     @Begin
  66         @Class_begin
  67         @End
  68 */
  69 class HashMap {
  70         public:
  71
  72         /**
  73                 @Begin
  74                 @Options:
  75                         LANG = CPP;
  76                         CLASS = HashMap;
  77                 @Global_define:
  78                         @DeclareVar:
  79                                 IntegerMap *__map;
  80                         @InitVar:
  81                                 __map = createIntegerMap();
  82                         @Finalize:
  83                                 if (__map)
  84                                         destroyIntegerMap(__map);
  85                                 return true;
  86                 @Happens_before: Put -> Get
  87                 @Commutativity: Put <-> Put: _Method1.key != _Method2.key
  88                 @Commutativity: Put <-> Get: _Method1.key != _Method2.key
  89                 @Commutativity: Get <-> Get: true
  90                 @End
  91         */
  92
  93         atomic<Entry*> *table;
  94
  95         int capacity;
  96         int size;
  97
  98         static const int CONCURRENCY_LEVEL = 4;
  99
 100         static const int SEGMENT_MASK = CONCURRENCY_LEVEL - 1;
 101
 102         Segment *segments[CONCURRENCY_LEVEL];
 103
 104         static const int DEFAULT_INITIAL_CAPACITY = 16;
 105
 106         // Not gonna consider resize now...
 107
 108         HashMap() {
 109                 /**
 110                         @Begin
 111                         @Entry_point
 112                         @End
 113                 */
 114                 this->size = 0;
 115                 this->capacity = DEFAULT_INITIAL_CAPACITY;
 116                 this->table = new atomic<Entry*>[capacity];
 117                 for (int i = 0; i < capacity; i++) {
 118                         atomic_init(&table[i], NULL);
 119                 }
 120                 for (int i = 0; i < CONCURRENCY_LEVEL; i++) {
 121                         segments[i] = new Segment;
 122                 }
 123         }
 124
 125         int hashKey(int key) {
 126                 return key;
 127         }
 128
 129
 130         /**
 131                 @Begin
 132                 @Interface: Get
 133                 @Commit_point_set: GetReadValue1 | GetReadValue2 | GetReadNothing
 134                 @ID: __RET__
 135                 @Action:
 136                         int res = getIntegerMap(__map, key);
 137                         //model_print("Get: key=%d, res=%d\n", key, res);
 138
 139                 @Post_check: __RET__ ? res == __RET__: true
 140                 @End
 141         */
 142         int get(int key) {
 143                 ASSERT (key);
 144                 int hash = hashKey(key);
 145
 146                 // Try first without locking...
 147                 atomic<Entry*> *tab = table;
 148                 int index = hash & (capacity - 1);
 149                 atomic<Entry*> *first = &tab[index];
 150                 Entry *e;
 151                 int res = 0;
 152
 153                 // Should be a load acquire
 154                 // This load action here makes it problematic for the SC analysis, what
 155                 // we need to do is as follows: if the get() method ever acquires the
 156                 // lock, we ignore this operation for the SC analysis, and otherwise we
 157                 // take it into consideration
 158
 159                 /**** UL (main.cc) ****/
 160                 Entry *firstPtr = first->load(acquire);
 161
 162                 e = firstPtr;
 163                 while (e != NULL) {
 164                         if (key, e->key) {
 165                                 /**** inadmissible (testcase1.cc) ****/
 166                                 res = e->value.load(seq_cst);
 167                                 /**
 168                                         @Begin
 169                                         @Commit_point_define_check: res != 0
 170                                         @Label: GetReadValue1
 171                                         @End
 172                                 */
 173                                 if (res != 0)
 174                                         return res;
 175                                 else
 176                                         break;
 177                         }
 178                         // Loading the next entry, this can be relaxed because the
 179                         // synchronization has been established by the load of head
 180                         e = e->next.load(relaxed);
 181                 }
 182
 183                 // Recheck under synch if key apparently not there or interference
 184                 Segment *seg = segments[hash & SEGMENT_MASK];
 185                 seg->lock(); // Critical region begins
 186                 // Not considering resize now, so ignore the reload of table...
 187
 188                 // Synchronized by locking, no need to be load acquire
 189                 Entry *newFirstPtr = first->load(relaxed);
 190                 if (e != NULL || firstPtr != newFirstPtr) {
 191                         e = newFirstPtr;
 192                         while (e != NULL) {
 193                                 if (key == e->key) {
 194                                         // Protected by lock, no need to be SC
 195                                         res = e->value.load(relaxed);
 196                                         /**
 197                                                 @Begin
 198                                                 @Commit_point_define_check: true
 199                                                 @Label: GetReadValue2
 200                                                 @End
 201                                         */
 202                                         seg->unlock(); // Critical region ends
 203                                         return res;
 204                                 }
 205                                 // Synchronized by locking
 206                                 e = e->next.load(relaxed);
 207                         }
 208                 }
 209                 seg->unlock(); // Critical region ends
 210                 /**
 211                         @Begin
 212                         @Commit_point_define_check: true
 213                         @Label: GetReadNothing
 214                         @End
 215                 */
 216                 return 0;
 217         }
 218
 219         /**
 220                 @Begin
 221                 @Interface: Put
 222                 @Commit_point_set: PutUpdateValue | PutInsertValue
 223                 @ID: value
 224                 @Action:
 225                         putIntegerMap(__map, key, value);
 226                         //model_print("Put: key=%d, val=%d\n", key, value);
 227                 @End
 228         */
 229         int put(int key, int value) {
 230                 ASSERT (key && value);
 231                 int hash = hashKey(key);
 232                 Segment *seg = segments[hash & SEGMENT_MASK];
 233                 atomic<Entry*> *tab;
 234
 235                 seg->lock(); // Critical region begins
 236                 tab = table;
 237                 int index = hash & (capacity - 1);
 238
 239                 atomic<Entry*> *first = &tab[index];
 240                 Entry *e;
 241                 int oldValue = 0;
 242
 243                 // The written of the entry is synchronized by locking
 244                 Entry *firstPtr = first->load(relaxed);
 245                 e = firstPtr;
 246                 while (e != NULL) {
 247                         if (key == e->key) {
 248                                 // This could be a relaxed (because locking synchronize
 249                                 // with the previous put()),  no need to be acquire
 250                                 oldValue = e->value.load(relaxed);
 251
 252                                 /**** inadmissible (testcase1.cc) ****/
 253                                 e->value.store(value, seq_cst);
 254                                 /**
 255                                         @Begin
 256                                         @Commit_point_define_check: true
 257                                         @Label: PutUpdateValue
 258                                         @End
 259                                 */
 260                                 seg->unlock(); // Don't forget to unlock before return
 261                                 return oldValue;
 262                         }
 263                         // Synchronized by locking
 264                         e = e->next.load(relaxed);
 265                 }
 266
 267                 // Add to front of list
 268                 Entry *newEntry = new Entry(hash, key, value, firstPtr);
 269
 270                 /**** UL (main.cc) ****/
 271                 // Publish the newEntry to others
 272                 first->store(newEntry, release);
 273                 /**
 274                         @Begin
 275                         @Commit_point_clear: true
 276                         @Label: PutClear
 277                         @End
 278                 */
 279
 280                 /**
 281                         @Begin
 282                         @Commit_point_define_check: true
 283                         @Label: PutInsertValue
 284                         @End
 285                 */
 286                 seg->unlock(); // Critical region ends
 287                 return 0;
 288         }
 289 };
 290 /**
 291     @Begin
 292         @Class_end
 293         @End
 294 */
 295
 296 #endif