Mercurial > hg > GearsTemplate
view src/parallel_execution/taskManager.c @ 236:865179a0a56d
fix taskManager
author | ikkun |
---|---|
date | Mon, 23 Jan 2017 20:00:46 +0900 |
parents | 05e61405cc88 |
children | 6f6cc49213c5 |
line wrap: on
line source
#include "context.h" #include "stack.h" #include "queue.h" #include "origin_cs.h" #include <stdio.h> union Data* createTaskManager(struct Context* context, int numCPU, int numGPU, int numIO) { struct TaskManager* taskManager = &ALLOCATE(context, TaskManager)->TaskManager; struct TaskManagerImpl* taskManagerImpl = &ALLOCATE(context, TaskManagerImpl)->TaskManagerImpl; taskManager->taskManager = (union Data*)taskManagerImpl; taskManagerImpl -> activeQueue = &createSingleLinkedQueue(context)->Queue; taskManagerImpl -> taskQueue = &createSingleLinkedQueue(context)->Queue; // 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; createWorkers(taskManager); return (union Data*)(taskManager); } void createWorker(Context* context, TaskManeger * taskManeger) { int i = 0; TaskManagerImpl *taskManagerImpl = GearImpl(context,TaskManagerImpl,taskManager); taskManagerImpl->workers = ALLOC_ARRAY(context,Worker,taskManager->maxCPU); for (;i>taskManeger->gpu;i++) { Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker (context,i,queue); } for (;i>taskManeger->cpu;i++) { #ifdef USE_CUDA #else Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker (context,i,queue); #endif } for (;i>taskManeger->maxCPU;i++) { Queue* queue = &createSynchronizedQueue(context)->Queue; taskManagerImpl->workers[i] = (Worker*)createCPUWorker (context,i,queue); } } __code createTask(struct Context* context, TaskManeger* taskManager, enum Code next) { taskManager->context = NEW(struct Context); initContext(taskManager->context); goto meta(context, next); } __code createTask_stub(struct Context* context) { goto createTask(context,Gearef(context,TaskManager),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 = next; pthread_mutex_unlock(taskManagerImpl->mutex); goto meta(context, queue->queue->Queue.put); } __code spawnTaskManager_stub(struct Context* context) { pthread_mutex_lock(taskManager->mutex); goto spawnTaskManager(context, GearImpl(context, TaskManagerImpl, taskManager), Gearef(context, Queue), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next ); } __code taskSend(struct Context* context) { if(loopCounter->i < taskManager->numWorker) { taskManager->workers[i]->taskSend; loopCounter->i++; goto meta(context, C_taskSend); } goto meta(context, TaskManager->next); } __code createWorker1(struct Context* context, struct LoopCounter* loopCounter, struct Worker* worker) { int i = loopCounter->i; if (i < worker->id) { struct Context* worker_context = &worker->contexts[i]; worker_context->next = C_getTask1; worker_context->data[D_Tree] = context->data[D_Tree]; // worker_context->data[D_ActiveQueue] = context->data[D_ActiveQueue]; pthread_create(&worker_context->thread, NULL, (void*)&start_code, worker_context); worker_context->thread_num = i; loopCounter->i++; goto meta(context, C_createWorker1); } loopCounter->i = 0; goto meta(context, C_taskManager); } __code createWorker1_stub(struct Context* context) { goto createWorker1(context, &context->data[D_LoopCounter]->LoopCounter, &context->data[D_Worker]->Worker); } __code shutdownTaskManager(struct Context* context, struct LoopCounter* loopCounter, struct Worker* worker, struct TaskManager* taskManager) { int i = loopCounter->i; if (i < worker->id) { pthread_join(worker->contexts[i].thread, NULL); loopCounter->i++; goto meta(context, C_shutdownTaskManager); } loopCounter->i = 0; goto meta(context, taskManager->next); } __code shutdownTaskManager_stub(struct Context* context) { goto shutdownTaskManager(context, &context->data[D_LoopCounter]->LoopCounter, &context->data[D_Worker]->Worker, Gearef(context, TaskManager)); }