piercedSync.cpp

/*---------------------------------------------------------------------------*\
 *
 *  bitpit
 *
 *  Copyright (C) 2015-2021 OPTIMAD engineering Srl
 *
 *  -------------------------------------------------------------------------
 *  License
 *  This file is part of bitpit.
 *
 *  bitpit is free software: you can redistribute it and/or modify it
 *  under the terms of the GNU Lesser General Public License v3 (LGPL)
 *  as published by the Free Software Foundation.
 *
 *  bitpit is distributed in the hope that it will be useful, but WITHOUT
 *  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 *  FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
 *  License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public License
 *  along with bitpit. If not, see <http://www.gnu.org/licenses/>.
 *
\*---------------------------------------------------------------------------*/
 
#include "bitpit_common.hpp"
 
#include "piercedSync.hpp"
 
namespace bitpit {


PiercedSyncAction::PiercedSyncAction(ActionType _type)
    : type(_type), data(nullptr)
{
}

PiercedSyncAction::PiercedSyncAction(const PiercedSyncAction &other)
{
    // Copy the items that need allocation
    std::unique_ptr<std::vector<std::size_t>> new_data;
    if (other.data) {
        new_data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(*(other.data)));
    }
 
    // Assign the new memory to the object
    type = other.type;
 
    for (std::size_t k = 0; k < INFO_COUNT; ++k) {
        info[k] = other.info[k];
    }
 
    data.swap(new_data);
}

PiercedSyncAction & PiercedSyncAction::operator=(const PiercedSyncAction &other)
{
    if (this != &other) {
        PiercedSyncAction temporary(other);
        temporary.swap(*this);
    }
 
    return *this;
}

void PiercedSyncAction::swap(PiercedSyncAction &other) noexcept
{
    std::swap(type, other.type);
    std::swap(info, other.info);
    data.swap(other.data);
}

void PiercedSyncAction::importData(std::vector<std::size_t> &&values)
{
    data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(std::move(values)));
}

void PiercedSyncAction::importData(const std::vector<std::size_t> &values)
{
    data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(values));
}

void PiercedSyncAction::restore(std::istream &stream)
{
    int actionType;
    utils::binary::read(stream, actionType);
    type = static_cast<ActionType>(actionType);
 
    for (int k = 0; k < INFO_COUNT; ++k) {
        utils::binary::read(stream, info[k]);
    }
 
    bool hasData;
    utils::binary::read(stream, hasData);
 
    if (hasData) {
        std::size_t dataSize;
        utils::binary::read(stream, dataSize);
        data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(dataSize));
        for (std::size_t k = 0; k < dataSize; ++k) {
            utils::binary::read(stream, (*data)[k]);
        }
    } else {
        data.reset();
    }
}

void PiercedSyncAction::dump(std::ostream &stream) const
{
    utils::binary::write(stream, static_cast<int>(type));
 
    for (int k = 0; k < INFO_COUNT; ++k) {
        utils::binary::write(stream, info[k]);
    }
 
    bool hasData = (data != nullptr);
    utils::binary::write(stream, hasData);
 
    if (hasData) {
        std::size_t dataSize = data->size();
        utils::binary::write(stream, dataSize);
        for (std::size_t k = 0; k < dataSize; ++k) {
            utils::binary::write(stream, (*data)[k]);
        }
    }
}


/**
* Constructor
*/
PiercedSyncSlave::PiercedSyncSlave()
{
}

void PiercedSyncSlave::swap(PiercedSyncSlave &other) noexcept
{
    BITPIT_UNUSED(other);
}


PiercedSyncMaster::PiercedSyncMaster()
    : m_syncEnabled(false)
{
    for (int k = SYNC_MODE_ITR_BEGIN; k != SYNC_MODE_ITR_END; ++k) {
        SyncMode mode = static_cast<SyncMode>(k);
        m_syncGroups[mode] = SyncGroup();
    }
}

void PiercedSyncMaster::swap(PiercedSyncMaster &other) noexcept
{
    std::swap(other.m_slaves, m_slaves);
    std::swap(other.m_syncGroups, m_syncGroups);
    std::swap(other.m_syncJournal, m_syncJournal);
}

void PiercedSyncMaster::registerSlave(PiercedSyncSlave *slave, SyncMode syncMode) const
{
    if (m_slaves.count(slave) > 0) {
        throw std::out_of_range("Slave is already registered");
    }
 
    // Register the slave
    m_slaves.insert({slave, syncMode});
 
    // Add the slave to the proper synchronization group
    SyncGroup &syncGroup = m_syncGroups.at(syncMode);
    syncGroup.push_back(slave);
 
    // If needed enable synchronization
    if (!isSyncEnabled() && syncMode != SYNC_MODE_DISABLED) {
        setSyncEnabled(true);
    }
}

