Leapfrog< Map >::TableMigration Class Reference

#include <leapfrog.h>

Inheritance diagram for Leapfrog< Map >::TableMigration:

Classes
struct	Source

Public Member Functions
void	destroy ()
Source *	getSources () const
bool	migrateRange (Table *srcTable, quint64 startIdx)
virtual void	run () override
	TableMigration (Map &map)
virtual	~TableMigration () override
Public Member Functions inherited from SimpleJobCoordinator::Job
virtual	~Job ()

Static Public Member Functions
static TableMigration *	create (Map &map, quint64 numSources)

Public Attributes
Table *	m_destination {nullptr}
Map &	m_map
quint64	m_numSources {0}
Atomic< bool >	m_overflowed
Atomic< qint64 >	m_unitsRemaining
Atomic< quint64 >	m_workerStatus

Detailed Description

template<class Map>
class Leapfrog< Map >::TableMigration

Definition at line 100 of file leapfrog.h.

Constructor & Destructor Documentation

TableMigration()

template<class Map >

Leapfrog< Map >::TableMigration::TableMigration ( Map & map )

inline

Definition at line 115 of file leapfrog.h.

                                 : m_map(map)
        {
        }

~TableMigration()

template<class Map >

virtual Leapfrog< Map >::TableMigration::~TableMigration ( )

inlineoverridevirtual

Definition at line 133 of file leapfrog.h.

        {
        }

Member Function Documentation

create()

template<class Map >

static TableMigration * Leapfrog< Map >::TableMigration::create ( Map & map, quint64 numSources )

inlinestatic

Definition at line 119 of file leapfrog.h.

        {
            TableMigration* migration =
                (TableMigration*) std::malloc(sizeof(TableMigration) + sizeof(TableMigration::Source) * numSources);
            new (migration) TableMigration(map);
 
            migration->m_workerStatus.storeNonatomic(0);
            migration->m_overflowed.storeNonatomic(false);
            migration->m_unitsRemaining.storeNonatomic(0);
            migration->m_numSources = numSources;
            // Caller is responsible for filling in sources & destination
            return migration;
        }

References Leapfrog< Map >::TableMigration::m_numSources, Leapfrog< Map >::TableMigration::m_overflowed, Leapfrog< Map >::TableMigration::m_unitsRemaining, Leapfrog< Map >::TableMigration::m_workerStatus, Atomic< T >::storeNonatomic(), and Leapfrog< Map >::TableMigration::TableMigration().

destroy()

template<class Map >

void Leapfrog< Map >::TableMigration::destroy ( )

inline

Definition at line 137 of file leapfrog.h.

        {
            // Destroy all source tables.
            for (quint64 i = 0; i < m_numSources; i++)
                if (getSources()[i].table)
                    getSources()[i].table->destroy();
            // Delete the migration object itself.
            this->TableMigration::~TableMigration();
            std::free(this);
        }

References Leapfrog< Map >::Table::destroy(), Leapfrog< Map >::TableMigration::getSources(), Leapfrog< Map >::TableMigration::m_numSources, Leapfrog< Map >::TableMigration::Source::table, and Leapfrog< Map >::TableMigration::~TableMigration().

getSources()

template<class Map >

Source * Leapfrog< Map >::TableMigration::getSources ( ) const

inline

Definition at line 148 of file leapfrog.h.

        {
            return (Source*)(this + 1);
        }

migrateRange()

template<class Map >

bool Leapfrog< Map >::TableMigration::migrateRange	(	Table *	srcTable,
		quint64	startIdx )

Definition at line 358 of file leapfrog.h.

