VuoThreadManager.cc

Go to the documentation of this file.

 
#include "VuoThreadManager.hh"
 
#include "VuoRuntimePersistentState.hh"
#include "VuoRuntimeState.hh"
#include "VuoRuntimeUtilities.hh"
#include "VuoEventLoop.h"
#include "VuoException.hh"
 
VuoThreadManager::Worker::Worker(dispatch_queue_t queue, void *context, void (*function)(void *),
                                 int minThreadsNeeded, int maxThreadsNeeded,
                                 unsigned long eventId, unsigned long compositionHash, int chainCount)
{
    this->queue = queue;
    this->context = context;
    this->function = function;
    this->isTrigger = true;
    this->minThreadsNeeded = minThreadsNeeded;
    this->maxThreadsNeeded = maxThreadsNeeded;
    this->eventId = eventId;
    this->compositionHash = compositionHash;
    this->chainIndex = ULONG_MAX;
    this->chainCount = chainCount;
    this->upstreamChainIndices = NULL;
    this->upstreamChainIndicesCount = 0;
}
 
VuoThreadManager::Worker::Worker(dispatch_queue_t queue, void *context, void (*function)(void *),
                                 int minThreadsNeeded, int maxThreadsNeeded,
                                 unsigned long eventId, unsigned long compositionHash, unsigned long chainIndex,
                                 unsigned long *upstreamChainIndices, int upstreamChainIndicesCount)
{
    this->queue = queue;
    this->context = context;
    this->function = function;
    this->isTrigger = false;
    this->minThreadsNeeded = minThreadsNeeded;
    this->maxThreadsNeeded = maxThreadsNeeded;
    this->eventId = eventId;
    this->compositionHash = compositionHash;
    this->chainIndex = chainIndex;
    this->chainCount = -1;
    this->upstreamChainIndices = upstreamChainIndices;
    this->upstreamChainIndicesCount = upstreamChainIndicesCount;
}
 
VuoThreadManager::WorkerQueue::WorkerQueue(void)
{
    first = NULL;
    last = NULL;
}
 
void VuoThreadManager::WorkerQueue::enqueue(Worker *worker)
{
    Node *node = new Node;
    node->worker = worker;
    node->next = NULL;
    node->prev = NULL;
 
    if (! first)
    {
        last = node;
    }
    else
    {
        first->prev = node;
        node->next = first;
    }
 
    first = node;
}
 
void VuoThreadManager::WorkerQueue::dequeue(Node *node)
{
    if (node == last)
        last = node->prev;
    else
        node->next->prev = node->prev;
 
    if (node == first)
        first = node->next;
    else
        node->prev->next = node->next;
 
    delete node;
}
 
VuoThreadManager::WorkerQueue::Node * VuoThreadManager::WorkerQueue::getNextOldest(Node *newest, Node *node)
{
    if (node == newest)
        return NULL;
    else
        return node->prev;
}
 
VuoThreadManager::WorkerQueue::Node * VuoThreadManager::WorkerQueue::getOldest(void)
{
    return last;
}
 
VuoThreadManager::WorkerQueue::Node * VuoThreadManager::WorkerQueue::getNewest(void)
{
    return first;
}
 
size_t VuoThreadManager::WorkerQueue::size(void)
{
    size_t s = 0;
    for (Node *curr = first; curr != NULL; curr = curr->next)
        ++s;
    return s;
}
 
VuoThreadManager::ThreadPool::ThreadPool(void)
{
    totalThreads = 0;
    threadsAvailable = 0;
    totalWorkers = 0;
}
 
void VuoThreadManager::ThreadPool::setTotalThreads(int totalThreads)
{
    this->totalThreads = totalThreads;
    this->threadsAvailable = totalThreads;
}
 
bool VuoThreadManager::ThreadPool::tryClaimThreads(int minThreadsNeeded, int maxThreadsNeeded, unsigned long workerId, int &threadsClaimed)
{
    if (minThreadsNeeded <= threadsAvailable)
    {
        threadsClaimed = min(threadsAvailable, maxThreadsNeeded);
        threadsAvailable -= threadsClaimed;
        workersClaimingThreads[workerId] = threadsClaimed;
        return true;
    }
    else
    {
        threadsClaimed = 0;
        return false;
    }
}
 
int VuoThreadManager::ThreadPool::getThreadsClaimed(unsigned long workerId)
{
    map<unsigned long, int>::iterator iter = workersClaimingThreads.find(workerId);
    if (iter != workersClaimingThreads.end())
        return iter->second;
    else
        throw VuoException("Couldn't find worker in thread pool to get its claimed threads.");
}
 
