VuoRunner.cc

Go to the documentation of this file.

 
#include "VuoRunner.hh"
#include "VuoFileUtilities.hh"
#include "VuoStringUtilities.hh"
#include "VuoEventLoop.h"
#include "VuoException.hh"
#include "VuoRuntime.h"
#include "VuoRunningCompositionLibraries.hh"
 
#include <CoreServices/CoreServices.h>
#include <pthread.h>
#include <stdio.h>
#include <dlfcn.h>
#include <sstream>
#include <copyfile.h>
#include <mach-o/loader.h>
#include <sys/proc_info.h>
#include <sys/stat.h>
 
void *VuoApp_mainThread = NULL; 
static const char *mainThreadChecker = "/Applications/Xcode.app/Contents/Developer/usr/lib/libMainThreadChecker.dylib";  
static int compositionReadRunnerWritePipe[2];  
static bool VuoRunner_isHostVDMX = false;  
 
static void VuoRunner_closeOnExec(int fd)
{
    int flags = fcntl(fd, F_GETFD);
    if (flags < 0)
    {
        VUserLog("Error: Couldn't get flags for desciptor %d: %s", fd, strerror(errno));
        return;
    }
 
    flags |= FD_CLOEXEC;
 
    if (fcntl(fd, F_SETFD, flags) != 0)
        VUserLog("Error: Couldn't set FD_CLOEXEC on descriptor %d: %s", fd, strerror(errno));
}
 
static void __attribute__((constructor)) VuoRunner_init()
{
    VuoApp_mainThread = (void *)pthread_self();
 
#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wdeprecated-declarations"
    // Calls _TSGetMainThread().
    // https://b33p.net/kosada/node/12944
    YieldToAnyThread();
#pragma clang diagnostic pop
 
    pipe(compositionReadRunnerWritePipe);
 
    // Ensure that the write end of this pipe gets closed upon fork()/exec(),
    // so child processes don't prop open this pipe,
    // which would prevent Vuo compositions from quitting when the VuoRunner process quits.
    VuoRunner_closeOnExec(compositionReadRunnerWritePipe[1]);
 
    if (VuoStringUtilities::makeFromCFString(CFBundleGetIdentifier(CFBundleGetMainBundle())) == "com.vidvox.VDMX5")
        VuoRunner_isHostVDMX = true;
}
 
static bool isMainThread(void)
{
    return VuoApp_mainThread == (void *)pthread_self();
}
 
static void VuoRunner_configureSocket(void *zmqSocket)
{
    int linger = 0;  // avoid having zmq_term block if the runner has tried to send a message on a broken connection
    zmq_setsockopt(zmqSocket, ZMQ_LINGER, &linger, sizeof linger);
}
 
class VuoRunner::Private
{
public:
    Private() :
        lastWidth(0),
        lastHeight(0)
    {
    }
 
    once_flag vuoImageFunctionsInitialized;  
    typedef void *(*vuoImageMakeFromJsonWithDimensionsType)(json_object *, unsigned int, unsigned int);  
    vuoImageMakeFromJsonWithDimensionsType vuoImageMakeFromJsonWithDimensions;  
    typedef json_object *(*vuoImageGetInterprocessJsonType)(void *);  
    vuoImageGetInterprocessJsonType vuoImageGetInterprocessJson;  
 
    uint64_t lastWidth;   
    uint64_t lastHeight;  
};
 
VuoRunner * VuoRunner::newSeparateProcessRunnerFromExecutable(string executablePath, string sourceDir,
                                                              bool continueIfRunnerDies, bool deleteExecutableWhenFinished)
{
    VuoRunner * vr = new VuoRunner();
    vr->executablePath = executablePath;
    vr->shouldContinueIfRunnerDies = continueIfRunnerDies;
    vr->shouldDeleteBinariesWhenFinished = deleteExecutableWhenFinished;
    vr->sourceDir = sourceDir;
    return vr;
}
 
VuoRunner * VuoRunner::newSeparateProcessRunnerFromDynamicLibrary(string compositionLoaderPath, string compositionDylibPath,
                                                                  const std::shared_ptr<VuoRunningCompositionLibraries> &runningCompositionLibraries,
                                                                  string sourceDir, bool continueIfRunnerDies, bool deleteDylibsWhenFinished)
{
    VuoRunner * vr = new VuoRunner();
    vr->executablePath = compositionLoaderPath;
    vr->dylibPath = compositionDylibPath;
    vr->dependencyLibraries = runningCompositionLibraries;
    vr->sourceDir = sourceDir;
    vr->shouldContinueIfRunnerDies = continueIfRunnerDies;
    vr->shouldDeleteBinariesWhenFinished = deleteDylibsWhenFinished;
    runningCompositionLibraries->setDeleteResourceLibraries(deleteDylibsWhenFinished);
    return vr;
}
 
VuoRunner * VuoRunner::newCurrentProcessRunnerFromDynamicLibrary(string dylibPath, string sourceDir,
                                                                 bool deleteDylibWhenFinished)
{
    VuoRunner * vr = new VuoRunner();
    vr->dylibPath = dylibPath;
    vr->shouldDeleteBinariesWhenFinished = deleteDylibWhenFinished;
    vr->sourceDir = sourceDir;
    return vr;
}
 
VuoRunner::~VuoRunner(void)
{
    dispatch_release(stoppedSemaphore);
    dispatch_release(terminatedZMQContextSemaphore);
    dispatch_release(beganListeningSemaphore);
    dispatch_release(endedListeningSemaphore);
    dispatch_release(lastFiredEventSemaphore);
    dispatch_release(delegateQueue);
    delete p;
}
 
void VuoRunner::setRuntimeChecking(bool runtimeCheckingEnabled)
{
    if (!stopped)
    {
        VUserLog("Error: Only call VuoRunner::setRuntimeChecking() prior to starting the composition.");
        return;
    }
 
    isRuntimeCheckingEnabled = runtimeCheckingEnabled && VuoFileUtilities::fileExists(mainThreadChecker);
}
 
VuoRunner::VuoRunner(void)
{
    p = new Private;
    dylibHandle = NULL;
    dependencyLibraries = NULL;
    shouldContinueIfRunnerDies = false;
    shouldDeleteBinariesWhenFinished = false;
    isRuntimeCheckingEnabled         = false;
    paused = true;
    stopped = true;
    lostContact = false;
    listenCanceled = false;
    stoppedSemaphore = dispatch_semaphore_create(1);
    terminatedZMQContextSemaphore = dispatch_semaphore_create(0);
    beganListeningSemaphore = dispatch_semaphore_create(0);
    endedListeningSemaphore = dispatch_semaphore_create(1);
    lastFiredEventSemaphore = dispatch_semaphore_create(0);
    lastFiredEventSignaled = false;
    controlQueue = dispatch_queue_create("org.vuo.runner.control", NULL);
    ZMQContext = NULL;
    ZMQSelfSend = NULL;
    ZMQSelfReceive = NULL;
    ZMQControl = NULL;
    ZMQTelemetry = NULL;
    ZMQLoaderControl = NULL;
    delegate = NULL;
    delegateQueue = dispatch_queue_create("org.vuo.runner.delegate", NULL);
    arePublishedInputPortsCached = false;
    arePublishedOutputPortsCached = false;
 
    static once_flag sleepHandlersInstalled;
    call_once(sleepHandlersInstalled, [](){
        VuoEventLoop_installSleepHandlers();
    });
}
 
void VuoRunner::start(void)
{
    try
    {
        startInternal();
 
        if (isInCurrentProcess())
        {
            dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
            dispatch_async(queue, ^{
                               unpause();
                           });
            while (paused)
            {
                drainMainDispatchQueue();
                usleep(USEC_PER_SEC / 1000);
            }
        }
        else
        {
            unpause();
        }
    }
    catch (VuoException &e)
    {
        stopBecauseLostContact(e.what());
    }
}
 
void VuoRunner::startPaused(void)
{
    try
    {
        startInternal();
    }
    catch (VuoException &e)
    {
        stopBecauseLostContact(e.what());
    }
}
 