{
    quint64 srcSizeMask = srcTable->sizeMask;
    quint64 endIdx = qMin(startIdx + TableMigrationUnitSize, srcSizeMask + 1);
    // Iterate over source range.
    for (quint64 srcIdx = startIdx; srcIdx < endIdx; srcIdx++) {
        CellGroup* srcGroup = srcTable->getCellGroups() + ((srcIdx & srcSizeMask) >> 2);
        Cell* srcCell = srcGroup->cells + (srcIdx & 3);
        Hash srcHash;
        Value srcValue;
        // Fetch the srcHash and srcValue.
        for (;;) {
            srcHash = srcCell->hash.load(Relaxed);
            if (srcHash == KeyTraits::NullHash) {
                // An unused cell. Try to put a Redirect marker in its value.
                srcValue =
                    srcCell->value.compareExchange(Value(ValueTraits::NullValue), Value(ValueTraits::Redirect), Relaxed);
                if (srcValue == Value(ValueTraits::Redirect)) {
                    // srcValue is already marked Redirect due to previous incomplete migration.
                    break;
                }
                if (srcValue == Value(ValueTraits::NullValue)) {
                    break; // Redirect has been placed. Break inner loop, continue outer loop.
                }
                // Otherwise, somebody just claimed the cell. Read srcHash again...
            } else {
                // Check for deleted/uninitialized value.
                srcValue = srcCell->value.load(Relaxed);
                if (srcValue == Value(ValueTraits::NullValue)) {
                    // Try to put a Redirect marker.
                    if (srcCell->value.compareExchangeStrong(srcValue, Value(ValueTraits::Redirect), Relaxed)) {
                        break; // Redirect has been placed. Break inner loop, continue outer loop.
                    }
 
                    if (srcValue == Value(ValueTraits::Redirect)) {
                        // FIXME: I don't think this will happen. Investigate & change to assert
                        break;
                    }
                } else if (srcValue == Value(ValueTraits::Redirect)) {
                    // srcValue is already marked Redirect due to previous incomplete migration.
                    break;
                }
 
                // We've got a key/value pair to migrate.
                // Reserve a destination cell in the destination.
#ifdef SANITY_CHECK
                KIS_ASSERT_RECOVER_NOOP(srcHash != KeyTraits::NullHash);
                KIS_ASSERT_RECOVER_NOOP(srcValue != Value(ValueTraits::NullValue));
                KIS_ASSERT_RECOVER_NOOP(srcValue != Value(ValueTraits::Redirect));
#endif // SANITY_CHECK
                Cell* dstCell;
                quint64 overflowIdx;
                InsertResult result = insertOrFind(srcHash, m_destination, dstCell, overflowIdx);
                // During migration, a hash can only exist in one place among all the source tables,
                // and it is only migrated by one thread. Therefore, the hash will never already exist
                // in the destination table:
#ifdef SANITY_CHECK
                KIS_ASSERT_RECOVER_NOOP(result != InsertResult_AlreadyFound);
#endif // SANITY_CHECK
                if (result == InsertResult_Overflow) {
                    // Destination overflow.
                    // This can happen for several reasons. For example, the source table could have
                    // existed of all deleted cells when it overflowed, resulting in a small destination
                    // table size, but then another thread could re-insert all the same hashes
                    // before the migration completed.
                    // Caller will cancel the current migration and begin a new one.
                    return false;
                }
                // Migrate the old value to the new cell.
                for (;;) {
                    // Copy srcValue to the destination.
                    dstCell->value.store(srcValue, Relaxed);
                    // Try to place a Redirect marker in srcValue.
                    Value doubleCheckedSrcValue = srcCell->value.compareExchange(srcValue, Value(ValueTraits::Redirect), Relaxed);
#ifdef SANITY_CHECK
                    KIS_ASSERT_RECOVER_NOOP(doubleCheckedSrcValue != Value(ValueTraits::Redirect)); // Only one thread can redirect a cell at a time.
#endif // SANITY_CHECK
                    if (doubleCheckedSrcValue == srcValue) {
                        // No racing writes to the src. We've successfully placed the Redirect marker.
                        // srcValue was non-NULL when we decided to migrate it, but it may have changed to NULL
                        // by a late-arriving erase.
                        if (srcValue == Value(ValueTraits::NullValue)) {
                            // racing update was erase", uptr(srcTable), srcIdx)
                        }
 
                        break;
                    }
                    // There was a late-arriving write (or erase) to the src. Migrate the new value and try again.
                    srcValue = doubleCheckedSrcValue;
                }
                // Cell successfully migrated. Proceed to next source cell.
                break;
            }
        }
    }
    // Range has been migrated successfully.
    return true;
}