void VuoThreadManager::ThreadPool::returnThreads(unsigned long workerId)
{
    map<unsigned long, int>::iterator iter = workersClaimingThreads.find(workerId);
    if (iter != workersClaimingThreads.end())
    {
        threadsAvailable += iter->second;
        if (totalWorkers > 0)
            workersCompleted.insert(iter->first);
        workersClaimingThreads.erase(iter);
    }
    else
        throw VuoException("Couldn't find worker in thread pool to return its threads.");
}
 
 
VuoThreadManager::VuoThreadManager(void)
{
    mainThreadPool.setTotalThreads(60);  // maximum number of worker threads in use simultaneously
    workersWaitingSync = dispatch_queue_create("org.vuo.runtime.workersWaiting", VuoEventLoop_getDispatchInteractiveAttribute());
    threadPoolSync = dispatch_queue_create("org.vuo.runtime.threadPool", VuoEventLoop_getDispatchInteractiveAttribute());
    workersUpdated = dispatch_semaphore_create(0);
    completed = dispatch_semaphore_create(0);
}
 
VuoThreadManager::~VuoThreadManager(void)
{
    dispatch_release(workersWaitingSync);
    dispatch_release(threadPoolSync);
    dispatch_release(workersUpdated);
}
 
void VuoThreadManager::enableSchedulingWorkers(void)
{
    mayMoreWorkersBeEnqueued = true;
    mayMoreWorkersBeDequeued = true;
 
    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_HIGH, 0);
    dispatch_async(queue, ^{
                       while (mayMoreWorkersBeDequeued)
                       {
                           vector<Worker *> workers = dequeueWorkers();
 
                           for (vector<Worker *>::iterator i = workers.begin(); i != workers.end(); ++i)
                           {
                               Worker *w = *i;
                               dispatch_async_f(w->queue, w->context, w->function);
                               delete w;
                           }
                       }
 
                       dispatch_semaphore_signal(completed);
                   });
}
 
void VuoThreadManager::disableSchedulingWorkers(void)
{
    dispatch_sync(workersWaitingSync, ^{
                      mayMoreWorkersBeEnqueued = false;
                  });
 
    dispatch_semaphore_signal(workersUpdated);
    dispatch_semaphore_wait(completed, DISPATCH_TIME_FOREVER);
 
    mainThreadPool.workersCompleted.clear();
}
 
vector<VuoThreadManager::Worker *> VuoThreadManager::dequeueWorkers(void)
{
    workersDequeued.clear();
    workersDequeuedNodes.clear();
 
    while (workersDequeued.empty() && mayMoreWorkersBeDequeued)
    {
        dispatch_semaphore_wait(workersUpdated, DISPATCH_TIME_FOREVER);
 
        __block WorkerQueue::Node *newest;
        __block WorkerQueue::Node *oldest;
        dispatch_sync(workersWaitingSync, ^{
                          newest = workersWaitingForThreads.getNewest();
                          oldest = workersWaitingForThreads.getOldest();
                      });
 
        // During this block, additional workers may be enqueued in `workersWaitingForThreads`,
        // but the section of the queue from `oldest` to `newest` remains unchanged.
        dispatch_sync(threadPoolSync, ^{
                          bool hasTriggerBeenSkipped = false;
 
                          for (WorkerQueue::Node *n = oldest; n != NULL; n = workersWaitingForThreads.getNextOldest(newest, n))
                          {
                              Worker *w = n->worker;
 
                              if (w->isTrigger)
                              {
                                  if (w->minThreadsNeeded >= 0 && w->maxThreadsNeeded >= 0)
                                  {
                                      // Trigger worker for new event
 
                                      if (! hasTriggerBeenSkipped)
                                      {
                                          int threadsClaimed;
                                          bool gotThreads = mainThreadPool.tryClaimThreads(w->minThreadsNeeded, w->maxThreadsNeeded, w->eventId, threadsClaimed);
                                          if (gotThreads)
                                          {
                                              ThreadPool &triggerThreadPool = triggerThreadPools[w->eventId][w->compositionHash];
                                              triggerThreadPool.setTotalThreads(threadsClaimed);
                                              triggerThreadPool.totalWorkers = w->chainCount;
                                              workersDequeuedNodes.push_back(n);
                                          }
                                          else
                                          {
                                              hasTriggerBeenSkipped = true;
                                          }
                                      }
                                  }
                                  else
                                  {
                                      // Trigger worker for published input trigger of a subcomposition node
 
                                      triggerThreadPools[w->eventId][w->compositionHash].totalWorkers = w->chainCount;
                                      workersDequeuedNodes.push_back(n);
                                  }
                              }
                              else
                              {
                                  // Chain worker
 
                                  ThreadPool &triggerThreadPool = triggerThreadPools[w->eventId][w->compositionHash];
 
                                  bool haveAllUpstreamChainsCompleted = true;
                                  for (int i = 0; i < w->upstreamChainIndicesCount; ++i)
                                  {
                                      unsigned long index = w->upstreamChainIndices[i];
 
                                      set<unsigned long>::iterator iter = triggerThreadPool.workersCompleted.find( index );
                                      if (iter == triggerThreadPool.workersCompleted.end())
                                      {
                                          haveAllUpstreamChainsCompleted = false;
                                          break;
                                      }
                                  }
 
                                  if (haveAllUpstreamChainsCompleted)
                                  {
                                      int threadsClaimed;
                                      bool gotThreads = triggerThreadPool.tryClaimThreads(w->minThreadsNeeded, w->maxThreadsNeeded, w->chainIndex, threadsClaimed);
                                      if (gotThreads)
                                      {
                                          workersDequeuedNodes.push_back(n);
 
                                          free(w->upstreamChainIndices);
                                          w->upstreamChainIndices = NULL;
                                      }
                                  }
                              }
 
                              if (n == newest) {
                                  break;
                              }
                          }
                      });
 
        dispatch_sync(workersWaitingSync, ^{
                          for (vector<WorkerQueue::Node *>::iterator i = workersDequeuedNodes.begin(); i != workersDequeuedNodes.end(); ++i)
                          {
                              WorkerQueue::Node *n = *i;
                              workersDequeued.push_back(n->worker);
                              workersWaitingForThreads.dequeue(n);
                          }
 
                          if (! mayMoreWorkersBeEnqueued && workersWaitingForThreads.first == NULL) {
                              mayMoreWorkersBeDequeued = false;
                          }
                      });
    }
 
    return workersDequeued;
}
 