void VuoRunner::copyDylibAndChangeId(string dylibPath, string &outputDylibPath)
{
    string directory, file, extension;
    VuoFileUtilities::splitPath(dylibPath, directory, file, extension);
 
    const int makeTmpFileExtension = 7;
    if (file.length() > makeTmpFileExtension)
    {
        // makeTmpFile() appends "-XXXXXX"; make room for that.
        string trimmedFile = file.substr(0, file.length() - makeTmpFileExtension);
 
        bool alreadyLoaded;
        do
        {
            outputDylibPath = VuoFileUtilities::makeTmpFile(trimmedFile, "dylib");
            alreadyLoaded = dlopen(outputDylibPath.c_str(), RTLD_NOLOAD);
        } while (alreadyLoaded);
    }
    else
    {
        // For short names, like those generated by VDMX, just replace the entire name with a hash.
        // https://b33p.net/kosada/node/12917
        bool alreadyLoaded;
        do
        {
            string hash = VuoStringUtilities::makeRandomHash(file.length());
            outputDylibPath = "/tmp/" + hash + ".dylib";
            alreadyLoaded = dlopen(outputDylibPath.c_str(), RTLD_NOLOAD);
        } while (alreadyLoaded);
    }
 
    string newDirectory, newFile, newExtension;
    VuoFileUtilities::splitPath(outputDylibPath, newDirectory, newFile, newExtension);
 
    if (newFile.length() > file.length())
        throw VuoException("The composition couldn't start because the uniqued dylib name (" + newFile + ") is longer than the original dylib name (" + file + ").");
 
    if (copyfile(dylibPath.c_str(), outputDylibPath.c_str(), NULL, COPYFILE_ALL))
        throw VuoException("The composition couldn't start because a copy of the dylib couldn't be made.");
 
    FILE *fp = fopen(outputDylibPath.c_str(), "r+b");
    if (!fp)
        throw VuoException("The composition couldn't start because the dylib's header couldn't be opened.");
    VuoDefer(^{ fclose(fp); });
 
    struct mach_header_64 header;
    if (fread(&header, sizeof(header), 1, fp) != 1)
        throw VuoException("The composition couldn't start because the dylib's header couldn't be read.");
 
    if (header.magic != MH_MAGIC_64
     || header.cputype != CPU_TYPE_X86_64)
        throw VuoException("The composition couldn't start because the dylib isn't an x86_64-only (non-fat) Mach-O binary.");
 
    for (int i = 0; i < header.ncmds; ++i)
    {
        struct load_command lc;
        if (fread(&lc, sizeof(lc), 1, fp) != 1)
            throw VuoException("The composition couldn't start because the dylib's command couldn't be read.");
 
        // VLog("cmd[%d]: %x (size %d)",i,lc.cmd,lc.cmdsize);
        if (lc.cmd == LC_ID_DYLIB)
        {
            fseek(fp, sizeof(struct dylib), SEEK_CUR);
 
            size_t nameLength = lc.cmdsize - sizeof(struct dylib_command);
            char *name = (char *)calloc(nameLength + 1, 1);
            if (fread(name, nameLength, 1, fp) != 1)
                throw VuoException("The composition couldn't start because the dylib's ID command couldn't be read.");
 
//          VLog("Changing name \"%s\" to \"%s\"…", name, outputDylibPath.c_str());
            fseek(fp, -nameLength, SEEK_CUR);
            bzero(name, nameLength);
            memcpy(name, outputDylibPath.c_str(), min(nameLength, outputDylibPath.length()));
            fwrite(name, nameLength, 1, fp);
            return;
        }
        else
            fseek(fp, lc.cmdsize-sizeof(lc), SEEK_CUR);
    }
 
    throw VuoException("The composition couldn't start because the dylib's LC_ID_DYLIB command couldn't be found.");
}
 
