Mercurial > hg > Members > Moririn
view src/parallel_execution/CPUWorker.cbc @ 462:8d7e5d48cad3
Running CPU examples
author | Tatsuki IHA <innparusu@cr.ie.u-ryukyu.ac.jp> |
---|---|
date | Wed, 20 Dec 2017 22:05:08 +0900 |
parents | 57c715bd6283 |
children | ac244346c85d |
line wrap: on
line source
#include "../context.h" #include "TaskManager.h" static void startWorker(Worker* worker); Worker* createCPUWorker(struct Context* context, int id, Queue* queue) { struct Worker* worker = new Worker(); struct CPUWorker* cpuWorker = new CPUWorker(); worker->worker = (union Data*)cpuWorker; worker->tasks = queue; cpuWorker->id = id; worker->taskReceive = C_taskReceiveWorker; worker->shutdown = C_shutdownWorker; pthread_create(&worker->thread, NULL, (void*)&startWorker, worker); return worker; } static void startWorker(Worker* worker) { CPUWorker* cpuWorker = (CPUWorker*)worker->worker; cpuWorker->context = NEW(struct Context); initContext(cpuWorker->context); Gearef(cpuWorker->context, Worker)->worker = (union Data*)worker; goto meta(cpuWorker->context, worker->taskReceive); } __code taskReceiveWorker(struct Worker* worker,struct Queue* queue) { queue->queue = (union Data*)worker->tasks; queue->next = C_getTask; goto meta(context, worker->tasks->take); } __code taskReceiveWorker_stub(struct Context* context) { goto taskReceiveWorker(context, &Gearef(context, Worker)->worker->Worker, Gearef(context, Queue)); } __code getTask(struct Worker* worker, struct Context* task) { if (!task) goto meta(context, worker->shutdown); // end thread task->worker = worker; enum Code taskCg = task->next; if (task->iterate) { task->next = C_iterateCommit; } else { task->next = C_odgCommit; // set CG after task exec } goto meta(task, taskCg); } __code getTask_stub(struct Context* context) { Worker* worker = &Gearef(context,Worker)->worker->Worker; struct Context* task = &Gearef(context, Queue)->data->Context; goto getTask(context, worker, task); } __code iterateCommit(struct Iterator* iterator) { iterator->iterator = (union Data*)context->iterator; iterator->task = context; iterator->next = C_odgCommit; iterator->whenWait = C_iterateCommit1; goto meta(context, context->iterator->barrier); } __code iterateCommit1(struct Context* task) { goto meta(context, C_taskReceiveWorker); } __code iterateCommit1_stub(struct Context* context) { // switch worker context struct Context* workerContext = context->worker->worker->CPUWorker.context; goto iterateCommit1(workerContext, context); } __code odgCommit(struct LoopCounter* loopCounter, struct Context* task, struct TaskManager* taskManager) { int i = loopCounter->i ; if (task->odg+i < task->maxOdg) { goto meta(task, C_odgCommit1); } loopCounter->i = 0; taskManager->taskManager = (union Data*)task->taskManager; taskManager->next = C_taskReceiveWorker; goto meta(context, task->taskManager->decrementTaskCount); } __code odgCommit_stub(struct Context* context) { // switch worker context struct Context* workerContext = context->worker->worker->CPUWorker.context; goto odgCommit(workerContext, Gearef(context, LoopCounter), context, Gearef(workerContext, TaskManager)); } __code odgCommit1(struct LoopCounter* loopCounter, struct Queue* queue) { int i = loopCounter->i ; queue->queue = (union Data*)GET_WAIT_LIST(context->data[context->odg+i]); queue->whenEmpty = C_odgCommit4; queue->next = C_odgCommit2; goto meta(context, queue->queue->Queue.isEmpty); } __code odgCommit1_stub(struct Context* context) { goto odgCommit1(context, Gearef(context, LoopCounter), Gearef(context, Queue)); } __code odgCommit2(struct Queue* queue) { queue->next = C_odgCommit3; goto meta(context, queue->queue->Queue.take); } __code odgCommit2_stub(struct Context* context) { goto odgCommit2(context, Gearef(context, Queue)); } __code odgCommit3(struct TaskManager* taskManager, struct Context* task) { if (__sync_fetch_and_sub(&task->idgCount, 1) == 1) { // atomic decrement idg counter(__sync_fetch_and_sub function return initial value of task->idgCount point) taskManager->taskManager = (union Data*)task->taskManager; taskManager->task = task; taskManager->next = C_odgCommit1; goto meta(context, task->taskManager->spawn); } goto meta(context, C_odgCommit1); } __code odgCommit3_stub(struct Context* context) { struct Context* task = &Gearef(context, Queue)->data->Context; goto odgCommit3(context, Gearef(context, TaskManager), task); } __code odgCommit4(struct LoopCounter* loopCounter) { loopCounter->i++; goto meta(context, C_odgCommit); } __code odgCommit4_stub(struct Context* context) { goto odgCommit4(context, Gearef(context, LoopCounter)); } __code shutdownWorker(struct CPUWorker* worker) { goto meta(context, C_exit_code); }