Mercurial > hg > Members > Moririn
annotate src/parallel_execution/TaskManagerImpl.cbc @ 317:51aa65676e37
CUDAtwice.cbc is called now
| author | Shinji KONO <kono@ie.u-ryukyu.ac.jp> |
|---|---|
| date | Wed, 15 Feb 2017 16:28:25 +0900 |
| parents | 54d203daf06b |
| children | a15511b1a6e0 |
| rev | line source |
|---|---|
| 269 | 1 #include "../context.h" |
| 278 | 2 |
| 269 | 3 #include <stdio.h> |
| 4 | |
| 5 void createWorkers(struct Context* context, TaskManager* taskManeger, TaskManagerImpl* taskManagerImpl); | |
| 6 | |
| 280 | 7 TaskManager* createTaskManagerImpl(struct Context* context, int numCPU, int numGPU, int numIO) { |
| 269 | 8 struct TaskManager* taskManager = new TaskManager(); |
| 9 // 0...numIO-1 IOProcessor | |
| 10 // numIO...numIO+numGPU-1 GPUProcessor | |
| 11 // numIO+numGPU...numIO+numGPU+numCPU-1 CPUProcessor | |
| 12 taskManager->io = 0; | |
| 13 taskManager->gpu = numIO; | |
| 14 taskManager->cpu = numIO+numGPU; | |
| 15 taskManager->maxCPU = numIO+numGPU+numCPU; | |
| 16 taskManager->createTask = C_createTask; | |
| 17 taskManager->spawn = C_spawnTaskManager; | |
| 18 taskManager->shutdown = C_shutdownTaskManager; | |
| 19 struct TaskManagerImpl* taskManagerImpl = new TaskManagerImpl(); | |
| 280 | 20 taskManagerImpl -> activeQueue = createSingleLinkedQueue(context); |
| 21 taskManagerImpl -> taskQueue = createSingleLinkedQueue(context); | |
| 269 | 22 taskManagerImpl -> numWorker = taskManager->maxCPU; |
| 23 createWorkers(context, taskManager, taskManagerImpl); | |
|
316
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
24 taskManager->taskManager = (union Data*)taskManagerImpl; |
| 269 | 25 return taskManager; |
| 26 } | |
| 27 | |
| 28 void createWorkers(struct Context* context, TaskManager* taskManager, TaskManagerImpl* taskManagerImpl) { | |
| 29 int i = 0; | |
| 30 taskManagerImpl->workers = (Worker**)ALLOC_ARRAY(context, Worker, taskManager->maxCPU); | |
| 31 for (;i<taskManager->gpu;i++) { | |
| 280 | 32 Queue* queue = createSynchronizedQueue(context); |
| 269 | 33 taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); |
| 34 } | |
| 35 for (;i<taskManager->cpu;i++) { | |
|
312
7dd5a7d52a67
USE_CUDAWorker flag only for CUDAtwice
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
288
diff
changeset
|
36 #ifdef USE_CUDAWorker |
|
7dd5a7d52a67
USE_CUDAWorker flag only for CUDAtwice
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
288
diff
changeset
|
37 Queue* queue = createSynchronizedQueue(context); |
|
316
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
38 // taskManagerImpl->workers[i] = (Worker*)createCUDAWorker(context, i, queue,0); |
|
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
39 taskManagerImpl->workers[i] = (Worker*)queue; |
| 269 | 40 #else |
| 280 | 41 Queue* queue = createSynchronizedQueue(context); |
| 269 | 42 taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); |
| 43 #endif | |
| 44 } | |
| 45 for (;i<taskManager->maxCPU;i++) { | |
| 280 | 46 Queue* queue = createSynchronizedQueue(context); |
| 269 | 47 taskManagerImpl->workers[i] = (Worker*)createCPUWorker(context, i, queue); |
| 48 } | |
| 49 } | |
| 50 | |
| 51 __code createTask(struct TaskManager* taskManager) { | |
|
316
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
52 TaskManager *t = (TaskManager *)taskManager->taskManager; |
|
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
53 TaskManagerImpl *im = (TaskManagerImpl *)t->taskManager; |
|
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
54 |
| 269 | 55 taskManager->context = NEW(struct Context); |
| 56 initContext(taskManager->context); | |
| 288 | 57 taskManager->context->taskManager = taskManager; |
|
316
54d203daf06b
CUDAtwice.cbc is called.
Shinji KONO <kono@ie.u-ryukyu.ac.jp>
parents:
312
diff
changeset
|
58 struct Queue* tasks = im->workers[0]->tasks; |
| 280 | 59 goto meta(context, C_setWorker); |
| 269 | 60 } |
| 61 | |
| 62 __code setWorker(struct TaskManagerImpl* taskManager, struct Context* task, __code next(...)) { | |
| 63 task->workerId = taskManager->sendWorkerIndex; | |
| 64 if(++taskManager->sendWorkerIndex >= taskManager->numWorker) { | |
| 65 taskManager->sendWorkerIndex = 0; | |
| 66 } | |
| 67 goto next(...); | |
| 68 } | |
| 69 | |
| 280 | 70 __code setWorker_stub(struct Context* context) { |
| 71 TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); | |
| 72 goto setWorker(context, taskManager, Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); | |
| 73 } | |
| 74 | |
| 269 | 75 __code spawnTaskManager(struct TaskManagerImpl* taskManager, struct Queue* queue, struct Context* task, __code next(...)) { |
| 76 if (task->idgCount == 0) { | |
| 77 // enqueue activeQueue | |
| 78 queue->queue = (union Data*)taskManager->activeQueue; | |
| 79 } else { | |
| 80 // enqueue waitQueue | |
| 81 queue->queue = (union Data*)taskManager->taskQueue; | |
| 82 } | |
| 83 queue->data = (union Data*)task; | |
| 84 queue->next = C_spawnTaskManager1; | |
| 85 goto meta(context, queue->queue->Queue.put); | |
| 86 } | |
| 87 | |
| 88 __code spawnTaskManager_stub(struct Context* context) { | |
| 89 TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); | |
| 90 pthread_mutex_lock(&taskManager->mutex); | |
| 91 goto spawnTaskManager(context, | |
| 92 taskManager, | |
| 93 Gearef(context, Queue), | |
| 94 Gearef(context, TaskManager)->context, | |
| 95 Gearef(context, TaskManager)->next); | |
| 96 } | |
| 97 | |
| 98 | |
| 99 __code spawnTaskManager1(struct TaskManagerImpl* taskManager) { | |
| 100 pthread_mutex_unlock(&taskManager->mutex); | |
| 280 | 101 goto meta(context, C_taskSend); |
| 269 | 102 } |
| 103 | |
| 104 __code spawnTaskManager1_stub(struct Context* context) { | |
| 105 TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); | |
| 106 goto spawnTaskManager1(context, | |
| 107 taskManager); | |
| 108 } | |
| 109 | |
| 110 __code taskSend(struct TaskManagerImpl* taskManager, struct Queue* queue) { | |
| 111 queue->queue = (union Data*)taskManager->activeQueue; | |
| 112 queue->next = C_taskSend1; | |
| 113 goto meta(context, taskManager->activeQueue->take); | |
| 114 } | |
| 115 | |
| 116 __code taskSend1(struct TaskManagerImpl* taskManager, struct Queue* queue, struct Context* task, __code next(...)) { | |
| 117 struct Queue* tasks = taskManager->workers[task->workerId]->tasks; | |
| 118 queue->queue = (union Data*)tasks; | |
| 119 queue->data = (union Data*)task; | |
| 120 queue->next = next; | |
| 121 goto meta(context, tasks->put); | |
| 122 } | |
| 123 | |
| 124 __code taskSend1_stub(struct Context* context) { | |
| 125 TaskManagerImpl* taskManager = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); | |
| 126 goto taskSend1(context, taskManager, Gearef(context, Queue), Gearef(context, TaskManager)->context, Gearef(context, TaskManager)->next); | |
| 127 } | |
| 128 | |
| 129 __code shutdownTaskManager(struct LoopCounter* loopCounter, struct TaskManager* taskManager, struct TaskManagerImpl* taskManagerImpl, struct Queue* queue, __code next(...)) { | |
| 130 int i = loopCounter->i; | |
| 131 if (taskManager->cpu <= i && i < taskManager->maxCPU) { | |
| 132 struct Queue* tasks = taskManagerImpl->workers[i]->tasks; | |
| 133 queue->queue = (union Data*)tasks; | |
| 134 queue->data = NULL; | |
|
282
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
135 queue->next = C_shutdownTaskManager1; |
| 269 | 136 goto meta(context, tasks->put); |
| 137 } | |
| 138 | |
| 139 loopCounter->i = 0; | |
|
282
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
140 goto meta(context, next); |
| 269 | 141 } |
| 142 | |
| 143 __code shutdownTaskManager_stub(struct Context* context) { | |
| 144 TaskManagerImpl* taskManagerImpl = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); | |
|
282
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
145 goto shutdownTaskManager(context, Gearef(context, LoopCounter), &Gearef(context, TaskManager)->taskManager->TaskManager, taskManagerImpl, Gearef(context, Queue), Gearef(context, TaskManager)->next); |
| 269 | 146 } |
|
282
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
147 |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
148 __code shutdownTaskManager1(struct LoopCounter* loopCounter, TaskManagerImpl* taskManagerImpl) { |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
149 int i = loopCounter->i; |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
150 pthread_join(taskManagerImpl->workers[i]->worker->CPUWorker.thread, NULL); |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
151 loopCounter->i++; |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
152 goto meta(context, C_shutdownTaskManager); |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
153 } |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
154 |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
155 __code shutdownTaskManager1_stub(struct Context* context) { |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
156 TaskManagerImpl* taskManagerImpl = (TaskManagerImpl*)GearImpl(context, TaskManager, taskManager); |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
157 goto shutdownTaskManager1(context, Gearef(context, LoopCounter), taskManagerImpl); |
|
a3448b0f0a56
Add input data gear
Tatsuki IHA <e125716@ie.u-ryukyu.ac.jp>
parents:
280
diff
changeset
|
158 } |