void VuoRunner::startInternal(void)
{
    stopped = false;
    dispatch_semaphore_wait(stoppedSemaphore, DISPATCH_TIME_FOREVER);
 
    ZMQContext = zmq_init(1);
 
    if (isInCurrentProcess())
    {
        // Start the composition in the current process.
 
        bool alreadyLoaded = dlopen(dylibPath.c_str(), RTLD_NOLOAD);
        if (alreadyLoaded)
        {
            // Each composition instance needs its own global variables.
            // Change the dylib's internal name, to convince dlopen() to load another instance of it.
 
            string uniquedDylibPath;
            copyDylibAndChangeId(dylibPath, uniquedDylibPath);
            VDebugLog("\"%s\" is already loaded, so I duplicated it and changed its LC_ID_DYLIB to \"%s\".", dylibPath.c_str(), uniquedDylibPath.c_str());
 
            if (shouldDeleteBinariesWhenFinished)
                remove(dylibPath.c_str());
 
            dylibPath = uniquedDylibPath;
            shouldDeleteBinariesWhenFinished = true;
        }
 
        dylibHandle = dlopen(dylibPath.c_str(), RTLD_NOW);
        if (!dylibHandle)
            throw VuoException("The composition couldn't start because the library '" + dylibPath + "' couldn't be loaded : " + dlerror());
 
        try
        {
            VuoInitInProcessType *vuoInitInProcess = (VuoInitInProcessType *)dlsym(dylibHandle, "vuoInitInProcess");
            if (! vuoInitInProcess)
                throw VuoException("The composition couldn't start because vuoInitInProcess() couldn't be found in '" + dylibPath + "' : " + dlerror());
 
            ZMQControlURL = "inproc://" + VuoFileUtilities::makeTmpFile("vuo-control", "");
            ZMQTelemetryURL = "inproc://" + VuoFileUtilities::makeTmpFile("vuo-telemetry", "");
 
            vuoInitInProcess(ZMQContext, ZMQControlURL.c_str(), ZMQTelemetryURL.c_str(), true, getpid(), -1, false,
                             sourceDir.c_str(), dylibHandle, NULL, false);
        }
        catch (VuoException &e)
        {
            VUserLog("error: %s", e.what());
            dlclose(dylibHandle);
            dylibHandle = NULL;
            throw;
        }
    }
    else
    {
        // Start the composition or composition loader in a new process.
 
        vector<string> args;
 
        string executableName;
        if (isUsingCompositionLoader())
        {
            // If we're using the loader, set the executable's display name to the dylib,
            // so that composition's name shows up in the process list.
            string dir, file, ext;
            VuoFileUtilities::splitPath(dylibPath, dir, file, ext);
            executableName = file;
        }
        else
        {
            string dir, file, ext;
            VuoFileUtilities::splitPath(executablePath, dir, file, ext);
            string executableName = file;
            if (! ext.empty())
                executableName += "." + ext;
        }
        args.push_back(executableName);
 
        // https://b33p.net/kosada/node/16374
        // The socket's full pathname (`sockaddr_un::sun_path`) must be 104 characters or less
        // (https://opensource.apple.com/source/xnu/xnu-2782.1.97/bsd/sys/un.h.auto.html).
        // "/Users/me/Library/Containers/com.apple.ScreenSaver.Engine.legacyScreenSaver/Data/vuo-telemetry-rr8Br3"
        // is 101 characters, which limits the username to 5 characters.
        // "/Users/me/Library/Containers/com.apple.ScreenSaver.Engine.legacyScreenSaver/Data/v-rr8Br3"
        // is 89 characters, which limits the username to 17 characters
        // (still not a lot, but more likely to work with typical macOS usernames).
        ZMQControlURL = "ipc://" + VuoFileUtilities::makeTmpFile("v", "");
        ZMQTelemetryURL = "ipc://" + VuoFileUtilities::makeTmpFile("v", "");
        args.push_back("--vuo-control=" + ZMQControlURL);
        args.push_back("--vuo-telemetry=" + ZMQTelemetryURL);
 
        {
            ostringstream oss;
            oss << getpid();
            args.push_back("--vuo-runner-pid=" + oss.str());
        }
 
        {
            ostringstream oss;
            oss << compositionReadRunnerWritePipe[0];
            args.push_back("--vuo-runner-pipe=" + oss.str());
        }
 
        if (shouldContinueIfRunnerDies)
            args.push_back("--vuo-continue-if-runner-dies");
 
        if (isUsingCompositionLoader())
        {
            ZMQLoaderControlURL = "ipc://" + VuoFileUtilities::makeTmpFile("v", "");
            args.push_back("--vuo-loader=" + ZMQLoaderControlURL);
        }
        else
            args.push_back("--vuo-pause");
 
        int fd[2];
        int ret = pipe(fd);
        if (ret)
            throw VuoException("The composition couldn't start because a pipe couldn't be opened : " + string(strerror(errno)));
 
        int argSize = args.size();
        char *argv[argSize + 1];
        for (size_t i = 0; i < argSize; ++i)
        {
            size_t mallocSize = args[i].length() + 1;
            argv[i]           = (char *)malloc(mallocSize);
            strlcpy(argv[i], args[i].c_str(), mallocSize);
        }
        argv[argSize] = NULL;
 
        string errorWorkingDirectory = "The composition couldn't start because the working directory couldn't be changed to '" + sourceDir + "' : ";
        string errorExecutable = "The composition couldn't start because the file '" + executablePath + "' couldn't be executed : ";
        string errorFork = "The composition couldn't start because the composition process couldn't be forked : ";
        const size_t ERROR_BUFFER_LEN = 256;
        char errorBuffer[ERROR_BUFFER_LEN];
 
        pipe(runnerReadCompositionWritePipe);
 
        pid_t childPid = fork();
        if (childPid == 0)
        {
            // There are only a limited set of functions you're allowed to call in the child process
            // after fork() and before exec(). Functions such as VUserLog() and exit() aren't allowed,
            // so instead we're calling alternatives such as write() and _exit().
 
            close(runnerReadCompositionWritePipe[0]);
 
            pid_t grandchildPid = fork();
            if (grandchildPid == 0)
            {
                close(fd[0]);
                close(fd[1]);
 
                // Set the current working directory to that of the source .vuo composition so that
                // relative URL paths are resolved correctly.
                if (!sourceDir.empty())
                {
                    ret = chdir(sourceDir.c_str());
                    if (ret)
                    {
                        strerror_r(errno, errorBuffer, ERROR_BUFFER_LEN);
                        write(STDERR_FILENO, errorWorkingDirectory.c_str(), errorWorkingDirectory.length());
                        write(STDERR_FILENO, errorBuffer, strlen(errorBuffer));
                        write(STDERR_FILENO, "\n", 1);
                        _exit(-1);
                    }
                }
 
                if (isRuntimeCheckingEnabled)
                {
                    const char *vuoRuntimeChecking = getenv("VUO_RUNTIME_CHECKING");
                    if (vuoRuntimeChecking)
                        setenv("DYLD_INSERT_LIBRARIES", vuoRuntimeChecking, 1);
                    else
                        setenv("DYLD_INSERT_LIBRARIES", mainThreadChecker, 1);
                }
 
                ret = execv(executablePath.c_str(), argv);
                if (ret)
                {
                    strerror_r(errno, errorBuffer, ERROR_BUFFER_LEN);
                    write(STDERR_FILENO, errorExecutable.c_str(), errorExecutable.length());
                    write(STDERR_FILENO, errorBuffer, strlen(errorBuffer));
                    write(STDERR_FILENO, "\n", 1);
                    for (size_t i = 0; i < argSize; ++i)
                        free(argv[i]);
                    _exit(-1);
                }
            }
            else if (grandchildPid > 0)
            {
                close(fd[0]);
 
                int ret = write(fd[1], &grandchildPid, sizeof(pid_t));
                if (ret != sizeof(pid_t))
                {
                    strerror_r(errno, errorBuffer, ERROR_BUFFER_LEN);
                    write(STDERR_FILENO, errorExecutable.c_str(), errorExecutable.length());
                    write(STDERR_FILENO, errorBuffer, strlen(errorBuffer));
                    write(STDERR_FILENO, "\n", 1);
                }
                close(fd[1]);
 
                _exit(0);
            }
            else
            {
                close(fd[0]);
                close(fd[1]);
 
                strerror_r(errno, errorBuffer, ERROR_BUFFER_LEN);
                write(STDERR_FILENO, errorFork.c_str(), errorFork.length());
                write(STDERR_FILENO, errorBuffer, strlen(errorBuffer));
                write(STDERR_FILENO, "\n", 1);
                _exit(-1);
            }
        }
        else if (childPid > 0)
        {
            close(fd[1]);
 
            // If this process launches compositions in addition to this one,
            // ensure they don't prop open this pipe,
            // which would prevent VuoRunner::stop's `read()` from terminating.
            VuoRunner_closeOnExec(runnerReadCompositionWritePipe[1]);
 
            for (size_t i = 0; i < argSize; ++i)
                free(argv[i]);
 
            pid_t grandchildPid = 0;
            int ret = read(fd[0], &grandchildPid, sizeof(pid_t));
            if (ret != sizeof(pid_t))
                throw VuoException("The composition couldn't start because the composition process id couldn't be obtained: " + string(strerror(errno)));
            close(fd[0]);
 
            // Reap the child process.
            int status;
            do {
                ret = waitpid(childPid, &status, 0);
            } while (ret == -1 && errno == EINTR);
            if (WIFEXITED(status) && WEXITSTATUS(status))
                throw VuoException("The composition couldn't start because the parent of the composition process exited with an error.");
            else if (WIFSIGNALED(status))
                throw VuoException("The composition couldn't start because the parent of the composition process exited abnormally : " + string(strsignal(WTERMSIG(status))));
 
            if (grandchildPid > 0)
                compositionPid = grandchildPid;
            else
                throw VuoException("The composition couldn't start because the composition process id couldn't be obtained");
        }
        else
        {
            for (size_t i = 0; i < argSize; ++i)
                free(argv[i]);
 
            throw VuoException("The composition couldn't start because the parent of the composition process couldn't be forked : " + string(strerror(errno)));
        }
    }
 
    // Connect to the composition loader (if any) and composition.
    if (isUsingCompositionLoader())
    {
        ZMQLoaderControl = zmq_socket(ZMQContext,ZMQ_REQ);
        VuoRunner_configureSocket(ZMQLoaderControl);
 
        // Try to connect to the composition loader. If at first we don't succeed, give the composition loader a little more time to set up the socket.
        int numTries = 0;
        while (zmq_connect(ZMQLoaderControl,ZMQLoaderControlURL.c_str()))
        {
            if (++numTries == 1000)
                throw VuoException("The composition couldn't start because the runner couldn't establish communication with the composition loader : " + string(strerror(errno)));
            usleep(USEC_PER_SEC / 1000);
        }
 
        // Actually start the composition.
        // Since we don't know how long this will take, the socket has an infinite timeout (https://b33p.net/kosada/vuo/vuo/-/issues/18450).
        replaceComposition(dylibPath, "");
    }
    else
    {
        __block string errorMessage;
        dispatch_sync(controlQueue, ^{
                          try {
                              setUpConnections();
                          } catch (VuoException &e) {
                              errorMessage = e.what();
                          }
                      });
        if (! errorMessage.empty())
            throw VuoException(errorMessage);
    }
}
 
void *VuoRunner_listen(void *context)
{
    pthread_detach(pthread_self());
    VuoRunner *runner = static_cast<VuoRunner *>(context);
    runner->listen();
    return NULL;
}
 
void VuoRunner::setUpConnections(void)
{
    ZMQControl = zmq_socket(ZMQContext,ZMQ_REQ);
    VuoRunner_configureSocket(ZMQControl);
 
    // Try to connect to the composition. If at first we don't succeed, give the composition a little more time to set up the socket.
    int numTries = 0;
    while (zmq_connect(ZMQControl,ZMQControlURL.c_str()))
    {
        if (++numTries == 1000)
            throw VuoException("The composition couldn't start because the runner couldn't establish communication to control the composition : " + string(strerror(errno)));
        usleep(USEC_PER_SEC / 1000);
    }
 
    // Cache published ports so they're available whenever a caller starts listening for published port value changes.
    arePublishedInputPortsCached = false;
    arePublishedOutputPortsCached = false;
    if (isInCurrentProcess())
    {
        dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
        __block string publishedPortsError;
        dispatch_async(queue, ^{
                           try {
                               getCachedPublishedPorts(false);
                               getCachedPublishedPorts(true);
                           } catch (VuoException &e) {
                               publishedPortsError = e.what();
                           }
                       });
        while (! (arePublishedInputPortsCached && arePublishedOutputPortsCached) )
        {
            drainMainDispatchQueue();
            usleep(USEC_PER_SEC / 1000);
 
            if (! publishedPortsError.empty())
                throw VuoException(publishedPortsError);
        }
    }
    else
    {
        getCachedPublishedPorts(false);
        getCachedPublishedPorts(true);
    }
 
    listenCanceled = false;
    dispatch_semaphore_wait(endedListeningSemaphore, DISPATCH_TIME_FOREVER);
 
    pthread_t listenThread;
    int ret = pthread_create(&listenThread, nullptr, &VuoRunner_listen, this);
    if (ret)
        throw VuoException(string("The composition couldn't start because the runner couldn't create a thread: ") + strerror(ret));
 
    dispatch_semaphore_wait(beganListeningSemaphore, DISPATCH_TIME_FOREVER);
    if (!listenError.empty())
        throw VuoException("The composition couldn't start because the runner couldn't establish communication to listen to the composition: " + listenError);
 
    vuoControlRequestSend(VuoControlRequestSlowHeartbeat,NULL,0);
    vuoControlReplyReceive(VuoControlReplyHeartbeatSlowed);
}
 
