benchmark/cliffc-hashtable/simplified_cliffc_hashtable.h

   1 #ifndef SIMPLIFIED_CLIFFC_HASHTABLE_H
   2 #define SIMPLIFIED_CLIFFC_HASHTABLE_H
   3
   4 #include <iostream>
   5 #include <atomic>
   6 #include <memory>
   7 #include <assert.h>
   8
   9 using namespace std;
  10
  11
  12
  13 /**
  14         This header file declares and defines a simplified version of Cliff Click's
  15         NonblockingHashMap. It contains all the necessary structrues and main
  16         functions. In simplified_cliffc_hashtable.cc file, it has the definition for
  17         the static fields.
  18 */
  19
  20 template<typename TypeK, typename TypeV>
  21 class cliffc_hashtable;
  22
  23 /**
  24         Corresponding the the Object[] array in Cliff Click's Java implementation.
  25         It keeps the first two slots for CHM (Hashtable control unit) and the hash
  26         records (an array of hash used for fast negative key-equality check).
  27 */
  28 struct kvs_data {
  29         int _size;
  30         atomic<void*> *_data;
  31
  32         kvs_data(int sz) {
  33                 _size = sz;
  34                 int real_size = sizeof(atomic<void*>) * 2 + 2;
  35                 _data = new atomic<void*>[real_size];
  36                 // The control block should be initialized in resize()
  37                 // Init the hash record array
  38                 int *hashes = new int[_size];
  39                 int i;
  40                 for (i = 0; i < _size; i++) {
  41                         hashes[i] = 0;
  42                 }
  43                 _data[1].store(hashes, memory_order_relaxed);
  44                 // Init the data to Null slot
  45                 for (i = 2; i < real_size; i++) {
  46                         _data[i].store(NULL, memory_order_relaxed);
  47                 }
  48         }
  49
  50         ~kvs_data() {
  51                 int *hashes = (int*) _data[1].load(memory_order_relaxed);
  52                 delete hashes;
  53                 delete[] _data;
  54         }
  55 };
  56
  57 struct slot {
  58         bool _prime;
  59         shared_ptr<void> _ptr;
  60
  61         slot(bool prime, shared_ptr<void> ptr) {
  62                 _prime = prime;
  63                 _ptr = ptr;
  64         }
  65
  66 };
  67
  68
  69 /**
  70         TypeK must have defined function "int hashCode()" which return the hash
  71         code for the its object, and "int equals(TypeK anotherKey)" which is
  72         used to judge equality.
  73         TypeK and TypeV should define their own copy constructor.
  74         To make the memory management safe and similar to Cliff Click's Java
  75         implementation, we use shared_ptr instead of normal pointer in terms of the
  76         pointers that point to TypeK and TypeV.
  77 */
  78 template<typename TypeK, typename TypeV>
  79 class cliffc_hashtable {
  80         /**
  81                 # The synchronization we have for the hashtable gives us the property of
  82                 # serializability, so we should have a sequential hashtable when we check the
  83                 # correctness. The key thing is to identify all the commit point.
  84
  85                 @Begin
  86                 @Options:
  87                         LANG = C;
  88                 @Global_define:
  89                         @DeclareVar:
  90                         spec_hashtable<TypeK, TypeV*> map;
  91                         spec_hashtable<TypeK, Tag> id_map;
  92                         Tag tag;
  93                         @InitVar:
  94                                 map = spec_hashtable<TypeK, TypeV*>();
  95                                 id_map = spec_hashtable<TypeK, TypeV*>();
  96                                 tag = Tag();
  97                         @DefineFunc:
  98                         static bool equals_val(TypeV *ptr1, TypeV *ptr2) {
  99                                 // ...
 100                         }
 101
 102                         # Update the tag for the current key slot if the corresponding tag
 103                         # is NULL, otherwise just return that tag. It will update the next
 104                         # available tag too if it requires a new tag for that key slot.
 105                         static Tag getKeyTag(TypeK &key) {
 106                                 if (id_map.get(key) == NULL) {
 107                                         Tag cur_tag = tag.current();
 108                                         id_map.put(key, cur_tag);
 109                                         tag.next();
 110                                         return cur_tag;
 111                                 } else {
 112                                         return id_map.get(key);
 113                                 }
 114                         }
 115
 116                 @Interface_cluster:
 117                         Read_interface = {
 118                                 Get,
 119                                 PutIfAbsent,
 120                                 RemoveAny,
 121                                 RemoveIfMatch,
 122                                 ReplaceAny,
 123                                 ReplaceIfMatch
 124                         }
 125
 126                         Write_interface = {
 127                                 Put,
 128                                 PutIfAbsent(COND_PutIfAbsentSucc),
 129                                 RemoveAny,
 130                                 RemoveIfMatch(COND_RemoveIfMatchSucc),
 131                                 ReplaceAny,
 132                                 ReplaceIfMatch(COND_ReplaceIfMatchSucc)
 133                         }
 134                 @Happens_before:
 135                         Write_interface -> Read_interface
 136                 @End
 137         */
 138
 139 friend class CHM;
 140         /**
 141                 The control structure for the hashtable
 142         */
 143         private:
 144         class CHM {
 145                 friend class cliffc_hashtable;
 146                 private:
 147                 atomic<kvs_data*> _newkvs;
 148
 149                 // Size of active K,V pairs
 150                 atomic_int _size;
 151
 152                 // Count of used slots
 153                 atomic_int _slots;
 154
 155                 // The next part of the table to copy
 156                 atomic_int _copy_idx;
 157
 158                 // Work-done reporting
 159                 atomic_int _copy_done;
 160
 161                 public:
 162                 CHM(int size) {
 163                         _size.store(size, memory_order_relaxed);
 164                         _slots.store(0, memory_order_relaxed);
 165
 166                         _copy_idx.store(0, memory_order_relaxed);
 167                         _copy_done.store(0, memory_order_release);
 168                 }
 169
 170                 ~CHM() {}
 171
 172                 private:
 173
 174                 // Heuristic to decide if the table is too full
 175                 bool table_full(int reprobe_cnt, int len) {
 176                         return
 177                                 reprobe_cnt >= REPROBE_LIMIT &&
 178                                 _slots.load(memory_order_relaxed) >= reprobe_limit(len);
 179                 }
 180
 181                 kvs_data* resize(cliffc_hashtable *topmap, kvs_data *kvs) {
 182                         kvs_data *newkvs = _newkvs.load(memory_order_acquire);
 183                         if (newkvs != NULL)
 184                                 return newkvs;
 185
 186                         // No copy in-progress, start one; Only double the table size
 187                         int oldlen = kvs->_size;
 188                         int sz = _size.load(memory_order_relaxed);
 189                         int newsz = sz;
 190
 191                         // Just follow Cliff Click's heuristic to decide the new size
 192                         if (sz >= (oldlen >> 2)) { // If we are 25% full
 193                                 newsz = oldlen << 1; // Double size
 194                                 if (sz >= (oldlen >> 1))
 195                                         newsz = oldlen << 2; // Double double size
 196                         }
 197
 198                         // We do not record the record timestamp
 199                         if (newsz <= oldlen) newsz = oldlen << 1;
 200                         // Do not shrink ever
 201                         if (newsz < oldlen) newsz = oldlen;
 202
 203                         // Last check cause the 'new' below is expensive
 204                         newkvs = _newkvs.load(memory_order_acquire);
 205                         if (newkvs != NULL) return newkvs;
 206
 207                         newkvs = new kvs_data(newsz);
 208                         void *chm = (void*) new CHM(sz);
 209                         newkvs->_data[0].store(chm, memory_order_relaxed);
 210
 211                         kvs_data *cur_newkvs;
 212                         // Another check after the slow allocation
 213                         if ((cur_newkvs = _newkvs.load(memory_order_acquire)) != NULL)
 214                                 return cur_newkvs;
 215                         // CAS the _newkvs to the allocated table
 216                         kvs_data *desired = (kvs_data*) NULL;
 217                         kvs_data *expected = (kvs_data*) newkvs;
 218                         if (!_newkvs.compare_exchange_strong(desired, expected, memory_order_release,
 219                                         memory_order_release)) {
 220                                 // Should clean the allocated area
 221                                 delete newkvs;
 222                                 newkvs = _newkvs.load(memory_order_acquire);
 223                         }
 224                         return newkvs;
 225                 }
 226
 227                 void help_copy_impl(cliffc_hashtable *topmap, kvs_data *oldkvs,
 228                         bool copy_all) {
 229                         assert (get_chm(oldkvs) == this);
 230                         kvs_data *newkvs = _newkvs.load(memory_order_acquire);
 231                         int oldlen = oldkvs->_size;
 232                         int min_copy_work = oldlen > 1024 ? 1024 : oldlen;
 233
 234                         // Just follow Cliff Click's code here
 235                         int panic_start = -1;
 236                         int copyidx;
 237                         while (_copy_done.load(memory_order_acquire) < oldlen) {
 238                                 copyidx = _copy_idx.load(memory_order_acquire);
 239                                 if (panic_start == -1) { // No painc
 240                                         copyidx = _copy_idx.load(memory_order_acquire);
 241                                         while (copyidx < (oldlen << 1) &&
 242                                                 !_copy_idx.compare_exchange_strong(copyidx, copyidx +
 243                                                         min_copy_work, memory_order_release, memory_order_release))
 244                                                 copyidx = _copy_idx.load(memory_order_relaxed);
 245                                         if (!(copyidx < (oldlen << 1)))
 246                                                 panic_start = copyidx;
 247                                 }
 248
 249                                 // Now copy the chunk of work we claimed
 250                                 int workdone = 0;
 251                                 for (int i = 0; i < min_copy_work; i++)
 252                                         if (copy_slot(topmap, (copyidx + i) & (oldlen - 1), oldkvs,
 253                                                 newkvs))
 254                                                 workdone++;
 255                                 if (workdone > 0)
 256                                         copy_check_and_promote(topmap, oldkvs, workdone);
 257
 258                                 copyidx += min_copy_work;
 259                                 if (!copy_all && panic_start == -1)
 260                                         return; // We are done with the work we claim
 261                         }
 262                         copy_check_and_promote(topmap, oldkvs, 0); // See if we can promote
 263                 }
 264
 265                 kvs_data* copy_slot_and_check(cliffc_hashtable *topmap, kvs_data
 266                         *oldkvs, int idx, void *should_help) {
 267                         kvs_data *newkvs = _newkvs.load(memory_order_acquire);
 268                         // We're only here cause the caller saw a Prime
 269                         if (copy_slot(topmap, idx, oldkvs, _newkvs))
 270                                 copy_check_and_promote(topmap, oldkvs, 1); // Record the slot copied
 271                         return (should_help == NULL) ? newkvs : topmap->help_copy(newkvs);
 272                 }
 273
 274                 void copy_check_and_promote(cliffc_hashtable *topmap, kvs_data*
 275                         oldkvs, int workdone) {
 276                         int oldlen = oldkvs->_size;
 277                         int copyDone = _copy_done.load(memory_order_relaxed);
 278                         if (workdone > 0) {
 279                                 while (true) {
 280                                         copyDone = _copy_done.load(memory_order_relaxed);
 281                                         if (_copy_done.compare_exchange_weak(copyDone, copyDone +
 282                                                 workdone, memory_order_relaxed, memory_order_relaxed))
 283                                                 break;
 284                                 }
 285                         }
 286
 287                         // Promote the new table to the current table
 288                         if (copyDone + workdone == oldlen &&
 289                                 topmap->_kvs.load(memory_order_acquire) == oldkvs)
 290                                 topmap->_kvs.compare_exchange_strong(oldkvs, _newkvs, memory_order_release,
 291                                         memory_order_release);
 292                 }
 293
 294                 bool copy_slot(cliffc_hashtable *topmap, int idx, kvs_data *oldkvs,
 295                         kvs_data *newkvs) {
 296                         slot *key_slot;
 297                         while ((key_slot = key(oldkvs, idx)) == NULL)
 298                                 CAS_key(oldkvs, idx, NULL, TOMBSTONE);
 299
 300                         // First CAS old to Prime
 301                         slot *oldval = val(oldkvs, idx, NULL);
 302                         while (!is_prime(oldval)) {
 303                                 slot *box = (oldval == NULL || oldval == TOMBSTONE)
 304                                         ? TOMBPRIME : new slot(true, oldval->_ptr);
 305                                 if (CAS_val(oldkvs, idx, oldval, box)) {
 306                                         if (box == TOMBPRIME)
 307                                                 return 1; // Copy done
 308                                         // Otherwise we CAS'd the box
 309                                         oldval = box; // Record updated oldval
 310                                         break;
 311                                 }
 312                                 oldval = val(oldkvs, idx, NULL); // Else re-try
 313                         }
 314
 315                         if (oldval == TOMBPRIME) return false; // Copy already completed here
 316
 317                         slot *old_unboxed = new slot(false, oldval->_ptr);
 318                         int copied_into_new = (putIfMatch(topmap, newkvs, key_slot, old_unboxed,
 319                                 NULL) == NULL);
 320
 321                         // Old value is exposed in the new table
 322                         while (!CAS_val(oldkvs, idx, oldval, TOMBPRIME))
 323                                 oldval = val(oldkvs, idx, NULL);
 324
 325                         return copied_into_new;
 326                 }
 327         };
 328
 329
 330
 331         private:
 332         static const int Default_Init_Size = 8; // Intial table size
 333
 334         static slot* const MATCH_ANY;
 335         static slot* const NO_MATCH_OLD;
 336
 337         static slot* const TOMBPRIME;
 338         static slot* const TOMBSTONE;
 339
 340         static const int REPROBE_LIMIT = 10; // Forces a table-resize
 341
 342         atomic<kvs_data*> _kvs;
 343
 344         public:
 345         cliffc_hashtable() {
 346                 // Should initialize the CHM for the construction of the table
 347                 // For other CHM in kvs_data, they should be initialzed in resize()
 348                 // because the size is determined dynamically
 349                 kvs_data *kvs = new kvs_data(Default_Init_Size);
 350                 void *chm = (void*) new CHM(0);
 351                 kvs->_data[0].store(chm, memory_order_relaxed);
 352                 _kvs.store(kvs, memory_order_release);
 353         }
 354
 355         cliffc_hashtable(int init_size) {
 356                 // Should initialize the CHM for the construction of the table
 357                 // For other CHM in kvs_data, they should be initialzed in resize()
 358                 // because the size is determined dynamically
 359                 kvs_data *kvs = new kvs_data(init_size);
 360                 void *chm = (void*) new CHM(0);
 361                 kvs->_data[0].store(chm, memory_order_relaxed);
 362                 _kvs.store(kvs, memory_order_release);
 363         }
 364
 365         /**
 366                 @Begin
 367                 @Interface: Get
 368                 @Commit_point_set: Read_Val_Point1 | Read_Val_Point2 | Read_Val_Point3
 369                 @ID: __sequential.getKeyTag(key)
 370                 @Action:
 371                         @DefineVar: TypeV *_Old_Val = __sequential.map.get(key)
 372                 @Post_check:
 373                         __sequential.equals_val(_Old_Val, __RET__)
 374                 @End
 375         */
 376         shared_ptr<TypeV> get(TypeK& key) {
 377                 void *key_ptr = (void*) new TypeK(key);
 378                 slot *key_slot = new slot(false, shared_ptr<void>(key_ptr));
 379                 int fullhash = hash(key_slot);
 380                 slot *V = get_impl(this, _kvs, key_slot, fullhash);
 381                 if (V == NULL) return NULL;
 382                 assert (!is_prime(V));
 383                 return static_pointer_cast<TypeV>(V->_ptr);
 384         }
 385
 386         /**
 387                 @Begin
 388                 @Interface: Put
 389                 @Commit_point_set: Write_Val_Point
 390                 @ID: __sequential.getKeyTag(key)
 391                 @Action:
 392                         # Remember this old value at checking point
 393                         @DefineVar: TypeV *_Old_Val = __sequential.map.get(key)
 394                         @Code: __sequential.map.put(key, &val);
 395                 @Post_check:
 396                         __sequential.equals_val(__RET__, _Old_Val)
 397                 @End
 398         */
 399         shared_ptr<TypeV> put(TypeK& key, TypeV& val) {
 400                 return putIfMatch(key, val, NO_MATCH_OLD);
 401         }
 402
 403         /**
 404                 @Begin
 405                 @Interface: PutIfAbsent
 406                 @Commit_point_set:
 407                         Write_Val_Point | PutIfAbsent_Fail_Point
 408                 @Condition: __sequential.map.get(key) == NULL
 409                 @HB_condition:
 410                         COND_PutIfAbsentSucc :: __RET__ == NULL
 411                 @ID: __sequential.getKeyTag(key)
 412                 @Action:
 413                         @DefineVar: TypeV *_Old_Val = __sequential.map.get(key)
 414                         @Code:
 415                         if (__COND_SAY__)
 416                                 __sequential.map.put(key, &value);
 417                 @Post_check:
 418                         __COND_SAY__ ? __RET__ == NULL : __sequential.equals_val(_Old_Val, __RET__)
 419                 @End
 420         */
 421         shared_ptr<TypeV> putIfAbsent(TypeK& key, TypeV& value) {
 422                 return putIfMatch(key, val, TOMBSTONE);
 423         }
 424
 425         /**
 426                 @Begin
 427                 @Interface: RemoveAny
 428                 @Commit_point_set: Write_Val_Point
 429                 @ID: __sequential.getKeyTag(key)
 430                 @Action:
 431                         @DefineVar: TypeV *_Old_Val = __sequential.map.get(key)
 432                         @Code: __sequential.map.put(key, NULL);
 433                 @Post_check:
 434                         __sequential.equals_val(__RET__, _Old_Val)
 435                 @End
 436         */
 437         shared_ptr<TypeV> remove(TypeK& key) {
 438                 return putIfMatch(key, TOMBSTONE, NO_MATCH_OLD);
 439         }
 440
 441         /**
 442                 @Begin
 443                 @Interface: RemoveIfMatch
 444                 @Commit_point_set:
 445                         Write_Val_Point | RemoveIfMatch_Fail_Point
 446                 @Condition:
 447                         __sequential.equals_val(__sequential.map.get(key), &val)
 448                 @HB_condition:
 449                         COND_RemoveIfMatchSucc :: __RET__ == true
 450                 @ID: __sequential.getKeyTag(key)
 451                 @Action:
 452                         @Code:
 453                         if (__COND_SAY__)
 454                                 __sequential.map.put(key, NULL);
 455                 @Post_check:
 456                         __COND_SAY__ ? __RET__ : !__RET__
 457                 @End
 458         */
 459         bool remove(TypeK& key, TypeV& val) {
 460                 slot *val_slot = val == NULL ? NULL : new slot(false, val);
 461                 return putIfMatch(key, TOMBSTONE, val) == val;
 462
 463         }
 464
 465         /**
 466                 @Begin
 467                 @Interface: ReplaceAny
 468                 @Commit_point_set:
 469                         Write_Val_Point
 470                 @ID: __sequential.getKeyTag(key)
 471                 @Action:
 472                         @DefineVar: TypeV *_Old_Val = __sequential.map.get(key)
 473                 @Post_check:
 474                         __sequential.equals_val(__RET__, _Old_Val)
 475                 @End
 476         */
 477         shared_ptr<TypeV> replace(TypeK& key, TypeV& val) {
 478                 return putIfMatch(key, val, MATCH_ANY);
 479         }
 480
 481         /**
 482                 @Begin
 483                 @Interface: ReplaceIfMatch
 484                 @Commit_point_set:
 485                         Write_Val_Point | ReplaceIfMatch_Fail_Point
 486                 @Condition:
 487                         __sequential.equals_val(__sequential.map.get(key), &oldval)
 488                 @HB_condition:
 489                         COND_ReplaceIfMatchSucc :: __RET__ == true
 490                 @ID: __sequential.getKeyTag(key)
 491                 @Action:
 492                         @Code:
 493                         if (__COND_SAY__)
 494                                 __sequential.map.put(key, &newval);
 495                 @Post_check:
 496                         __COND_SAY__ ? __RET__ : !__RET__
 497                 @End
 498         */
 499         bool replace(TypeK& key, TypeV& oldval, TypeV& newval) {
 500                 return putIfMatch(key, newval, oldval) == oldval;
 501         }
 502
 503         private:
 504         static CHM* get_chm(kvs_data* kvs) {
 505                 return (CHM*) kvs->_data[0].load(memory_order_relaxed);
 506         }
 507
 508         static int* get_hashes(kvs_data *kvs) {
 509                 return (int *) kvs->_data[1].load(memory_order_relaxed);
 510         }
 511
 512         // Preserve happens-before semantics on newly inserted keys
 513         static inline slot* key(kvs_data *kvs, int idx) {
 514                 assert (idx >= 0 && idx < kvs->_size);
 515                 // Corresponding to the volatile read in get_impl() and putIfMatch in
 516                 // Cliff Click's Java implementation
 517                 return (slot*) kvs->_data[idx * 2 + 2].load(memory_order_acquire);
 518         }
 519
 520         /**
 521                 The atomic operation in val() function is a "potential" commit point,
 522                 which means in some case it is a real commit point while it is not for
 523                 some other cases. This so happens because the val() function is such a
 524                 fundamental function that many internal operation will call. Our
 525                 strategy is that we label any potential commit points and check if they
 526                 really are the commit points later.
 527         */
 528         // Preserve happens-before semantics on newly inserted values
 529         static inline slot* val(kvs_data *kvs, int idx) {
 530                 assert (idx >= 0 && idx < kvs->_size);
 531                 // Corresponding to the volatile read in get_impl() and putIfMatch in
 532                 // Cliff Click's Java implementation
 533                 slot *res = (slot*) kvs->_data[idx * 2 + 3].load(memory_order_acquire);
 534                 /**
 535                         @Begin
 536                         # This is a complicated potential commit point since many many functions are
 537                         # calling val().
 538                         @Potential_commit_point_define: true
 539                         @Label: Read_Val_Point
 540                         @End
 541                 */
 542                 return res;
 543
 544
 545         }
 546
 547         static int hash(slot *key_slot) {
 548                 assert(key_slot != NULL && key_slot->_ptr != NULL);
 549                 shared_ptr<TypeK> key = static_pointer_cast<TypeK>(key_slot->_ptr);
 550                 int h = key->hashCode();
 551                 // Spread bits according to Cliff Click's code
 552                 h += (h << 15) ^ 0xffffcd7d;
 553                 h ^= (h >> 10);
 554                 h += (h << 3);
 555                 h ^= (h >> 6);
 556                 h += (h << 2) + (h << 14);
 557                 return h ^ (h >> 16);
 558         }
 559
 560         // Heuristic to decide if reprobed too many times.
 561         // Be careful here: Running over the limit on a 'get' acts as a 'miss'; on a
 562         // put it triggers a table resize. Several places MUST have exact agreement.
 563         static int reprobe_limit(int len) {
 564                 return REPROBE_LIMIT + (len >> 2);
 565         }
 566
 567         static inline bool is_prime(slot *val) {
 568                 return (val != NULL) && val->_prime;
 569         }
 570
 571         // Check for key equality. Try direct pointer comparison first (fast
 572         // negative teset) and then the full 'equals' call
 573         static bool keyeq(slot *K, slot *key_slot, int *hashes, int hash,
 574                 int fullhash) {
 575                 // Caller should've checked this.
 576                 assert (K != NULL);
 577                 shared_ptr<TypeK> key_ptr = static_pointer_cast<TypeK>(key_slot->_ptr);
 578                 return
 579                         K == key_slot ||
 580                                 ((hashes[hash] == 0 || hashes[hash] == fullhash) &&
 581                                 K != TOMBSTONE &&
 582                                 key_ptr->equals(K->_ptr));
 583         }
 584
 585         static bool valeq(slot *val_slot1, slot *val_slot2) {
 586                 assert (val_slot1 != NULL);
 587                 shared_ptr<TypeK> ptr1 = static_pointer_cast<TypeV>(val_slot1->_ptr);
 588                 if (val_slot2 == NULL || ptr1 == NULL) return false;
 589                 return ptr1->equals(val_slot2->_ptr);
 590         }
 591
 592         // Together with key() preserve the happens-before relationship on newly
 593         // inserted keys
 594         static inline bool CAS_key(kvs_data *kvs, int idx, void *expected, void *desired) {
 595                 return kvs->_data[2 * idx + 2].compare_exchange_strong(expected,
 596                         desired, memory_order_release, memory_order_release);
 597         }
 598
 599         /**
 600                 Same as the val() function, we only label the CAS operation as the
 601                 potential commit point.
 602         */
 603         // Together with val() preserve the happens-before relationship on newly
 604         // inserted values
 605         static inline bool CAS_val(kvs_data *kvs, int idx, void *expected, void
 606                 *desired) {
 607                 bool res =  kvs->_data[2 * idx + 3].compare_exchange_strong(expected,
 608                         desired, memory_order_release, memory_order_release);
 609                 /**
 610                         # If it is a successful put instead of a copy or any other internal
 611                         # operantions, expected != NULL
 612                         @Begin
 613                         @Potential_commit_point_define: __ATOMIC_RET__ == true
 614                         @Label: Write_Val_Point
 615                         @End
 616                 */
 617                 return res;
 618         }
 619
 620         slot* get_impl(cliffc_hashtable *topmap, kvs_data *kvs, slot* key_slot, int
 621                 fullhash) {
 622                 int len = kvs->_size;
 623                 CHM *chm = get_chm(kvs);
 624                 int *hashes = get_hashes(kvs);
 625
 626                 int idx = fullhash & (len - 1);
 627                 int reprobe_cnt = 0;
 628                 while (true) {
 629                         slot *K = key(kvs, idx);
 630                         slot *V = val(kvs, idx);
 631                         /**
 632                                 @Begin
 633                                 @Commit_point_define: V == NULL
 634                                 @Potential_commit_point_label: Read_Val_Point
 635                                 @Label: Get_Success_Point_1
 636                                 @End
 637                         */
 638
 639                         if (V == NULL) return NULL; // A miss
 640
 641                         if (keyeq(K, key_slot, hashes, idx, fullhash)) {
 642                                 // Key hit! Check if table-resize in progress
 643                                 if (!is_prime(V)) {
 644                                         /**
 645                                                 @Begin
 646                                                 @Commit_point_define: true
 647                                                 @Potential_commit_point_label: Read_Val_Point
 648                                                 @Label: Get_Success_Point_2
 649                                                 @End
 650                                         */
 651                                         return (V == TOMBSTONE) ? NULL : V; // Return this value
 652                                 }
 653                                 // Otherwise, finish the copy & retry in the new table
 654                                 return get_impl(topmap, chm->copy_slot_and_check(topmap, kvs,
 655                                         idx, key_slot), key_slot, fullhash);
 656                         }
 657
 658                         if (++reprobe_cnt >= REPROBE_LIMIT ||
 659                                 key_slot == TOMBSTONE) {
 660                                 // Retry in new table
 661                                 // Atomic read (acquire) can be here
 662                                 kvs_data *newkvs = chm->_newkvs.load(memory_order_acquire);
 663                                 /**
 664                                         @Begin
 665                                         @Commit_point_define_check: newkvs == NULL
 666                                         @Label: Get_Success_Point_3
 667                                         @End
 668                                 */
 669                                 return newkvs == NULL ? NULL : get_impl(topmap,
 670                                         topmap->help_copy(newkvs), key_slot, fullhash);
 671                         }
 672
 673                         idx = (idx + 1) & (len - 1); // Reprobe by 1
 674                 }
 675         }
 676
 677         // A wrapper of the essential function putIfMatch()
 678         shared_ptr<TypeV> putIfMatch(TypeK& key, TypeV& value, slot *old_val) {
 679                 // TODO: Should throw an exception rather return NULL
 680                 if (old_val == NULL) {
 681                         return NULL;
 682                 }
 683                 void *key_ptr = (void*) new TypeK(key);
 684                 slot *key_slot = new slot(false, shared_ptr<void>(key_ptr));
 685
 686                 void *val_ptr = (void*) new TypeV(value);
 687                 slot *value_slot = new slot(false, shared_ptr<void>(val_ptr));
 688                 slot *res = putIfMatch(this, _kvs, key_slot, value_slot, old_val);
 689                 // Only when copy_slot() call putIfMatch() will it return NULL
 690                 assert (res != NULL);
 691                 assert (!is_prime(res));
 692                 return res == TOMBSTONE ? NULL : static_pointer_cast<TypeV>(res->_ptr);
 693         }
 694
 695         /**
 696                 Put, Remove, PutIfAbsent, etc will call this function. Return the old
 697                 value. If the returned value is equals to the expVal (or expVal is
 698                 NO_MATCH_OLD), then this function puts the val_slot to the table 'kvs'.
 699                 Only copy_slot will pass a NULL expVal, and putIfMatch only returns a
 700                 NULL if passed a NULL expVal.
 701         */
 702         static slot* putIfMatch(cliffc_hashtable *topmap, kvs_data *kvs, slot
 703                 *key_slot, slot *val_slot, slot *expVal) {
 704                 assert (val_slot != NULL);
 705                 assert (!is_prime(val_slot));
 706                 assert (!is_prime(expVal));
 707
 708                 int fullhash = hash(key_slot);
 709                 int len = kvs->_size;
 710                 CHM *chm = get_chm(kvs);
 711                 int *hashes = get_hashes(kvs);
 712                 int idx = fullhash & (len - 1);
 713
 714                 // Claim a key slot
 715                 int reprobe_cnt = 0;
 716                 slot *K;
 717                 slot *V;
 718                 kvs_data *newkvs;
 719
 720                 while (true) { // Spin till we get a key slot
 721                         K = key(kvs, idx);
 722                         V = val(kvs, idx, NULL);
 723                         if (K == NULL) { // Get a free slot
 724                                 if (val_slot == TOMBSTONE) return val_slot;
 725                                 // Claim the null key-slot
 726                                 if (CAS_key(kvs, idx, NULL, key_slot)) {
 727                                         chm->_slots.fetch_add(1, memory_order_relaxed); // Inc key-slots-used count
 728                                         hashes[idx] = fullhash; // Memorize full hash
 729                                         break;
 730                                 }
 731                                 K = key(kvs, idx); // CAS failed, get updated value
 732                                 assert (K != NULL);
 733                         }
 734
 735                         // Key slot not null, there exists a Key here
 736                         if (keyeq(K, key_slot, hashes, idx, fullhash))
 737                                 break; // Got it
 738
 739                         // Notice that the logic here should be consistent with that of get.
 740                         // The first predicate means too many reprobes means nothing in the
 741                         // old table.
 742                         if (++reprobe_cnt >= reprobe_limit(len) ||
 743                                 K == TOMBSTONE) { // Found a Tombstone key, no more keys
 744                                 newkvs = chm->resize(topmap, kvs);
 745                                 // Help along an existing copy
 746                                 if (expVal != NULL) topmap->help_copy(newkvs);
 747                                 return putIfMatch(topmap, newkvs, key_slot, val_slot, expVal);
 748                         }
 749
 750                         idx = (idx + 1) & (len - 1); // Reprobe
 751                 } // End of spinning till we get a Key slot
 752
 753                 if (val_slot == V) return V; // Fast cutout for no-change
 754
 755                 // Here it tries to resize cause it doesn't want other threads to stop
 756                 // its progress (eagerly try to resize soon)
 757                 newkvs = chm->_newkvs.load(memory_order_acquire);
 758                 if (newkvs == NULL &&
 759                         ((V == NULL && chm->table_full(reprobe_cnt, len)) || is_prime(V)))
 760                         newkvs = chm->resize(topmap, kvs); // Force the copy to start
 761
 762                 // Finish the copy and then put it in the new table
 763                 if (newkvs != NULL)
 764                         return putIfMatch(topmap, chm->copy_slot_and_check(topmap, kvs, idx,
 765                                 expVal), key_slot, val_slot, expVal);
 766
 767                 // Decided to update the existing table
 768                 while (true) {
 769                         assert (!is_prime(V));
 770
 771                         if (expVal != NO_MATCH_OLD &&
 772                                 V != expVal &&
 773                                 (expVal != MATCH_ANY || V == TOMBSTONE || V == NULL) &&
 774                                 !(V == NULL && expVal == TOMBSTONE) &&
 775                                 (expVal == NULL || !valeq(expVal, V))) {
 776                                 /**
 777                                         @Begin
 778                                         @Commit_point_define: expVal == TOMBSTONE
 779                                         @Potential_commit_point_label: Read_Val_Point
 780                                         @Label: PutIfAbsent_Fail_Point
 781                                                 # This is a check for the PutIfAbsent() when the value
 782                                                 # is not absent
 783                                         @End
 784                                 */
 785                                 /**
 786                                         @Begin
 787                                         @Commit_point_define: expVal != NULL && val_slot == TOMBSTONE
 788                                         @Potential_commit_point_label: Read_Val_Point
 789                                         @Label: RemoveIfMatch_Fail_Point
 790                                         @End
 791                                 */
 792                                 /**
 793                                         @Begin
 794                                         @Commit_point_define: !valeq(expVal, V)
 795                                         @Potential_commit_point_label: Read_Val_Point
 796                                         @Label: ReplaceIfMatch_Fail_Point
 797                                         @End
 798                                 */
 799                                 return V; // Do not update!
 800                         }
 801
 802                         if (CAS_val(kvs, idx, V, val_slot)) {
 803                                 /**
 804                                         @Begin
 805                                         # The only point where a successful put happens
 806                                         @Commit_point_define: true
 807                                         @Potential_commit_point_label: Write_Val_Point
 808                                         @Label: Write_Success_Point
 809                                         @End
 810                                 */
 811                                 if (expVal != NULL) { // Not called by a table-copy
 812                                         // CAS succeeded, should adjust size
 813                                         // Both normal put's and table-copy calls putIfMatch, but
 814                                         // table-copy does not increase the number of live K/V pairs
 815                                         if ((V == NULL || V == TOMBSTONE) &&
 816                                                 val_slot != TOMBSTONE)
 817                                                 chm->_size.fetch_add(1, memory_order_relaxed);
 818                                         if (!(V == NULL || V == TOMBSTONE) &&
 819                                                 val_slot == TOMBSTONE)
 820                                                 chm->_size.fetch_add(-1, memory_order_relaxed);
 821                                 }
 822                                 return (V == NULL && expVal != NULL) ? TOMBSTONE : V;
 823                         }
 824                         // Else CAS failed
 825                         V = val(kvs, idx, NULL);
 826                         if (is_prime(V))
 827                                 return putIfMatch(topmap, chm->copy_slot_and_check(topmap, kvs,
 828                                         idx, expVal), key_slot, val_slot, expVal);
 829                 }
 830         }
 831
 832         // Help along an existing table-resize. This is a fast cut-out wrapper.
 833         kvs_data* help_copy(kvs_data *helper) {
 834                 kvs_data *topkvs = _kvs.load(memory_order_acquire);
 835                 CHM *topchm = get_chm(topkvs);
 836                 // No cpy in progress
 837                 if (topchm->_newkvs.load(memory_order_acquire) == NULL) return helper;
 838                 topchm->help_copy_impl(this, topkvs, false);
 839                 return helper;
 840         }
 841 };
 842
 843 #endif