void VuoThreadManager::scheduleTriggerWorker(dispatch_queue_t queue, void *context, void (*function)(void *),
                                             int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId,
                                             const char *compositionIdentifier, int chainCount)
{
    // Assumes mainThreadPool.totalThreads is constant, thus safe to access outside of threadPoolSync.
    int adjustedMinThreadsNeeded = minThreadsNeeded;
    if (adjustedMinThreadsNeeded > mainThreadPool.totalThreads) {
        adjustedMinThreadsNeeded = mainThreadPool.totalThreads;
        VUserLog("Warning: Couldn't allocate as many threads to a trigger worker as it requires.");
    }
 
    unsigned long compositionHash = VuoRuntimeUtilities::hash(compositionIdentifier);
    Worker *worker = new Worker(queue, context, function, adjustedMinThreadsNeeded, maxThreadsNeeded, eventId, compositionHash, chainCount);
 
    dispatch_sync(workersWaitingSync, ^{
                       workersWaitingForThreads.enqueue(worker);
                   });
 
    dispatch_semaphore_signal(workersUpdated);
}
 
void VuoThreadManager::scheduleChainWorker(dispatch_queue_t queue, void *context, void (*function)(void *),
                                           int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, const char *compositionIdentifier,
                                           unsigned long chainIndex, unsigned long *upstreamChainIndices, int upstreamChainIndicesCount)
{
    unsigned long compositionHash = VuoRuntimeUtilities::hash(compositionIdentifier);
    Worker *worker = new Worker(queue, context, function, minThreadsNeeded, maxThreadsNeeded, eventId, compositionHash, chainIndex, upstreamChainIndices, upstreamChainIndicesCount);
 
    dispatch_sync(workersWaitingSync, ^{
                      workersWaitingForThreads.enqueue(worker);
                  });
 
    dispatch_semaphore_signal(workersUpdated);
}
 
void VuoThreadManager::grantThreadsToChain(int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, const char *compositionIdentifier,
                                           unsigned long chainIndex)
{
    unsigned long compositionHash = VuoRuntimeUtilities::hash(compositionIdentifier);
 
    dispatch_sync(threadPoolSync, ^{
                      ThreadPool &triggerThreadPool = triggerThreadPools[eventId][compositionHash];
                      int threadsClaimed;
                      bool gotThreads = triggerThreadPool.tryClaimThreads(minThreadsNeeded, maxThreadsNeeded, chainIndex, threadsClaimed);
                      if (! gotThreads) {
                          throw VuoException("Not enough threads available in the thread pool to execute the chain.");
                      }
                  });
 
    dispatch_semaphore_signal(workersUpdated);
}
 