void VuoRunner::runOnMainThread(void)
{
    if (! isInCurrentProcess())
        throw VuoException("The composition is not running in the current process. Only use this function if the composition was constructed with newCurrentProcessRunnerFromDynamicLibrary().");
 
    if (! isMainThread())
        throw VuoException("This is not the main thread. Only call this function from the main thread.");
 
    while (! stopped)
        VuoEventLoop_processEvent(VuoEventLoop_WaitIndefinitely);
}
 
void VuoRunner::drainMainDispatchQueue(void)
{
    if (! isInCurrentProcess())
        throw VuoException("The composition is not running in the current process. Only use this function if the composition was constructed with newCurrentProcessRunnerFromDynamicLibrary().");
 
    if (! isMainThread())
        throw VuoException("This is not the main thread. Only call this function from the main thread.");
 
    VuoEventLoop_processEvent(VuoEventLoop_RunOnce);
}
 
void VuoRunner::pause(void)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          vuoControlRequestSend(VuoControlRequestCompositionPause,NULL,0);
                          vuoControlReplyReceive(VuoControlReplyCompositionPaused);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
 
                      paused = true;
                  });
}
 
void VuoRunner::unpause(void)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          vuoControlRequestSend(VuoControlRequestCompositionUnpause,NULL,0);
                          vuoControlReplyReceive(VuoControlReplyCompositionUnpaused);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
 
                      paused = false;
                  });
}
 
void VuoRunner::replaceComposition(string compositionDylibPath, string compositionDiff)
{
    if (! isUsingCompositionLoader())
        throw VuoException("The runner is not using a composition loader. Only use this function if the composition was constructed with newSeparateProcessRunnerFromDynamicLibrary().");
 
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      VUserLog("Loading composition…");
 
                      if (dylibPath != compositionDylibPath)
                      {
                          if (shouldDeleteBinariesWhenFinished)
                          {
                              remove(dylibPath.c_str());
                          }
 
                          dylibPath = compositionDylibPath;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          if (! paused)
                          {
                              VUserLog("    Pausing…");
                              vuoControlRequestSend(VuoControlRequestCompositionPause,NULL,0);
                              vuoControlReplyReceive(VuoControlReplyCompositionPaused);
                          }
 
                          cleanUpConnections();
 
                          vector<string> dependencyDylibPathsRemoved = dependencyLibraries->dequeueLibrariesToUnload();
                          vector<string> dependencyDylibPathsAdded = dependencyLibraries->dequeueLibrariesToLoad();
 
                          unsigned int messageCount = 4 + dependencyDylibPathsAdded.size() + dependencyDylibPathsRemoved.size();
                          zmq_msg_t *messages = (zmq_msg_t *)malloc(messageCount * sizeof(zmq_msg_t));
                          int index = 0;
 
                          vuoInitMessageWithString(&messages[index++], dylibPath.c_str());
 
                          vuoInitMessageWithInt(&messages[index++], dependencyDylibPathsAdded.size());
                          for (vector<string>::iterator i = dependencyDylibPathsAdded.begin(); i != dependencyDylibPathsAdded.end(); ++i) {
                              vuoInitMessageWithString(&messages[index++], (*i).c_str());
                          }
 
                          vuoInitMessageWithInt(&messages[index++], dependencyDylibPathsRemoved.size());
                          for (vector<string>::iterator i = dependencyDylibPathsRemoved.begin(); i != dependencyDylibPathsRemoved.end(); ++i) {
                              vuoInitMessageWithString(&messages[index++], (*i).c_str());
                          }
 
                          vuoInitMessageWithString(&messages[index], compositionDiff.c_str());
 
                          if (! paused)
                              VUserLog("    Replacing composition…");
 
                          vuoLoaderControlRequestSend(VuoLoaderControlRequestCompositionReplace,messages,messageCount);
                          vuoLoaderControlReplyReceive(VuoLoaderControlReplyCompositionReplaced);
 
                          setUpConnections();
 
                          if (! paused)
                          {
                              VUserLog("    Unpausing…");
                              vuoControlRequestSend(VuoControlRequestCompositionUnpause,NULL,0);
                              vuoControlReplyReceive(VuoControlReplyCompositionUnpaused);
                          }
 
                          VUserLog("    Done.");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::stop(void)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      // Only tell the composition to stop if it hasn't already ended on its own.
                      if (! lostContact)
                      {
                          try
                          {
                              int timeoutInSeconds = (isInCurrentProcess() ? -1 : 5);
                              zmq_msg_t messages[3];
                              vuoInitMessageWithInt(&messages[0], timeoutInSeconds);
                              vuoInitMessageWithBool(&messages[1], false); // isBeingReplaced
                              vuoInitMessageWithBool(&messages[2], !isInCurrentProcess()); // isLastEverInProcess
                              vuoControlRequestSend(VuoControlRequestCompositionStop, messages, 3);
 
                              if (isInCurrentProcess() && isMainThread())
                              {
                                  // If VuoRunner::stop() is blocking the main thread, wait for the composition's reply on another thread, and drain the main queue, in case the composition needs to shut down stuff that requires the main queue.
                                  __block bool replyReceived = false;
                                  dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
                                      vuoMemoryBarrier();
                                      try
                                      {
                                          vuoControlReplyReceive(VuoControlReplyCompositionStopping);
                                      }
                                      catch (...)
                                      {
                                          // do nothing; doesn't matter if connection timed out
                                      }
                                      replyReceived = true;
                                  });
                                  while (!replyReceived)
                                  {
                                      drainMainDispatchQueue();
                                      usleep(USEC_PER_SEC / 1000);
                                  }
                                  vuoMemoryBarrier();
                              }
                              else
                                  vuoControlReplyReceive(VuoControlReplyCompositionStopping);
                          }
                          catch (...)
                          {
                              // do nothing; doesn't matter if connection timed out
                          }
                      }
 
                      cleanUpConnections();
 
                      if (isUsingCompositionLoader() && ZMQLoaderControl)
                      {
                          zmq_close(ZMQLoaderControl);
                          ZMQLoaderControl = NULL;
                      }
 
                      if (isInCurrentProcess() && dylibHandle)
                      {
                          VuoInitInProcessType *vuoInitInProcess = (VuoInitInProcessType *)dlsym(dylibHandle, "vuoInitInProcess");
                          if (vuoInitInProcess)  // Avoid double jeopardy if startInternal() already failed for missing vuoInitInProcess.
                          {
                              VuoFiniType *vuoFini = (VuoFiniType *)dlsym(dylibHandle, "vuoFini");
                              if (! vuoFini)
                              {
                                  VUserLog("The composition couldn't stop because vuoFini() couldn't be found in '%s' : %s", dylibPath.c_str(), dlerror());
                                  return;
                              }
                              void *runtimeState = vuoFini();
 
                              VuoFiniRuntimeStateType *vuoFiniRuntimeState = (VuoFiniRuntimeStateType *)dlsym(dylibHandle, "vuoFiniRuntimeState");
                              if (! vuoFiniRuntimeState)
                              {
                                  VUserLog("The composition couldn't stop because vuoFiniRuntimeState() couldn't be found in '%s' : %s", dylibPath.c_str(), dlerror());
                                  return;
                              }
                              vuoFiniRuntimeState(runtimeState);
                          }
 
                          dlclose(dylibHandle);
                          dylibHandle = NULL;
                      }
                      else if (isInCurrentProcess() && !dylibHandle)
                      {
                          // If the dylib isn't open, the composition isn't running, so there's nothing to clean up.
                      }
                      else
                      {
                          char buf[1];
                          close(runnerReadCompositionWritePipe[1]);
 
                          if (! lostContact)
                          {
                              // Wait for child process to end.
                              // Can't use waitpid() since it only waits on child processes, yet compositionPid is a grandchild.
                              // Instead, do a blocking read() — the grandchild never writes anything to the pipe, and when the grandchild exits,
                              // read() will return EOF (since it was the last process that had it open for writing).
                              read(runnerReadCompositionWritePipe[0], &buf, 1);
                          }
 
                          close(runnerReadCompositionWritePipe[0]);
 
                          if (! lostContact)
                          {
                              zmq_term(ZMQContext);
                              ZMQContext = NULL;
                          }
                          else
                          {
                              dispatch_semaphore_wait(terminatedZMQContextSemaphore, DISPATCH_TIME_FOREVER);
                          }
                      }
 
                      if (shouldDeleteBinariesWhenFinished)
                      {
                          if (isUsingCompositionLoader())
                          {
                              remove(dylibPath.c_str());
                          }
                          else if (isInCurrentProcess())
                          {
                              remove(dylibPath.c_str());
                          }
                          else
                          {
                              remove(executablePath.c_str());
                          }
                      }
 
                      dependencyLibraries = nullptr;  // release shared_ptr
 
                      stopped = true;
                      dispatch_semaphore_signal(stoppedSemaphore);
                      VuoEventLoop_break();
                  });
}
 
