Mercurial > hg > Members > Moririn
view src/parallel_execution/taskManager.c @ 269:5170539348ec
rename TaskManagerImpl.cbc
author | mir3636 |
---|---|
date | Sun, 29 Jan 2017 22:15:32 +0900 |
parents | cd3486e4ba70 |
children |
line wrap: on
line source
#include "context.h" #include "stack.h" #include "queue.h" #include "worker.h" #include "origin_cs.h" #include <stdio.h> void createWorkers(struct Context* context, TaskManager* taskManeger, TaskManagerImpl* taskManagerImpl); union Data* createTaskManager(struct Context* context, int numCPU, int numGPU, int numIO) { struct TaskManager* taskManager = &ALLOCATE(context, TaskManager)->TaskManager; // 0...numIO-1 IOProcessor // numIO...numIO+numGPU-1 GPUProcessor // numIO+numGPU...numIO+numGPU+numCPU-1 CPUProcessor taskManager->io = 0; taskManager->gpu = numIO; taskManager->cpu = numIO+numGPU; taskManager->maxCPU = numIO+numGPU+numCPU; taskManager->createTask = C_createTask; taskManager->spawn = C_spawnTaskManager; taskManager->shutdown = C_shutdownTaskManager; struct TaskManagerImpl* taskManagerImpl = &ALLOCATE(context, TaskManagerImpl)->TaskManagerImpl; taskManager->taskManager = (union Data*)taskManagerImpl; taskManagerImpl -> activeQueue = &createSingleLinkedQueue(context)->Queue; taskManagerImpl -> taskQueue = &createSingleLinkedQueue(context)->Queue; taskManagerImpl -> numWorker = taskManager->maxCPU; createWorkers(context, taskManager, taskManagerImpl); return (union Data*)(taskManager); } void createWorkers(struct Context* context, TaskManager* taskManager, TaskManagerImpl* taskManagerImpl) { int i = 0; taskManagerImpl->workers = (Worker**)ALLOC_ARRAY(context, Worker, taskManager->maxCPU); for (;i<taskManager->gpu;i++) { Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); } for (;i<taskManager->cpu;i++) { #ifdef USE_CUDA #else Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); #endif } for (;i<taskManager->maxCPU;i++) { Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); } } __code createTask(struct Context* context, TaskManager* taskManager) { taskManager->context = NEW(struct Context); initContext(taskManager->context); goto meta(context, C_setWorker); } __code createTask_stub(struct Context* context) { goto createTask(context, Gearef(context,TaskManager)); } __code setWorker(struct Context* context, TaskManagerImpl* taskManager, struct Context* task, enum Code next) { task->workerId = taskManager->sendWorkerIndex; if(++taskManager->sendWorkerIndex >= taskManager->numWorker) { taskManager->sendWorkerIndex = 0; } goto meta(context, next); } __code setWorker_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto setWorker(context, taskManager, Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); } __code spawnTaskManager(struct Context* context, struct TaskManagerImpl* taskManager, struct Queue* queue, struct Context* task, enum Code next) { if (task->idgCount == 0) { // enqueue activeQueue queue->queue = (union Data*)taskManager->activeQueue; } else { // enqueue waitQueue queue->queue = (union Data*)taskManager->taskQueue; } queue->data = (union Data*)task; queue->next = C_spawnTaskManager1; goto meta(context, queue->queue->Queue.put); } __code spawnTaskManager_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); pthread_mutex_lock(&taskManager->mutex); goto spawnTaskManager(context, taskManager, Gearef(context, Queue), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); } __code spawnTaskManager1(struct Context* context, struct TaskManagerImpl* taskManager) { pthread_mutex_unlock(&taskManager->mutex); goto meta(context, C_taskSend); } __code spawnTaskManager1_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto spawnTaskManager1(context, taskManager); } __code taskSend(struct Context* context, TaskManagerImpl* taskManager, Queue* queue) { queue->queue = (union Data*)taskManager->activeQueue; queue->next = C_taskSend1; goto meta(context, taskManager->activeQueue->take); } __code taskSend_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto taskSend(context, taskManager, Gearef(context, Queue)); } __code taskSend1(struct Context* context, TaskManagerImpl* taskManager, Queue* queue, struct Context* task, enum Code next) { struct Queue* tasks = taskManager->workers[task->workerId]->tasks; queue->queue = (union Data*)tasks; queue->data = (union Data*)task; queue->next = next; pthread_cond_signal(&taskManager->workers[task->workerId]->worker->CPUWorker.cond); goto meta(context, tasks->put); } __code taskSend1_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto taskSend1(context, taskManager, Gearef(context, Queue), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); } __code shutdownTaskManager(struct Context* context, struct LoopCounter* loopCounter, TaskManager* taskManager, TaskManagerImpl* taskManagerImpl, Queue* queue, enum Code next) { int i = loopCounter->i; if (taskManager->cpu <= i && i < taskManager->maxCPU) { struct Queue* tasks = taskManagerImpl->workers[i]->tasks; queue->queue = (union Data*)tasks; queue->data = NULL; queue->next = next; goto meta(context, tasks->put); pthread_join(taskManagerImpl->workers[i]->worker->CPUWorker.thread, NULL); loopCounter->i++; goto meta(context, C_shutdownTaskManager); } loopCounter->i = 0; goto meta(context, taskManager->next); } __code shutdownTaskManager_stub(struct Context* context) { TaskManagerImpl* taskManagerImpl = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto shutdownTaskManager(context, Gearef(context, LoopCounter), Gearef(context, TaskManager), taskManagerImpl, Gearef(context, Queue), Gearef(context, TaskManager)->next); }