Max 5 API Reference

parallel.c

/*

parallel processing api

try to keep as simple and general as possible, 
more complicated/specific behavior can be added
on top of the following

example applications
- poly~
- jit.matrix operators
- other parallelizable tasks 


- t_parallel_task: object which manages overall workload and workjer threads
    state:
    - maximum worker count
    - worker entry proc
    - thread priority (making use of thread pools for efficiency sake, 
        currenlty 3 thread pools: audio (high), scheduler (med), queue (low))
    - task global storage, user defined
    - state related to benchmarking/load balancing
    
    verbs:
    - run task
    - cancel task

- t_parallel_worker: one worker thread
    - workerid
    - worker entry proc
    - pointer to parent task object
    - state related to benchmarking
    - worker specific data?
    verbs:
    - run worker
    - some means of benchmarking


TODO:
- t_parallel_iterator: utility wrapper for parallel task for iterating on a dataset
    could use object methods for ease of use

    - means of negotiating behavior(initial callback before spawning threads)
    - setup initial worker state(secondary callback once for each thread)
    - shutdown per worker state(can cleanup any data needed)
    - shutdown of task execution handled in the function
    - notification of completion if complete asynchronous?


*/

// for use inside an external
#include "ext.h"
#include "ext_obex.h"
#include "ext_systhread.h"
#include "ext_common.h"

#include "parallel.h"


//private

t_class *_parallel_task_class=NULL;
t_class *_parallel_worker_class=NULL;

// thread pool stuff. can improve after initial pass

typedef struct _parallel_thread
{
    t_systhread         thread;
    t_systhread_mutex   mutex;          // establishes external ownership of thread
    t_systhread_cond    cond;
    t_parallel_worker   *worker;        
    long                state;
    struct _parallel_thread_pool *pool; //parent    
} t_parallel_thread;

typedef struct _parallel_thread_pool
{
    t_parallel_thread   threads[PARALLEL_MAX_WORKERS];
    long                count;
    t_systhread_mutex   count_mutex;
    t_systhread_cond    count_cond;
} t_parallel_thread_pool;
    
static t_parallel_thread_pool _pool_low;
static t_parallel_thread_pool _pool_med;
static t_parallel_thread_pool _pool_high;

void parallel_thread_init(t_parallel_thread *x);
void parallel_threadproc(t_parallel_thread *x);
void parallel_thread_pool_init(t_parallel_thread_pool *pool);

// do we want to add support for "barriers"?
// see p. 215 in "High Performance Computing", Dowd + Severance 

void parallel_init(void)
{
    long i,attrflags=0;
    t_class *c;
    t_object *attr;
        
// CLASS DEFINITION: parallel_task
    c = class_new("parallel_task", (method)parallel_task_new, 
            (method)parallel_task_free, sizeof(t_parallel_task), (method)NULL, A_CANT, 0L);
            
    class_obexoffset_set(c, calcoffset(t_parallel_task, obex));

/*  TODO:
    attr = attr_offset_new("workercount",_sym_long,attrflags,
        (method)0L,(method)parallel_task_workercount, 
        calcoffset(t_parallel_task, workercount));
    class_addattr(c,attr);
*/
    attr = attr_offset_new("benchmark",gensym("char"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_task, benchmark));
    class_addattr(c,attr);
    attr = attr_offset_new("priority",gensym("char"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_task, priority));
    class_addattr(c,attr);
    attr = attr_offset_new("iteration",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_task, iteration));
    class_addattr(c,attr);
    attr = attr_offset_new("begintime",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_task, begintime));
    class_addattr(c,attr);
    attr = attr_offset_new("endtime",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_task, endtime));
    class_addattr(c,attr);

    class_addmethod(c, (method)parallel_task_execute,       "execute", 0L);
    class_addmethod(c, (method)parallel_task_cancel,        "cancel", 0L);
    class_addmethod(c, (method)parallel_task_workerproc,    "workerproc", A_CANT, 0L);
    class_addmethod(c, (method)parallel_task_data,      "data", A_CANT, 0L);
    
    _parallel_task_class = c;
    class_register(CLASS_NOBOX,_parallel_task_class);