References Leapfrog< Map >::CellGroup::cells, Atomic< T >::compareExchange(), Atomic< T >::compareExchangeStrong(), Leapfrog< Map >::Table::getCellGroups(), Leapfrog< Map >::Cell::hash, Leapfrog< Map >::insertOrFind(), Leapfrog< Map >::InsertResult_AlreadyFound, Leapfrog< Map >::InsertResult_Overflow, KIS_ASSERT_RECOVER_NOOP, Atomic< T >::load(), Leapfrog< Map >::TableMigration::m_destination, Relaxed, Leapfrog< Map >::Table::sizeMask, Atomic< T >::store(), Leapfrog< Map >::TableMigrationUnitSize, and Leapfrog< Map >::Cell::value.

run()

template<class Map >

void Leapfrog< Map >::TableMigration::run ( )

overridevirtual

Implements SimpleJobCoordinator::Job.

Definition at line 458 of file leapfrog.h.

{
#ifdef SANITY_CHECK
    KIS_ASSERT_RECOVER_NOOP(m_map.getGC().sanityRawPointerAccessLocked());
#endif // SANITY_CHECK
 
 
    // Conditionally increment the shared # of workers.
    quint64 probeStatus = m_workerStatus.load(Relaxed);
    do {
        if (probeStatus & 1) {
            // End flag is already set, so do nothing.
            return;
        }
    } while (!m_workerStatus.compareExchangeWeak(probeStatus, probeStatus + 2, Relaxed, Relaxed));
    // # of workers has been incremented, and the end flag is clear.
#ifdef SANITY_CHECK
    KIS_ASSERT_RECOVER_NOOP((probeStatus & 1) == 0);
#endif // SANITY_CHECK
 
    // Iterate over all source tables.
    for (quint64 s = 0; s < m_numSources; s++) {
        Source& source = getSources()[s];
        // Loop over all migration units in this source table.
        for (;;) {
            if (m_workerStatus.load(Relaxed) & 1) {
                goto endMigration;
            }
            quint64 startIdx = source.sourceIndex.fetchAdd(TableMigrationUnitSize, Relaxed);
            if (startIdx >= source.table->sizeMask + 1)
                break; // No more migration units in this table. Try next source table.
            bool overflowed = !migrateRange(source.table, startIdx);
            if (overflowed) {
                // *** FAILED MIGRATION ***
                // TableMigration failed due to destination table overflow.
                // No other thread can declare the migration successful at this point, because *this* unit will never complete,
                // hence m_unitsRemaining won't reach zero.
                // However, multiple threads can independently detect a failed migration at the same time.
                // The reason we store overflowed in a shared variable is because we can must flush all the worker threads before
                // we can safely deal with the overflow. Therefore, the thread that detects the failure is often different from
                // the thread
                // that deals with it.
                bool oldOverflowed = m_overflowed.exchange(overflowed, Relaxed);
                if (oldOverflowed) {
                    // race to set m_overflowed
                }
 
                m_workerStatus.fetchOr(1, Relaxed);
                goto endMigration;
            }
 
            qint64 prevRemaining = m_unitsRemaining.fetchSub(1, Relaxed);
#ifdef SANITY_CHECK
            KIS_ASSERT_RECOVER_NOOP(prevRemaining > 0);
#endif // SANITY_CHECK
            if (prevRemaining == 1) {
                // *** SUCCESSFUL MIGRATION ***
                // That was the last chunk to migrate.
                m_workerStatus.fetchOr(1, Relaxed);
                goto endMigration;
            }
        }
    }
 
endMigration:
    // Decrement the shared # of workers.
    probeStatus = m_workerStatus.fetchSub(2, AcquireRelease); // AcquireRelease makes all previous writes visible to the last worker thread.
    if (probeStatus >= 4) {
        // There are other workers remaining. Return here so that only the very last worker will proceed.
        return;
    }
 
    // We're the very last worker thread.
    // Perform the appropriate post-migration step depending on whether the migration succeeded or failed.
#ifdef SANITY_CHECK
    KIS_ASSERT_RECOVER_NOOP(probeStatus == 3);
#endif // SANITY_CHECK
    bool overflowed = m_overflowed.loadNonatomic(); // No racing writes at this point
    if (!overflowed) {
        // The migration succeeded. This is the most likely outcome. Publish the new subtree.
        m_map.publishTableMigration(this);
        // End the jobCoodinator.
        getSources()[0].table->jobCoordinator.end();
    } else {
        // The migration failed due to the overflow of the destination table.
        Table* origTable = getSources()[0].table;
        QMutexLocker guard(&origTable->mutex);
        SimpleJobCoordinator::Job* checkedJob = origTable->jobCoordinator.loadConsume();
 
        if (checkedJob != this) {
            // a new TableMigration was already started
        } else {
            TableMigration* migration = TableMigration::create(m_map, m_numSources + 1);
            // Double the destination table size.
            migration->m_destination = Table::create((m_destination->sizeMask + 1) * 2);
            // Transfer source tables to the new migration.
            for (quint64 i = 0; i < m_numSources; i++) {
                migration->getSources()[i].table = getSources()[i].table;
                getSources()[i].table = NULL;
                migration->getSources()[i].sourceIndex.storeNonatomic(0);
            }
 
            migration->getSources()[m_numSources].table = m_destination;
            migration->getSources()[m_numSources].sourceIndex.storeNonatomic(0);
            // Calculate total number of migration units to move.
            quint64 unitsRemaining = 0;
            for (quint64 s = 0; s < migration->m_numSources; s++) {
                unitsRemaining += migration->getSources()[s].table->getNumMigrationUnits();
            }
 
            migration->m_unitsRemaining.storeNonatomic(unitsRemaining);
            // Publish the new migration.
            origTable->jobCoordinator.storeRelease(migration);
        }
    }
 
    // We're done with this TableMigration. Queue it for GC.
    m_map.getGC().enqueue(&TableMigration::destroy, this, true);
}

