blob: 3a276b55ad570ef88302b51b32243f4d261b4475 [file] [log] [blame]
Jim Cownie5e8470a2013-09-27 10:38:44 +00001/*
2 * kmp_taskq.c -- TASKQ support for OpenMP.
Jim Cownie4cc4bb42014-10-07 16:25:50 +00003 * $Revision: 43389 $
4 * $Date: 2014-08-11 10:54:01 -0500 (Mon, 11 Aug 2014) $
Jim Cownie5e8470a2013-09-27 10:38:44 +00005 */
6
7
8//===----------------------------------------------------------------------===//
9//
10// The LLVM Compiler Infrastructure
11//
12// This file is dual licensed under the MIT and the University of Illinois Open
13// Source Licenses. See LICENSE.txt for details.
14//
15//===----------------------------------------------------------------------===//
16
17
18#include "kmp.h"
19#include "kmp_i18n.h"
20#include "kmp_io.h"
21#include "kmp_error.h"
22
23#define MAX_MESSAGE 512
24
25/* ------------------------------------------------------------------------ */
26/* ------------------------------------------------------------------------ */
27
28/*
29 * Taskq routines and global variables
30 */
31
32#define KMP_DEBUG_REF_CTS(x) KF_TRACE(1, x);
33
34#define THREAD_ALLOC_FOR_TASKQ
35
Jim Cownie5e8470a2013-09-27 10:38:44 +000036static int
37in_parallel_context( kmp_team_t *team )
38{
39 return ! team -> t.t_serialized;
40}
41
42static void
43__kmp_taskq_eo( int *gtid_ref, int *cid_ref, ident_t *loc_ref )
44{
45 int gtid = *gtid_ref;
46 int tid = __kmp_tid_from_gtid( gtid );
47 kmp_uint32 spins;
48 kmp_uint32 my_token;
49 kmpc_task_queue_t *taskq;
50 kmp_taskq_t *tq = & __kmp_threads[gtid] -> th.th_team -> t.t_taskq;
51
52 if ( __kmp_env_consistency_check )
53 __kmp_push_sync( gtid, ct_ordered_in_taskq, loc_ref, NULL );
54
55 if ( ! __kmp_threads[ gtid ]-> th.th_team -> t.t_serialized ) {
56 KMP_MB(); /* Flush all pending memory write invalidates. */
57
58 /* GEH - need check here under stats to make sure */
59 /* inside task (curr_thunk[*tid_ref] != NULL) */
60
61 my_token =tq->tq_curr_thunk[ tid ]-> th_tasknum;
62
63 taskq = tq->tq_curr_thunk[ tid ]-> th.th_shareds -> sv_queue;
64
65 KMP_WAIT_YIELD(&taskq->tq_tasknum_serving, my_token, KMP_EQ, NULL);
66 KMP_MB();
67 }
68}
69
70static void
71__kmp_taskq_xo( int *gtid_ref, int *cid_ref, ident_t *loc_ref )
72{
73 int gtid = *gtid_ref;
74 int tid = __kmp_tid_from_gtid( gtid );
75 kmp_uint32 my_token;
76 kmp_taskq_t *tq = & __kmp_threads[gtid] -> th.th_team -> t.t_taskq;
77
78 if ( __kmp_env_consistency_check )
79 __kmp_pop_sync( gtid, ct_ordered_in_taskq, loc_ref );
80
81 if ( ! __kmp_threads[ gtid ]-> th.th_team -> t.t_serialized ) {
82 KMP_MB(); /* Flush all pending memory write invalidates. */
83
84 /* GEH - need check here under stats to make sure */
85 /* inside task (curr_thunk[tid] != NULL) */
86
87 my_token = tq->tq_curr_thunk[ tid ]->th_tasknum;
88
89 KMP_MB(); /* Flush all pending memory write invalidates. */
90
91 tq->tq_curr_thunk[ tid ]-> th.th_shareds -> sv_queue -> tq_tasknum_serving = my_token + 1;
92
93 KMP_MB(); /* Flush all pending memory write invalidates. */
94 }
95}
96
97static void
98__kmp_taskq_check_ordered( kmp_int32 gtid, kmpc_thunk_t *thunk )
99{
100 kmp_uint32 spins;
101 kmp_uint32 my_token;
102 kmpc_task_queue_t *taskq;
103
104 /* assume we are always called from an active parallel context */
105
106 KMP_MB(); /* Flush all pending memory write invalidates. */
107
108 my_token = thunk -> th_tasknum;
109
110 taskq = thunk -> th.th_shareds -> sv_queue;
111
112 if(taskq->tq_tasknum_serving <= my_token) {
113 KMP_WAIT_YIELD(&taskq->tq_tasknum_serving, my_token, KMP_GE, NULL);
114 KMP_MB();
115 taskq->tq_tasknum_serving = my_token +1;
116 KMP_MB();
117 }
118}
119
120static void
121__kmp_dump_TQF(kmp_int32 flags)
122{
123 if (flags & TQF_IS_ORDERED)
124 __kmp_printf("ORDERED ");
125 if (flags & TQF_IS_LASTPRIVATE)
126 __kmp_printf("LAST_PRIV ");
127 if (flags & TQF_IS_NOWAIT)
128 __kmp_printf("NOWAIT ");
129 if (flags & TQF_HEURISTICS)
130 __kmp_printf("HEURIST ");
131 if (flags & TQF_INTERFACE_RESERVED1)
132 __kmp_printf("RESERV1 ");
133 if (flags & TQF_INTERFACE_RESERVED2)
134 __kmp_printf("RESERV2 ");
135 if (flags & TQF_INTERFACE_RESERVED3)
136 __kmp_printf("RESERV3 ");
137 if (flags & TQF_INTERFACE_RESERVED4)
138 __kmp_printf("RESERV4 ");
139 if (flags & TQF_IS_LAST_TASK)
140 __kmp_printf("LAST_TASK ");
141 if (flags & TQF_TASKQ_TASK)
142 __kmp_printf("TASKQ_TASK ");
143 if (flags & TQF_RELEASE_WORKERS)
144 __kmp_printf("RELEASE ");
145 if (flags & TQF_ALL_TASKS_QUEUED)
146 __kmp_printf("ALL_QUEUED ");
147 if (flags & TQF_PARALLEL_CONTEXT)
148 __kmp_printf("PARALLEL ");
149 if (flags & TQF_DEALLOCATED)
150 __kmp_printf("DEALLOC ");
151 if (!(flags & (TQF_INTERNAL_FLAGS|TQF_INTERFACE_FLAGS)))
152 __kmp_printf("(NONE)");
153}
154
155static void
156__kmp_dump_thunk( kmp_taskq_t *tq, kmpc_thunk_t *thunk, kmp_int32 global_tid )
157{
158 int i;
159 int nproc = __kmp_threads[global_tid] -> th.th_team -> t.t_nproc;
160
161 __kmp_printf("\tThunk at %p on (%d): ", thunk, global_tid);
162
163 if (thunk != NULL) {
164 for (i = 0; i < nproc; i++) {
165 if( tq->tq_curr_thunk[i] == thunk ) {
166 __kmp_printf("[%i] ", i);
167 }
168 }
169 __kmp_printf("th_shareds=%p, ", thunk->th.th_shareds);
170 __kmp_printf("th_task=%p, ", thunk->th_task);
171 __kmp_printf("th_encl_thunk=%p, ", thunk->th_encl_thunk);
172 __kmp_printf("th_status=%d, ", thunk->th_status);
173 __kmp_printf("th_tasknum=%u, ", thunk->th_tasknum);
174 __kmp_printf("th_flags="); __kmp_dump_TQF(thunk->th_flags);
175 }
176
177 __kmp_printf("\n");
178}
179
180static void
181__kmp_dump_thunk_stack(kmpc_thunk_t *thunk, kmp_int32 thread_num)
182{
183 kmpc_thunk_t *th;
184
185 __kmp_printf(" Thunk stack for T#%d: ", thread_num);
186
187 for (th = thunk; th != NULL; th = th->th_encl_thunk )
188 __kmp_printf("%p ", th);
189
190 __kmp_printf("\n");
191}
192
193static void
194__kmp_dump_task_queue( kmp_taskq_t *tq, kmpc_task_queue_t *queue, kmp_int32 global_tid )
195{
196 int qs, count, i;
197 kmpc_thunk_t *thunk;
198 kmpc_task_queue_t *taskq;
199
200 __kmp_printf("Task Queue at %p on (%d):\n", queue, global_tid);
201
202 if (queue != NULL) {
203 int in_parallel = queue->tq_flags & TQF_PARALLEL_CONTEXT;
204
205 if ( __kmp_env_consistency_check ) {
206 __kmp_printf(" tq_loc : ");
207 }
208 if (in_parallel) {
209
210 //if (queue->tq.tq_parent != 0)
211 //__kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
212
213 //__kmp_acquire_lock(& queue->tq_link_lck, global_tid);
214
215 KMP_MB(); /* make sure data structures are in consistent state before querying them */
216 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
217
218 __kmp_printf(" tq_parent : %p\n", queue->tq.tq_parent);
219 __kmp_printf(" tq_first_child : %p\n", queue->tq_first_child);
220 __kmp_printf(" tq_next_child : %p\n", queue->tq_next_child);
221 __kmp_printf(" tq_prev_child : %p\n", queue->tq_prev_child);
222 __kmp_printf(" tq_ref_count : %d\n", queue->tq_ref_count);
223
224 //__kmp_release_lock(& queue->tq_link_lck, global_tid);
225
226 //if (queue->tq.tq_parent != 0)
227 //__kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
228
229 //__kmp_acquire_lock(& queue->tq_free_thunks_lck, global_tid);
230 //__kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
231
232 KMP_MB(); /* make sure data structures are in consistent state before querying them */
233 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
234 }
235
236 __kmp_printf(" tq_shareds : ");
237 for (i=0; i<((queue == tq->tq_root) ? queue->tq_nproc : 1); i++)
238 __kmp_printf("%p ", queue->tq_shareds[i].ai_data);
239 __kmp_printf("\n");
240
241 if (in_parallel) {
242 __kmp_printf(" tq_tasknum_queuing : %u\n", queue->tq_tasknum_queuing);
243 __kmp_printf(" tq_tasknum_serving : %u\n", queue->tq_tasknum_serving);
244 }
245
246 __kmp_printf(" tq_queue : %p\n", queue->tq_queue);
247 __kmp_printf(" tq_thunk_space : %p\n", queue->tq_thunk_space);
248 __kmp_printf(" tq_taskq_slot : %p\n", queue->tq_taskq_slot);
249
250 __kmp_printf(" tq_free_thunks : ");
251 for (thunk = queue->tq_free_thunks; thunk != NULL; thunk = thunk->th.th_next_free )
252 __kmp_printf("%p ", thunk);
253 __kmp_printf("\n");
254
255 __kmp_printf(" tq_nslots : %d\n", queue->tq_nslots);
256 __kmp_printf(" tq_head : %d\n", queue->tq_head);
257 __kmp_printf(" tq_tail : %d\n", queue->tq_tail);
258 __kmp_printf(" tq_nfull : %d\n", queue->tq_nfull);
259 __kmp_printf(" tq_hiwat : %d\n", queue->tq_hiwat);
260 __kmp_printf(" tq_flags : "); __kmp_dump_TQF(queue->tq_flags);
261 __kmp_printf("\n");
262
263 if (in_parallel) {
264 __kmp_printf(" tq_th_thunks : ");
265 for (i = 0; i < queue->tq_nproc; i++) {
266 __kmp_printf("%d ", queue->tq_th_thunks[i].ai_data);
267 }
268 __kmp_printf("\n");
269 }
270
271 __kmp_printf("\n");
272 __kmp_printf(" Queue slots:\n");
273
274
275 qs = queue->tq_tail;
276 for ( count = 0; count < queue->tq_nfull; ++count ) {
277 __kmp_printf("(%d)", qs);
278 __kmp_dump_thunk( tq, queue->tq_queue[qs].qs_thunk, global_tid );
279 qs = (qs+1) % queue->tq_nslots;
280 }
281
282 __kmp_printf("\n");
283
284 if (in_parallel) {
285 if (queue->tq_taskq_slot != NULL) {
286 __kmp_printf(" TaskQ slot:\n");
287 __kmp_dump_thunk( tq, (kmpc_thunk_t *) queue->tq_taskq_slot, global_tid );
288 __kmp_printf("\n");
289 }
290 //__kmp_release_lock(& queue->tq_queue_lck, global_tid);
291 //__kmp_release_lock(& queue->tq_free_thunks_lck, global_tid);
292 }
293 }
294
295 __kmp_printf(" Taskq freelist: ");
296
297 //__kmp_acquire_lock( & tq->tq_freelist_lck, global_tid );
298
299 KMP_MB(); /* make sure data structures are in consistent state before querying them */
300 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
301
302 for( taskq = tq->tq_freelist; taskq != NULL; taskq = taskq->tq.tq_next_free )
303 __kmp_printf("%p ", taskq);
304
305 //__kmp_release_lock( & tq->tq_freelist_lck, global_tid );
306
307 __kmp_printf("\n\n");
308}
309
310static void
311__kmp_aux_dump_task_queue_tree( kmp_taskq_t *tq, kmpc_task_queue_t *curr_queue, kmp_int32 level, kmp_int32 global_tid )
312{
313 int i, count, qs;
314 int nproc = __kmp_threads[global_tid] -> th.th_team -> t.t_nproc;
315 kmpc_task_queue_t *queue = curr_queue;
316
317 if (curr_queue == NULL)
318 return;
319
320 __kmp_printf(" ");
321
322 for (i=0; i<level; i++)
323 __kmp_printf(" ");
324
325 __kmp_printf("%p", curr_queue);
326
327 for (i = 0; i < nproc; i++) {
328 if( tq->tq_curr_thunk[i] && tq->tq_curr_thunk[i]->th.th_shareds->sv_queue == curr_queue ) {
329 __kmp_printf(" [%i]", i);
330 }
331 }
332
333 __kmp_printf(":");
334
335 //__kmp_acquire_lock(& curr_queue->tq_queue_lck, global_tid);
336
337 KMP_MB(); /* make sure data structures are in consistent state before querying them */
338 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
339
340 qs = curr_queue->tq_tail;
341
342 for ( count = 0; count < curr_queue->tq_nfull; ++count ) {
343 __kmp_printf("%p ", curr_queue->tq_queue[qs].qs_thunk);
344 qs = (qs+1) % curr_queue->tq_nslots;
345 }
346
347 //__kmp_release_lock(& curr_queue->tq_queue_lck, global_tid);
348
349 __kmp_printf("\n");
350
351 if (curr_queue->tq_first_child) {
352 //__kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
353
354 KMP_MB(); /* make sure data structures are in consistent state before querying them */
355 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
356
357 if (curr_queue->tq_first_child) {
358 for(queue = (kmpc_task_queue_t *)curr_queue->tq_first_child;
359 queue != NULL;
360 queue = queue->tq_next_child) {
361 __kmp_aux_dump_task_queue_tree( tq, queue, level+1, global_tid );
362 }
363 }
364
365 //__kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
366 }
367}
368
369static void
370__kmp_dump_task_queue_tree( kmp_taskq_t *tq, kmpc_task_queue_t *tqroot, kmp_int32 global_tid)
371{
372 __kmp_printf("TaskQ Tree at root %p on (%d):\n", tqroot, global_tid);
373
374 __kmp_aux_dump_task_queue_tree( tq, tqroot, 0, global_tid );
375
376 __kmp_printf("\n");
377}
378
379/* --------------------------------------------------------------------------- */
380
381/*
382 New taskq storage routines that try to minimize overhead of mallocs but
383 still provide cache line alignment.
384*/
385
386
387static void *
388__kmp_taskq_allocate(size_t size, kmp_int32 global_tid)
389{
390 void *addr, *orig_addr;
391 size_t bytes;
392
393 KB_TRACE( 5, ("__kmp_taskq_allocate: called size=%d, gtid=%d\n", (int) size, global_tid ) );
394
395 bytes = sizeof(void *) + CACHE_LINE + size;
396
397#ifdef THREAD_ALLOC_FOR_TASKQ
398 orig_addr = (void *) __kmp_thread_malloc( __kmp_thread_from_gtid(global_tid), bytes );
399#else
400 KE_TRACE( 10, ("%%%%%% MALLOC( %d )\n", bytes ) );
401 orig_addr = (void *) KMP_INTERNAL_MALLOC( bytes );
402#endif /* THREAD_ALLOC_FOR_TASKQ */
403
404 if (orig_addr == 0)
405 KMP_FATAL( OutOfHeapMemory );
406
407 addr = orig_addr;
408
409 if (((kmp_uintptr_t) addr & ( CACHE_LINE - 1 )) != 0) {
410 KB_TRACE( 50, ("__kmp_taskq_allocate: adjust for cache alignment\n" ) );
411 addr = (void *) (((kmp_uintptr_t) addr + CACHE_LINE) & ~( CACHE_LINE - 1 ));
412 }
413
414 (* (void **) addr) = orig_addr;
415
416 KB_TRACE( 10, ("__kmp_taskq_allocate: allocate: %p, use: %p - %p, size: %d, gtid: %d\n",
417 orig_addr, ((void **) addr) + 1, ((char *)(((void **) addr) + 1)) + size-1,
418 (int) size, global_tid ));
419
420 return ( ((void **) addr) + 1 );
421}
422
423static void
424__kmpc_taskq_free(void *p, kmp_int32 global_tid)
425{
426 KB_TRACE( 5, ("__kmpc_taskq_free: called addr=%p, gtid=%d\n", p, global_tid ) );
427
428 KB_TRACE(10, ("__kmpc_taskq_free: freeing: %p, gtid: %d\n", (*( ((void **) p)-1)), global_tid ));
429
430#ifdef THREAD_ALLOC_FOR_TASKQ
431 __kmp_thread_free( __kmp_thread_from_gtid(global_tid), *( ((void **) p)-1) );
432#else
433 KMP_INTERNAL_FREE( *( ((void **) p)-1) );
434#endif /* THREAD_ALLOC_FOR_TASKQ */
435}
436
437/* --------------------------------------------------------------------------- */
438
439/*
440 * Keep freed kmpc_task_queue_t on an internal freelist and recycle since
441 * they're of constant size.
442 */
443
444static kmpc_task_queue_t *
445__kmp_alloc_taskq ( kmp_taskq_t *tq, int in_parallel, kmp_int32 nslots, kmp_int32 nthunks,
446 kmp_int32 nshareds, kmp_int32 nproc, size_t sizeof_thunk,
447 size_t sizeof_shareds, kmpc_thunk_t **new_taskq_thunk, kmp_int32 global_tid )
448{
449 kmp_int32 i;
450 size_t bytes;
451 kmpc_task_queue_t *new_queue;
452 kmpc_aligned_shared_vars_t *shared_var_array;
453 char *shared_var_storage;
454 char *pt; /* for doing byte-adjusted address computations */
455
456 __kmp_acquire_lock( & tq->tq_freelist_lck, global_tid );
457
458 KMP_MB(); /* make sure data structures are in consistent state before querying them */
459 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
460
461 if( tq->tq_freelist ) {
462 new_queue = tq -> tq_freelist;
463 tq -> tq_freelist = tq -> tq_freelist -> tq.tq_next_free;
464
465 KMP_DEBUG_ASSERT(new_queue->tq_flags & TQF_DEALLOCATED);
466
467 new_queue->tq_flags = 0;
468
469 __kmp_release_lock( & tq->tq_freelist_lck, global_tid );
470 }
471 else {
472 __kmp_release_lock( & tq->tq_freelist_lck, global_tid );
473
474 new_queue = (kmpc_task_queue_t *) __kmp_taskq_allocate (sizeof (kmpc_task_queue_t), global_tid);
475 new_queue->tq_flags = 0;
476 }
477
478 /* space in the task queue for queue slots (allocate as one big chunk */
479 /* of storage including new_taskq_task space) */
480
481 sizeof_thunk += (CACHE_LINE - (sizeof_thunk % CACHE_LINE)); /* pad to cache line size */
482 pt = (char *) __kmp_taskq_allocate (nthunks * sizeof_thunk, global_tid);
483 new_queue->tq_thunk_space = (kmpc_thunk_t *)pt;
484 *new_taskq_thunk = (kmpc_thunk_t *)(pt + (nthunks - 1) * sizeof_thunk);
485
486 /* chain the allocated thunks into a freelist for this queue */
487
488 new_queue->tq_free_thunks = (kmpc_thunk_t *)pt;
489
490 for (i = 0; i < (nthunks - 2); i++) {
491 ((kmpc_thunk_t *)(pt+i*sizeof_thunk))->th.th_next_free = (kmpc_thunk_t *)(pt + (i+1)*sizeof_thunk);
492#ifdef KMP_DEBUG
493 ((kmpc_thunk_t *)(pt+i*sizeof_thunk))->th_flags = TQF_DEALLOCATED;
494#endif
495 }
496
497 ((kmpc_thunk_t *)(pt+(nthunks-2)*sizeof_thunk))->th.th_next_free = NULL;
498#ifdef KMP_DEBUG
499 ((kmpc_thunk_t *)(pt+(nthunks-2)*sizeof_thunk))->th_flags = TQF_DEALLOCATED;
500#endif
501
502 /* initialize the locks */
503
504 if (in_parallel) {
505 __kmp_init_lock( & new_queue->tq_link_lck );
506 __kmp_init_lock( & new_queue->tq_free_thunks_lck );
507 __kmp_init_lock( & new_queue->tq_queue_lck );
508 }
509
510 /* now allocate the slots */
511
512 bytes = nslots * sizeof (kmpc_aligned_queue_slot_t);
513 new_queue->tq_queue = (kmpc_aligned_queue_slot_t *) __kmp_taskq_allocate( bytes, global_tid );
514
515 /* space for array of pointers to shared variable structures */
516 sizeof_shareds += sizeof(kmpc_task_queue_t *);
517 sizeof_shareds += (CACHE_LINE - (sizeof_shareds % CACHE_LINE)); /* pad to cache line size */
518
519 bytes = nshareds * sizeof (kmpc_aligned_shared_vars_t);
520 shared_var_array = (kmpc_aligned_shared_vars_t *) __kmp_taskq_allocate ( bytes, global_tid);
521
522 bytes = nshareds * sizeof_shareds;
523 shared_var_storage = (char *) __kmp_taskq_allocate ( bytes, global_tid);
524
525 for (i=0; i<nshareds; i++) {
526 shared_var_array[i].ai_data = (kmpc_shared_vars_t *) (shared_var_storage + i*sizeof_shareds);
527 shared_var_array[i].ai_data->sv_queue = new_queue;
528 }
529 new_queue->tq_shareds = shared_var_array;
530
531
532 /* array for number of outstanding thunks per thread */
533
534 if (in_parallel) {
535 bytes = nproc * sizeof(kmpc_aligned_int32_t);
536 new_queue->tq_th_thunks = (kmpc_aligned_int32_t *) __kmp_taskq_allocate ( bytes, global_tid);
537 new_queue->tq_nproc = nproc;
538
539 for (i=0; i<nproc; i++)
540 new_queue->tq_th_thunks[i].ai_data = 0;
541 }
542
543 return new_queue;
544}
545
546static void
547__kmp_free_taskq (kmp_taskq_t *tq, kmpc_task_queue_t *p, int in_parallel, kmp_int32 global_tid)
548{
549 __kmpc_taskq_free(p->tq_thunk_space, global_tid);
550 __kmpc_taskq_free(p->tq_queue, global_tid);
551
552 /* free shared var structure storage */
553 __kmpc_taskq_free((void *) p->tq_shareds[0].ai_data, global_tid);
554
555 /* free array of pointers to shared vars storage */
556 __kmpc_taskq_free(p->tq_shareds, global_tid);
557
558#ifdef KMP_DEBUG
559 p->tq_first_child = NULL;
560 p->tq_next_child = NULL;
561 p->tq_prev_child = NULL;
562 p->tq_ref_count = -10;
563 p->tq_shareds = NULL;
564 p->tq_tasknum_queuing = 0;
565 p->tq_tasknum_serving = 0;
566 p->tq_queue = NULL;
567 p->tq_thunk_space = NULL;
568 p->tq_taskq_slot = NULL;
569 p->tq_free_thunks = NULL;
570 p->tq_nslots = 0;
571 p->tq_head = 0;
572 p->tq_tail = 0;
573 p->tq_nfull = 0;
574 p->tq_hiwat = 0;
575
576 if (in_parallel) {
577 int i;
578
579 for (i=0; i<p->tq_nproc; i++)
580 p->tq_th_thunks[i].ai_data = 0;
581 }
582 if ( __kmp_env_consistency_check )
583 p->tq_loc = NULL;
584 KMP_DEBUG_ASSERT( p->tq_flags & TQF_DEALLOCATED );
585 p->tq_flags = TQF_DEALLOCATED;
586#endif /* KMP_DEBUG */
587
588 if (in_parallel) {
589 __kmpc_taskq_free(p->tq_th_thunks, global_tid);
590 __kmp_destroy_lock(& p->tq_link_lck);
591 __kmp_destroy_lock(& p->tq_queue_lck);
592 __kmp_destroy_lock(& p->tq_free_thunks_lck);
593 }
594#ifdef KMP_DEBUG
595 p->tq_th_thunks = NULL;
596#endif /* KMP_DEBUG */
597
598 KMP_MB(); /* make sure data structures are in consistent state before querying them */
599 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
600
601 __kmp_acquire_lock( & tq->tq_freelist_lck, global_tid );
602 p->tq.tq_next_free = tq->tq_freelist;
603
604 tq->tq_freelist = p;
605 __kmp_release_lock( & tq->tq_freelist_lck, global_tid );
606}
607
608/*
609 * Once a group of thunks has been allocated for use in a particular queue,
610 * these are managed via a per-queue freelist.
611 * We force a check that there's always a thunk free if we need one.
612 */
613
614static kmpc_thunk_t *
615__kmp_alloc_thunk (kmpc_task_queue_t *queue, int in_parallel, kmp_int32 global_tid)
616{
617 kmpc_thunk_t *fl;
618
619 if (in_parallel) {
620 __kmp_acquire_lock(& queue->tq_free_thunks_lck, global_tid);
621
622 KMP_MB(); /* make sure data structures are in consistent state before querying them */
623 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
624 }
625
626 fl = queue->tq_free_thunks;
627
628 KMP_DEBUG_ASSERT (fl != NULL);
629
630 queue->tq_free_thunks = fl->th.th_next_free;
631 fl->th_flags = 0;
632
633 if (in_parallel)
634 __kmp_release_lock(& queue->tq_free_thunks_lck, global_tid);
635
636 return fl;
637}
638
639static void
640__kmp_free_thunk (kmpc_task_queue_t *queue, kmpc_thunk_t *p, int in_parallel, kmp_int32 global_tid)
641{
642#ifdef KMP_DEBUG
643 p->th_task = 0;
644 p->th_encl_thunk = 0;
645 p->th_status = 0;
646 p->th_tasknum = 0;
647 /* Also could zero pointers to private vars */
648#endif
649
650 if (in_parallel) {
651 __kmp_acquire_lock(& queue->tq_free_thunks_lck, global_tid);
652
653 KMP_MB(); /* make sure data structures are in consistent state before querying them */
654 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
655 }
656
657 p->th.th_next_free = queue->tq_free_thunks;
658 queue->tq_free_thunks = p;
659
660#ifdef KMP_DEBUG
661 p->th_flags = TQF_DEALLOCATED;
662#endif
663
664 if (in_parallel)
665 __kmp_release_lock(& queue->tq_free_thunks_lck, global_tid);
666}
667
668/* --------------------------------------------------------------------------- */
669
670/* returns nonzero if the queue just became full after the enqueue */
671
672static kmp_int32
673__kmp_enqueue_task ( kmp_taskq_t *tq, kmp_int32 global_tid, kmpc_task_queue_t *queue, kmpc_thunk_t *thunk, int in_parallel )
674{
675 kmp_int32 ret;
676
677 /* dkp: can we get around the lock in the TQF_RELEASE_WORKERS case (only the master is executing then) */
678 if (in_parallel) {
679 __kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
680
681 KMP_MB(); /* make sure data structures are in consistent state before querying them */
682 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
683 }
684
685 KMP_DEBUG_ASSERT (queue->tq_nfull < queue->tq_nslots); /* check queue not full */
686
687 queue->tq_queue[(queue->tq_head)++].qs_thunk = thunk;
688
689 if (queue->tq_head >= queue->tq_nslots)
690 queue->tq_head = 0;
691
692 (queue->tq_nfull)++;
693
694 KMP_MB(); /* to assure that nfull is seen to increase before TQF_ALL_TASKS_QUEUED is set */
695
696 ret = (in_parallel) ? (queue->tq_nfull == queue->tq_nslots) : FALSE;
697
698 if (in_parallel) {
699 /* don't need to wait until workers are released before unlocking */
700 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
701
702 if( tq->tq_global_flags & TQF_RELEASE_WORKERS ) {
703 /* If just creating the root queue, the worker threads are waiting at */
704 /* a join barrier until now, when there's something in the queue for */
705 /* them to do; release them now to do work. */
706 /* This should only be done when this is the first task enqueued, */
707 /* so reset the flag here also. */
708
709 tq->tq_global_flags &= ~TQF_RELEASE_WORKERS; /* no lock needed, workers are still in spin mode */
710
711 KMP_MB(); /* avoid releasing barrier twice if taskq_task switches threads */
712
713 __kmpc_end_barrier_master( NULL, global_tid);
714 }
715 }
716
717 return ret;
718}
719
720static kmpc_thunk_t *
721__kmp_dequeue_task (kmp_int32 global_tid, kmpc_task_queue_t *queue, int in_parallel)
722{
723 kmpc_thunk_t *pt;
724 int tid = __kmp_tid_from_gtid( global_tid );
725
726 KMP_DEBUG_ASSERT (queue->tq_nfull > 0); /* check queue not empty */
727
728 if (queue->tq.tq_parent != NULL && in_parallel) {
729 int ct;
730 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
731 ct = ++(queue->tq_ref_count);
732 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
733 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p inc %d\n",
734 __LINE__, global_tid, queue, ct));
735 }
736
737 pt = queue->tq_queue[(queue->tq_tail)++].qs_thunk;
738
739 if (queue->tq_tail >= queue->tq_nslots)
740 queue->tq_tail = 0;
741
742 if (in_parallel) {
743 queue->tq_th_thunks[tid].ai_data++;
744
745 KMP_MB(); /* necessary so ai_data increment is propagated to other threads immediately (digital) */
746
747 KF_TRACE(200, ("__kmp_dequeue_task: T#%d(:%d) now has %d outstanding thunks from queue %p\n",
748 global_tid, tid, queue->tq_th_thunks[tid].ai_data, queue));
749 }
750
751 (queue->tq_nfull)--;
752
753#ifdef KMP_DEBUG
754 KMP_MB();
755
756 /* necessary so (queue->tq_nfull > 0) above succeeds after tq_nfull is decremented */
757
758 KMP_DEBUG_ASSERT(queue->tq_nfull >= 0);
759
760 if (in_parallel) {
761 KMP_DEBUG_ASSERT(queue->tq_th_thunks[tid].ai_data <= __KMP_TASKQ_THUNKS_PER_TH);
762 }
763#endif
764
765 return pt;
766}
767
768/*
769 * Find the next (non-null) task to dequeue and return it.
770 * This is never called unless in_parallel=TRUE
771 *
772 * Here are the rules for deciding which queue to take the task from:
773 * 1. Walk up the task queue tree from the current queue's parent and look
774 * on the way up (for loop, below).
775 * 2. Do a depth-first search back down the tree from the root and
Jim Cownie4cc4bb42014-10-07 16:25:50 +0000776 * look (find_task_in_descendant_queue()).
Jim Cownie5e8470a2013-09-27 10:38:44 +0000777 *
778 * Here are the rules for deciding which task to take from a queue
779 * (__kmp_find_task_in_queue ()):
780 * 1. Never take the last task from a queue if TQF_IS_LASTPRIVATE; this task
781 * must be staged to make sure we execute the last one with
782 * TQF_IS_LAST_TASK at the end of task queue execution.
783 * 2. If the queue length is below some high water mark and the taskq task
784 * is enqueued, prefer running the taskq task.
785 * 3. Otherwise, take a (normal) task from the queue.
786 *
787 * If we do all this and return pt == NULL at the bottom of this routine,
788 * this means there are no more tasks to execute (except possibly for
789 * TQF_IS_LASTPRIVATE).
790 */
791
792static kmpc_thunk_t *
793__kmp_find_task_in_queue (kmp_int32 global_tid, kmpc_task_queue_t *queue)
794{
795 kmpc_thunk_t *pt = NULL;
796 int tid = __kmp_tid_from_gtid( global_tid );
797
798 /* To prevent deadlock from tq_queue_lck if queue already deallocated */
799 if ( !(queue->tq_flags & TQF_DEALLOCATED) ) {
800
801 __kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
802
803 /* Check again to avoid race in __kmpc_end_taskq() */
804 if ( !(queue->tq_flags & TQF_DEALLOCATED) ) {
805
806 KMP_MB(); /* make sure data structures are in consistent state before querying them */
807 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
808
809 if ((queue->tq_taskq_slot != NULL) && (queue->tq_nfull <= queue->tq_hiwat)) {
810 /* if there's enough room in the queue and the dispatcher */
811 /* (taskq task) is available, schedule more tasks */
812 pt = (kmpc_thunk_t *) queue->tq_taskq_slot;
813 queue->tq_taskq_slot = NULL;
814 }
815 else if (queue->tq_nfull == 0 ||
816 queue->tq_th_thunks[tid].ai_data >= __KMP_TASKQ_THUNKS_PER_TH) {
817 /* do nothing if no thunks available or this thread can't */
818 /* run any because it already is executing too many */
819
820 pt = NULL;
821 }
822 else if (queue->tq_nfull > 1) {
823 /* always safe to schedule a task even if TQF_IS_LASTPRIVATE */
824
825 pt = __kmp_dequeue_task (global_tid, queue, TRUE);
826 }
827 else if (!(queue->tq_flags & TQF_IS_LASTPRIVATE)) {
828 /* one thing in queue, always safe to schedule if !TQF_IS_LASTPRIVATE */
829
830 pt = __kmp_dequeue_task (global_tid, queue, TRUE);
831 }
832 else if (queue->tq_flags & TQF_IS_LAST_TASK) {
833 /* TQF_IS_LASTPRIVATE, one thing in queue, kmpc_end_taskq_task() */
834 /* has been run so this is last task, run with TQF_IS_LAST_TASK so */
835 /* instrumentation does copy-out. */
836
837 pt = __kmp_dequeue_task (global_tid, queue, TRUE);
838 pt->th_flags |= TQF_IS_LAST_TASK; /* don't need test_then_or since already locked */
839 }
840 }
841
842 /* GEH - What happens here if is lastprivate, but not last task? */
843 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
844 }
845
846 return pt;
847}
848
849/*
850 * Walk a tree of queues starting at queue's first child
851 * and return a non-NULL thunk if one can be scheduled.
852 * Must only be called when in_parallel=TRUE
853 */
854
855static kmpc_thunk_t *
856__kmp_find_task_in_descendant_queue (kmp_int32 global_tid, kmpc_task_queue_t *curr_queue)
857{
858 kmpc_thunk_t *pt = NULL;
859 kmpc_task_queue_t *queue = curr_queue;
860
861 if (curr_queue->tq_first_child != NULL) {
862 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
863
864 KMP_MB(); /* make sure data structures are in consistent state before querying them */
865 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
866
867 queue = (kmpc_task_queue_t *) curr_queue->tq_first_child;
868 if (queue == NULL) {
869 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
870 return NULL;
871 }
872
873 while (queue != NULL) {
874 int ct;
875 kmpc_task_queue_t *next;
876
877 ct= ++(queue->tq_ref_count);
878 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
879 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p inc %d\n",
880 __LINE__, global_tid, queue, ct));
881
882 pt = __kmp_find_task_in_queue (global_tid, queue);
883
884 if (pt != NULL) {
885 int ct;
886
887 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
888
889 KMP_MB(); /* make sure data structures are in consistent state before querying them */
890 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
891
892 ct = --(queue->tq_ref_count);
893 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
894 __LINE__, global_tid, queue, ct));
895 KMP_DEBUG_ASSERT( queue->tq_ref_count >= 0 );
896
897 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
898
899 return pt;
900 }
901
902 /* although reference count stays active during descendant walk, shouldn't matter */
903 /* since if children still exist, reference counts aren't being monitored anyway */
904
905 pt = __kmp_find_task_in_descendant_queue (global_tid, queue);
906
907 if (pt != NULL) {
908 int ct;
909
910 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
911
912 KMP_MB(); /* make sure data structures are in consistent state before querying them */
913 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
914
915 ct = --(queue->tq_ref_count);
916 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
917 __LINE__, global_tid, queue, ct));
918 KMP_DEBUG_ASSERT( ct >= 0 );
919
920 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
921
922 return pt;
923 }
924
925 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
926
927 KMP_MB(); /* make sure data structures are in consistent state before querying them */
928 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
929
930 next = queue->tq_next_child;
931
932 ct = --(queue->tq_ref_count);
933 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
934 __LINE__, global_tid, queue, ct));
935 KMP_DEBUG_ASSERT( ct >= 0 );
936
937 queue = next;
938 }
939
940 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
941 }
942
943 return pt;
944}
945
946/*
947 * Walk up the taskq tree looking for a task to execute.
948 * If we get to the root, search the tree for a descendent queue task.
949 * Must only be called when in_parallel=TRUE
950 */
951
952static kmpc_thunk_t *
953__kmp_find_task_in_ancestor_queue (kmp_taskq_t *tq, kmp_int32 global_tid, kmpc_task_queue_t *curr_queue)
954{
955 kmpc_task_queue_t *queue;
956 kmpc_thunk_t *pt;
957
958 pt = NULL;
959
960 if (curr_queue->tq.tq_parent != NULL) {
961 queue = curr_queue->tq.tq_parent;
962
963 while (queue != NULL) {
964 if (queue->tq.tq_parent != NULL) {
965 int ct;
966 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
967
968 KMP_MB(); /* make sure data structures are in consistent state before querying them */
969 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
970
971 ct = ++(queue->tq_ref_count);
972 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
973 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p inc %d\n",
974 __LINE__, global_tid, queue, ct));
975 }
976
977 pt = __kmp_find_task_in_queue (global_tid, queue);
978 if (pt != NULL) {
979 if (queue->tq.tq_parent != NULL) {
980 int ct;
981 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
982
983 KMP_MB(); /* make sure data structures are in consistent state before querying them */
984 /* Seems to work without this call for digital/alpha, needed for IBM/RS6000 */
985
986 ct = --(queue->tq_ref_count);
987 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
988 __LINE__, global_tid, queue, ct));
989 KMP_DEBUG_ASSERT( ct >= 0 );
990
991 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
992 }
993
994 return pt;
995 }
996
997 if (queue->tq.tq_parent != NULL) {
998 int ct;
999 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1000
1001 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1002 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1003
1004 ct = --(queue->tq_ref_count);
1005 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
1006 __LINE__, global_tid, queue, ct));
1007 KMP_DEBUG_ASSERT( ct >= 0 );
1008 }
1009 queue = queue->tq.tq_parent;
1010
1011 if (queue != NULL)
1012 __kmp_release_lock(& queue->tq_link_lck, global_tid);
1013 }
1014
1015 }
1016
1017 pt = __kmp_find_task_in_descendant_queue( global_tid, tq->tq_root );
1018
1019 return pt;
1020}
1021
1022static int
1023__kmp_taskq_tasks_finished (kmpc_task_queue_t *queue)
1024{
1025 int i;
1026
1027 /* KMP_MB(); *//* is this really necessary? */
1028
1029 for (i=0; i<queue->tq_nproc; i++) {
1030 if (queue->tq_th_thunks[i].ai_data != 0)
1031 return FALSE;
1032 }
1033
1034 return TRUE;
1035}
1036
1037static int
1038__kmp_taskq_has_any_children (kmpc_task_queue_t *queue)
1039{
1040 return (queue->tq_first_child != NULL);
1041}
1042
1043static void
1044__kmp_remove_queue_from_tree( kmp_taskq_t *tq, kmp_int32 global_tid, kmpc_task_queue_t *queue, int in_parallel )
1045{
1046#ifdef KMP_DEBUG
1047 kmp_int32 i;
1048 kmpc_thunk_t *thunk;
1049#endif
1050
1051 KF_TRACE(50, ("Before Deletion of TaskQ at %p on (%d):\n", queue, global_tid));
1052 KF_DUMP(50, __kmp_dump_task_queue( tq, queue, global_tid ));
1053
1054 /* sub-queue in a recursion, not the root task queue */
1055 KMP_DEBUG_ASSERT (queue->tq.tq_parent != NULL);
1056
1057 if (in_parallel) {
1058 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1059
1060 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1061 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1062 }
1063
1064 KMP_DEBUG_ASSERT (queue->tq_first_child == NULL);
1065
1066 /* unlink queue from its siblings if any at this level */
1067 if (queue->tq_prev_child != NULL)
1068 queue->tq_prev_child->tq_next_child = queue->tq_next_child;
1069 if (queue->tq_next_child != NULL)
1070 queue->tq_next_child->tq_prev_child = queue->tq_prev_child;
1071 if (queue->tq.tq_parent->tq_first_child == queue)
1072 queue->tq.tq_parent->tq_first_child = queue->tq_next_child;
1073
1074 queue->tq_prev_child = NULL;
1075 queue->tq_next_child = NULL;
1076
1077 if (in_parallel) {
1078 kmp_uint32 spins;
1079
1080 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p waiting for ref_count of %d to reach 1\n",
1081 __LINE__, global_tid, queue, queue->tq_ref_count));
1082
1083 /* wait until all other threads have stopped accessing this queue */
1084 while (queue->tq_ref_count > 1) {
1085 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1086
1087 KMP_WAIT_YIELD((volatile kmp_uint32*)&queue->tq_ref_count, 1, KMP_LE, NULL);
1088
1089 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1090
1091 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1092 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1093 }
1094
1095 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1096 }
1097
1098 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p freeing queue\n",
1099 __LINE__, global_tid, queue));
1100
1101#ifdef KMP_DEBUG
1102 KMP_DEBUG_ASSERT(queue->tq_flags & TQF_ALL_TASKS_QUEUED);
1103 KMP_DEBUG_ASSERT(queue->tq_nfull == 0);
1104
1105 for (i=0; i<queue->tq_nproc; i++) {
1106 KMP_DEBUG_ASSERT(queue->tq_th_thunks[i].ai_data == 0);
1107 }
1108
1109 i = 0;
1110 for (thunk=queue->tq_free_thunks; thunk != NULL; thunk=thunk->th.th_next_free)
1111 ++i;
1112
1113 KMP_ASSERT (i == queue->tq_nslots + (queue->tq_nproc * __KMP_TASKQ_THUNKS_PER_TH));
1114#endif
1115
1116 /* release storage for queue entry */
1117 __kmp_free_taskq ( tq, queue, TRUE, global_tid );
1118
1119 KF_TRACE(50, ("After Deletion of TaskQ at %p on (%d):\n", queue, global_tid));
1120 KF_DUMP(50, __kmp_dump_task_queue_tree( tq, tq->tq_root, global_tid ));
1121}
1122
1123/*
1124 * Starting from indicated queue, proceed downward through tree and
1125 * remove all taskqs which are finished, but only go down to taskqs
1126 * which have the "nowait" clause present. Assume this is only called
1127 * when in_parallel=TRUE.
1128 */
1129
1130static void
1131__kmp_find_and_remove_finished_child_taskq( kmp_taskq_t *tq, kmp_int32 global_tid, kmpc_task_queue_t *curr_queue )
1132{
1133 kmpc_task_queue_t *queue = curr_queue;
1134
1135 if (curr_queue->tq_first_child != NULL) {
1136 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
1137
1138 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1139 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1140
1141 queue = (kmpc_task_queue_t *) curr_queue->tq_first_child;
1142 if (queue != NULL) {
1143 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
1144 return;
1145 }
1146
1147 while (queue != NULL) {
1148 kmpc_task_queue_t *next;
1149 int ct = ++(queue->tq_ref_count);
1150 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p inc %d\n",
1151 __LINE__, global_tid, queue, ct));
1152
1153
1154 /* although reference count stays active during descendant walk, */
1155 /* shouldn't matter since if children still exist, reference */
1156 /* counts aren't being monitored anyway */
1157
1158 if (queue->tq_flags & TQF_IS_NOWAIT) {
1159 __kmp_find_and_remove_finished_child_taskq ( tq, global_tid, queue );
1160
1161 if ((queue->tq_flags & TQF_ALL_TASKS_QUEUED) && (queue->tq_nfull == 0) &&
1162 __kmp_taskq_tasks_finished(queue) && ! __kmp_taskq_has_any_children(queue)) {
1163
1164 /*
1165 Only remove this if we have not already marked it for deallocation.
1166 This should prevent multiple threads from trying to free this.
1167 */
1168
1169 if ( __kmp_test_lock(& queue->tq_queue_lck, global_tid) ) {
1170 if ( !(queue->tq_flags & TQF_DEALLOCATED) ) {
1171 queue->tq_flags |= TQF_DEALLOCATED;
1172 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
1173
1174 __kmp_remove_queue_from_tree( tq, global_tid, queue, TRUE );
1175
1176 /* Can't do any more here since can't be sure where sibling queue is so just exit this level */
1177 return;
1178 }
1179 else {
1180 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
1181 }
1182 }
1183 /* otherwise, just fall through and decrement reference count */
1184 }
1185 }
1186
1187 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
1188
1189 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1190 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1191
1192 next = queue->tq_next_child;
1193
1194 ct = --(queue->tq_ref_count);
1195 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
1196 __LINE__, global_tid, queue, ct));
1197 KMP_DEBUG_ASSERT( ct >= 0 );
1198
1199 queue = next;
1200 }
1201
1202 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
1203 }
1204}
1205
1206/*
1207 * Starting from indicated queue, proceed downward through tree and
1208 * remove all taskq's assuming all are finished and
1209 * assuming NO other threads are executing at this point.
1210 */
1211
1212static void
1213__kmp_remove_all_child_taskq( kmp_taskq_t *tq, kmp_int32 global_tid, kmpc_task_queue_t *queue )
1214{
1215 kmpc_task_queue_t *next_child;
1216
1217 queue = (kmpc_task_queue_t *) queue->tq_first_child;
1218
1219 while (queue != NULL) {
1220 __kmp_remove_all_child_taskq ( tq, global_tid, queue );
1221
1222 next_child = queue->tq_next_child;
1223 queue->tq_flags |= TQF_DEALLOCATED;
1224 __kmp_remove_queue_from_tree ( tq, global_tid, queue, FALSE );
1225 queue = next_child;
1226 }
1227}
1228
1229static void
1230__kmp_execute_task_from_queue( kmp_taskq_t *tq, ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *thunk, int in_parallel )
1231{
1232 kmpc_task_queue_t *queue = thunk->th.th_shareds->sv_queue;
1233 kmp_int32 tid = __kmp_tid_from_gtid( global_tid );
1234
1235 KF_TRACE(100, ("After dequeueing this Task on (%d):\n", global_tid));
1236 KF_DUMP(100, __kmp_dump_thunk( tq, thunk, global_tid ));
1237 KF_TRACE(100, ("Task Queue: %p looks like this (%d):\n", queue, global_tid));
1238 KF_DUMP(100, __kmp_dump_task_queue( tq, queue, global_tid ));
1239
1240 /*
1241 * For the taskq task, the curr_thunk pushes and pop pairs are set up as follows:
1242 *
1243 * happens exactly once:
1244 * 1) __kmpc_taskq : push (if returning thunk only)
1245 * 4) __kmpc_end_taskq_task : pop
1246 *
1247 * optionally happens *each* time taskq task is dequeued/enqueued:
1248 * 2) __kmpc_taskq_task : pop
1249 * 3) __kmp_execute_task_from_queue : push
1250 *
1251 * execution ordering: 1,(2,3)*,4
1252 */
1253
1254 if (!(thunk->th_flags & TQF_TASKQ_TASK)) {
1255 kmp_int32 index = (queue == tq->tq_root) ? tid : 0;
1256 thunk->th.th_shareds = (kmpc_shared_vars_t *) queue->tq_shareds[index].ai_data;
1257
1258 if ( __kmp_env_consistency_check ) {
1259 __kmp_push_workshare( global_tid,
1260 (queue->tq_flags & TQF_IS_ORDERED) ? ct_task_ordered : ct_task,
1261 queue->tq_loc );
1262 }
1263 }
1264 else {
1265 if ( __kmp_env_consistency_check )
1266 __kmp_push_workshare( global_tid, ct_taskq, queue->tq_loc );
1267 }
1268
1269 if (in_parallel) {
1270 thunk->th_encl_thunk = tq->tq_curr_thunk[tid];
1271 tq->tq_curr_thunk[tid] = thunk;
1272
1273 KF_DUMP( 200, __kmp_dump_thunk_stack( tq->tq_curr_thunk[tid], global_tid ));
1274 }
1275
1276 KF_TRACE( 50, ("Begin Executing Thunk %p from queue %p on (%d)\n", thunk, queue, global_tid));
1277 thunk->th_task (global_tid, thunk);
1278 KF_TRACE( 50, ("End Executing Thunk %p from queue %p on (%d)\n", thunk, queue, global_tid));
1279
1280 if (!(thunk->th_flags & TQF_TASKQ_TASK)) {
1281 if ( __kmp_env_consistency_check )
1282 __kmp_pop_workshare( global_tid, (queue->tq_flags & TQF_IS_ORDERED) ? ct_task_ordered : ct_task,
1283 queue->tq_loc );
1284
1285 if (in_parallel) {
1286 tq->tq_curr_thunk[tid] = thunk->th_encl_thunk;
1287 thunk->th_encl_thunk = NULL;
1288 KF_DUMP( 200, __kmp_dump_thunk_stack( tq->tq_curr_thunk[tid], global_tid ));
1289 }
1290
1291 if ((thunk->th_flags & TQF_IS_ORDERED) && in_parallel) {
1292 __kmp_taskq_check_ordered(global_tid, thunk);
1293 }
1294
1295 __kmp_free_thunk (queue, thunk, in_parallel, global_tid);
1296
1297 KF_TRACE(100, ("T#%d After freeing thunk: %p, TaskQ looks like this:\n", global_tid, thunk));
1298 KF_DUMP(100, __kmp_dump_task_queue( tq, queue, global_tid ));
1299
1300 if (in_parallel) {
1301 KMP_MB(); /* needed so thunk put on free list before outstanding thunk count is decremented */
1302
1303 KMP_DEBUG_ASSERT(queue->tq_th_thunks[tid].ai_data >= 1);
1304
1305 KF_TRACE( 200, ("__kmp_execute_task_from_queue: T#%d has %d thunks in queue %p\n",
1306 global_tid, queue->tq_th_thunks[tid].ai_data-1, queue));
1307
1308 queue->tq_th_thunks[tid].ai_data--;
1309
1310 /* KMP_MB(); */ /* is MB really necessary ? */
1311 }
1312
1313 if (queue->tq.tq_parent != NULL && in_parallel) {
1314 int ct;
1315 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1316 ct = --(queue->tq_ref_count);
1317 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1318 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p dec %d\n",
1319 __LINE__, global_tid, queue, ct));
1320 KMP_DEBUG_ASSERT( ct >= 0 );
1321 }
1322 }
1323}
1324
1325/* --------------------------------------------------------------------------- */
1326
1327/* starts a taskq; creates and returns a thunk for the taskq_task */
1328/* also, returns pointer to shared vars for this thread in "shareds" arg */
1329
1330kmpc_thunk_t *
1331__kmpc_taskq( ident_t *loc, kmp_int32 global_tid, kmpc_task_t taskq_task,
1332 size_t sizeof_thunk, size_t sizeof_shareds,
1333 kmp_int32 flags, kmpc_shared_vars_t **shareds )
1334{
1335 int in_parallel;
1336 kmp_int32 nslots, nthunks, nshareds, nproc;
1337 kmpc_task_queue_t *new_queue, *curr_queue;
1338 kmpc_thunk_t *new_taskq_thunk;
1339 kmp_info_t *th;
1340 kmp_team_t *team;
1341 kmp_taskq_t *tq;
1342 kmp_int32 tid;
1343
1344 KE_TRACE( 10, ("__kmpc_taskq called (%d)\n", global_tid));
1345
1346 th = __kmp_threads[ global_tid ];
1347 team = th -> th.th_team;
1348 tq = & team -> t.t_taskq;
1349 nproc = team -> t.t_nproc;
1350 tid = __kmp_tid_from_gtid( global_tid );
1351
1352 /* find out whether this is a parallel taskq or serialized one. */
1353 in_parallel = in_parallel_context( team );
1354
1355 if( ! tq->tq_root ) {
1356 if (in_parallel) {
1357 /* Vector ORDERED SECTION to taskq version */
1358 th->th.th_dispatch->th_deo_fcn = __kmp_taskq_eo;
1359
1360 /* Vector ORDERED SECTION to taskq version */
1361 th->th.th_dispatch->th_dxo_fcn = __kmp_taskq_xo;
1362 }
1363
1364 if (in_parallel) {
1365 /* This shouldn't be a barrier region boundary, it will confuse the user. */
1366 /* Need the boundary to be at the end taskq instead. */
1367 if ( __kmp_barrier( bs_plain_barrier, global_tid, TRUE, 0, NULL, NULL )) {
1368 /* Creating the active root queue, and we are not the master thread. */
1369 /* The master thread below created the queue and tasks have been */
1370 /* enqueued, and the master thread released this barrier. This */
1371 /* worker thread can now proceed and execute tasks. See also the */
1372 /* TQF_RELEASE_WORKERS which is used to handle this case. */
1373
1374 *shareds = (kmpc_shared_vars_t *) tq->tq_root->tq_shareds[tid].ai_data;
1375
1376 KE_TRACE( 10, ("__kmpc_taskq return (%d)\n", global_tid));
1377
1378 return NULL;
1379 }
1380 }
1381
1382 /* master thread only executes this code */
1383
1384 if( tq->tq_curr_thunk_capacity < nproc ) {
1385 int i;
1386
1387 if(tq->tq_curr_thunk)
1388 __kmp_free(tq->tq_curr_thunk);
1389 else {
1390 /* only need to do this once at outer level, i.e. when tq_curr_thunk is still NULL */
1391 __kmp_init_lock( & tq->tq_freelist_lck );
1392 }
1393
1394 tq->tq_curr_thunk = (kmpc_thunk_t **) __kmp_allocate( nproc * sizeof(kmpc_thunk_t *) );
1395 tq -> tq_curr_thunk_capacity = nproc;
1396 }
1397
1398 if (in_parallel)
1399 tq->tq_global_flags = TQF_RELEASE_WORKERS;
1400 }
1401
1402 /* dkp: in future, if flags & TQF_HEURISTICS, will choose nslots based */
1403 /* on some heuristics (e.g., depth of queue nesting?). */
1404
1405 nslots = (in_parallel) ? (2 * nproc) : 1;
1406
1407 /* There must be nproc * __KMP_TASKQ_THUNKS_PER_TH extra slots for pending */
1408 /* jobs being executed by other threads, and one extra for taskq slot */
1409
1410 nthunks = (in_parallel) ? (nslots + (nproc * __KMP_TASKQ_THUNKS_PER_TH) + 1) : nslots + 2;
1411
1412 /* Only the root taskq gets a per-thread array of shareds. */
1413 /* The rest of the taskq's only get one copy of the shared vars. */
1414
1415 nshareds = ( !tq->tq_root && in_parallel) ? nproc : 1;
1416
1417 /* create overall queue data structure and its components that require allocation */
1418
1419 new_queue = __kmp_alloc_taskq ( tq, in_parallel, nslots, nthunks, nshareds, nproc,
1420 sizeof_thunk, sizeof_shareds, &new_taskq_thunk, global_tid );
1421
1422 /* rest of new_queue initializations */
1423
1424 new_queue->tq_flags = flags & TQF_INTERFACE_FLAGS;
1425
1426 if (in_parallel) {
1427 new_queue->tq_tasknum_queuing = 0;
1428 new_queue->tq_tasknum_serving = 0;
1429 new_queue->tq_flags |= TQF_PARALLEL_CONTEXT;
1430 }
1431
1432 new_queue->tq_taskq_slot = NULL;
1433 new_queue->tq_nslots = nslots;
1434 new_queue->tq_hiwat = HIGH_WATER_MARK (nslots);
1435 new_queue->tq_nfull = 0;
1436 new_queue->tq_head = 0;
1437 new_queue->tq_tail = 0;
1438 new_queue->tq_loc = loc;
1439
1440 if ((new_queue->tq_flags & TQF_IS_ORDERED) && in_parallel) {
1441 /* prepare to serve the first-queued task's ORDERED directive */
1442 new_queue->tq_tasknum_serving = 1;
1443
1444 /* Vector ORDERED SECTION to taskq version */
1445 th->th.th_dispatch->th_deo_fcn = __kmp_taskq_eo;
1446
1447 /* Vector ORDERED SECTION to taskq version */
1448 th->th.th_dispatch->th_dxo_fcn = __kmp_taskq_xo;
1449 }
1450
1451 /* create a new thunk for the taskq_task in the new_queue */
1452 *shareds = (kmpc_shared_vars_t *) new_queue->tq_shareds[0].ai_data;
1453
1454 new_taskq_thunk->th.th_shareds = *shareds;
1455 new_taskq_thunk->th_task = taskq_task;
1456 new_taskq_thunk->th_flags = new_queue->tq_flags | TQF_TASKQ_TASK;
1457 new_taskq_thunk->th_status = 0;
1458
1459 KMP_DEBUG_ASSERT (new_taskq_thunk->th_flags & TQF_TASKQ_TASK);
1460
1461 /* KMP_MB(); */ /* make sure these inits complete before threads start using this queue (necessary?) */
1462
1463 /* insert the new task queue into the tree, but only after all fields initialized */
1464
1465 if (in_parallel) {
1466 if( ! tq->tq_root ) {
1467 new_queue->tq.tq_parent = NULL;
1468 new_queue->tq_first_child = NULL;
1469 new_queue->tq_next_child = NULL;
1470 new_queue->tq_prev_child = NULL;
1471 new_queue->tq_ref_count = 1;
1472 tq->tq_root = new_queue;
1473 }
1474 else {
1475 curr_queue = tq->tq_curr_thunk[tid]->th.th_shareds->sv_queue;
1476 new_queue->tq.tq_parent = curr_queue;
1477 new_queue->tq_first_child = NULL;
1478 new_queue->tq_prev_child = NULL;
1479 new_queue->tq_ref_count = 1; /* for this the thread that built the queue */
1480
1481 KMP_DEBUG_REF_CTS(("line %d gtid %d: Q %p alloc %d\n",
1482 __LINE__, global_tid, new_queue, new_queue->tq_ref_count));
1483
1484 __kmp_acquire_lock(& curr_queue->tq_link_lck, global_tid);
1485
1486 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1487 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1488
1489 new_queue->tq_next_child = (struct kmpc_task_queue_t *) curr_queue->tq_first_child;
1490
1491 if (curr_queue->tq_first_child != NULL)
1492 curr_queue->tq_first_child->tq_prev_child = new_queue;
1493
1494 curr_queue->tq_first_child = new_queue;
1495
1496 __kmp_release_lock(& curr_queue->tq_link_lck, global_tid);
1497 }
1498
1499 /* set up thunk stack only after code that determines curr_queue above */
1500 new_taskq_thunk->th_encl_thunk = tq->tq_curr_thunk[tid];
1501 tq->tq_curr_thunk[tid] = new_taskq_thunk;
1502
1503 KF_DUMP( 200, __kmp_dump_thunk_stack( tq->tq_curr_thunk[tid], global_tid ));
1504 }
1505 else {
1506 new_taskq_thunk->th_encl_thunk = 0;
1507 new_queue->tq.tq_parent = NULL;
1508 new_queue->tq_first_child = NULL;
1509 new_queue->tq_next_child = NULL;
1510 new_queue->tq_prev_child = NULL;
1511 new_queue->tq_ref_count = 1;
1512 }
1513
1514#ifdef KMP_DEBUG
1515 KF_TRACE(150, ("Creating TaskQ Task on (%d):\n", global_tid));
1516 KF_DUMP(150, __kmp_dump_thunk( tq, new_taskq_thunk, global_tid ));
1517
1518 if (in_parallel) {
1519 KF_TRACE(25, ("After TaskQ at %p Creation on (%d):\n", new_queue, global_tid));
1520 } else {
1521 KF_TRACE(25, ("After Serial TaskQ at %p Creation on (%d):\n", new_queue, global_tid));
1522 }
1523
1524 KF_DUMP(25, __kmp_dump_task_queue( tq, new_queue, global_tid ));
1525
1526 if (in_parallel) {
1527 KF_DUMP(50, __kmp_dump_task_queue_tree( tq, tq->tq_root, global_tid ));
1528 }
1529#endif /* KMP_DEBUG */
1530
1531 if ( __kmp_env_consistency_check )
1532 __kmp_push_workshare( global_tid, ct_taskq, new_queue->tq_loc );
1533
1534 KE_TRACE( 10, ("__kmpc_taskq return (%d)\n", global_tid));
1535
1536 return new_taskq_thunk;
1537}
1538
1539
1540/* ends a taskq; last thread out destroys the queue */
1541
1542void
1543__kmpc_end_taskq(ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *taskq_thunk)
1544{
1545#ifdef KMP_DEBUG
1546 kmp_int32 i;
1547#endif
1548 kmp_taskq_t *tq;
1549 int in_parallel;
1550 kmp_info_t *th;
1551 kmp_int32 is_outermost;
1552 kmpc_task_queue_t *queue;
1553 kmpc_thunk_t *thunk;
1554 int nproc;
1555
1556 KE_TRACE( 10, ("__kmpc_end_taskq called (%d)\n", global_tid));
1557
1558 tq = & __kmp_threads[global_tid] -> th.th_team -> t.t_taskq;
1559 nproc = __kmp_threads[global_tid] -> th.th_team -> t.t_nproc;
1560
1561 /* For the outermost taskq only, all but one thread will have taskq_thunk == NULL */
1562 queue = (taskq_thunk == NULL) ? tq->tq_root : taskq_thunk->th.th_shareds->sv_queue;
1563
1564 KE_TRACE( 50, ("__kmpc_end_taskq queue=%p (%d) \n", queue, global_tid));
1565 is_outermost = (queue == tq->tq_root);
1566 in_parallel = (queue->tq_flags & TQF_PARALLEL_CONTEXT);
1567
1568 if (in_parallel) {
1569 kmp_uint32 spins;
1570
1571 /* this is just a safeguard to release the waiting threads if */
1572 /* the outermost taskq never queues a task */
1573
1574 if (is_outermost && (KMP_MASTER_GTID( global_tid ))) {
1575 if( tq->tq_global_flags & TQF_RELEASE_WORKERS ) {
1576 /* no lock needed, workers are still in spin mode */
1577 tq->tq_global_flags &= ~TQF_RELEASE_WORKERS;
1578
1579 __kmp_end_split_barrier( bs_plain_barrier, global_tid );
1580 }
1581 }
1582
1583 /* keep dequeueing work until all tasks are queued and dequeued */
1584
1585 do {
1586 /* wait until something is available to dequeue */
1587 KMP_INIT_YIELD(spins);
1588
1589 while ( (queue->tq_nfull == 0)
1590 && (queue->tq_taskq_slot == NULL)
1591 && (! __kmp_taskq_has_any_children(queue) )
1592 && (! (queue->tq_flags & TQF_ALL_TASKS_QUEUED) )
1593 ) {
Jim Cownie5e8470a2013-09-27 10:38:44 +00001594 KMP_YIELD_WHEN( TRUE, spins );
1595 }
1596
1597 /* check to see if we can execute tasks in the queue */
1598 while ( ( (queue->tq_nfull != 0) || (queue->tq_taskq_slot != NULL) )
1599 && (thunk = __kmp_find_task_in_queue(global_tid, queue)) != NULL
1600 ) {
1601 KF_TRACE(50, ("Found thunk: %p in primary queue %p (%d)\n", thunk, queue, global_tid));
1602 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1603 }
1604
1605 /* see if work found can be found in a descendant queue */
1606 if ( (__kmp_taskq_has_any_children(queue))
1607 && (thunk = __kmp_find_task_in_descendant_queue(global_tid, queue)) != NULL
1608 ) {
1609
1610 KF_TRACE(50, ("Stole thunk: %p in descendant queue: %p while waiting in queue: %p (%d)\n",
1611 thunk, thunk->th.th_shareds->sv_queue, queue, global_tid ));
1612
1613 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1614 }
1615
1616 } while ( (! (queue->tq_flags & TQF_ALL_TASKS_QUEUED))
1617 || (queue->tq_nfull != 0)
1618 );
1619
1620 KF_TRACE(50, ("All tasks queued and dequeued in queue: %p (%d)\n", queue, global_tid));
1621
1622 /* wait while all tasks are not finished and more work found
1623 in descendant queues */
1624
1625 while ( (!__kmp_taskq_tasks_finished(queue))
1626 && (thunk = __kmp_find_task_in_descendant_queue(global_tid, queue)) != NULL
1627 ) {
1628
1629 KF_TRACE(50, ("Stole thunk: %p in descendant queue: %p while waiting in queue: %p (%d)\n",
1630 thunk, thunk->th.th_shareds->sv_queue, queue, global_tid));
1631
1632 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1633 }
1634
1635 KF_TRACE(50, ("No work found in descendent queues or all work finished in queue: %p (%d)\n", queue, global_tid));
1636
1637 if (!is_outermost) {
1638 /* need to return if NOWAIT present and not outermost taskq */
1639
1640 if (queue->tq_flags & TQF_IS_NOWAIT) {
1641 __kmp_acquire_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1642 queue->tq_ref_count--;
1643 KMP_DEBUG_ASSERT( queue->tq_ref_count >= 0 );
1644 __kmp_release_lock(& queue->tq.tq_parent->tq_link_lck, global_tid);
1645
1646 KE_TRACE( 10, ("__kmpc_end_taskq return for nowait case (%d)\n", global_tid));
1647
1648 return;
1649 }
1650
1651 __kmp_find_and_remove_finished_child_taskq( tq, global_tid, queue );
1652
1653 /* WAIT until all tasks are finished and no child queues exist before proceeding */
1654 KMP_INIT_YIELD(spins);
1655
1656 while (!__kmp_taskq_tasks_finished(queue) || __kmp_taskq_has_any_children(queue)) {
1657 thunk = __kmp_find_task_in_ancestor_queue( tq, global_tid, queue );
1658
1659 if (thunk != NULL) {
1660 KF_TRACE(50, ("Stole thunk: %p in ancestor queue: %p while waiting in queue: %p (%d)\n",
1661 thunk, thunk->th.th_shareds->sv_queue, queue, global_tid));
1662 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1663 }
1664
1665 KMP_YIELD_WHEN( thunk == NULL, spins );
1666
1667 __kmp_find_and_remove_finished_child_taskq( tq, global_tid, queue );
1668 }
1669
1670 __kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
1671 if ( !(queue->tq_flags & TQF_DEALLOCATED) ) {
1672 queue->tq_flags |= TQF_DEALLOCATED;
1673 }
1674 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
1675
1676 /* only the allocating thread can deallocate the queue */
1677 if (taskq_thunk != NULL) {
1678 __kmp_remove_queue_from_tree( tq, global_tid, queue, TRUE );
1679 }
1680
1681 KE_TRACE( 10, ("__kmpc_end_taskq return for non_outermost queue, wait case (%d)\n", global_tid));
1682
1683 return;
1684 }
1685
1686 /* Outermost Queue: steal work from descendants until all tasks are finished */
1687
1688 KMP_INIT_YIELD(spins);
1689
1690 while (!__kmp_taskq_tasks_finished(queue)) {
1691 thunk = __kmp_find_task_in_descendant_queue(global_tid, queue);
1692
1693 if (thunk != NULL) {
1694 KF_TRACE(50, ("Stole thunk: %p in descendant queue: %p while waiting in queue: %p (%d)\n",
1695 thunk, thunk->th.th_shareds->sv_queue, queue, global_tid));
1696
1697 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1698 }
1699
1700 KMP_YIELD_WHEN( thunk == NULL, spins );
1701 }
1702
1703 /* Need this barrier to prevent destruction of queue before threads have all executed above code */
1704 /* This may need to be done earlier when NOWAIT is implemented for the outermost level */
1705
1706 if ( !__kmp_barrier( bs_plain_barrier, global_tid, TRUE, 0, NULL, NULL )) {
1707 /* the queue->tq_flags & TQF_IS_NOWAIT case is not yet handled here; */
1708 /* for right now, everybody waits, and the master thread destroys the */
1709 /* remaining queues. */
1710
1711 __kmp_remove_all_child_taskq( tq, global_tid, queue );
1712
1713 /* Now destroy the root queue */
1714 KF_TRACE(100, ("T#%d Before Deletion of top-level TaskQ at %p:\n", global_tid, queue ));
1715 KF_DUMP(100, __kmp_dump_task_queue( tq, queue, global_tid ));
1716
1717#ifdef KMP_DEBUG
1718 /* the root queue entry */
1719 KMP_DEBUG_ASSERT ((queue->tq.tq_parent == NULL) && (queue->tq_next_child == NULL));
1720
1721 /* children must all be gone by now because of barrier above */
1722 KMP_DEBUG_ASSERT (queue->tq_first_child == NULL);
1723
1724 for (i=0; i<nproc; i++) {
1725 KMP_DEBUG_ASSERT(queue->tq_th_thunks[i].ai_data == 0);
1726 }
1727
1728 for (i=0, thunk=queue->tq_free_thunks; thunk != NULL; i++, thunk=thunk->th.th_next_free);
1729
1730 KMP_DEBUG_ASSERT (i == queue->tq_nslots + (nproc * __KMP_TASKQ_THUNKS_PER_TH));
1731
1732 for (i = 0; i < nproc; i++) {
1733 KMP_DEBUG_ASSERT( ! tq->tq_curr_thunk[i] );
1734 }
1735#endif
1736 /* unlink the root queue entry */
1737 tq -> tq_root = NULL;
1738
1739 /* release storage for root queue entry */
1740 KF_TRACE(50, ("After Deletion of top-level TaskQ at %p on (%d):\n", queue, global_tid));
1741
1742 queue->tq_flags |= TQF_DEALLOCATED;
1743 __kmp_free_taskq ( tq, queue, in_parallel, global_tid );
1744
1745 KF_DUMP(50, __kmp_dump_task_queue_tree( tq, tq->tq_root, global_tid ));
1746
1747 /* release the workers now that the data structures are up to date */
1748 __kmp_end_split_barrier( bs_plain_barrier, global_tid );
1749 }
1750
1751 th = __kmp_threads[ global_tid ];
1752
1753 /* Reset ORDERED SECTION to parallel version */
1754 th->th.th_dispatch->th_deo_fcn = 0;
1755
1756 /* Reset ORDERED SECTION to parallel version */
1757 th->th.th_dispatch->th_dxo_fcn = 0;
1758 }
1759 else {
1760 /* in serial execution context, dequeue the last task */
1761 /* and execute it, if there were any tasks encountered */
1762
1763 if (queue->tq_nfull > 0) {
1764 KMP_DEBUG_ASSERT(queue->tq_nfull == 1);
1765
1766 thunk = __kmp_dequeue_task(global_tid, queue, in_parallel);
1767
1768 if (queue->tq_flags & TQF_IS_LAST_TASK) {
1769 /* TQF_IS_LASTPRIVATE, one thing in queue, __kmpc_end_taskq_task() */
1770 /* has been run so this is last task, run with TQF_IS_LAST_TASK so */
1771 /* instrumentation does copy-out. */
1772
1773 /* no need for test_then_or call since already locked */
1774 thunk->th_flags |= TQF_IS_LAST_TASK;
1775 }
1776
1777 KF_TRACE(50, ("T#%d found thunk: %p in serial queue: %p\n", global_tid, thunk, queue));
1778
1779 __kmp_execute_task_from_queue( tq, loc, global_tid, thunk, in_parallel );
1780 }
1781
1782 /* destroy the unattached serial queue now that there is no more work to do */
1783 KF_TRACE(100, ("Before Deletion of Serialized TaskQ at %p on (%d):\n", queue, global_tid));
1784 KF_DUMP(100, __kmp_dump_task_queue( tq, queue, global_tid ));
1785
1786#ifdef KMP_DEBUG
1787 i = 0;
1788 for (thunk=queue->tq_free_thunks; thunk != NULL; thunk=thunk->th.th_next_free)
1789 ++i;
1790 KMP_DEBUG_ASSERT (i == queue->tq_nslots + 1);
1791#endif
1792 /* release storage for unattached serial queue */
1793 KF_TRACE(50, ("Serialized TaskQ at %p deleted on (%d).\n", queue, global_tid));
1794
1795 queue->tq_flags |= TQF_DEALLOCATED;
1796 __kmp_free_taskq ( tq, queue, in_parallel, global_tid );
1797 }
1798
1799 KE_TRACE( 10, ("__kmpc_end_taskq return (%d)\n", global_tid));
1800}
1801
1802/* Enqueues a task for thunk previously created by __kmpc_task_buffer. */
1803/* Returns nonzero if just filled up queue */
1804
1805kmp_int32
1806__kmpc_task(ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *thunk)
1807{
1808 kmp_int32 ret;
1809 kmpc_task_queue_t *queue;
1810 int in_parallel;
1811 kmp_taskq_t *tq;
1812
1813 KE_TRACE( 10, ("__kmpc_task called (%d)\n", global_tid));
1814
1815 KMP_DEBUG_ASSERT (!(thunk->th_flags & TQF_TASKQ_TASK)); /* thunk->th_task is a regular task */
1816
1817 tq = &__kmp_threads[global_tid] -> th.th_team -> t.t_taskq;
1818 queue = thunk->th.th_shareds->sv_queue;
1819 in_parallel = (queue->tq_flags & TQF_PARALLEL_CONTEXT);
1820
1821 if (in_parallel && (thunk->th_flags & TQF_IS_ORDERED))
1822 thunk->th_tasknum = ++queue->tq_tasknum_queuing;
1823
1824 /* For serial execution dequeue the preceding task and execute it, if one exists */
1825 /* This cannot be the last task. That one is handled in __kmpc_end_taskq */
1826
1827 if (!in_parallel && queue->tq_nfull > 0) {
1828 kmpc_thunk_t *prev_thunk;
1829
1830 KMP_DEBUG_ASSERT(queue->tq_nfull == 1);
1831
1832 prev_thunk = __kmp_dequeue_task(global_tid, queue, in_parallel);
1833
1834 KF_TRACE(50, ("T#%d found thunk: %p in serial queue: %p\n", global_tid, prev_thunk, queue));
1835
1836 __kmp_execute_task_from_queue( tq, loc, global_tid, prev_thunk, in_parallel );
1837 }
1838
1839 /* The instrumentation sequence is: __kmpc_task_buffer(), initialize private */
1840 /* variables, __kmpc_task(). The __kmpc_task_buffer routine checks that the */
1841 /* task queue is not full and allocates a thunk (which is then passed to */
1842 /* __kmpc_task()). So, the enqueue below should never fail due to a full queue. */
1843
1844 KF_TRACE(100, ("After enqueueing this Task on (%d):\n", global_tid));
1845 KF_DUMP(100, __kmp_dump_thunk( tq, thunk, global_tid ));
1846
1847 ret = __kmp_enqueue_task ( tq, global_tid, queue, thunk, in_parallel );
1848
1849 KF_TRACE(100, ("Task Queue looks like this on (%d):\n", global_tid));
1850 KF_DUMP(100, __kmp_dump_task_queue( tq, queue, global_tid ));
1851
1852 KE_TRACE( 10, ("__kmpc_task return (%d)\n", global_tid));
1853
1854 return ret;
1855}
1856
1857/* enqueues a taskq_task for thunk previously created by __kmpc_taskq */
1858/* this should never be called unless in a parallel context */
1859
1860void
1861__kmpc_taskq_task(ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *thunk, kmp_int32 status)
1862{
1863 kmpc_task_queue_t *queue;
1864 kmp_taskq_t *tq = &__kmp_threads[global_tid] -> th.th_team -> t.t_taskq;
1865 int tid = __kmp_tid_from_gtid( global_tid );
1866
1867 KE_TRACE( 10, ("__kmpc_taskq_task called (%d)\n", global_tid));
1868 KF_TRACE(100, ("TaskQ Task argument thunk on (%d):\n", global_tid));
1869 KF_DUMP(100, __kmp_dump_thunk( tq, thunk, global_tid ));
1870
1871 queue = thunk->th.th_shareds->sv_queue;
1872
1873 if ( __kmp_env_consistency_check )
1874 __kmp_pop_workshare( global_tid, ct_taskq, loc );
1875
1876 /* thunk->th_task is the taskq_task */
1877 KMP_DEBUG_ASSERT (thunk->th_flags & TQF_TASKQ_TASK);
1878
1879 /* not supposed to call __kmpc_taskq_task if it's already enqueued */
1880 KMP_DEBUG_ASSERT (queue->tq_taskq_slot == NULL);
1881
1882 /* dequeue taskq thunk from curr_thunk stack */
1883 tq->tq_curr_thunk[tid] = thunk->th_encl_thunk;
1884 thunk->th_encl_thunk = NULL;
1885
1886 KF_DUMP( 200, __kmp_dump_thunk_stack( tq->tq_curr_thunk[tid], global_tid ));
1887
1888 thunk->th_status = status;
1889
1890 KMP_MB(); /* flush thunk->th_status before taskq_task enqueued to avoid race condition */
1891
1892 /* enqueue taskq_task in thunk into special slot in queue */
1893 /* GEH - probably don't need to lock taskq slot since only one */
1894 /* thread enqueues & already a lock set at dequeue point */
1895
1896 queue->tq_taskq_slot = thunk;
1897
1898 KE_TRACE( 10, ("__kmpc_taskq_task return (%d)\n", global_tid));
1899}
1900
1901/* ends a taskq_task; done generating tasks */
1902
1903void
1904__kmpc_end_taskq_task(ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *thunk)
1905{
1906 kmp_taskq_t *tq;
1907 kmpc_task_queue_t *queue;
1908 int in_parallel;
1909 int tid;
1910
1911 KE_TRACE( 10, ("__kmpc_end_taskq_task called (%d)\n", global_tid));
1912
1913 tq = &__kmp_threads[global_tid] -> th.th_team -> t.t_taskq;
1914 queue = thunk->th.th_shareds->sv_queue;
1915 in_parallel = (queue->tq_flags & TQF_PARALLEL_CONTEXT);
1916 tid = __kmp_tid_from_gtid( global_tid );
1917
1918 if ( __kmp_env_consistency_check )
1919 __kmp_pop_workshare( global_tid, ct_taskq, loc );
1920
1921 if (in_parallel) {
1922#if KMP_ARCH_X86 || \
1923 KMP_ARCH_X86_64
1924
1925 KMP_TEST_THEN_OR32( &queue->tq_flags, (kmp_int32) TQF_ALL_TASKS_QUEUED );
1926#else
1927 {
1928 __kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
1929
1930 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1931 /* Seems to work fine without this call for digital/alpha, needed for IBM/RS6000 */
1932
1933 queue->tq_flags |= TQF_ALL_TASKS_QUEUED;
1934
1935 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
1936 }
1937#endif
1938 }
1939
1940 if (thunk->th_flags & TQF_IS_LASTPRIVATE) {
1941 /* Normally, __kmp_find_task_in_queue() refuses to schedule the last task in the */
1942 /* queue if TQF_IS_LASTPRIVATE so we can positively identify that last task */
1943 /* and run it with its TQF_IS_LAST_TASK bit turned on in th_flags. When */
1944 /* __kmpc_end_taskq_task() is called we are done generating all the tasks, so */
1945 /* we know the last one in the queue is the lastprivate task. Mark the queue */
1946 /* as having gotten to this state via tq_flags & TQF_IS_LAST_TASK; when that */
1947 /* task actually executes mark it via th_flags & TQF_IS_LAST_TASK (this th_flags */
1948 /* bit signals the instrumented code to do copy-outs after execution). */
1949
1950 if (! in_parallel) {
1951 /* No synchronization needed for serial context */
1952 queue->tq_flags |= TQF_IS_LAST_TASK;
1953 }
1954 else {
1955#if KMP_ARCH_X86 || \
1956 KMP_ARCH_X86_64
1957
1958 KMP_TEST_THEN_OR32( &queue->tq_flags, (kmp_int32) TQF_IS_LAST_TASK );
1959#else
1960 {
1961 __kmp_acquire_lock(& queue->tq_queue_lck, global_tid);
1962
1963 KMP_MB(); /* make sure data structures are in consistent state before querying them */
1964 /* Seems to work without this call for digital/alpha, needed for IBM/RS6000 */
1965
1966 queue->tq_flags |= TQF_IS_LAST_TASK;
1967
1968 __kmp_release_lock(& queue->tq_queue_lck, global_tid);
1969 }
1970#endif
1971 /* to prevent race condition where last task is dequeued but */
1972 /* flag isn't visible yet (not sure about this) */
1973 KMP_MB();
1974 }
1975 }
1976
1977 /* dequeue taskq thunk from curr_thunk stack */
1978 if (in_parallel) {
1979 tq->tq_curr_thunk[tid] = thunk->th_encl_thunk;
1980 thunk->th_encl_thunk = NULL;
1981
1982 KF_DUMP( 200, __kmp_dump_thunk_stack( tq->tq_curr_thunk[tid], global_tid ));
1983 }
1984
1985 KE_TRACE( 10, ("__kmpc_end_taskq_task return (%d)\n", global_tid));
1986}
1987
1988/* returns thunk for a regular task based on taskq_thunk */
1989/* (__kmpc_taskq_task does the analogous thing for a TQF_TASKQ_TASK) */
1990
1991kmpc_thunk_t *
1992__kmpc_task_buffer(ident_t *loc, kmp_int32 global_tid, kmpc_thunk_t *taskq_thunk, kmpc_task_t task)
1993{
1994 kmp_taskq_t *tq;
1995 kmpc_task_queue_t *queue;
1996 kmpc_thunk_t *new_thunk;
1997 int in_parallel;
1998
1999 KE_TRACE( 10, ("__kmpc_task_buffer called (%d)\n", global_tid));
2000
2001 KMP_DEBUG_ASSERT (taskq_thunk->th_flags & TQF_TASKQ_TASK); /* taskq_thunk->th_task is the taskq_task */
2002
2003 tq = &__kmp_threads[global_tid] -> th.th_team -> t.t_taskq;
2004 queue = taskq_thunk->th.th_shareds->sv_queue;
2005 in_parallel = (queue->tq_flags & TQF_PARALLEL_CONTEXT);
2006
2007 /* The instrumentation sequence is: __kmpc_task_buffer(), initialize private */
2008 /* variables, __kmpc_task(). The __kmpc_task_buffer routine checks that the */
2009 /* task queue is not full and allocates a thunk (which is then passed to */
2010 /* __kmpc_task()). So, we can pre-allocate a thunk here assuming it will be */
2011 /* the next to be enqueued in __kmpc_task(). */
2012
2013 new_thunk = __kmp_alloc_thunk (queue, in_parallel, global_tid);
2014 new_thunk->th.th_shareds = (kmpc_shared_vars_t *) queue->tq_shareds[0].ai_data;
2015 new_thunk->th_encl_thunk = NULL;
2016 new_thunk->th_task = task;
2017
2018 /* GEH - shouldn't need to lock the read of tq_flags here */
2019 new_thunk->th_flags = queue->tq_flags & TQF_INTERFACE_FLAGS;
2020
2021 new_thunk->th_status = 0;
2022
2023 KMP_DEBUG_ASSERT (!(new_thunk->th_flags & TQF_TASKQ_TASK));
2024
2025 KF_TRACE(100, ("Creating Regular Task on (%d):\n", global_tid));
2026 KF_DUMP(100, __kmp_dump_thunk( tq, new_thunk, global_tid ));
2027
2028 KE_TRACE( 10, ("__kmpc_task_buffer return (%d)\n", global_tid));
2029
2030 return new_thunk;
2031}
2032
2033/* --------------------------------------------------------------------------- */