VuoNodeSynchronization.cc

Go to the documentation of this file.

 
#include <sstream>
#include "VuoNodeSynchronization.hh"
#include "VuoNodeRegistry.hh"
#include "VuoRuntimeState.hh"
#include "VuoRuntimePersistentState.hh"
 
void VuoNodeSynchronization::lockNodes(VuoCompositionState *compositionState, unsigned long *nodeIndices, unsigned long nodeCount, unsigned long eventId)
{
    for (unsigned long i = 0; i < nodeCount; ++i)
        lockNode(compositionState, nodeIndices[i], eventId);
}
 
void VuoNodeSynchronization::lockNode(VuoCompositionState *compositionState, unsigned long nodeIndex, unsigned long eventId)
{
    recordWaiting(compositionState, eventId, nodeIndex);
 
    {
        NodeContext *nodeContext = vuoGetNodeContext(compositionState, nodeIndex);
 
        std::unique_lock<std::mutex> lock(* static_cast<std::mutex *>(nodeContext->nodeMutex));
        static_cast<std::condition_variable *>(nodeContext->nodeConditionVariable)->wait(lock, [nodeContext, eventId]()
        {
            if (nodeContext->claimingEventId == 0)
                nodeContext->claimingEventId = eventId;
 
            return nodeContext->claimingEventId == eventId;
        });
    }
 
    recordLocked(compositionState, eventId, nodeIndex);
}
 
void VuoNodeSynchronization::unlockNodes(VuoCompositionState *compositionState, unsigned long *nodeIndices, unsigned long nodeCount)
{
    for (unsigned long i = 0; i < nodeCount; ++i)
        unlockNode(compositionState, nodeIndices[i]);
}
 
void VuoNodeSynchronization::unlockNode(VuoCompositionState *compositionState, unsigned long nodeIndex)
{
    unsigned long eventId = 0;
 
    {
        NodeContext *nodeContext = vuoGetNodeContext(compositionState, nodeIndex);
 
        std::unique_lock<std::mutex> lock(* static_cast<std::mutex *>(nodeContext->nodeMutex));
 
        eventId = nodeContext->claimingEventId;
        nodeContext->claimingEventId = 0;
 
        static_cast<std::condition_variable *>(nodeContext->nodeConditionVariable)->notify_all();
    }
 
    recordUnlocked(compositionState, eventId, nodeIndex);
}
 
void VuoNodeSynchronization::recordWaiting(VuoCompositionState *compositionState, unsigned long eventId, unsigned long nodeIndex)
{
    if (! VuoIsDebugEnabled())
        return;
 
    std::lock_guard<std::mutex> lock(statusMutex);
 
    auto i = statusWaiting.find(compositionState->compositionIdentifier);
    if (i != statusWaiting.end())
    {
        auto j = i->second.find(eventId);
        if (j != i->second.end())
        {
            auto k = std::find(j->second.begin(), j->second.end(), nodeIndex);
            if (k != j->second.end())
                return;
        }
    }
 
    statusWaiting[compositionState->compositionIdentifier][eventId].push_back(nodeIndex);
}
 
void VuoNodeSynchronization::recordLocked(VuoCompositionState *compositionState, unsigned long eventId, unsigned long nodeIndex)
{
    if (! VuoIsDebugEnabled())
        return;
 
    std::lock_guard<std::mutex> lock(statusMutex);
 
    auto i = statusWaiting.find(compositionState->compositionIdentifier);
    if (i != statusWaiting.end())
    {
        auto j = i->second.find(eventId);
        if (j != i->second.end())
        {
            auto k = std::find(j->second.begin(), j->second.end(), nodeIndex);
            if (k != j->second.end())
            {
                j->second.erase(k);
                if (j->second.empty())
                    i->second.erase(j);
                if (i->second.empty())
                    statusWaiting.erase(i);
            }
        }
    }
 
    i = statusLocked.find(compositionState->compositionIdentifier);
    if (i != statusLocked.end())
    {
        auto j = i->second.find(eventId);
        if (j != i->second.end())
        {
            auto k = std::find(j->second.begin(), j->second.end(), nodeIndex);
            if (k != j->second.end())
                return;
        }
    }
 
    statusLocked[compositionState->compositionIdentifier][eventId].push_back(nodeIndex);
}
 
void VuoNodeSynchronization::recordUnlocked(VuoCompositionState *compositionState, unsigned long eventId, unsigned long nodeIndex)
{
    if (! VuoIsDebugEnabled())
        return;
 
    std::lock_guard<std::mutex> lock(statusMutex);
 
    auto i = statusLocked.find(compositionState->compositionIdentifier);
    if (i != statusLocked.end())
    {
        auto j = i->second.find(eventId);
        if (j != i->second.end())
        {
            auto k = std::find(j->second.begin(), j->second.end(), nodeIndex);
            if (k != j->second.end())
            {
                j->second.erase(k);
                if (j->second.empty())
                    i->second.erase(j);
                if (i->second.empty())
                    statusLocked.erase(i);
 
                return;
            }
        }
    }
 
    VDebugLog("Error: Node %lu was unlocked without having been locked", nodeIndex);
}
 
void VuoNodeSynchronization::print(void)
{
    if (! VuoIsDebugEnabled())
    {
        VUserLog("Enable debug mode to record status.");
        return;
    }
 
    std::lock_guard<std::mutex> lock(statusMutex);
 
    ostringstream oss;
    oss << "=== locked ===" << endl << endl;
    for (auto i : statusLocked)
    {
        oss << i.first << ":" << endl;
        for (auto j : i.second)
        {
            oss << "   eventId " << j.first << ":";
            for (unsigned long nodeIndex : j.second)
                oss << " " << nodeIndex;
            oss << endl;
        }
        oss << endl;
    }
    oss << endl;
 
    oss << "=== waiting to lock ===" << endl << endl;
    for (auto i : statusWaiting)
    {
        oss << i.first << ":" << endl;
        for (auto j : i.second)
        {
            oss << "   eventId " << j.first << ":";
            for (unsigned long nodeIndex : j.second)
                oss << " " << nodeIndex;
            oss << endl;
        }
        oss << endl;
    }
    oss << endl;
 
    VUserLog("\n%s", oss.str().c_str());
}
 
extern "C"
{
 
void vuoLockNodes(VuoCompositionState *compositionState, unsigned long *nodeIndices, unsigned long nodeCount, unsigned long eventId)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    runtimeState->persistentState->nodeSynchronization->lockNodes(compositionState, nodeIndices, nodeCount, eventId);
}
 
void vuoLockNode(VuoCompositionState *compositionState, unsigned long nodeIndex, unsigned long eventId)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    runtimeState->persistentState->nodeSynchronization->lockNode(compositionState, nodeIndex, eventId);
}
 
void vuoUnlockNodes(VuoCompositionState *compositionState, unsigned long *nodeIndices, unsigned long nodeCount)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    runtimeState->persistentState->nodeSynchronization->unlockNodes(compositionState, nodeIndices, nodeCount);
}
 
void vuoUnlockNode(VuoCompositionState *compositionState, unsigned long nodeIndex)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    runtimeState->persistentState->nodeSynchronization->unlockNode(compositionState, nodeIndex);
}
 
}