void VuoRunner::cleanUpConnections(void)
{
    if (! ZMQControl)
        return;
 
    zmq_close(ZMQControl);
    ZMQControl = NULL;
 
    if (ZMQSelfSend)
        // Break out of zmq_poll().
        vuoSend("VuoRunner::ZMQSelfSend", ZMQSelfSend, 0, nullptr, 0, false, nullptr);
 
    dispatch_semaphore_wait(endedListeningSemaphore, DISPATCH_TIME_FOREVER);
    dispatch_semaphore_signal(endedListeningSemaphore);
}
 
void VuoRunner::waitUntilStopped(void)
{
    dispatch_retain(stoppedSemaphore);
    dispatch_semaphore_wait(stoppedSemaphore, DISPATCH_TIME_FOREVER);
    dispatch_semaphore_signal(stoppedSemaphore);
    dispatch_release(stoppedSemaphore);
}
 
void VuoRunner::setInputPortValue(string compositionIdentifier, string portIdentifier, json_object *value)
{
    const char *valueAsString = json_object_to_json_string_ext(value, JSON_C_TO_STRING_PLAIN);
 
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[3];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoInitMessageWithString(&messages[2], valueAsString);
                          vuoControlRequestSend(VuoControlRequestInputPortValueModify, messages, 3);
                          vuoControlReplyReceive(VuoControlReplyInputPortValueModified);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::fireTriggerPortEvent(string compositionIdentifier, string portIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestTriggerPortFireEvent, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyTriggerPortFiredEvent);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
json_object * VuoRunner::getInputPortValue(string compositionIdentifier, string portIdentifier)
{
    __block string valueAsString;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[3];
                          vuoInitMessageWithBool(&messages[0], !isInCurrentProcess());
                          vuoInitMessageWithString(&messages[1], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[2], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestInputPortValueRetrieve, messages, 3);
                          vuoControlReplyReceive(VuoControlReplyInputPortValueRetrieved);
                          valueAsString = receiveString("null");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return json_tokener_parse(valueAsString.c_str());
}
 
json_object * VuoRunner::getOutputPortValue(string compositionIdentifier, string portIdentifier)
{
    __block string valueAsString;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[3];
                          vuoInitMessageWithBool(&messages[0], !isInCurrentProcess());
                          vuoInitMessageWithString(&messages[1], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[2], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestOutputPortValueRetrieve, messages, 3);
                          vuoControlReplyReceive(VuoControlReplyOutputPortValueRetrieved);
                          valueAsString = receiveString("null");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return json_tokener_parse(valueAsString.c_str());
}
 
string VuoRunner::getInputPortSummary(string compositionIdentifier, string portIdentifier)
{
    __block string summary;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestInputPortSummaryRetrieve, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyInputPortSummaryRetrieved);
                          summary = receiveString("");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return summary;
}
 
string VuoRunner::getOutputPortSummary(string compositionIdentifier, string portIdentifier)
{
    __block string summary;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestOutputPortSummaryRetrieve, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyOutputPortSummaryRetrieved);
                          summary = receiveString("");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return summary;
}
 
string VuoRunner::subscribeToInputPortTelemetry(string compositionIdentifier, string portIdentifier)
{
    __block string summary;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestInputPortTelemetrySubscribe, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyInputPortTelemetrySubscribed);
                          summary = receiveString("");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return summary;
}
 
string VuoRunner::subscribeToOutputPortTelemetry(string compositionIdentifier, string portIdentifier)
{
    __block string summary;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestOutputPortTelemetrySubscribe, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyOutputPortTelemetrySubscribed);
                          summary = receiveString("");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return summary;
}
 
