VuoThreadManager.hh

Go to the documentation of this file.

 
#pragma once
 
#include <dispatch/dispatch.h>
#include <map>
#include <set>
#include <vector>
using namespace std;
 
#include "VuoCompositionState.h"
#include "VuoHeap.h"
 
class VuoThreadManager
{
private:
    class Worker
    {
    public:
        dispatch_queue_t queue;  
        void *context;  
        void (*function)(void *);  
 
        bool isTrigger;  
 
        int minThreadsNeeded;  
        int maxThreadsNeeded;  
 
        unsigned long eventId;  
        unsigned long compositionHash;  
 
        int chainCount;  
 
        unsigned long chainIndex;  
        unsigned long *upstreamChainIndices;  
        int upstreamChainIndicesCount;  
 
        Worker(dispatch_queue_t queue, void *context, void (*function)(void *), int minThreadsNeeded, int maxThreadsNeeded,
               unsigned long eventId, unsigned long compositionHash, int chainCount);
        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);
    };
 
    class WorkerQueue
    {
    public:
        struct Node
        {
            Worker *worker;  
            Node *next;  
            Node *prev;  
        };
 
        Node *first;  
        Node *last;  
 
        WorkerQueue(void);
        void enqueue(Worker *worker);
        void dequeue(Node *node);
        Node * getNextOldest(Node *newest, Node *node);
        Node * getOldest(void);
        Node * getNewest(void);
        size_t size(void);
    };
 
    class ThreadPool
    {
    public:
        int totalThreads;  
        int threadsAvailable;  
        map<unsigned long, int> workersClaimingThreads;  
 
        int totalWorkers;  
        set<unsigned long> workersCompleted;  
 
        ThreadPool(void);
        void setTotalThreads(int totalThreads);
        bool tryClaimThreads(int minThreadsNeeded, int maxThreadsNeeded, unsigned long workerId, int &threadsClaimed) VuoWarnUnusedResult;
        int getThreadsClaimed(unsigned long workerId);
        void returnThreads(unsigned long workerId);
    };
 
    WorkerQueue workersWaitingForThreads;  
    ThreadPool mainThreadPool;  
    map<unsigned long, map<unsigned long, ThreadPool> > triggerThreadPools;  
    dispatch_queue_t workersWaitingSync;  
    dispatch_queue_t threadPoolSync;  
    dispatch_semaphore_t workersUpdated;  
    dispatch_semaphore_t completed;  
    bool mayMoreWorkersBeEnqueued;  
    bool mayMoreWorkersBeDequeued;  
 
    vector<Worker *> workersDequeued;  
    vector<WorkerQueue::Node *> workersDequeuedNodes;  
 
    vector<Worker *> dequeueWorkers(void);
 
public:
    VuoThreadManager(void);
    ~VuoThreadManager(void);
    void enableSchedulingWorkers(void);
    void disableSchedulingWorkers(void);
    void scheduleTriggerWorker(dispatch_queue_t queue, void *context, void (*function)(void *),
                               int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, const char *compositionIdentifier,
                               int chainCount);
    void 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);
    void grantThreadsToChain(int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, const char *compositionIdentifier,
                             unsigned long chainIndex);
    void grantThreadsToSubcomposition(unsigned long eventId, const char *compositionIdentifier, unsigned long chainIndex, const char *subcompositionIdentifier);
    void returnThreadsForTriggerWorker(unsigned long eventId);
    void returnThreadsForChainWorker(unsigned long eventId, const char *compositionIdentifier, unsigned long chainIndex);
};
 
extern "C"
{
void vuoScheduleTriggerWorker(VuoCompositionState *compositionState, dispatch_queue_t queue, void *context, void (*function)(void *),
                              int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, int 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);
void vuoGrantThreadsToChain(VuoCompositionState *compositionState,
                            int minThreadsNeeded, int maxThreadsNeeded, unsigned long eventId, unsigned long chainIndex);
void vuoGrantThreadsToSubcomposition(VuoCompositionState *compositionState, unsigned long eventId, unsigned long chainIndex,
                                     char *subcompositionIdentifier);
void vuoReturnThreadsForTriggerWorker(VuoCompositionState *compositionState, unsigned long eventId);
void vuoReturnThreadsForChainWorker(VuoCompositionState *compositionState, unsigned long eventId, unsigned long chainIndex);
}