#include "../context.h" #interface "TaskManager.h" #interface "Iterator.h" #interface "Queue.h" #interface "Worker.h" #include #include void createWorkers(struct Context* context, MCTaskManagerImpl* taskManager); TaskManager* createMCTaskManagerImpl(struct Context* context, int numCPU, int numGPU, int numIO) { struct TaskManager* taskManager = new TaskManager(); taskManager->spawnTasks = C_spawnTasksMCTaskManagerImpl; taskManager->spawn = C_spawnMCTaskManagerImpl; taskManager->shutdown = C_shutdownMCTaskManagerImpl; taskManager->incrementTaskCount = C_incrementTaskCountMCTaskManagerImpl; taskManager->decrementTaskCount = C_decrementTaskCountMCTaskManagerImpl; taskManager->setWaitTask = C_setWaitTaskMCTaskManagerImpl; struct MCTaskManagerImpl* taskManagerImpl = new MCTaskManagerImpl(); // 0...numIO-1 IOProcessor // numIO...numIO+numGPU-1 GPUProcessor // numIO+numGPU...numIO+numGPU+numCPU-1 CPUProcessor taskManagerImpl->io = 0; taskManagerImpl->gpu = numIO; taskManagerImpl->cpu = numIO+numGPU; taskManagerImpl->maxCPU = numIO+numGPU+numCPU; taskManagerImpl->numWorker = taskManagerImpl->maxCPU; taskManagerImpl->sendGPUWorkerIndex = taskManagerImpl->gpu; taskManagerImpl->sendCPUWorkerIndex = taskManagerImpl->cpu; taskManagerImpl->taskCount = 0; taskManagerImpl->loopCounter = 0; taskManagerImpl->mem = 0; context->taskManager = taskManager; taskManager->taskManager = (union Data*)taskManagerImpl; createWorkers(context, taskManagerImpl); return taskManager; } void createWorkers(struct Context* context, MCTaskManagerImpl* taskManager) { int i = 0; taskManager->workers = (Worker**)ALLOCATE_PTR_ARRAY(context, Worker, taskManager->maxCPU); for (;igpu;i++) { Queue* queue = createSynchronizedQueue(context); taskManager->workers[i] = (Worker*)createCPUWorker(context, i, queue); } for (;icpu;i++) { Queue* queue = createSynchronizedQueue(context); #ifdef USE_CUDAWorker taskManager->workers[i] = (Worker*)createCUDAWorker(context, i, queue,0); #else taskManager->workers[i] = (Worker*)createMCWorker(context, i, queue); #endif } for (;imaxCPU;i++) { Queue* queue = createSynchronizedQueue(context); taskManager->workers[i] = (Worker*)createMCWorker(context, i, queue); } } __code spawnTasksMCTaskManagerImpl(struct MCTaskManagerImpl* taskManager, struct Element* taskList, __code next1(...)) { taskManager->taskList = taskList; goto spawnTasksMCTaskManagerImpl1(); } __code spawnTasksMCTaskManagerImpl1(struct MCTaskManagerImpl* taskManagerImpl, struct TaskManager* taskManager) { if (taskManagerImpl->taskList == NULL) { goto spawnTasksMCTaskManagerImpl2(); } struct Context* task = (struct Context*)taskManagerImpl->taskList->data; taskManagerImpl->taskList = taskManagerImpl->taskList->next; goto taskManager->setWaitTask(task, spawnTasksMCTaskManagerImpl1); } __code spawnTasksMCTaskManagerImpl1_stub(struct Context* context) { MCTaskManagerImpl* taskManagerImpl = (MCTaskManagerImpl*)GearImpl(context, TaskManager, taskManager); TaskManager* taskManager = &Gearef(context, TaskManager)->taskManager->TaskManager; goto spawnTasksMCTaskManagerImpl1(context, taskManagerImpl, taskManager); } __code spawnTasksMCTaskManagerImpl2(struct MCTaskManagerImpl* taskManager, struct Element* taskList, __code next1(...)) { taskManager->taskList = taskList; goto spawnTasksMCTaskManagerImpl3(); } __code spawnTasksMCTaskManagerImpl3(struct MCTaskManagerImpl* taskManagerImpl, __code next1(...), struct TaskManager* taskManager) { if (taskManagerImpl->taskList == NULL) { struct Queue* tasks = taskManagerImpl->workers[0]->tasks; // printf("put NULL\n"); goto tasks->put(NULL, shutdownMCTaskManagerImpl1); } struct Context* task = (struct Context*)taskManagerImpl->taskList->data; taskManagerImpl->taskList = taskManagerImpl->taskList->next; goto taskManager->spawn(task, spawnTasksMCTaskManagerImpl3); } __code spawnTasksMCTaskManagerImpl3_stub(struct Context* context) { MCTaskManagerImpl* taskManagerImpl = (MCTaskManagerImpl*)GearImpl(context, TaskManager, taskManager); enum Code next1 = Gearef(context, TaskManager)->next1; TaskManager* taskManager = &Gearef(context, TaskManager)->taskManager->TaskManager; goto spawnTasksMCTaskManagerImpl3(context, taskManagerImpl, next1, taskManager); } __code setWaitTaskMCTaskManagerImpl(struct MCTaskManagerImpl* taskManager, struct Context* task, __code next(...)) { int i = taskManager->loopCounter; if (task->idg+i < task->maxIdg) { struct Queue* queue = GET_WAIT_LIST(task->data[task->idg + i]); taskManager->loopCounter++; goto queue->put(task, setWaitTaskMCTaskManagerImpl); } taskManager->loopCounter = 0; goto incrementTaskCountMCTaskManagerImpl(); } __code incrementTaskCountMCTaskManagerImpl(struct MCTaskManagerImpl* taskManager, __code next(...)) { __sync_fetch_and_add(&taskManager->taskCount, 1); goto next(...); } __code decrementTaskCountMCTaskManagerImpl(struct MCTaskManagerImpl* taskManager, __code next(...)) { __sync_fetch_and_sub(&taskManager->taskCount, 1); goto next(...); } __code spawnMCTaskManagerImpl(struct MCTaskManagerImpl* taskManagerImpl, struct Context* task, __code next(...), struct TaskManager* taskManager) { if (task == NULL) { // get all task for model checking // start model checker // we should goto odgCommitMCWorker foreach mctask Gearef(context, Worker)->worker = (union Data*)taskManagerImpl->workers[0]; goto startModelChecker(Gearef(context, Worker)->worker); } task->taskManager = taskManager; if (task->idgCount == 0) { // iterator task is normal task until spawned if (task->iterator != NULL && task->iterate == 0) { pthread_mutex_unlock(&taskManagerImpl->mutex); struct Iterator* iterator = task->iterator; goto iterator->exec(task, taskManagerImpl->cpu - taskManagerImpl->gpu, next(...)); } goto taskSend(); } pthread_mutex_unlock(&taskManagerImpl->mutex); goto next(...); } __code spawnMCTaskManagerImpl_stub(struct Context* context) { MCTaskManagerImpl* taskManagerImpl = (MCTaskManagerImpl*)GearImpl(context, TaskManager, taskManager); struct Context* task = Gearef(context, TaskManager)->task; TaskManager* taskManager = &Gearef(context, TaskManager)->taskManager->TaskManager; goto spawnMCTaskManagerImpl(context, taskManagerImpl, task, Gearef(context, TaskManager)->next, taskManager); } __code taskSend(struct MCTaskManagerImpl* taskManager, struct Context* task, __code next(...)) { // set workerId if (task->gpu) { goto taskSend1(); } else { goto taskSend2(); } } __code taskSend1(struct MCTaskManagerImpl* taskManager, struct Context* task, __code next(...)) { int workerId = taskManager->sendGPUWorkerIndex; if (++taskManager->sendGPUWorkerIndex >= taskManager->cpu) { taskManager->sendGPUWorkerIndex = taskManager->gpu; } pthread_mutex_unlock(&taskManager->mutex); struct Queue* queue = taskManager->workers[workerId]->tasks; // printf("tasks->put taskSend1\n"); goto queue->put(task, next(...)); } __code taskSend2(struct MCTaskManagerImpl* taskManager, struct Context* task, __code next(...)) { int workerId = taskManager->sendCPUWorkerIndex; if (++taskManager->sendCPUWorkerIndex >= taskManager->maxCPU) { taskManager->sendCPUWorkerIndex = taskManager->cpu; } pthread_mutex_unlock(&taskManager->mutex); struct Queue* queue = taskManager->workers[workerId]->tasks; // printf("tasks->put taskSend2\n"); goto queue->put(task, next(...)); } __code shutdownMCTaskManagerImpl(struct MCTaskManagerImpl* taskManager, __code next(...)) { if (taskManager->taskCount != 0) { usleep(1000); goto shutdownMCTaskManagerImpl(); } int workerId = taskManager->sendCPUWorkerIndex; struct Queue* tasks = taskManager->workers[workerId]->tasks; // printf("tasks->put shutdown\n"); goto tasks->put(NULL, shutdownMCTaskManagerImpl1); } __code shutdownMCTaskManagerImpl1(struct MCTaskManagerImpl* taskManager, __code next(...)) { taskManager->loopCounter++; goto shutdownMCTaskManagerImpl(); } __code shutdownMCTaskManagerImpl2(struct MCTaskManagerImpl* taskManager, __code next(...)) { int i = taskManager->loopCounter; if (i < taskManager->numWorker) { pthread_join(taskManager->workers[i]->thread, NULL); taskManager->loopCounter++; goto shutdownMCTaskManagerImpl2(); } taskManager->loopCounter = 0; goto next(...); }