void VuoRunner::unsubscribeFromInputPortTelemetry(string compositionIdentifier, string portIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestInputPortTelemetryUnsubscribe, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyInputPortTelemetryUnsubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::unsubscribeFromOutputPortTelemetry(string compositionIdentifier, string portIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoInitMessageWithString(&messages[1], portIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestOutputPortTelemetryUnsubscribe, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyOutputPortTelemetryUnsubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::subscribeToEventTelemetry(string compositionIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[1];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestEventTelemetrySubscribe, messages, 1);
                          vuoControlReplyReceive(VuoControlReplyEventTelemetrySubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::unsubscribeFromEventTelemetry(string compositionIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[1];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestEventTelemetryUnsubscribe, messages, 1);
                          vuoControlReplyReceive(VuoControlReplyEventTelemetryUnsubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::subscribeToAllTelemetry(string compositionIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[1];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestAllTelemetrySubscribe, messages, 1);
                          vuoControlReplyReceive(VuoControlReplyAllTelemetrySubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::unsubscribeFromAllTelemetry(string compositionIdentifier)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[1];
                          vuoInitMessageWithString(&messages[0], compositionIdentifier.c_str());
                          vuoControlRequestSend(VuoControlRequestAllTelemetryUnsubscribe, messages, 1);
                          vuoControlReplyReceive(VuoControlReplyAllTelemetryUnsubscribed);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::setPublishedInputPortValues(map<Port *, json_object *> portsAndValuesToSet)
{
    if (VuoRunner_isHostVDMX)
        for (auto i : portsAndValuesToSet)
        {
            string portName = i.first->getName();
            if (portName == "width")
                p->lastWidth = json_object_get_int64(i.second);
            else if (portName == "height")
                p->lastHeight = json_object_get_int64(i.second);
            else if (portName == "image" || portName == "startImage")
            {
                json_object *o;
                if (json_object_object_get_ex(i.second, "pixelsWide", &o))
                    p->lastWidth = json_object_get_int64(o);
                if (json_object_object_get_ex(i.second, "pixelsHigh", &o))
                    p->lastHeight = json_object_get_int64(o);
            }
        }
 
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          int messageCount = portsAndValuesToSet.size() * 2;
                          zmq_msg_t messages[messageCount];
 
                          int i = 0;
                          for (auto &kv : portsAndValuesToSet)
                          {
                              vuoInitMessageWithString(&messages[i++], kv.first->getName().c_str());
                              vuoInitMessageWithString(&messages[i++], json_object_to_json_string_ext(kv.second, JSON_C_TO_STRING_PLAIN));
                          }
 
                          vuoControlRequestSend(VuoControlRequestPublishedInputPortValueModify, messages, messageCount);
                          vuoControlReplyReceive(VuoControlReplyPublishedInputPortValueModified);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::firePublishedInputPortEvent(VuoRunner::Port *port)
{
    set<VuoRunner::Port *> portAsSet;
    portAsSet.insert(port);
    firePublishedInputPortEvent(portAsSet);
}
 
void VuoRunner::firePublishedInputPortEvent(const set<Port *> &ports)
{
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      lastFiredEventSignaled = false;
 
                      try
                      {
                          size_t messageCount = ports.size() + 1;
                          zmq_msg_t messages[messageCount];
 
                          vuoInitMessageWithInt(&messages[0], ports.size());
                          int i = 1;
                          for (VuoRunner::Port *port : ports) {
                              vuoInitMessageWithString(&messages[i++], port->getName().c_str());
                          }
 
                          vuoControlRequestSend(VuoControlRequestPublishedInputPortFireEvent, messages, messageCount);
                          vuoControlReplyReceive(VuoControlReplyPublishedInputPortFiredEvent);
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
}
 
void VuoRunner::waitForFiredPublishedInputPortEvent(void)
{
    saturating_semaphore_wait(lastFiredEventSemaphore, &lastFiredEventSignaled);
}
 
json_object * VuoRunner::getPublishedInputPortValue(VuoRunner::Port *port)
{
    __block string valueAsString;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithBool(&messages[0], !isInCurrentProcess());
                          vuoInitMessageWithString(&messages[1], port->getName().c_str());
                          vuoControlRequestSend(VuoControlRequestPublishedInputPortValueRetrieve, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyPublishedInputPortValueRetrieved);
                          valueAsString = receiveString("null");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
    return json_tokener_parse(valueAsString.c_str());
}
 
json_object * VuoRunner::getPublishedOutputPortValue(VuoRunner::Port *port)
{
    __block string valueAsString;
    dispatch_sync(controlQueue, ^{
                      if (stopped || lostContact) {
                          return;
                      }
 
                      vuoMemoryBarrier();
 
                      try
                      {
                          zmq_msg_t messages[2];
                          vuoInitMessageWithBool(&messages[0], !isInCurrentProcess());
                          vuoInitMessageWithString(&messages[1], port->getName().c_str());
                          vuoControlRequestSend(VuoControlRequestPublishedOutputPortValueRetrieve, messages, 2);
                          vuoControlReplyReceive(VuoControlReplyPublishedOutputPortValueRetrieved);
                          valueAsString = receiveString("null");
                      }
                      catch (VuoException &e)
                      {
                          stopBecauseLostContact(e.what());
                      }
                  });
 
    // https://b33p.net/kosada/node/17535
    json_object *js = json_tokener_parse(valueAsString.c_str());
    if (VuoRunner_isHostVDMX && port->getName() == "outputImage")
    {
        json_object *o;
        uint64_t actualWidth = 0;
        if (json_object_object_get_ex(js, "pixelsWide", &o))
            actualWidth = json_object_get_int64(o);
        uint64_t actualHeight = 0;
        if (json_object_object_get_ex(js, "pixelsHigh", &o))
            actualHeight = json_object_get_int64(o);
 
        if (p->lastWidth && p->lastHeight
            && (actualWidth != p->lastWidth || actualHeight != p->lastHeight))
        {
            call_once(p->vuoImageFunctionsInitialized, [=](){
                p->vuoImageMakeFromJsonWithDimensions = (Private::vuoImageMakeFromJsonWithDimensionsType)dlsym(RTLD_SELF, "VuoImage_makeFromJsonWithDimensions");
                if (!p->vuoImageMakeFromJsonWithDimensions)
                {
                    VUserLog("Error: Couldn't find VuoImage_makeFromJsonWithDimensions.");
                    return;
                }
 
                p->vuoImageGetInterprocessJson = (Private::vuoImageGetInterprocessJsonType)dlsym(RTLD_SELF, "VuoImage_getInterprocessJson");
                if (!p->vuoImageGetInterprocessJson)
                {
                    VUserLog("Error: Couldn't find VuoImage_getInterprocessJson.");
                    return;
                }
            });
 
            if (p->vuoImageMakeFromJsonWithDimensions && p->vuoImageGetInterprocessJson)
            {
                void *vi = p->vuoImageMakeFromJsonWithDimensions(js, p->lastWidth, p->lastHeight);
                return p->vuoImageGetInterprocessJson(vi);
            }
        }
    }
 
    return js;
}
 
vector<VuoRunner::Port *> VuoRunner::getCachedPublishedPorts(bool input)
{
    // Caching not only provides faster access (without zmq messages),
    // but also ensures that the VuoRunner::Port pointers passed to
    // VuoRunnerDelegate::receivedTelemetryPublishedOutputPortUpdated are consistent.
 
    if (input)
    {
        if (! arePublishedInputPortsCached)
        {
            publishedInputPorts = refreshPublishedPorts(true);
            arePublishedInputPortsCached = true;
        }
        return publishedInputPorts;
    }
    else
    {
        if (! arePublishedOutputPortsCached)
        {
            publishedOutputPorts = refreshPublishedPorts(false);
            arePublishedOutputPortsCached = true;
        }
        return publishedOutputPorts;
    }
}
 
vector<VuoRunner::Port *> VuoRunner::refreshPublishedPorts(bool input)
{
    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
    dispatch_source_t timeout = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, DISPATCH_TIMER_STRICT, queue);
    dispatch_source_set_timer(timeout, dispatch_time(DISPATCH_TIME_NOW, 5*NSEC_PER_SEC), NSEC_PER_SEC, NSEC_PER_SEC/10);
    dispatch_source_set_event_handler(timeout, ^{
                                          stopBecauseLostContact("The connection between the composition and runner timed out when trying to receive the list of published ports");
                                          dispatch_source_cancel(timeout);
    });
    dispatch_resume(timeout);
 
    vector<VuoRunner::Port *> ports;
 
    try
    {
        vuoMemoryBarrier();
 
        enum VuoControlRequest requests[4];
        enum VuoControlReply replies[4];
        if (input)
        {
            requests[0] = VuoControlRequestPublishedInputPortNamesRetrieve;
            requests[1] = VuoControlRequestPublishedInputPortTypesRetrieve;
            requests[2] = VuoControlRequestPublishedInputPortDetailsRetrieve;
            replies[0] = VuoControlReplyPublishedInputPortNamesRetrieved;
            replies[1] = VuoControlReplyPublishedInputPortTypesRetrieved;
            replies[2] = VuoControlReplyPublishedInputPortDetailsRetrieved;
        }
        else
        {
            requests[0] = VuoControlRequestPublishedOutputPortNamesRetrieve;
            requests[1] = VuoControlRequestPublishedOutputPortTypesRetrieve;
            requests[2] = VuoControlRequestPublishedOutputPortDetailsRetrieve;
            replies[0] = VuoControlReplyPublishedOutputPortNamesRetrieved;
            replies[1] = VuoControlReplyPublishedOutputPortTypesRetrieved;
            replies[2] = VuoControlReplyPublishedOutputPortDetailsRetrieved;
        }
 
        vector<string> names;
        vector<string> types;
        vector<string> details;
 
        for (int i = 0; i < 3; ++i)
        {
            vuoControlRequestSend(requests[i], NULL, 0);
            vuoControlReplyReceive(replies[i]);
            vector<string> messageStrings = receiveListOfStrings();
            if (i == 0)
                names = messageStrings;
            else if (i == 1)
                types = messageStrings;
            else
                details = messageStrings;
        }
 
        for (size_t i = 0; i < names.size() && i < types.size() && i < details.size(); ++i)
        {
            VuoRunner::Port *port = new Port(names[i], types[i], json_tokener_parse(details[i].c_str()));
            ports.push_back(port);
        }
    }
    catch (...)
    {
        dispatch_source_cancel(timeout);
        dispatch_release(timeout);
        throw;
    }
 
    dispatch_source_cancel(timeout);
    dispatch_release(timeout);
 
    return ports;
}
 
vector<VuoRunner::Port *> VuoRunner::getPublishedInputPorts(void)
{
    return getCachedPublishedPorts(true);
}
 
vector<VuoRunner::Port *> VuoRunner::getPublishedOutputPorts(void)
{
    return getCachedPublishedPorts(false);
}
 
VuoRunner::Port * VuoRunner::getPublishedInputPortWithName(string name)
{
    vector<VuoRunner::Port *> inputPorts = getPublishedInputPorts();
    for (vector<VuoRunner::Port *>::iterator i = inputPorts.begin(); i != inputPorts.end(); ++i)
        if ((*i)->getName() == name)
            return *i;
 
    return NULL;
}
 
VuoRunner::Port * VuoRunner::getPublishedOutputPortWithName(string name)
{
    vector<VuoRunner::Port *> outputPorts = getPublishedOutputPorts();
    for (vector<VuoRunner::Port *>::iterator i = outputPorts.begin(); i != outputPorts.end(); ++i)
        if ((*i)->getName() == name)
            return *i;
 
    return NULL;
}
 
void VuoRunner::listen()
{
    // Name this thread.
    {
        const char *compositionName = dylibPath.empty() ? executablePath.c_str() : dylibPath.c_str();
 
        // Trim the path, if present.
        if (const char *lastSlash = strrchr(compositionName, '/'))
            compositionName = lastSlash + 1;
 
        char threadName[MAXTHREADNAMESIZE];
        snprintf(threadName, MAXTHREADNAMESIZE, "org.vuo.runner.telemetry: %s", compositionName);
        pthread_setname_np(threadName);
    }
 
    ZMQSelfReceive = zmq_socket(ZMQContext, ZMQ_PAIR);
    VuoRunner_configureSocket(ZMQSelfReceive);
    if (zmq_bind(ZMQSelfReceive, "inproc://vuo-runner-self") != 0)
    {
        listenError = strerror(errno);
        dispatch_semaphore_signal(beganListeningSemaphore);
        return;
    }
 
    ZMQSelfSend = zmq_socket(ZMQContext, ZMQ_PAIR);
    VuoRunner_configureSocket(ZMQSelfSend);
    if (zmq_connect(ZMQSelfSend, "inproc://vuo-runner-self") != 0)
    {
        listenError = strerror(errno);
        dispatch_semaphore_signal(beganListeningSemaphore);
        return;
    }
 
    {
        ZMQTelemetry = zmq_socket(ZMQContext,ZMQ_SUB);
        VuoRunner_configureSocket(ZMQTelemetry);
        if(zmq_connect(ZMQTelemetry,ZMQTelemetryURL.c_str()))
        {
            listenError = strerror(errno);
            dispatch_semaphore_signal(beganListeningSemaphore);
            return;
        }
 
        const int highWaterMark = 0;  // no limit
        if(zmq_setsockopt(ZMQTelemetry,ZMQ_RCVHWM,&highWaterMark,sizeof(highWaterMark)))
        {
            listenError = strerror(errno);
            dispatch_semaphore_signal(beganListeningSemaphore);
            return;
        }
    }
 
    {
        // subscribe to all types of telemetry
        char type = VuoTelemetryHeartbeat;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryNodeExecutionStarted;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryNodeExecutionFinished;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryInputPortsUpdated;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryOutputPortsUpdated;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryPublishedOutputPortsUpdated;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryEventFinished;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryEventDropped;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryError;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
        type = VuoTelemetryStopRequested;
        zmq_setsockopt(ZMQTelemetry, ZMQ_SUBSCRIBE, &type, sizeof type);
    }
 
    {
        // Wait until the connection is established, as evidenced by a heartbeat telemetry message
        // being received from the composition. This is necessary because the ØMQ API doesn't provide
        // any way to tell when a SUB socket is ready to receive messages, and if you call zmq_poll()
        // on it before it's ready, then it might miss messages that came in while it was still trying
        // to get ready. (The zmq_connect() function doesn't make any guarantees about the socket being ready.
        // It just starts some setup that may continue asynchronously after zmq_connect() has returned.)
        // To avoid missing important telemetry messages from the composition, we make sure that the
        // runner doesn't tell the composition to unpause until the runner has verified that it's
        // receiving heartbeat telemetry messages. http://zguide.zeromq.org/page:all#Node-Coordination
        zmq_pollitem_t items[]=
        {
            {ZMQTelemetry,0,ZMQ_POLLIN,0},
        };
        int itemCount = 1;
        long timeout = -1;
        zmq_poll(items,itemCount,timeout);
    }
 
    dispatch_semaphore_signal(beganListeningSemaphore);
 
    bool pendingCancel = false;
    while(! listenCanceled)
    {
        zmq_pollitem_t items[]=
        {
            {ZMQTelemetry,0,ZMQ_POLLIN,0},
            {ZMQSelfReceive,0,ZMQ_POLLIN,0},
        };
        int itemCount = 2;
 
        // Wait 1 second.  If no telemetry was received in that second, we probably lost contact with the composition.
        long timeout = pendingCancel ? 100 : 1000;
        zmq_poll(items,itemCount,timeout);
        if(items[0].revents & ZMQ_POLLIN)
        {
            // Receive telemetry type.
            char type = vuoReceiveInt(ZMQTelemetry, NULL);
 
            // Receive telemetry arguments and forward to VuoRunnerDelegate.
            switch (type)
            {
                case VuoTelemetryHeartbeat:
                {
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->receivedTelemetryStats(0, 0);
                                  });
                    break;
                }
                case VuoTelemetryNodeExecutionStarted:
                {
                    char *compositionIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    char *nodeIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->receivedTelemetryNodeExecutionStarted(compositionIdentifier, nodeIdentifier);
                                  });
                    free(compositionIdentifier);
                    free(nodeIdentifier);
                    break;
                }
                case VuoTelemetryNodeExecutionFinished:
                {
                    char *compositionIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    char *nodeIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->receivedTelemetryNodeExecutionFinished(compositionIdentifier, nodeIdentifier);
                                  });
                    free(compositionIdentifier);
                    free(nodeIdentifier);
                    break;
                }
                case VuoTelemetryInputPortsUpdated:
                {
                    while (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                    {
                        char *compositionIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                        if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                        {
                            char *portIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                            if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                            {
                                bool receivedEvent = vuoReceiveBool(ZMQTelemetry, NULL);
                                if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                                {
                                    bool receivedData = vuoReceiveBool(ZMQTelemetry, NULL);
                                    if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                                    {
                                        string portDataSummary;
                                        char *s = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                                        if (s)
                                        {
                                            portDataSummary = s;
                                            free(s);
                                        }
                                        else
                                            portDataSummary = "";
 
                                        dispatch_sync(delegateQueue, ^{
                                                          if (delegate)
                                                              delegate->receivedTelemetryInputPortUpdated(compositionIdentifier, portIdentifier, receivedEvent, receivedData, portDataSummary);
                                                      });
                                    }
                                }
                            }
                            free(portIdentifier);
                        }
                        free(compositionIdentifier);
                    }
                    break;
                }
                case VuoTelemetryOutputPortsUpdated:
                {
                    while (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                    {
                        char *compositionIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                        if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                        {
                            char *portIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                            if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                            {
                                bool sentEvent = vuoReceiveBool(ZMQTelemetry, NULL);
                                if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                                {
                                    bool sentData = vuoReceiveBool(ZMQTelemetry, NULL);
                                    if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                                    {
                                        string portDataSummary;
                                        char *s = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                                        if (s)
                                        {
                                            portDataSummary = s;
                                            free(s);
                                        }
                                        else
                                            portDataSummary = "";
 
                                        dispatch_sync(delegateQueue, ^{
                                                          if (delegate)
                                                              delegate->receivedTelemetryOutputPortUpdated(compositionIdentifier, portIdentifier, sentEvent, sentData, portDataSummary);
                                                      });
                                    }
                                }
                            }
                            free(portIdentifier);
                        }
                        free(compositionIdentifier);
                    }
                    break;
                }
                case VuoTelemetryPublishedOutputPortsUpdated:
                {
                    while (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                    {
                        char *portIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                        if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                        {
                            bool sentData = vuoReceiveBool(ZMQTelemetry, NULL);
                            if (VuoTelemetry_hasMoreToReceive(ZMQTelemetry))
                            {
                                string portDataSummary;
                                char *s = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                                if (s)
                                {
                                    portDataSummary = s;
                                    free(s);
                                }
                                else
                                    portDataSummary = "";
 
                                Port *port = getPublishedOutputPortWithName(portIdentifier);
 
                                dispatch_sync(delegateQueue, ^{
                                                  if (delegate)
                                                      delegate->receivedTelemetryPublishedOutputPortUpdated(port, sentData, portDataSummary);
                                              });
                            }
                        }
                        free(portIdentifier);
                    }
                    break;
                }
                case VuoTelemetryEventFinished:
                {
                    saturating_semaphore_signal(lastFiredEventSemaphore, &lastFiredEventSignaled);
                    break;
                }
                case VuoTelemetryEventDropped:
                {
                    char *compositionIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    char *portIdentifier = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->receivedTelemetryEventDropped(compositionIdentifier, portIdentifier);
                                  });
                    free(compositionIdentifier);
                    free(portIdentifier);
                    break;
                }
                case VuoTelemetryError:
                {
                    char *message = vuoReceiveAndCopyString(ZMQTelemetry, NULL);
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->receivedTelemetryError( string(message) );
                                  });
                    free(message);
                    break;
                }
                case VuoTelemetryStopRequested:
                {
                    dispatch_sync(delegateQueue, ^{
                                      if (delegate)
                                          delegate->lostContactWithComposition();
                                  });
                    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
                                       stop();
                                   });
                    break;
                }
                default:
                    VUserLog("Error: Unknown telemetry message type: %d", type);
                    break;
            }
        }
        else if (! listenCanceled)  // Either the 1-second timeout elapsed, or we got a stop-listening message from ZMQSelfSend
        {
            if (items[1].revents & ZMQ_POLLIN)
            {
                // This is a stop-listening message.
                vuoReceiveInt(ZMQSelfReceive, NULL);
 
                // Drain any remaining telemetry messages.
                pendingCancel = true;
            }
 
            else if (pendingCancel)
                listenCanceled = true;
 
            else
            {
                // Timeout.
                // Could happen if the composition crashed, or if the system fell asleep then hibernated (standby mode).
                // If it's a crash we should disconnect; if it's hibernation we should ignore the timeout and try zmq_poll again.
                if (VuoEventLoop_isSystemAsleep())
                    VDebugLog("zmq_poll timed out, but system is sleeping so I'll try again.");
                else if (VuoLog_isDebuggerAttached())
                    VDebugLog("zmq_poll timed out, but a debugger is attached to the host so I'll try again.");
                else
                {
                    listenCanceled = true;
                    string dir, file, ext;
                    VuoFileUtilities::splitPath(executablePath, dir, file, ext);
                    stopBecauseLostContact("The connection between the composition ('" + file + "') and runner timed out while listening for telemetry.");
                }
            }
        }
    }
 
    zmq_close(ZMQTelemetry);
    ZMQTelemetry = NULL;
 
    zmq_close(ZMQSelfSend);
    ZMQSelfSend = NULL;
    zmq_close(ZMQSelfReceive);
    ZMQSelfReceive = NULL;
 
    dispatch_semaphore_signal(endedListeningSemaphore);
    return;
}
 