// CLASS DEFINITION: parallel_worker
    c = class_new("parallel_worker", (method)parallel_worker_new, 
            (method)parallel_worker_free, sizeof(t_parallel_worker), (method)NULL, A_CANT, 0L);
            
    class_obexoffset_set(c, calcoffset(t_parallel_worker, obex));

    attr = attr_offset_new("id",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_worker, id));
    class_addattr(c,attr);
    attr = attr_offset_new("begintime",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_worker, begintime));
    class_addattr(c,attr);
    attr = attr_offset_new("endtime",gensym("long"),attrflags,
        (method)0L,(method)0L, calcoffset(t_parallel_worker, endtime));
    class_addattr(c,attr);

    class_addmethod(c, (method)parallel_worker_execute,     "execute", 0L);
    
    _parallel_worker_class = c;
    class_register(CLASS_NOBOX,_parallel_worker_class);


// thread pool stuff. can improve after initial pass    
    parallel_thread_pool_init(&_pool_low);
    parallel_thread_pool_init(&_pool_med);
    parallel_thread_pool_init(&_pool_high); 
}


void parallel_thread_pool_init(t_parallel_thread_pool *pool)
{
    long i;
    
    for (i=0;i<PARALLEL_MAX_WORKERS;i++) {
        pool->threads[i].thread  = NULL;
        systhread_mutex_new(&pool->threads[i].mutex,0);
        systhread_cond_new(&pool->threads[i].cond,0);
        pool->threads[i].worker  = NULL;
        pool->threads[i].state = 0;
        pool->threads[i].pool = pool;
    }
    pool->count = 0;
    systhread_mutex_new(&pool->count_mutex,0);
    systhread_cond_new(&pool->count_cond,0);

}

long parallel_processorcount(void)
{
#ifdef MAC_VERSION
    return MPProcessorsScheduled();
#else
    SYSTEM_INFO info;
    GetSystemInfo(&info);
    return info.dwNumberOfProcessors; 
#endif
}

t_parallel_task *parallel_task_new(void *data, method workerproc, long workercount)
{
    long i;
    t_parallel_task *x=NULL;
    t_parallel_worker *w=NULL;
        
    if (workerproc && (x=(t_parallel_task *)object_alloc(_parallel_task_class))) {
        x->data         = data;
        x->workerproc   = workerproc;
        if (workercount<=0)
            workercount = parallel_processorcount();
        x->workercount  = MIN(workercount,PARALLEL_MAX_WORKERS);
        x->priority     = PARALLEL_PRIORITY_DEFAULT;
        x->benchmark    = 0;
        x->iteration    = 0;
        x->begintime    = 0;
        x->endtime      = 0;
        x->cancel       = 0;
        x->workers      = linklist_new();
        if (!x->workers) {
            freeobject((t_object *)x);
            x=NULL; 
        }
        for (i=0;i<x->workercount;i++) {
            w = parallel_worker_new(data,workerproc,x);
            if (w) {
                w->id = i;
                // set any other state
                linklist_append(x->workers,w);      
            } else {
                freeobject((t_object *)x);
                x=NULL; 
            }
        }
    }
    return x;
}

t_max_err parallel_task_execute(t_parallel_task *x)
{
    long i;
    t_parallel_thread_pool *pool;
    t_parallel_worker *worker;
    
    switch (x->priority) {
    case PARALLEL_PRIORITY_HIGH: 
        pool = &_pool_high;
        break;
    case PARALLEL_PRIORITY_MEDIUM: 
        pool = &_pool_med;
        break;
    case PARALLEL_PRIORITY_LOW: 
    default:
        pool = &_pool_low;
    }   

    x->begintime = x->benchmark ? systimer_gettime() : 0.;

    if (pool->count!=0) {
        error("our pool count isn't zero (%d), must be calling from the wrong thread...living dangerously",pool->count);    
        pool->count = 0;
    }
    
    for (i=0;i<(x->workercount-1);i++) {
        //lock and increment thread count
        systhread_mutex_lock(pool->count_mutex);
        pool->count++;
        systhread_mutex_unlock(pool->count_mutex);
        
        //lock worker to modify worker state. will block if worker is currently executing
        systhread_mutex_lock(pool->threads[i].mutex);
        parallel_thread_init((pool->threads)+i); // create systhread if not yet
        pool->threads[i].worker = (t_parallel_worker *)linklist_getindex(x->workers,i);
        pool->threads[i].state = PARALLEL_STATE_RUN;
        systhread_mutex_unlock(pool->threads[i].mutex);
        //signal the worker
        systhread_cond_signal(pool->threads[i].cond);
    }
    
    worker = (t_parallel_worker *)linklist_getindex(x->workers,x->workercount-1);   
    parallel_worker_execute(worker);

    // wait for all threads to complete
    systhread_mutex_lock(pool->count_mutex);
    while (pool->count>0) {
        systhread_cond_wait(pool->count_cond,pool->count_mutex);    
    }   
    systhread_mutex_unlock(pool->count_mutex);
    
    x->endtime = x->benchmark ? systimer_gettime() : 0.;
    
    return MAX_ERR_NONE;
}

