VuoRuntimeState.cc

Go to the documentation of this file.

 
#include "VuoRuntimeState.hh"
 
#include <dlfcn.h>
#include <signal.h>
#include <sstream>
#include "VuoCompositionDiff.hh"
#include "VuoEventLoop.h"
#include "VuoException.hh"
#include "VuoHeap.h"
#include "VuoNodeRegistry.hh"
#include "VuoRuntimeCommunicator.hh"
#include "VuoRuntimePersistentState.hh"
#include "VuoThreadManager.hh"
 
VuoRuntimeState::VuoRuntimeState(void)
{
    persistentState = NULL;
}
 
VuoRuntimeState::~VuoRuntimeState(void)
{
    delete persistentState;
}
 
void VuoRuntimeState::init(void *zmqContext, const char *controlUrl, const char *telemetryUrl, bool isPaused,
                           pid_t runnerPid, int runnerPipe, bool continueIfRunnerDies, const char *workingDirectory,
                           void *compositionBinaryHandle)
{
    if (! persistentState)
    {
        persistentState = new VuoRuntimePersistentState(workingDirectory);
        persistentState->runtimeState = this;
    }
 
    this->_isPaused = isPaused;
    hasBeenUnpaused = false;
    _isStopped = false;
    wasStopCompositionCalled = false;
    this->continueIfRunnerDies = continueIfRunnerDies;
 
    this->runnerPid = runnerPid;
 
    terminationDisabledCount = 0;
    terminationDisabledQueue = dispatch_queue_create("org.vuo.runtime.terminationDisabled", NULL);
 
    stopQueue = dispatch_queue_create("org.vuo.runtime.stop", NULL);
 
    waitForStopTimer = NULL;
    waitForStopCanceledSemaphore = NULL;
 
    vuoSetup = NULL;
    vuoCleanup = NULL;
    vuoInstanceInit = NULL;
    vuoInstanceFini = NULL;
    vuoInstanceTriggerStart = NULL;
    vuoInstanceTriggerStop = NULL;
 
    updateCompositionSymbols(compositionBinaryHandle);
    persistentState->communicator->openConnection(zmqContext, controlUrl, telemetryUrl, runnerPipe);
}
 
void VuoRuntimeState::fini(void)
{
    vuoMemoryBarrier();
    persistentState->communicator->stopSendingAndCleanUpHeartbeat();
    persistentState->communicator->cleanUpControl();
 
    if (waitForStopTimer)
    {
        dispatch_source_cancel(waitForStopTimer);
        dispatch_semaphore_wait(waitForStopCanceledSemaphore, DISPATCH_TIME_FOREVER);
        dispatch_release(waitForStopTimer);
        dispatch_release(waitForStopCanceledSemaphore);
    }
 
    persistentState->communicator->closeConnection();
 
    dispatch_release(stopQueue);
}
 