void VuoRunner::vuoControlRequestSend(enum VuoControlRequest request, zmq_msg_t *messages, unsigned int messageCount)
{
    char *error = NULL;
    vuoSend("runner VuoControl",ZMQControl,request,messages,messageCount,false,&error);
 
    if (error)
    {
        string e(error);
        free(error);
        throw VuoException(e);
    }
}
 
void VuoRunner::vuoLoaderControlRequestSend(enum VuoLoaderControlRequest request, zmq_msg_t *messages, unsigned int messageCount)
{
    char *error = NULL;
    vuoSend("runner VuoLoaderControl",ZMQLoaderControl,request,messages,messageCount,false,&error);
 
    if (error)
    {
        string e(error);
        free(error);
        throw VuoException(e);
    }
}
 
void VuoRunner::vuoControlReplyReceive(enum VuoControlReply expectedReply)
{
    char *error = NULL;
    int reply = vuoReceiveInt(ZMQControl, &error);
 
    if (error)
    {
        string e(error);
        free(error);
        ostringstream oss;
        oss << e << " (expected " << expectedReply << ")";
        throw VuoException(oss.str());
    }
    else if (reply != expectedReply)
    {
        ostringstream oss;
        oss << "The runner received the wrong message from the composition (expected " << expectedReply << ", received " << reply << ")";
        throw VuoException(oss.str());
    }
}
 
