junction/ConcurrentMap_LeapFrog.h

   1 /*------------------------------------------------------------------------
   2   Junction: Concurrent data structures in C++
   3   Copyright (c) 2016 Jeff Preshing
   4
   5   Distributed under the Simplified BSD License.
   6   Original location: https://github.com/preshing/junction
   7
   8   This software is distributed WITHOUT ANY WARRANTY; without even the
   9   implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  10   See the LICENSE file for more information.
  11 ------------------------------------------------------------------------*/
  12
  13 #ifndef JUNCTION_CONCURRENTMAP_LEAPFROG_H
  14 #define JUNCTION_CONCURRENTMAP_LEAPFROG_H
  15
  16 #include <junction/Core.h>
  17 #include <junction/details/LeapFrog.h>
  18 #include <junction/QSBR.h>
  19 #include <turf/Heap.h>
  20 #include <turf/Trace.h>
  21
  22 namespace junction {
  23
  24 TURF_TRACE_DECLARE(ConcurrentMap_LeapFrog, 17)
  25
  26 template <typename K, typename V, class KT = DefaultKeyTraits<K>, class VT = DefaultValueTraits<V>>
  27 class ConcurrentMap_LeapFrog {
  28 public:
  29     typedef K Key;
  30     typedef V Value;
  31     typedef KT KeyTraits;
  32     typedef VT ValueTraits;
  33     typedef typename turf::util::BestFit<Key>::Unsigned Hash;
  34     typedef details::LeapFrog<ConcurrentMap_LeapFrog> Details;
  35
  36 private:
  37     turf::Atomic<typename Details::Table*> m_root;
  38
  39 public:
  40     ConcurrentMap_LeapFrog(ureg capacity) : m_root(Details::Table::create(capacity)) {
  41     }
  42
  43     ~ConcurrentMap_LeapFrog() {
  44         typename Details::Table* table = m_root.loadNonatomic();
  45         table->destroy();
  46     }
  47
  48     // publishTableMigration() is called by exactly one thread from Details::TableMigration::run()
  49     // after all the threads participating in the migration have completed their work.
  50     void publishTableMigration(typename Details::TableMigration* migration) {
  51         // There are no racing calls to this function.
  52         typename Details::Table* oldRoot = m_root.loadNonatomic();
  53         m_root.store(migration->m_destination, turf::Release);
  54         TURF_ASSERT(oldRoot == migration->getSources()[0].table);
  55         // Caller will GC the TableMigration and the source table.
  56     }
  57
  58     // A Mutator represents a known cell in the hash table.
  59     // It's meant for manipulations within a temporary function scope.
  60     // Obviously you must not call QSBR::Update while holding a Mutator.
  61     // Any operation that modifies the table (exchangeValue, eraseValue)
  62     // may be forced to follow a redirected cell, which changes the Mutator itself.
  63     // Note that even if the Mutator was constructed from an existing cell,
  64     // exchangeValue() can still trigger a resize if the existing cell was previously marked deleted,
  65     // or if another thread deletes the key between the two steps.
  66     class Mutator {
  67     private:
  68         friend class ConcurrentMap_LeapFrog;
  69
  70         ConcurrentMap_LeapFrog& m_map;
  71         typename Details::Table* m_table;
  72         typename Details::Cell* m_cell;
  73         Value m_value;
  74
  75         // Constructor: Find existing cell
  76         Mutator(ConcurrentMap_LeapFrog& map, Key key, bool) : m_map(map), m_value(Value(ValueTraits::NullValue)) {
  77             TURF_TRACE(ConcurrentMap_LeapFrog, 0, "[Mutator] find constructor called", uptr(m_table), uptr(key));
  78             Hash hash = KeyTraits::hash(key);
  79             for (;;) {
  80                 m_table = m_map.m_root.load(turf::Consume);
  81                 m_cell = Details::find(hash, m_table);
  82                 if (!m_cell)
  83                     return;
  84                 m_value = m_cell->value.load(turf::Consume);
  85                 if (m_value != Value(ValueTraits::Redirect))
  86                     return;                        // Found an existing value
  87                 // We've encountered a Redirect value. Help finish the migration.
  88                 TURF_TRACE(ConcurrentMap_LeapFrog, 1, "[Mutator] find was redirected", uptr(m_table), 0);
  89                 m_table->jobCoordinator.participate();
  90                 // Try again using the latest root.
  91             }
  92         }
  93
  94         // Constructor: Insert cell
  95         Mutator(ConcurrentMap_LeapFrog& map, Key key) : m_map(map), m_table(map.m_root.load(turf::Consume)), m_value(Value(ValueTraits::NullValue)) {
  96             TURF_TRACE(ConcurrentMap_LeapFrog, 2, "[Mutator] insert constructor called", uptr(m_table), uptr(key));
  97             Hash hash = KeyTraits::hash(key);
  98             for (;;) {
  99                 m_table = m_map.m_root.load(turf::Consume);
 100                 ureg overflowIdx;
 101                 switch (Details::insert(hash, m_table, m_cell, overflowIdx)) {   // Modifies m_cell
 102                     case Details::InsertResult_InsertedNew: {
 103                         // We've inserted a new cell. Don't load m_cell->value.
 104                         return;
 105                     }
 106                     case Details::InsertResult_AlreadyFound: {
 107                         // The hash was already found in the table.
 108                         m_value = m_cell->value.load(turf::Consume);
 109                         if (m_value == Value(ValueTraits::Redirect)) {
 110                             // We've encountered a Redirect value.
 111                             TURF_TRACE(ConcurrentMap_LeapFrog, 3, "[Mutator] insert was redirected", uptr(m_table), uptr(m_value));
 112                             break;   // Help finish the migration.
 113                         }
 114                         return;     // Found an existing value
 115                     }
 116                     case Details::InsertResult_Overflow: {
 117                         Details::beginTableMigration(m_map, m_table, overflowIdx);
 118                         break;
 119                     }
 120                 }
 121                 // A migration has been started (either by us, or another thread). Participate until it's complete.
 122                 m_table->jobCoordinator.participate();
 123                 // Try again using the latest root.
 124             }
 125         }
 126
 127     public:
 128         Value getValue() const {
 129             // Return previously loaded value. Don't load it again.
 130             return Value(m_value);
 131         }
 132
 133         Value exchangeValue(Value desired) {
 134             TURF_ASSERT(desired != Value(ValueTraits::NullValue));
 135             TURF_ASSERT(m_cell);    // Cell must have been found or inserted
 136             TURF_TRACE(ConcurrentMap_LeapFrog, 4, "[Mutator::exchangeValue] called", uptr(m_table), uptr(m_value));
 137             for (;;) {
 138                 Value oldValue = m_value;
 139                 if (m_cell->value.compareExchangeStrong(m_value, desired, turf::ConsumeRelease)) {
 140                     // Exchange was successful. Return previous value.
 141                     TURF_TRACE(ConcurrentMap_LeapFrog, 5, "[Mutator::exchangeValue] exchanged Value", uptr(m_value), uptr(desired));
 142                     Value result = m_value;
 143                     m_value = desired;  // Leave the mutator in a valid state
 144                     return result;
 145                 }
 146                 // The CAS failed and m_value has been updated with the latest value.
 147                 if (m_value != Value(ValueTraits::Redirect)) {
 148                     TURF_TRACE(ConcurrentMap_LeapFrog, 6, "[Mutator::exchangeValue] detected race to write value", uptr(m_table), uptr(m_value));
 149                     if (oldValue == Value(ValueTraits::NullValue) && m_value != Value(ValueTraits::NullValue)) {
 150                         TURF_TRACE(ConcurrentMap_LeapFrog, 7, "[Mutator::exchangeValue] racing write inserted new value", uptr(m_table), uptr(m_value));
 151                     }
 152                     // There was a racing write (or erase) to this cell.
 153                     // Pretend we exchanged with ourselves, and just let the racing write win.
 154                     return desired;
 155                 }
 156                 // We've encountered a Redirect value. Help finish the migration.
 157                 TURF_TRACE(ConcurrentMap_LeapFrog, 8, "[Mutator::exchangeValue] was redirected", uptr(m_table), uptr(m_value));
 158                 Hash hash = m_cell->hash.load(turf::Relaxed);
 159                 for (;;) {
 160                     // Help complete the migration.
 161                     m_table->jobCoordinator.participate();
 162                     // Try again in the new table.
 163                     m_table = m_map.m_root.load(turf::Consume);
 164                     m_value = Value(ValueTraits::NullValue);
 165                     ureg overflowIdx;
 166                     switch (Details::insert(hash, m_table, m_cell, overflowIdx)) {   // Modifies m_cell
 167                     case Details::InsertResult_AlreadyFound:
 168                         m_value = m_cell->value.load(turf::Consume);
 169                         if (m_value == Value(ValueTraits::Redirect)) {
 170                             TURF_TRACE(ConcurrentMap_LeapFrog, 9, "[Mutator::exchangeValue] was re-redirected", uptr(m_table), uptr(m_value));
 171                             break;
 172                         }
 173                         goto breakOuter;
 174                     case Details::InsertResult_InsertedNew:
 175                         goto breakOuter;
 176                     case Details::InsertResult_Overflow:
 177                         TURF_TRACE(ConcurrentMap_LeapFrog, 10, "[Mutator::exchangeValue] overflow after redirect", uptr(m_table), overflowIdx);
 178                         Details::beginTableMigration(m_map, m_table, overflowIdx);
 179                         break;
 180                     }
 181                     // We were redirected... again
 182                 }
 183             breakOuter:
 184                 ;
 185                 // Try again in the new table.
 186             }
 187         }
 188
 189         void setValue(Value desired) {
 190             exchangeValue(desired);
 191         }
 192
 193         Value eraseValue() {
 194             TURF_ASSERT(m_cell);    // Cell must have been found or inserted
 195             TURF_TRACE(ConcurrentMap_LeapFrog, 11, "[Mutator::eraseValue] called", uptr(m_table), uptr(m_cell));
 196             for (;;) {
 197                 if (m_value == Value(ValueTraits::NullValue))
 198                     return Value(m_value);
 199                 TURF_ASSERT(m_cell);    // m_value is non-NullValue, therefore cell must have been found or inserted.
 200                 if (m_cell->value.compareExchangeStrong(m_value, Value(ValueTraits::NullValue), turf::Consume)) {
 201                     // Exchange was successful and a non-NULL value was erased and returned by reference in m_value.
 202                     TURF_ASSERT(m_value != ValueTraits::NullValue);   // Implied by the test at the start of the loop.
 203                     Value result = m_value;
 204                     m_value = Value(ValueTraits::NullValue);   // Leave the mutator in a valid state
 205                     return result;
 206                 }
 207                 // The CAS failed and m_value has been updated with the latest value.
 208                 TURF_TRACE(ConcurrentMap_LeapFrog, 12, "[Mutator::eraseValue] detected race to write value", uptr(m_table), uptr(m_cell));
 209                 if (m_value != Value(ValueTraits::Redirect)) {
 210                     // There was a racing write (or erase) to this cell.
 211                     // Pretend we erased nothing, and just let the racing write win.
 212                     return Value(ValueTraits::NullValue);
 213                 }
 214                 // We've been redirected to a new table.
 215                 TURF_TRACE(ConcurrentMap_LeapFrog, 13, "[Mutator::eraseValue] was redirected", uptr(m_table), uptr(m_cell));
 216                 Hash hash = m_cell->hash.load(turf::Relaxed);           // Re-fetch hash
 217                 for (;;) {
 218                     // Help complete the migration.
 219                     m_table->jobCoordinator.participate();
 220                     // Try again in the new table.
 221                     m_table = m_map.m_root.load(turf::Consume);
 222                     m_cell = Details::find(hash, m_table);
 223                     if (!m_cell) {
 224                         m_value = Value(ValueTraits::NullValue);
 225                         return m_value;
 226                     }
 227                     m_value = m_cell->value.load(turf::Relaxed);
 228                     if (m_value != Value(ValueTraits::Redirect))
 229                         break;
 230                     TURF_TRACE(ConcurrentMap_LeapFrog, 14, "[Mutator::eraseValue] was re-redirected", uptr(m_table), uptr(m_cell));
 231                 }
 232             }
 233         }
 234     };
 235
 236     Mutator insert(Key key) {
 237         return Mutator(*this, key);
 238     }
 239
 240     Mutator find(Key key) {
 241         return Mutator(*this, key, false);
 242     }
 243
 244     // Lookup without creating a temporary Mutator.
 245     Value get(Key key) {
 246         Hash hash = KeyTraits::hash(key);
 247         TURF_TRACE(ConcurrentMap_LeapFrog, 15, "[get] called", uptr(hash), 0);
 248         for (;;) {
 249             typename Details::Table* table = m_root.load(turf::Consume);
 250             typename Details::Cell* cell = Details::find(hash, table);
 251             if (!cell)
 252                 return Value(ValueTraits::NullValue);
 253             Value value = cell->value.load(turf::Consume);
 254             if (value != Value(ValueTraits::Redirect))
 255                 return value;                        // Found an existing value
 256             // We've been redirected to a new table. Help with the migration.
 257             TURF_TRACE(ConcurrentMap_LeapFrog, 16, "[get] was redirected", uptr(table), uptr(hash));
 258             table->jobCoordinator.participate();
 259             // Try again in the new table.
 260         }
 261     }
 262
 263     Value insert(Key key, Value desired) {
 264         Mutator iter(*this, key);
 265         return iter.exchangeValue(desired);
 266     }
 267
 268     Value exchange(Key key, Value desired) {
 269         Mutator iter(*this, key);
 270         return iter.exchangeValue(desired);
 271     }
 272
 273     Value erase(Key key) {
 274         Mutator iter(*this, key, false);
 275         return iter.eraseValue();
 276     }
 277
 278     // The easiest way to implement an Iterator is to prevent all Redirects.
 279     // The currrent Iterator does that by forbidding concurrent inserts.
 280     // To make it work with concurrent inserts, we'd need a way to block TableMigrations.
 281     class Iterator {
 282     private:
 283         typename Details::Table* m_table;
 284         ureg m_idx;
 285         Key m_hash;
 286         Value m_value;
 287
 288     public:
 289         Iterator(ConcurrentMap_LeapFrog& map) {
 290             // Since we've forbidden concurrent inserts (for now), nonatomic would suffice here, but let's plan ahead:
 291             m_table = map.m_root.load(turf::Consume);
 292             m_idx = -1;
 293             next();
 294         }
 295
 296         void next() {
 297             TURF_ASSERT(m_table);
 298             TURF_ASSERT(isValid() || m_idx == -1);  // Either the Iterator is already valid, or we've just started iterating.
 299             while (++m_idx <= m_table->sizeMask) {
 300                 // Index still inside range of table.
 301                 typename Details::CellGroup* group = m_table->getCellGroups() + (m_idx >> 2);
 302                 typename Details::Cell *cell = group->cells + (m_idx & 3);
 303                 m_hash = cell->hash.load(turf::Relaxed);
 304                 if (m_hash != KeyTraits::NullHash) {
 305                     // Cell has been reserved.
 306                     m_value = cell->value.load(turf::Relaxed);
 307                     TURF_ASSERT(m_value != Value(ValueTraits::Redirect));
 308                     if (m_value != Value(ValueTraits::NullValue))
 309                         return;     // Yield this cell.
 310                 }
 311             }
 312             // That's the end of the map.
 313             m_hash = KeyTraits::NullHash;
 314             m_value = ValueTraits::NullValue;
 315         }
 316
 317         bool isValid() const {
 318             return m_value != Value(ValueTraits::NullValue);
 319         }
 320
 321         Key getKey() const {
 322             TURF_ASSERT(isValid());
 323             // Since we've forbidden concurrent inserts (for now), nonatomic would suffice here, but let's plan ahead:
 324             return KeyTraits::dehash(m_hash);
 325         }
 326
 327         Value getValue() const {
 328             TURF_ASSERT(isValid());
 329             return m_value;
 330         }
 331     };
 332 };
 333
 334 } // namespace junction
 335
 336 #endif // JUNCTION_CONCURRENTMAP_LEAPFROG_H