t_max_err parallel_task_cancel(t_parallel_task *x)
{
    x->cancel = 1;
    return MAX_ERR_NONE;
}

void parallel_task_free(t_parallel_task *x)
{
    if (x->workers)
        object_free(x->workers);
}

void parallel_task_benchprint(t_parallel_task *x)
{
    t_parallel_worker *w;
    double avg=0,tmp;
    long i;
    
    jit_object_post((t_object *)x,"PARALLEL BENCH");
    jit_object_post((t_object *)x,"PROCESSOR COUNT: %d", parallel_processorcount());    
    jit_object_post((t_object *)x,"WORKER COUNT: %d", x->workercount);
    jit_object_post((t_object *)x,"TOTAL TASK TIME:  %f", x->endtime - x->begintime);
    for (i=0;i<x->workercount;i++) {
        if (x->workers&&(w=(t_parallel_worker *)linklist_getindex(x->workers,i))) {
            tmp = w->endtime - w->begintime;
            jit_object_post((t_object *)x,"WORKER[%d] TIME:  %f", i, tmp);
            avg += tmp;
        } else {
            jit_object_post((t_object *)x,"error trying to print worker[%d]",i);
        }
    }
    avg = avg/(double)x->workercount;
    jit_object_post((t_object *)x,"AVERAGE WORKER TIME: %f", avg);  
}                       

void parallel_task_data(t_parallel_task *x, void * data)
{
    t_parallel_worker *w;
    long i;
    
    x->data = data;
    for (i=0;i<x->workercount;i++) {
        if (x->workers&&(w=(t_parallel_worker *)linklist_getindex(x->workers,i))) {
            w->data = data;
        }
    }
}

void parallel_task_workerproc(t_parallel_task *x, method workerproc)
{
    t_parallel_worker *w;
    long i;
    
    x->workerproc = workerproc;
    for (i=0;i<x->workercount;i++) {
        if (x->workers&&(w=(t_parallel_worker *)linklist_getindex(x->workers,i))) {
            w->workerproc = workerproc;
        }
    }
}

// TODO: parallel_task_workercount

t_parallel_worker *parallel_worker_new(void *data, method workerproc, t_parallel_task *task)
{
    t_parallel_worker *x=NULL;
        
    if (workerproc && task && (x=(t_parallel_worker *)object_alloc(_parallel_worker_class))) {
        x->data         = data;
        x->workerproc   = workerproc;
        x->task         = task;
        x->begintime    = 0;
        x->endtime      = 0;
    }
    return x;
}

t_max_err parallel_worker_execute(t_parallel_worker *x)
{
    x->begintime = x->task->benchmark ? systimer_gettime() : 0.;            
    x->workerproc(x);   
    x->endtime = x->task->benchmark ? systimer_gettime() : 0.;
    
    return MAX_ERR_NONE;
}

void parallel_worker_free(t_parallel_worker *x)
{
    // nada for now
}

void parallel_thread_init(t_parallel_thread *x)
{
    if (!x->thread)
        systhread_create((method)parallel_threadproc,x,0,0,0,&x->thread);
}

void parallel_threadproc(t_parallel_thread *x)
{
    systhread_mutex_lock(x->mutex);
    while (TRUE) {
        while (x->state!=PARALLEL_STATE_RUN) {
            systhread_cond_wait(x->cond,x->mutex);
        }
        //if (x->state==PARALLEL_STATE_RUN) {
            parallel_worker_execute(x->worker);
            x->state=PARALLEL_STATE_DONE;
            systhread_mutex_lock(x->pool->count_mutex);
            x->pool->count--;
            systhread_mutex_unlock(x->pool->count_mutex);
            systhread_cond_broadcast(x->pool->count_cond); // could just use signal here since only one thread waiting
        //} 
    }
    systhread_mutex_unlock(x->mutex);
}

Copyright © 2008, Cycling '74