void VuoThreadManager::grantThreadsToSubcomposition(unsigned long eventId, const char *compositionIdentifier, unsigned long chainIndex,
                                                    const char *subcompositionIdentifier)
{
    unsigned long compositionHash = VuoRuntimeUtilities::hash(compositionIdentifier);
    unsigned long subcompositionHash = VuoRuntimeUtilities::hash(subcompositionIdentifier);
 
    dispatch_sync(threadPoolSync, ^{
                      map<unsigned long, ThreadPool> &eventThreadPools = triggerThreadPools[eventId];
                      int threadsClaimedForChain = eventThreadPools[compositionHash].getThreadsClaimed(chainIndex);
                      eventThreadPools[subcompositionHash].setTotalThreads(threadsClaimedForChain);
                  });
}
 
void VuoThreadManager::returnThreadsForTriggerWorker(unsigned long eventId)
{
    dispatch_sync(threadPoolSync, ^{
                       triggerThreadPools.erase(eventId);
                       mainThreadPool.returnThreads(eventId);
                   });
 
    dispatch_semaphore_signal(workersUpdated);
}
 
void VuoThreadManager::returnThreadsForChainWorker(unsigned long eventId, const char *compositionIdentifier, unsigned long chainIndex)
{
    unsigned long compositionHash = VuoRuntimeUtilities::hash(compositionIdentifier);
 
    dispatch_sync(threadPoolSync, ^{
                      ThreadPool &triggerThreadPool = triggerThreadPools[eventId][compositionHash];
                      triggerThreadPool.returnThreads(chainIndex);
 
                      // If this was the last chain for the event (for the current subcomposition or overall),
                      // erase the thread pool entries for the event.
                      set<unsigned long> &workersCompleted = triggerThreadPool.workersCompleted;
                      if (workersCompleted.size() == triggerThreadPool.totalWorkers)
                      {
                          map<unsigned long, ThreadPool> &eventThreadPools = triggerThreadPools[eventId];
                          eventThreadPools.erase(compositionHash);
                          if (eventThreadPools.empty())
                          {
                              triggerThreadPools.erase(eventId);
                              mainThreadPool.returnThreads(eventId);
                          }
                      }
                  });
 
    dispatch_semaphore_signal(workersUpdated);
}
 
extern "C"
{
 
void vuoScheduleTriggerWorker(VuoCompositionState *compositionState, dispatch_queue_t queue, void *context, void (*function)(void *),
                              int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, int chainCount)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    const char *compositionIdentifier = compositionState->compositionIdentifier;
    runtimeState->persistentState->threadManager->scheduleTriggerWorker(queue, context, function,
                                                                        minThreadsNeeded, maxThreadsNeeded, eventId,
                                                                        compositionIdentifier, chainCount);
}
 
void vuoScheduleChainWorker(VuoCompositionState *compositionState, dispatch_queue_t queue, void *context, void (*function)(void *),
                            int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, unsigned long chainIndex,
                            unsigned long *upstreamChainIndices, int upstreamChainIndicesCount)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    const char *compositionIdentifier = compositionState->compositionIdentifier;
    runtimeState->persistentState->threadManager->scheduleChainWorker(queue, context, function,
                                                                      minThreadsNeeded, maxThreadsNeeded, eventId, compositionIdentifier,
                                                                      chainIndex, upstreamChainIndices, upstreamChainIndicesCount);
}
 
void vuoGrantThreadsToChain(VuoCompositionState *compositionState,
                            int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, unsigned long chainIndex)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    const char *compositionIdentifier = compositionState->compositionIdentifier;
    runtimeState->persistentState->threadManager->grantThreadsToChain(minThreadsNeeded, maxThreadsNeeded, eventId, compositionIdentifier,
                                                                      chainIndex);
}
 
void vuoGrantThreadsToSubcomposition(VuoCompositionState *compositionState, unsigned long eventId, unsigned long chainIndex,
                                     char *subcompositionIdentifier)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    const char *compositionIdentifier = compositionState->compositionIdentifier;
    runtimeState->persistentState->threadManager->grantThreadsToSubcomposition(eventId, compositionIdentifier, chainIndex,
                                                                               subcompositionIdentifier);
}
 
void vuoReturnThreadsForTriggerWorker(VuoCompositionState *compositionState, unsigned long eventId)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    runtimeState->persistentState->threadManager->returnThreadsForTriggerWorker(eventId);
}
 
void vuoReturnThreadsForChainWorker(VuoCompositionState *compositionState, unsigned long eventId, unsigned long chainIndex)
{
    VuoRuntimeState *runtimeState = (VuoRuntimeState *)compositionState->runtimeState;
    const char *compositionIdentifier = compositionState->compositionIdentifier;
    runtimeState->persistentState->threadManager->returnThreadsForChainWorker(eventId, compositionIdentifier, chainIndex);
}
 
}