92 for (uint i = 0; i < workers; i++) {
93 tc->do_thread(worker(i));
94 }
95 }
96
97 static void run_foreground_task_if_needed(AbstractGangTask* task, uint num_workers,
98 bool add_foreground_work) {
99 if (add_foreground_work) {
100 log_develop_trace(gc, workgang)("Running work gang: %s task: %s worker: foreground",
101 Thread::current()->name(), task->name());
102 task->work(num_workers);
103 log_develop_trace(gc, workgang)("Finished work gang: %s task: %s worker: foreground "
104 "thread: " PTR_FORMAT, Thread::current()->name(), task->name(), p2i(Thread::current()));
105 }
106 }
107
108 // WorkGang dispatcher implemented with semaphores.
109 //
110 // Semaphores don't require the worker threads to re-claim the lock when they wake up.
111 // This helps lowering the latency when starting and stopping the worker threads.
112 class SemaphoreGangTaskDispatcher : public GangTaskDispatcher {
113 // The task currently being dispatched to the GangWorkers.
114 AbstractGangTask* _task;
115
116 volatile uint _started;
117 volatile uint _not_finished;
118
119 // Semaphore used to start the GangWorkers.
120 Semaphore* _start_semaphore;
121 // Semaphore used to notify the coordinator that all workers are done.
122 Semaphore* _end_semaphore;
123
124 public:
125 SemaphoreGangTaskDispatcher() :
126 _task(NULL),
127 _started(0),
128 _not_finished(0),
129 _start_semaphore(new Semaphore()),
130 _end_semaphore(new Semaphore())
131 { }
132
133 ~SemaphoreGangTaskDispatcher() {
134 delete _start_semaphore;
135 delete _end_semaphore;
136 }
137
138 void coordinator_execute_on_workers(AbstractGangTask* task, uint num_workers, bool add_foreground_work) {
139 // No workers are allowed to read the state variables until they have been signaled.
140 _task = task;
141 _not_finished = num_workers;
142
143 // Dispatch 'num_workers' number of tasks.
144 _start_semaphore->signal(num_workers);
145
146 run_foreground_task_if_needed(task, num_workers, add_foreground_work);
147
148 // Wait for the last worker to signal the coordinator.
149 _end_semaphore->wait();
150
151 // No workers are allowed to read the state variables after the coordinator has been signaled.
152 assert(_not_finished == 0, "%d not finished workers?", _not_finished);
153 _task = NULL;
154 _started = 0;
155
156 }
157
158 WorkData worker_wait_for_task() {
159 // Wait for the coordinator to dispatch a task.
160 _start_semaphore->wait();
161
162 uint num_started = Atomic::add(&_started, 1u);
163
164 // Subtract one to get a zero-indexed worker id.
165 uint worker_id = num_started - 1;
166
167 return WorkData(_task, worker_id);
168 }
169
170 void worker_done_with_task() {
171 // Mark that the worker is done with the task.
172 // The worker is not allowed to read the state variables after this line.
173 uint not_finished = Atomic::sub(&_not_finished, 1u);
174
175 // The last worker signals to the coordinator that all work is completed.
176 if (not_finished == 0) {
177 _end_semaphore->signal();
178 }
179 }
180 };
181
182 WorkGang::WorkGang(const char* name,
183 uint workers,
184 bool are_GC_task_threads,
185 bool are_ConcurrentGC_threads) :
186 AbstractWorkGang(name, workers, are_GC_task_threads, are_ConcurrentGC_threads),
187 _dispatcher(new SemaphoreGangTaskDispatcher())
188 { }
189
190 WorkGang::~WorkGang() {
191 delete _dispatcher;
192 }
193
194 AbstractGangWorker* WorkGang::allocate_worker(uint worker_id) {
195 return new GangWorker(this, worker_id);
196 }
197
198 void WorkGang::run_task(AbstractGangTask* task) {
199 run_task(task, active_workers());
200 }
201
202 void WorkGang::run_task(AbstractGangTask* task, uint num_workers, bool add_foreground_work) {
203 guarantee(num_workers <= total_workers(),
204 "Trying to execute task %s with %u workers which is more than the amount of total workers %u.",
205 task->name(), num_workers, total_workers());
206 guarantee(num_workers > 0, "Trying to execute task %s with zero workers", task->name());
207 uint old_num_workers = _active_workers;
|
92 for (uint i = 0; i < workers; i++) {
93 tc->do_thread(worker(i));
94 }
95 }
96
97 static void run_foreground_task_if_needed(AbstractGangTask* task, uint num_workers,
98 bool add_foreground_work) {
99 if (add_foreground_work) {
100 log_develop_trace(gc, workgang)("Running work gang: %s task: %s worker: foreground",
101 Thread::current()->name(), task->name());
102 task->work(num_workers);
103 log_develop_trace(gc, workgang)("Finished work gang: %s task: %s worker: foreground "
104 "thread: " PTR_FORMAT, Thread::current()->name(), task->name(), p2i(Thread::current()));
105 }
106 }
107
108 // WorkGang dispatcher implemented with semaphores.
109 //
110 // Semaphores don't require the worker threads to re-claim the lock when they wake up.
111 // This helps lowering the latency when starting and stopping the worker threads.
112 class GangTaskDispatcher : public CHeapObj<mtGC> {
113 // The task currently being dispatched to the GangWorkers.
114 AbstractGangTask* _task;
115
116 volatile uint _started;
117 volatile uint _not_finished;
118
119 // Semaphore used to start the GangWorkers.
120 Semaphore* _start_semaphore;
121 // Semaphore used to notify the coordinator that all workers are done.
122 Semaphore* _end_semaphore;
123
124 public:
125 GangTaskDispatcher() :
126 _task(NULL),
127 _started(0),
128 _not_finished(0),
129 _start_semaphore(new Semaphore()),
130 _end_semaphore(new Semaphore())
131 { }
132
133 ~GangTaskDispatcher() {
134 delete _start_semaphore;
135 delete _end_semaphore;
136 }
137
138 // Coordinator API.
139
140 // Distributes the task out to num_workers workers.
141 // Returns when the task has been completed by all workers.
142 void coordinator_execute_on_workers(AbstractGangTask* task, uint num_workers, bool add_foreground_work) {
143 // No workers are allowed to read the state variables until they have been signaled.
144 _task = task;
145 _not_finished = num_workers;
146
147 // Dispatch 'num_workers' number of tasks.
148 _start_semaphore->signal(num_workers);
149
150 run_foreground_task_if_needed(task, num_workers, add_foreground_work);
151
152 // Wait for the last worker to signal the coordinator.
153 _end_semaphore->wait();
154
155 // No workers are allowed to read the state variables after the coordinator has been signaled.
156 assert(_not_finished == 0, "%d not finished workers?", _not_finished);
157 _task = NULL;
158 _started = 0;
159
160 }
161
162 // Worker API.
163
164 // Waits for a task to become available to the worker.
165 // Returns when the worker has been assigned a task.
166 WorkData worker_wait_for_task() {
167 // Wait for the coordinator to dispatch a task.
168 _start_semaphore->wait();
169
170 uint num_started = Atomic::add(&_started, 1u);
171
172 // Subtract one to get a zero-indexed worker id.
173 uint worker_id = num_started - 1;
174
175 return WorkData(_task, worker_id);
176 }
177
178 // Signal to the coordinator that the worker is done with the assigned task.
179 void worker_done_with_task() {
180 // Mark that the worker is done with the task.
181 // The worker is not allowed to read the state variables after this line.
182 uint not_finished = Atomic::sub(&_not_finished, 1u);
183
184 // The last worker signals to the coordinator that all work is completed.
185 if (not_finished == 0) {
186 _end_semaphore->signal();
187 }
188 }
189 };
190
191 WorkGang::WorkGang(const char* name,
192 uint workers,
193 bool are_GC_task_threads,
194 bool are_ConcurrentGC_threads) :
195 AbstractWorkGang(name, workers, are_GC_task_threads, are_ConcurrentGC_threads),
196 _dispatcher(new GangTaskDispatcher())
197 { }
198
199 WorkGang::~WorkGang() {
200 delete _dispatcher;
201 }
202
203 AbstractGangWorker* WorkGang::allocate_worker(uint worker_id) {
204 return new GangWorker(this, worker_id);
205 }
206
207 void WorkGang::run_task(AbstractGangTask* task) {
208 run_task(task, active_workers());
209 }
210
211 void WorkGang::run_task(AbstractGangTask* task, uint num_workers, bool add_foreground_work) {
212 guarantee(num_workers <= total_workers(),
213 "Trying to execute task %s with %u workers which is more than the amount of total workers %u.",
214 task->name(), num_workers, total_workers());
215 guarantee(num_workers > 0, "Trying to execute task %s with zero workers", task->name());
216 uint old_num_workers = _active_workers;
|