References AcquireRelease, Leapfrog< Map >::TableMigration::create(), Leapfrog< Map >::Table::create(), Leapfrog< Map >::TableMigration::destroy(), Leapfrog< Map >::Table::getNumMigrationUnits(), Leapfrog< Map >::TableMigration::getSources(), Leapfrog< Map >::Table::jobCoordinator, KIS_ASSERT_RECOVER_NOOP, SimpleJobCoordinator::loadConsume(), Leapfrog< Map >::TableMigration::m_destination, Leapfrog< Map >::TableMigration::m_numSources, Leapfrog< Map >::TableMigration::m_unitsRemaining, Leapfrog< Map >::Table::mutex, Relaxed, source(), Leapfrog< Map >::TableMigration::Source::sourceIndex, Atomic< T >::storeNonatomic(), SimpleJobCoordinator::storeRelease(), Leapfrog< Map >::TableMigration::Source::table, and Leapfrog< Map >::TableMigrationUnitSize.

Member Data Documentation

m_destination

template<class Map >

Table* Leapfrog< Map >::TableMigration::m_destination {nullptr}

Definition at line 109 of file leapfrog.h.

{nullptr};

m_map

template<class Map >

Map& Leapfrog< Map >::TableMigration::m_map

Definition at line 108 of file leapfrog.h.

m_numSources

template<class Map >

quint64 Leapfrog< Map >::TableMigration::m_numSources {0}

Definition at line 113 of file leapfrog.h.

{0};

m_overflowed

template<class Map >

Atomic<bool> Leapfrog< Map >::TableMigration::m_overflowed

Definition at line 111 of file leapfrog.h.

m_unitsRemaining

template<class Map >

Atomic<qint64> Leapfrog< Map >::TableMigration::m_unitsRemaining

Definition at line 112 of file leapfrog.h.

m_workerStatus

template<class Map >

Atomic<quint64> Leapfrog< Map >::TableMigration::m_workerStatus

Definition at line 110 of file leapfrog.h.

---

The documentation for this class was generated from the following file:

• libs/image/3rdparty/lock_free_map/leapfrog.h