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 = Details::InitialSize) : 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(0), 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                 Value value = m_cell->value.load(turf::Consume);
  85                 if (value != Value(ValueTraits::Redirect)) {
  86                     // Found an existing value
  87                     m_value = value;
  88                     return;
  89                 }
  90                 // We've encountered a Redirect value. Help finish the migration.
  91                 TURF_TRACE(ConcurrentMap_Leapfrog, 1, "[Mutator] find was redirected", uptr(m_table), 0);
  92                 m_table->jobCoordinator.participate();
  93                 // Try again using the latest root.
  94             }
  95         }
  96
  97         // Constructor: Insert or find cell
  98         Mutator(ConcurrentMap_Leapfrog& map, Key key) : m_map(map), m_value(Value(ValueTraits::NullValue)) {
  99             TURF_TRACE(ConcurrentMap_Leapfrog, 2, "[Mutator] insertOrFind constructor called", uptr(0), uptr(key));
 100             Hash hash = KeyTraits::hash(key);
 101             for (;;) {
 102                 m_table = m_map.m_root.load(turf::Consume);
 103                 ureg overflowIdx;
 104                 switch (Details::insertOrFind(hash, m_table, m_cell, overflowIdx)) { // Modifies m_cell
 105                 case Details::InsertResult_InsertedNew: {
 106                     // We've inserted a new cell. Don't load m_cell->value.
 107                     return;
 108                 }
 109                 case Details::InsertResult_AlreadyFound: {
 110                     // The hash was already found in the table.
 111                     Value value = m_cell->value.load(turf::Consume);
 112                     if (value == Value(ValueTraits::Redirect)) {
 113                         // We've encountered a Redirect value.
 114                         TURF_TRACE(ConcurrentMap_Leapfrog, 3, "[Mutator] insertOrFind was redirected", uptr(m_table), uptr(m_value));
 115                         break; // Help finish the migration.
 116                     }
 117                     // Found an existing value
 118                     m_value = value;
 119                     return;
 120                 }
 121                 case Details::InsertResult_Overflow: {
 122                     // Unlike ConcurrentMap_Linear, we don't need to keep track of & pass a "mustDouble" flag.
 123                     // Passing overflowIdx is sufficient to prevent an infinite loop here.
 124                     // It defines the start of the range of cells to check while estimating total cells in use.
 125                     // After the first migration, deleted keys are purged, so if we hit this line during the
 126                     // second loop iteration, every cell in the range will be in use, thus the estimate will be 100%.
 127                     // (Concurrent deletes could result in further iterations, but it will eventually settle.)
 128                     Details::beginTableMigration(m_map, m_table, overflowIdx);
 129                     break;
 130                 }
 131                 }
 132                 // A migration has been started (either by us, or another thread). Participate until it's complete.
 133                 m_table->jobCoordinator.participate();
 134                 // Try again using the latest root.
 135             }
 136         }
 137
 138     public:
 139         Value getValue() const {
 140             // Return previously loaded value. Don't load it again.
 141             return Value(m_value);
 142         }
 143
 144         Value exchangeValue(Value desired) {
 145             TURF_ASSERT(desired != Value(ValueTraits::NullValue));
 146             TURF_ASSERT(desired != Value(ValueTraits::Redirect));
 147             TURF_ASSERT(m_cell); // Cell must have been found or inserted
 148             TURF_TRACE(ConcurrentMap_Leapfrog, 4, "[Mutator::exchangeValue] called", uptr(m_table), uptr(m_value));
 149             for (;;) {
 150                 Value oldValue = m_value;
 151                 if (m_cell->value.compareExchangeStrong(m_value, desired, turf::ConsumeRelease)) {
 152                     // Exchange was successful. Return previous value.
 153                     TURF_TRACE(ConcurrentMap_Leapfrog, 5, "[Mutator::exchangeValue] exchanged Value", uptr(m_value),
 154                                uptr(desired));
 155                     Value result = m_value;
 156                     m_value = desired; // Leave the mutator in a valid state
 157                     return result;
 158                 }
 159                 // The CAS failed and m_value has been updated with the latest value.
 160                 if (m_value != Value(ValueTraits::Redirect)) {
 161                     TURF_TRACE(ConcurrentMap_Leapfrog, 6, "[Mutator::exchangeValue] detected race to write value", uptr(m_table),
 162                                uptr(m_value));
 163                     if (oldValue == Value(ValueTraits::NullValue) && m_value != Value(ValueTraits::NullValue)) {
 164                         TURF_TRACE(ConcurrentMap_Leapfrog, 7, "[Mutator::exchangeValue] racing write inserted new value",
 165                                    uptr(m_table), uptr(m_value));
 166                     }
 167                     // There was a racing write (or erase) to this cell.
 168                     // Pretend we exchanged with ourselves, and just let the racing write win.
 169                     return desired;
 170                 }
 171                 // We've encountered a Redirect value. Help finish the migration.
 172                 TURF_TRACE(ConcurrentMap_Leapfrog, 8, "[Mutator::exchangeValue] was redirected", uptr(m_table), uptr(m_value));
 173                 Hash hash = m_cell->hash.load(turf::Relaxed);
 174                 for (;;) {
 175                     // Help complete the migration.
 176                     m_table->jobCoordinator.participate();
 177                     // Try again in the new table.
 178                     m_table = m_map.m_root.load(turf::Consume);
 179                     m_value = Value(ValueTraits::NullValue);
 180                     ureg overflowIdx;
 181                     switch (Details::insertOrFind(hash, m_table, m_cell, overflowIdx)) { // Modifies m_cell
 182                     case Details::InsertResult_AlreadyFound:
 183                         m_value = m_cell->value.load(turf::Consume);
 184                         if (m_value == Value(ValueTraits::Redirect)) {
 185                             TURF_TRACE(ConcurrentMap_Leapfrog, 9, "[Mutator::exchangeValue] was re-redirected", uptr(m_table),
 186                                        uptr(m_value));
 187                             break;
 188                         }
 189                         goto breakOuter;
 190                     case Details::InsertResult_InsertedNew:
 191                         goto breakOuter;
 192                     case Details::InsertResult_Overflow:
 193                         TURF_TRACE(ConcurrentMap_Leapfrog, 10, "[Mutator::exchangeValue] overflow after redirect", uptr(m_table),
 194                                    overflowIdx);
 195                         Details::beginTableMigration(m_map, m_table, overflowIdx);
 196                         break;
 197                     }
 198                     // We were redirected... again
 199                 }
 200             breakOuter:;
 201                 // Try again in the new table.
 202             }
 203         }
 204
 205         void assignValue(Value desired) {
 206             exchangeValue(desired);
 207         }
 208
 209         Value eraseValue() {
 210             TURF_ASSERT(m_cell); // Cell must have been found or inserted
 211             TURF_TRACE(ConcurrentMap_Leapfrog, 11, "[Mutator::eraseValue] called", uptr(m_table), uptr(m_cell));
 212             for (;;) {
 213                 if (m_value == Value(ValueTraits::NullValue))
 214                     return Value(m_value);
 215                 TURF_ASSERT(m_cell); // m_value is non-NullValue, therefore cell must have been found or inserted.
 216                 if (m_cell->value.compareExchangeStrong(m_value, Value(ValueTraits::NullValue), turf::Consume)) {
 217                     // Exchange was successful and a non-NULL value was erased and returned by reference in m_value.
 218                     TURF_ASSERT(m_value != ValueTraits::NullValue); // Implied by the test at the start of the loop.
 219                     Value result = m_value;
 220                     m_value = Value(ValueTraits::NullValue); // Leave the mutator in a valid state
 221                     return result;
 222                 }
 223                 // The CAS failed and m_value has been updated with the latest value.
 224                 TURF_TRACE(ConcurrentMap_Leapfrog, 12, "[Mutator::eraseValue] detected race to write value", uptr(m_table),
 225                            uptr(m_cell));
 226                 if (m_value != Value(ValueTraits::Redirect)) {
 227                     // There was a racing write (or erase) to this cell.
 228                     // Pretend we erased nothing, and just let the racing write win.
 229                     return Value(ValueTraits::NullValue);
 230                 }
 231                 // We've been redirected to a new table.
 232                 TURF_TRACE(ConcurrentMap_Leapfrog, 13, "[Mutator::eraseValue] was redirected", uptr(m_table), uptr(m_cell));
 233                 Hash hash = m_cell->hash.load(turf::Relaxed); // Re-fetch hash
 234                 for (;;) {
 235                     // Help complete the migration.
 236                     m_table->jobCoordinator.participate();
 237                     // Try again in the new table.
 238                     m_table = m_map.m_root.load(turf::Consume);
 239                     m_cell = Details::find(hash, m_table);
 240                     if (!m_cell) {
 241                         m_value = Value(ValueTraits::NullValue);
 242                         return m_value;
 243                     }
 244                     m_value = m_cell->value.load(turf::Relaxed);
 245                     if (m_value != Value(ValueTraits::Redirect))
 246                         break;
 247                     TURF_TRACE(ConcurrentMap_Leapfrog, 14, "[Mutator::eraseValue] was re-redirected", uptr(m_table),
 248                                uptr(m_cell));
 249                 }
 250             }
 251         }
 252     };
 253
 254     Mutator insertOrFind(Key key) {
 255         return Mutator(*this, key);
 256     }
 257
 258     Mutator find(Key key) {
 259         return Mutator(*this, key, false);
 260     }
 261
 262     // Lookup without creating a temporary Mutator.
 263     Value get(Key key) {
 264         Hash hash = KeyTraits::hash(key);
 265         TURF_TRACE(ConcurrentMap_Leapfrog, 15, "[get] called", uptr(this), uptr(hash));
 266         for (;;) {
 267             typename Details::Table* table = m_root.load(turf::Consume);
 268             typename Details::Cell* cell = Details::find(hash, table);
 269             if (!cell)
 270                 return Value(ValueTraits::NullValue);
 271             Value value = cell->value.load(turf::Consume);
 272             if (value != Value(ValueTraits::Redirect))
 273                 return value; // Found an existing value
 274             // We've been redirected to a new table. Help with the migration.
 275             TURF_TRACE(ConcurrentMap_Leapfrog, 16, "[get] was redirected", uptr(table), uptr(hash));
 276             table->jobCoordinator.participate();
 277             // Try again in the new table.
 278         }
 279     }
 280
 281     Value assign(Key key, Value desired) {
 282         Mutator iter(*this, key);
 283         return iter.exchangeValue(desired);
 284     }
 285
 286     Value exchange(Key key, Value desired) {
 287         Mutator iter(*this, key);
 288         return iter.exchangeValue(desired);
 289     }
 290
 291     Value erase(Key key) {
 292         Mutator iter(*this, key, false);
 293         return iter.eraseValue();
 294     }
 295
 296     // The easiest way to implement an Iterator is to prevent all Redirects.
 297     // The currrent Iterator does that by forbidding concurrent inserts.
 298     // To make it work with concurrent inserts, we'd need a way to block TableMigrations.
 299     class Iterator {
 300     private:
 301         typename Details::Table* m_table;
 302         ureg m_idx;
 303         Key m_hash;
 304         Value m_value;
 305
 306     public:
 307         Iterator(ConcurrentMap_Leapfrog& map) {
 308             // Since we've forbidden concurrent inserts (for now), nonatomic would suffice here, but let's plan ahead:
 309             m_table = map.m_root.load(turf::Consume);
 310             m_idx = -1;
 311             next();
 312         }
 313
 314         void next() {
 315             TURF_ASSERT(m_table);
 316             TURF_ASSERT(isValid() || m_idx == -1); // Either the Iterator is already valid, or we've just started iterating.
 317             while (++m_idx <= m_table->sizeMask) {
 318                 // Index still inside range of table.
 319                 typename Details::CellGroup* group = m_table->getCellGroups() + (m_idx >> 2);
 320                 typename Details::Cell* cell = group->cells + (m_idx & 3);
 321                 m_hash = cell->hash.load(turf::Relaxed);
 322                 if (m_hash != KeyTraits::NullHash) {
 323                     // Cell has been reserved.
 324                     m_value = cell->value.load(turf::Relaxed);
 325                     TURF_ASSERT(m_value != Value(ValueTraits::Redirect));
 326                     if (m_value != Value(ValueTraits::NullValue))
 327                         return; // Yield this cell.
 328                 }
 329             }
 330             // That's the end of the map.
 331             m_hash = KeyTraits::NullHash;
 332             m_value = Value(ValueTraits::NullValue);
 333         }
 334
 335         bool isValid() const {
 336             return m_value != Value(ValueTraits::NullValue);
 337         }
 338
 339         Key getKey() const {
 340             TURF_ASSERT(isValid());
 341             // Since we've forbidden concurrent inserts (for now), nonatomic would suffice here, but let's plan ahead:
 342             return KeyTraits::dehash(m_hash);
 343         }
 344
 345         Value getValue() const {
 346             TURF_ASSERT(isValid());
 347             return m_value;
 348         }
 349     };
 350 };
 351
 352 } // namespace junction
 353
 354 #endif // JUNCTION_CONCURRENTMAP_LEAPFROG_H