void VuoRuntimeState::updateCompositionSymbols(void *compositionBinaryHandle)
{
    ostringstream errorMessage;
 
    vuoSetup = (vuoSetupType) dlsym(compositionBinaryHandle, "vuoSetup");
    if (! vuoSetup)
    {
        errorMessage << "The composition couldn't be started because its vuoSetup() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    vuoCleanup = (vuoCleanupType) dlsym(compositionBinaryHandle, "vuoCleanup");
    if (! vuoCleanup)
    {
        errorMessage << "The composition couldn't be started because its vuoCleanup() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    vuoInstanceInit = (vuoInstanceInitType) dlsym(compositionBinaryHandle, "vuoInstanceInit");
    if (! vuoInstanceInit)
    {
        errorMessage << "The composition couldn't be started because its vuoInstanceInit() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    vuoInstanceFini = (vuoInstanceFiniType) dlsym(compositionBinaryHandle, "vuoInstanceFini");
    if (! vuoInstanceFini)
    {
        errorMessage << "The composition couldn't be started because its vuoInstanceFini() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    vuoInstanceTriggerStart = (vuoInstanceTriggerStartType) dlsym(compositionBinaryHandle, "vuoInstanceTriggerStart");
    if (! vuoInstanceTriggerStart)
    {
        errorMessage << "The composition couldn't be started because its vuoInstanceTriggerStart() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    vuoInstanceTriggerStop = (vuoInstanceTriggerStopType) dlsym(compositionBinaryHandle, "vuoInstanceTriggerStop");
    if (! vuoInstanceTriggerStop)
    {
        errorMessage << "The composition couldn't be started because its vuoInstanceTriggerStop() function couldn't be found : " << dlerror();
        throw VuoException(errorMessage.str());
    }
 
    persistentState->communicator->updateCompositionSymbols(compositionBinaryHandle);
    persistentState->nodeRegistry->updateCompositionSymbols(compositionBinaryHandle);
}
 
bool VuoRuntimeState::isPaused(void)
{
    return _isPaused;
}
 
bool VuoRuntimeState::isStopped(void)
{
    return _isStopped;
}
 
pid_t VuoRuntimeState::getRunnerPid(void)
{
    return runnerPid;
}
 
bool VuoRuntimeState::mayBeTerminated(void)
{
    __block bool ret;
    dispatch_sync(terminationDisabledQueue, ^{
                      ret = (terminationDisabledCount == 0);
                  });
 
    return ret;
}
 
void VuoRuntimeState::disableTermination(void)
{
    dispatch_sync(terminationDisabledQueue, ^{
                      ++terminationDisabledCount;
                  });
}
 
void VuoRuntimeState::enableTermination(void)
{
    dispatch_sync(terminationDisabledQueue, ^{
                      --terminationDisabledCount;
                      if (terminationDisabledCount < 0)
                          terminationDisabledCount = 0;
                  });
}
 
void VuoRuntimeState::startComposition(void)
{
    persistentState->threadManager->enableSchedulingWorkers();
 
    vuoSetup();
 
    if (! _isPaused)
    {
        // Wait to call vuoInstanceInit() until the first time the composition enters an unpaused state.
        // If vuoInstanceInit() were called immediately when the composition is started in a paused state(),
        // then the vuo.event.fireOnStart node's fired event would be ignored.
 
        hasBeenUnpaused = true;
        __block bool initDone = false;
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
                           vuoInstanceInit();
                           vuoInstanceTriggerStart();
                           initDone = true;
                           VuoEventLoop_break();
                       });
        while (! initDone)
            VuoEventLoop_processEvent(VuoEventLoop_WaitIndefinitely);
    }
 
    persistentState->communicator->startSendingHeartbeat();
    persistentState->communicator->startListeningForControl();
 
    if (! continueIfRunnerDies)
        persistentState->communicator->startListeningForRunnerExit();
}
 
void VuoRuntimeState::pauseComposition(void)
{
    _isPaused = true;
    vuoInstanceTriggerStop();
}
 
void VuoRuntimeState::unpauseComposition(void)
{
    _isPaused = false;
 
    if (! hasBeenUnpaused)
    {
        hasBeenUnpaused = true;
        vuoInstanceInit();
    }
    vuoInstanceTriggerStart();
}
 
void VuoRuntimeState::stopCompositionAsOrderedByRunner(bool isBeingReplaced, int timeoutInSeconds, bool isLastEverInProcess)
{
    if (! isBeingReplaced)
    {
        persistentState->compositionDiff->setDiff(NULL);
    }
 
    stopComposition(isBeingReplaced, timeoutInSeconds);
 
    if (timeoutInSeconds >= 0)
    {
        VUOLOG_PROFILE_BEGIN(mainQueue);
        dispatch_sync(dispatch_get_main_queue(), ^{
            VUOLOG_PROFILE_END(mainQueue);
            VuoEventLoop_processEvent(VuoEventLoop_RunOnce);
        });
 
        VuoCompositionState compositionState = { (void *)this, "" };
        vuoAddCompositionStateToThreadLocalStorage(&compositionState);
 
        // https://b33p.net/kosada/node/12912
        if (isLastEverInProcess)
        {
            typedef void (*VuoImageTextCacheFiniType)(void);
            VuoImageTextCacheFiniType vuoImageTextCacheFini = (VuoImageTextCacheFiniType) dlsym(RTLD_DEFAULT, "VuoImageTextCache_fini");
            if (vuoImageTextCacheFini)
                vuoImageTextCacheFini();
        }
 
        VuoHeap_report();
 
        vuoRemoveCompositionStateFromThreadLocalStorage();
    }
}
 
void VuoRuntimeState::stopCompositionAsOrderedByComposition(void)
{
    persistentState->setStopRequested(true);
 
    dispatch_async(persistentState->communicator->getControlQueue(), ^{
                       if (persistentState->communicator->hasZmqConnection())
                       {
                           persistentState->communicator->sendStopRequested();
 
                           // If we haven't received a response to VuoTelemetryStopRequested within 2 seconds, stop anyway.
                           waitForStopTimer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, DISPATCH_TIMER_STRICT, persistentState->communicator->getControlQueue());
                           dispatch_source_set_timer(waitForStopTimer, dispatch_time(DISPATCH_TIME_NOW, NSEC_PER_SEC * 2.), NSEC_PER_SEC * 2, NSEC_PER_SEC/10);
                           waitForStopCanceledSemaphore = dispatch_semaphore_create(0);
 
                           dispatch_source_set_event_handler(waitForStopTimer, ^{
                               stopComposition(false, 5);
                               dispatch_source_cancel(waitForStopTimer);
                           });
 
                           dispatch_source_set_cancel_handler(waitForStopTimer, ^{
                               dispatch_semaphore_signal(waitForStopCanceledSemaphore);
                           });
 
                           dispatch_resume(waitForStopTimer);
                       }
                       else
                       {
                           stopComposition(false, 5);
                       }
                   });
 
    persistentState->communicator->interruptListeningForControl();
}
 
void VuoRuntimeState::stopComposition(bool isBeingReplaced, int timeoutInSeconds)
{
    dispatch_retain(stopQueue);  // Prevent "Release of a locked queue" error if fini() releases stopQueue while this function is on it (https://b33p.net/kosada/node/15438)
 
    dispatch_sync(stopQueue, ^{
 
        // If we're stopping due to a user request, and we've already stopped, don't try to stop again.
        // (The 2-second timer in stopCompositionAsOrderedByComposition() might ding while the previous call is still in progress.)
        if (wasStopCompositionCalled)
            return;
        wasStopCompositionCalled = true;
 
        killProcessAfterTimeout(timeoutInSeconds);
 
        dispatch_sync(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
            if (hasBeenUnpaused)
            {
                if (!_isPaused)
                {
                    _isPaused = true;
                    vuoInstanceTriggerStop();
                }
 
                vuoInstanceFini();  // Called on a non-main thread to avoid deadlock with VuoRuntimeCommunicator::sendTelemetry().
            }
        });
 
        if (!isBeingReplaced)
            persistentState->callFiniCallbacks();
 
        vuoCleanup();
 
        persistentState->threadManager->disableSchedulingWorkers();
 
        if (persistentState->communicator->hasZmqConnection())
            persistentState->communicator->stopListeningForControl();
        else
            breakOutOfEventLoop();
    });
 
    dispatch_release(stopQueue);
}
 
void VuoRuntimeState::breakOutOfEventLoop(void)
{
    _isStopped = true;
    VuoEventLoop_break();
}
 
void VuoRuntimeState::killProcessAfterTimeout(int timeoutInSeconds)
{
    if (timeoutInSeconds < 0)
        return;
 
    if (runnerPid == getpid())
        // If the runner is in the same process as the composition (and possibly other current-process compositions),
        // it would be overkill to kill all of it.
        return;
 
    dispatch_after(dispatch_time(DISPATCH_TIME_NOW, timeoutInSeconds * NSEC_PER_SEC),
                   dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
 
        // Since below we need to use the main thread to check whether the composition can be terminated,
        // set a backup timer that fires if the main thread isn't responding.
        dispatch_source_t backupTimer = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, DISPATCH_TIMER_STRICT, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0));
        dispatch_source_set_timer(backupTimer, dispatch_time(DISPATCH_TIME_NOW, timeoutInSeconds * NSEC_PER_SEC), timeoutInSeconds * NSEC_PER_SEC, NSEC_PER_SEC/10);
        dispatch_source_set_event_handler(backupTimer, ^{
            persistentState->communicator->sendCompositionStoppingAndCloseControl();
            VUserLog("Warning: Waited %d seconds for the composition to cleanly shut down, but it's still running. Now I'm force-quitting it.", timeoutInSeconds * 2);
            kill(getpid(), SIGKILL);
        });
        dispatch_resume(backupTimer);
 
        __block bool eventLoopMayBeTerminated;
        dispatch_sync(dispatch_get_main_queue(), ^{
            eventLoopMayBeTerminated = VuoEventLoop_mayBeTerminated();
        });
        if (mayBeTerminated() && eventLoopMayBeTerminated)
        {
            persistentState->communicator->sendCompositionStoppingAndCloseControl();
            VUserLog("Warning: Waited %d seconds for the composition to cleanly shut down, but it's still running. Now I'm force-quitting it.", timeoutInSeconds);
            kill(getpid(), SIGKILL);
        }
 
        dispatch_source_cancel(backupTimer);
        dispatch_release(backupTimer);
    });
}
 
extern "C"
{
 
bool vuoIsPaused(VuoCompositionState *compositionState)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    return runtimeState->isPaused();
}
 
}  // extern "C"