Mercurial > hg > Members > Moririn
view src/parallel_execution/TaskManagerImpl.cbc @ 395:864cd4e346e9
Fix taskManager
author | Tatsuki IHA <innparusu@cr.ie.u-ryukyu.ac.jp> |
---|---|
date | Tue, 01 Aug 2017 20:09:33 +0900 |
parents | 99c50356d917 |
children | fc4fcd441700 |
line wrap: on
line source
#include "../context.h" #include <stdio.h> void createWorkers(struct Context* context, TaskManager* taskManeger, TaskManagerImpl* taskManagerImpl); TaskManager* createTaskManagerImpl(struct Context* context, int numCPU, int numGPU, int numIO) { struct TaskManager* taskManager = new 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->spawnTasks = C_spawnTasksTaskManagerImpl; taskManager->spawn = C_spawnTaskManagerImpl; taskManager->shutdown = C_shutdownTaskManagerImpl; struct TaskManagerImpl* taskManagerImpl = new TaskManagerImpl(); taskManagerImpl -> activeQueue = createSingleLinkedQueue(context); taskManagerImpl -> taskQueue = createSingleLinkedQueue(context); taskManagerImpl -> numWorker = taskManager->maxCPU; taskManagerImpl -> loopCounter = new LoopCounter(); taskManagerImpl -> loopCounter -> i = 0; createWorkers(context, taskManager, taskManagerImpl); taskManager->taskManager = (union Data*)taskManagerImpl; return 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); taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); } for (;i<taskManager->cpu;i++) { #ifdef USE_CUDAWorker Queue* queue = createSynchronizedQueue(context); #ifndef USE_CUDA_MAIN_THREAD taskManagerImpl->workers[i] = (Worker*)createCUDAWorker(context, i, queue,0); #else taskManagerImpl->workers[i] = (Worker*)queue; #endif #else Queue* queue = createSynchronizedQueue(context); taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); #endif } for (;i<taskManager->maxCPU;i++) { Queue* queue = createSynchronizedQueue(context); taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); } } __code spawnTasksTaskManagerImpl(struct TaskManagerImpl* taskManagerImpl, struct Context** tasks, __code next1(...), struct TaskManager* taskManager) { int i = taskManagerImpl->loopCounter->i; if(i < GET_SIZE(tasks)) { struct Context* task = tasks[i]; taskManagerImpl->loopCounter->i++; task->taskManager = &taskManager->taskManager->TaskManager; taskManager->context = task; taskManager->next = C_spawnTasksTaskManagerImpl; goto meta(context, C_setWaitTask); } taskManagerImpl->loopCounter->i = 0; goto meta(context, C_spawnTasksTaskManagerImpl1); } __code spawnTasksTaskManagerImpl_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto spawnTasksTaskManagerImpl(context, taskManager, Gearef(context, TaskManager)->tasks, Gearef(context, TaskManager)->next1, Gearef(context, TaskManager)); } __code spawnTasksTaskManagerImpl1(struct TaskManagerImpl* taskManagerImpl, struct Context** tasks, __code next1(...), struct TaskManager* taskManager) { int i = taskManagerImpl->loopCounter->i; if(i < GET_SIZE(tasks)) { struct Context* task = tasks[i]; taskManagerImpl->loopCounter->i++; taskManager->context = task; taskManager->next = C_spawnTasksTaskManagerImpl1; goto meta(context, C_spawnTaskManagerImpl); } taskManagerImpl->loopCounter->i = 0; goto meta(context, next1); } __code spawnTasksTaskManagerImpl1_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto spawnTasksTaskManagerImpl1(context, taskManager, Gearef(context, TaskManager)->tasks, Gearef(context, TaskManager)->next1, Gearef(context, TaskManager)); } __code setWaitTask(struct Queue* queue, struct LoopCounter* loopCounter, struct Context* task, __code next(...)) { int i = loopCounter->i; if(task->idg+i < task->maxIdg) { queue->queue = (Data *)GET_WAIT_LIST(task->data[task->idg + i]); queue->next = C_setWaitTask; queue->data = (Data *)task; loopCounter->i++; goto meta(context, queue->queue->Queue.put); } loopCounter->i = 0; goto next(...); } __code setWaitTask_stub(struct Context* context) { struct Context* task = Gearef(context, TaskManager)->context; goto setWaitTask(context, Gearef(context, Queue), Gearef(task, LoopCounter), task, Gearef(context, TaskManager)->next); } __code spawnTaskManagerImpl(struct TaskManagerImpl* taskManager, struct Iterator* iterator, struct Context* task, __code next(...)) { if (task->idgCount == 0) { if(task->iterator != NULL && task->iterate == 0) { iterator->iterator = (union Data*)task->iterator; iterator->task = task; iterator->next = next; pthread_mutex_unlock(&taskManager->mutex); goto meta(context, task->iterator->exec); } goto meta(context, C_taskSend); } pthread_mutex_unlock(&taskManager->mutex); goto next(...); } __code spawnTaskManagerImpl_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); pthread_mutex_lock(&taskManager->mutex); goto spawnTaskManagerImpl(context, taskManager, Gearef(context, Iterator), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); } __code taskSend(struct TaskManagerImpl* taskManager, struct Queue* queue, struct Context* task, __code next(...)) { // set workerId task->workerId = taskManager->sendWorkerIndex; if(++taskManager->sendWorkerIndex >= taskManager->numWorker) { taskManager->sendWorkerIndex = 0; } struct Queue* tasks = taskManager->workers[task->workerId]->tasks; queue->queue = (union Data*)tasks; queue->data = (union Data*)task; queue->next = next; pthread_mutex_unlock(&taskManager->mutex); goto meta(context, tasks->put); } __code taskSend_stub(struct Context* context) { TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto taskSend(context, taskManager, Gearef(context, Queue), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); } __code shutdownTaskManagerImpl(struct LoopCounter* loopCounter, struct TaskManager* taskManager, struct TaskManagerImpl* taskManagerImpl, struct Queue* queue, __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 = C_shutdownTaskManagerImpl1; goto meta(context, tasks->put); } loopCounter->i = 0; goto meta(context, next); } __code shutdownTaskManagerImpl_stub(struct Context* context) { TaskManagerImpl* taskManagerImpl = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto shutdownTaskManagerImpl(context, Gearef(context, LoopCounter), &Gearef(context, TaskManager)->taskManager->TaskManager, taskManagerImpl, Gearef(context, Queue), Gearef(context, TaskManager)->next); } __code shutdownTaskManagerImpl1(struct LoopCounter* loopCounter, TaskManagerImpl* taskManagerImpl) { int i = loopCounter->i; pthread_join(taskManagerImpl->workers[i]->worker->CPUWorker.thread, NULL); loopCounter->i++; goto meta(context, C_shutdownTaskManagerImpl); } __code shutdownTaskManagerImpl1_stub(struct Context* context) { TaskManagerImpl* taskManagerImpl = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); goto shutdownTaskManagerImpl1(context, Gearef(context, LoopCounter), taskManagerImpl); }