void VuoRunner::vuoLoaderControlReplyReceive(enum VuoLoaderControlReply expectedReply)
{
    char *error = NULL;
    int reply = vuoReceiveInt(ZMQLoaderControl, &error);
 
    if (error)
    {
        string e(error);
        free(error);
        ostringstream oss;
        oss << e << " (expected " << expectedReply << ")";
        throw VuoException(oss.str());
    }
    else if (reply != expectedReply)
    {
        ostringstream oss;
        oss << "The runner received the wrong message from the composition loader (expected " << expectedReply << ", received " << reply << ")";
        throw VuoException(oss.str());
    }
}
 
string VuoRunner::receiveString(string fallbackIfNull)
{
    char *error = NULL;
    char *s = vuoReceiveAndCopyString(ZMQControl, &error);
 
    if (error)
    {
        string e(error);
        free(error);
        throw VuoException(e);
    }
 
    string ret;
    if (s)
    {
        ret = s;
        free(s);
    }
    else
        ret = fallbackIfNull;
 
    return ret;
}
 
vector<string> VuoRunner::receiveListOfStrings(void)
{
    vector<string> messageStrings;
    while (VuoTelemetry_hasMoreToReceive(ZMQControl))
    {
        string s = receiveString("");
        messageStrings.push_back(s);
    }
    return messageStrings;
}
 
void VuoRunner::saturating_semaphore_signal(dispatch_semaphore_t dsema, bool *signaled)
{
    if (__sync_bool_compare_and_swap(signaled, false, true))
        dispatch_semaphore_signal(dsema);
}
 
void VuoRunner::saturating_semaphore_wait(dispatch_semaphore_t dsema, bool *signaled)
{
    *signaled = false;
    dispatch_semaphore_wait(dsema, DISPATCH_TIME_FOREVER);
}
 
bool VuoRunner::isStopped(void)
{
    return stopped;
}
 
bool VuoRunner::isInCurrentProcess(void)
{
    return executablePath.empty();
}
 
bool VuoRunner::isUsingCompositionLoader(void)
{
    return ! executablePath.empty() && ! dylibPath.empty();
}
 
void VuoRunner::setDelegate(VuoRunnerDelegate *delegate)
{
    dispatch_sync(delegateQueue, ^{
                      this->delegate = delegate;
                  });
}
 
void VuoRunner::stopBecauseLostContact(string errorMessage)
{
    __block bool alreadyLostContact;
    dispatch_sync(delegateQueue, ^{
                      alreadyLostContact = lostContact;
                      lostContact = true;
                  });
 
    if (alreadyLostContact)
        return;
 
    saturating_semaphore_signal(lastFiredEventSemaphore, &lastFiredEventSignaled);
 
    dispatch_sync(delegateQueue, ^{
                      if (delegate)
                          delegate->lostContactWithComposition();
                  });
 
    dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
                       stop();
                   });
 
    if (! isInCurrentProcess())
    {
        // Normally, stop() is responsible for terminating the ZMQ context.
        // But, if stopBecauseLostContact() is called, it takes the responsibility away from stop().
        // If there's an in-progress zmq_recv() call, stop() will get stuck waiting on controlQueue, so
        // the below call to terminate the ZMQ context interrupts zmq_recv() and allows stop() to proceed.
        dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
                           vuoMemoryBarrier();
 
                           zmq_term(ZMQContext);
                           ZMQContext = NULL;
                           dispatch_semaphore_signal(terminatedZMQContextSemaphore);
                       });
    }
 
    VUserLog("%s", errorMessage.c_str());
}
 
pid_t VuoRunner::getCompositionPid()
{
    return compositionPid;
}
 
VuoRunner::Port::Port(string name, string type, json_object *details)
{
    this->name = name;
    this->type = type;
    this->details = details;
}
 
string VuoRunner::Port::getName(void)
{
    return name;
}
 
string VuoRunner::Port::getType(void)
{
    return type;
}
 
json_object *VuoRunner::Port::getDetails(void)
{
    return details;
}
 
VuoRunnerDelegate::~VuoRunnerDelegate() { }  // Fixes "undefined symbols" error