void PiercedSyncMaster::processSyncAction(const PiercedSyncAction &action)
{
    if (!isSyncEnabled()) {
        return;
    }
 
    // Synchronize concurrent slaves
    for (PiercedSyncSlave *slave : m_syncGroups.at(PiercedSyncMaster::SYNC_MODE_CONCURRENT)) {
        commitSyncAction(slave, action);
    }
 
    // If there are journaled slaves the synchronization action needs to be
    // journaled
    if (m_syncGroups.at(PiercedSyncMaster::SYNC_MODE_JOURNALED).size() > 0) {
        journalSyncAction(action);
    }
}

void PiercedSyncMaster::commitSyncAction(PiercedSyncSlave *slave, const PiercedSyncAction &action) const
{
    slave->commitSyncAction(action);
}

void PiercedSyncMaster::journalSyncAction(const PiercedSyncAction &action)
{
    // Get information about previous action
    PiercedSyncAction *previousAction = nullptr;
    if (!m_syncJournal.empty()) {
        previousAction = &(m_syncJournal.back());
    }
 
    PiercedSyncAction::ActionType previousActionType = PiercedSyncAction::TYPE_UNDEFINED;
    if (previousAction) {
        previousActionType = previousAction->type;
    }
 
    // Journal current action
    //
    // If possible, current action will be merged with the previous one.
    switch (action.type) {
 
    case PiercedSyncAction::TYPE_CLEAR:
        m_syncJournal.resize(1);
        m_syncJournal[0] = action;
        break;
 
    case PiercedSyncAction::TYPE_APPEND:
        if (previousActionType == PiercedSyncAction::TYPE_APPEND) {
            previousAction->type = PiercedSyncAction::TYPE_RESIZE;
            previousAction->info[PiercedSyncAction::INFO_SIZE] = action.info[PiercedSyncAction::INFO_POS] + 1;
            previousAction->info[PiercedSyncAction::INFO_POS]  = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_RESIZE) {
            previousAction->info[PiercedSyncAction::INFO_SIZE] = action.info[PiercedSyncAction::INFO_POS] + 1;
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_RESIZE:
        if (previousActionType == PiercedSyncAction::TYPE_APPEND) {
            previousAction->type = PiercedSyncAction::TYPE_RESIZE;
            previousAction->info[PiercedSyncAction::INFO_SIZE] = action.info[PiercedSyncAction::INFO_SIZE];
            previousAction->info[PiercedSyncAction::INFO_POS]  = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_RESIZE) {
            previousAction->info[PiercedSyncAction::INFO_SIZE] = action.info[PiercedSyncAction::INFO_SIZE];
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_PIERCE:
        if (previousActionType == PiercedSyncAction::TYPE_PIERCE) {
            previousAction->type = PiercedSyncAction::TYPE_PIERCE_MULTIPLE;
            previousAction->data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(2));
            (*(previousAction->data))[0] = previousAction->info[PiercedSyncAction::INFO_POS];
            (*(previousAction->data))[1] = action.info[PiercedSyncAction::INFO_POS];
            previousAction->info[PiercedSyncAction::INFO_POS]      = std::numeric_limits<std::size_t>::max();
            previousAction->info[PiercedSyncAction::INFO_POS_NEXT] = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_PIERCE_MULTIPLE) {
            previousAction->data->push_back(action.info[PiercedSyncAction::INFO_POS]);
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_PIERCE_MULTIPLE:
        if (previousActionType == PiercedSyncAction::TYPE_PIERCE) {
            previousAction->type = PiercedSyncAction::TYPE_PIERCE_MULTIPLE;
            previousAction->data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(action.data->size() + 1));
            (*(previousAction->data))[0] = previousAction->info[PiercedSyncAction::INFO_POS];
            previousAction->data->insert(previousAction->data->begin() + 1, action.data->begin(), action.data->end());
            previousAction->info[PiercedSyncAction::INFO_POS]      = std::numeric_limits<std::size_t>::max();
            previousAction->info[PiercedSyncAction::INFO_POS_NEXT] = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_PIERCE_MULTIPLE) {
            previousAction->data->insert(previousAction->data->begin(), action.data->begin(), action.data->end());
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_OVERWRITE:
        if (previousActionType == PiercedSyncAction::TYPE_OVERWRITE) {
            previousAction->type = PiercedSyncAction::TYPE_OVERWRITE_MULTIPLE;
            previousAction->data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(2));
            (*(previousAction->data))[0] = previousAction->info[PiercedSyncAction::INFO_POS];
            (*(previousAction->data))[1] = action.info[PiercedSyncAction::INFO_POS];
            previousAction->info[PiercedSyncAction::INFO_POS] = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_OVERWRITE_MULTIPLE) {
            previousAction->data->push_back(action.info[PiercedSyncAction::INFO_POS]);
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_OVERWRITE_MULTIPLE:
        if (previousActionType == PiercedSyncAction::TYPE_OVERWRITE) {
            previousAction->type = PiercedSyncAction::TYPE_OVERWRITE_MULTIPLE;
            previousAction->data = std::unique_ptr<std::vector<std::size_t>>(new std::vector<std::size_t>(action.data->size() + 1));
            (*(previousAction->data))[0] = previousAction->info[PiercedSyncAction::INFO_POS];
            previousAction->data->insert(previousAction->data->begin() + 1, action.data->begin(), action.data->end());
            previousAction->info[PiercedSyncAction::INFO_POS] = std::numeric_limits<std::size_t>::max();
        } else if (previousActionType == PiercedSyncAction::TYPE_OVERWRITE_MULTIPLE) {
            previousAction->data->insert(previousAction->data->begin(), action.data->begin(), action.data->end());
        } else {
            m_syncJournal.push_back(action);
        }
        break;
 
    case PiercedSyncAction::TYPE_NOOP:
        break;
 
    default:
        m_syncJournal.push_back(action);
        break;
 
    }
}

/**
* Unregister the specified slave
*
* \param slave is the slave that will be unregistered
*/
void PiercedSyncMaster::unregisterSlave(const PiercedSyncSlave *slave) const
{
    // Check if the slave was actually registered
    if (!isSlaveRegistered(slave)) {
        return;
    }
 
    // Remove the slave from the synchronization group
    SyncMode syncMode = getSlaveSyncMode(slave);
    SyncGroup &syncGroup = m_syncGroups.at(syncMode);
    for (auto itr = syncGroup.begin(); itr != syncGroup.end(); ++itr) {
        PiercedSyncSlave *item = *itr;
        if (item == slave) {
            syncGroup.erase(itr);
            break;
        }
    }
 
    // Unregister the slave
    m_slaves.erase(const_cast<PiercedSyncSlave *>(slave));
}

bool PiercedSyncMaster::isSlaveRegistered(const PiercedSyncSlave *slave) const
{
    return (m_slaves.count(const_cast<PiercedSyncSlave *>(slave)) > 0);
}

PiercedSyncMaster::SyncMode PiercedSyncMaster::getSlaveSyncMode(const PiercedSyncSlave *slave) const
{
    return m_slaves.at(const_cast<PiercedSyncSlave *>(slave));
}

bool PiercedSyncMaster::isSlaveSynced(const PiercedSyncSlave *slave) const
{
    SyncMode syncMode = getSlaveSyncMode(slave);
    if (syncMode == SYNC_MODE_CONCURRENT) {
        return true;
    } else if (syncMode == SYNC_MODE_JOURNALED) {
        return m_syncJournal.empty();
    } else {
        return false;
    }
}

void PiercedSyncMaster::sync()
{
    // Only journaled slaved need to be synchronized
    for (PiercedSyncSlave *slave : m_syncGroups.at(SYNC_MODE_JOURNALED)) {
        for (const PiercedSyncAction &action : m_syncJournal) {
            commitSyncAction(slave, action);
        }
    }
 
    // Clear the sync journal
    m_syncJournal.clear();
    m_syncJournal.shrink_to_fit();
}

bool PiercedSyncMaster::isSynced() const
{
    for (const PiercedSyncSlave *slave : m_syncGroups.at(SYNC_MODE_JOURNALED)) {
        if (!isSlaveSynced(slave)) {
            return false;
        }
    }
 
    return true;
}

void PiercedSyncMaster::setSyncEnabled(bool enabled) const
{
    m_syncEnabled = enabled;
}

bool PiercedSyncMaster::isSyncEnabled() const
{
    return m_syncEnabled;
}

void PiercedSyncMaster::restore(std::istream &stream)
{
    std::size_t journalSize;
    utils::binary::read(stream, journalSize);
    m_syncJournal.resize(journalSize);
    for (PiercedSyncAction &action : m_syncJournal) {
        action.restore(stream);
    }
}

void PiercedSyncMaster::dump(std::ostream &stream) const
{
    std::size_t journalSize = m_syncJournal.size();
    utils::binary::write(stream, journalSize);
    for (const PiercedSyncAction &action : m_syncJournal) {
        action.dump(stream);
    }
}
 
}