diff --git a/docs/doxygen/task-batching.md b/docs/doxygen/task-batching.md index 24e2c5d96..d5e500904 100644 --- a/docs/doxygen/task-batching.md +++ b/docs/doxygen/task-batching.md @@ -1,3 +1,7 @@ + + Task Batching {#task_batching} ============== @@ -61,9 +65,13 @@ Recommended collection helper The preferred interface for GPU submit hooks is `parsec_gpu_task_collect_batch()`. The runtime passes the submit hook a -singleton `parsec_gpu_task_t *gpu_task`. The hook calls the collector with a -callback that decides, for each task currently pending on the same stream, -whether that candidate can be added to the batch headed by `gpu_task`. +singleton `parsec_gpu_task_t *gpu_task` on its initial invocation. The hook +calls the collector with a callback that decides, for each task currently +pending on the same stream, whether that candidate can be added to the batch +headed by `gpu_task`. After a batched hook returns `PARSEC_HOOK_RETURN_AGAIN`, +the continuation invocation receives the same intact ring. Calling the +collector again is safe: it returns the existing follower count without +scanning or modifying the pending FIFO. The callback has the type `parsec_gpu_task_batch_cb_t` and receives: @@ -73,12 +81,19 @@ The callback has the type `parsec_gpu_task_batch_cb_t` and receives: The callback return value controls the iterator: -- negative: stop immediately and return that error code; -- zero: remove `candidate` from the pending FIFO and append it to - `batch_head`'s task ring; -- positive: leave `candidate` pending and continue to the next pending task. +- `PARSEC_GPU_TASK_BATCH_ACCEPT`: remove `candidate` from the pending FIFO and + append it to `batch_head`'s task ring; +- `PARSEC_GPU_TASK_BATCH_REJECT`: leave `candidate` pending and continue to the + next pending task; +- `PARSEC_GPU_TASK_BATCH_STOP`: end collection successfully, leaving the + current candidate and all unvisited candidates pending. -The callback must not modify `gpu_stream->fifo_pending` directly. +The collector does not impose a scan or batch-size bound. The submit hook owns +that policy and its callback should return `PARSEC_GPU_TASK_BATCH_STOP` once it +has accepted enough work. The callback executes while +`gpu_stream->fifo_pending` is locked, so it must remain short and nonblocking. +It must not modify or acquire that FIFO, call the collector recursively, or +otherwise reenter pending-task operations on the same stream. Example: @@ -93,9 +108,9 @@ gemm_batch_match(parsec_gpu_task_t *candidate, if( (batch_head->ec->task_class == candidate->ec->task_class) && (batch_head->ec->selected_chore == candidate->ec->selected_chore) && (batch_head->ec->selected_device == candidate->ec->selected_device) ) { - return 0; + return PARSEC_GPU_TASK_BATCH_ACCEPT; } - return 1; + return PARSEC_GPU_TASK_BATCH_REJECT; } int @@ -129,20 +144,183 @@ gemm_kernel_cuda(parsec_device_gpu_module_t *gpu_device, } ``` -`parsec_gpu_task_collect_batch()` returns the number of additional tasks appended -to the ring on success, or the negative callback error. A return value of 0 -means no task was batched, either because no compatible pending task was found, -because batching is disabled or unsupported by the head task's selected device, -or because the head task's selected incarnation is not batch-capable. Tasks -accepted before an error remain attached to `gpu_task`; tasks not accepted -remain in `gpu_stream->fifo_pending`. +`parsec_gpu_task_collect_batch()` returns the number of additional tasks +appended to the ring on success, including when the callback stops collection. +A return value of 0 means no task was batched, either because the callback +stopped before accepting one, no compatible pending task was found, batching +is disabled or unsupported by the head task's selected device, or the head +task's selected incarnation is not batch-capable. Any callback value outside +the three actions above returns `PARSEC_HOOK_RETURN_ERROR`. Tasks accepted +before that error remain attached to `gpu_task`; tasks not accepted remain in +`gpu_stream->fifo_pending`. The submit hook does not need a completion callback merely to return the ring to the runtime. If a batched submit hook returns a non-singleton task ring, the GPU progress engine automatically chains that ring into the next stream's pending FIFO after the recorded device event completes. The normal data retrieval, epilog, ownership, pushout, and task completion paths then process the tasks one -at a time. +at a time. Every stream insertion merges all ring members according to the +stream's priority policy; equal-priority members retain their ring order. + +The submit result settles ownership of the complete ring: + +- `PARSEC_HOOK_RETURN_DONE` commits every member to the submitted device work; +- `PARSEC_HOOK_RETURN_AGAIN` commits the ring as continuation state, waits for + work queued by the hook, and re-enters the hook with the same ring after the + stream event completes. The hook must have submitted progress for every ring + member before returning `AGAIN`; +- `PARSEC_HOOK_RETURN_NEXT` immediately restores the followers and applies the + existing singleton `NEXT` path to the head; +- `PARSEC_HOOK_RETURN_ASYNC` transfers every execution context in the ring to + the hook, which must eventually complete or reschedule each context. The GPU + engine releases all corresponding device-task wrappers; +- `PARSEC_HOOK_RETURN_ERROR` and `PARSEC_HOOK_RETURN_DISABLE` quiesce the + affected execution stream and clean every member of the failed batch before + propagating the terminal status. + +Device-wide recovery after `PARSEC_HOOK_RETURN_DISABLE` is not implemented. +The terminal path settles the batch that observed the error, but it does not +drain other stream FIFOs, event slots, or the device-wide pending queue. + +Because `AGAIN` preserves a collected ring, hooks must check predictable +resource constraints before calling `parsec_gpu_task_collect_batch()`. Returning +`AGAIN` before collection is a normal singleton retry; returning it afterward +means that the collected batch has started and must continue as one unit. + +The runtime never disbands a ring on the `AGAIN` path. A submit hook that no +longer wants to continue the batch must split the ring itself before returning: +detach the followers, restore the head as a valid singleton, and explicitly +give every detached wrapper a new owner. For example, followers may be returned +to `gpu_stream->fifo_pending` using that stream's priority-preserving insertion +policy, or the hook may explicitly retain responsibility for completing the +execution contexts and releasing their wrappers. Merely breaking the links is +insufficient because it leaves the followers unreachable. If the hook returns +`AGAIN` after disbanding, only the ring or singleton it leaves attached to +`gpu_task` is retained by the event. + +Ownership examples +------------------ + +### Coroutine-style `AGAIN` + +Calling the collector on every coroutine entry is safe. The initial call may +build a batch; continuation calls only report the followers already attached +to the same ring. The application-specific progress routine must submit work +for every member before reporting that another continuation is needed: + +```c +int +submit_coroutine_batch(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *gpu_task, + parsec_gpu_exec_stream_t *gpu_stream) +{ + int nb_batched; + + (void)gpu_device; + nb_batched = parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + batch_match, NULL); + if( nb_batched < 0 ) { + return nb_batched; + } + + /* Application code: resume and submit one progress step for every member + * of gpu_task's ring. The same ring returns here after AGAIN completes. + */ + return submit_coroutine_step(gpu_task, gpu_stream) + ? PARSEC_HOOK_RETURN_AGAIN + : PARSEC_HOOK_RETURN_DONE; +} +``` + +### Batched `ASYNC` + +`ASYNC` transfers every execution context, not the device-task wrappers, to the +hook. Save or enqueue each `ec` before returning; the GPU engine releases the +wrappers after the hook returns. The new owner must eventually complete or +reschedule every saved context: + +```c +parsec_gpu_task_t *current = gpu_task; + +do { + parsec_gpu_task_t *next = + (parsec_gpu_task_t *)current->list_item.list_next; + parsec_task_t *ec = current->ec; + + PARSEC_LIST_ITEM_SINGLETON(ec); + async_owner_enqueue(ec); /* Eventually complete or reschedule ec. */ + current = next; +} while( current != gpu_task ); + +return PARSEC_HOOK_RETURN_ASYNC; +``` + +The asynchronous owner must retain the execution contexts, not pointers to +`parsec_gpu_task_t`, because those wrappers become invalid after the return. + +### Manually disbanding an `AGAIN` ring + +`parsec_list_item_ring_chop()` reconnects the followers but deliberately leaves +the removed head's links invalid. Restore the head as a singleton, refresh the +followers' priority snapshots, and return the follower ring to the pending +stream with its configured ordering policy. Call this helper from the submit +hook after the collector returns, never from the collector callback while the +pending FIFO is locked: + +```c +static void +requeue_batch_followers(parsec_gpu_exec_stream_t *gpu_stream, + parsec_gpu_task_t *gpu_task) +{ + parsec_list_item_t *followers; + + followers = parsec_list_item_ring_chop(&gpu_task->list_item); + PARSEC_LIST_ITEM_SINGLETON(&gpu_task->list_item); + if( NULL == followers ) { + return; + } + +#if PARSEC_GPU_USE_PRIORITIES + parsec_gpu_task_t *first = (parsec_gpu_task_t *)followers; + parsec_gpu_task_t *current = first; + + do { + current->priority = current->ec->priority; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != first ); + parsec_list_chain_sorted(gpu_stream->fifo_pending, followers, + offsetof(parsec_gpu_task_t, priority)); +#else + parsec_list_chain_back(gpu_stream->fifo_pending, followers); +#endif +} +``` + +After this helper, `gpu_task` is the only member retained by an `AGAIN` return. +The hook must submit progress for that head before returning `AGAIN`; the +followers have already received a new owner through `fifo_pending`. + +Profiling semantics +------------------- + +GPU execution profiling is owned by the common device runtime, not by the +submit hook. After a hook first returns `PARSEC_HOOK_RETURN_DONE` or +`PARSEC_HOOK_RETURN_AGAIN`, the runtime emits one task start for every member of +the finalized ring. It emits one matching task end for every member when the +submitted execution finally completes. A task therefore contributes exactly +one start/end pair whether it was submitted alone or as part of a batch. + +An `AGAIN` event only resumes the same logical execution. The runtime keeps all +member intervals open across any number of repeated `AGAIN` continuations and +does not emit progress markers for those intermediate steps. A committed +non-singleton ring has already been fully accounted, so repeated continuations +do not rescan every member looking for new profiling starts. + +The open profiling state belongs to each device-task wrapper. If a hook +manually disbands an `AGAIN` ring and returns followers to the GPU stream's +pending FIFO, their intervals follow them and close when those wrappers +eventually complete. A hook must not simply release an open detached wrapper, +because that would leave its profiling interval without a matching end. Iterating over the returned ring -------------------------------- diff --git a/parsec/class/list_item.h b/parsec/class/list_item.h index 30bbc23a4..b7ad48940 100644 --- a/parsec/class/list_item.h +++ b/parsec/class/list_item.h @@ -185,8 +185,11 @@ parsec_list_item_ring_merge( parsec_list_item_t* ring1, * Removes an item from a ring of items. * * @details - * item must belong to a ring. It is singletoned, and the ring without - * item is returned. + * item must belong to a ring. Its neighbors are reconnected and the ring + * without item is returned, but item's own list_next and list_prev are not + * made consistent with that ring. In release builds they retain their old, + * now broken linkage; paranoid builds poison them instead. The caller must + * reinitialize item before using its list links again. * @param[inout] item the item from the ring of items to be removed. * @return the rest of the ring * @remark diff --git a/parsec/include/parsec/execution_stream.h b/parsec/include/parsec/execution_stream.h index b6ef3770d..53b347af3 100644 --- a/parsec/include/parsec/execution_stream.h +++ b/parsec/include/parsec/execution_stream.h @@ -48,8 +48,12 @@ struct parsec_execution_stream_s { void *scheduler_object; - /* The task to be executed next by this execution_stream. Beware as this bypasses - * the scheduler decision. + /* One ready task reserved for this execution stream. It bypasses the + * scheduler and is invisible to other workers until consumed or flushed. + * Once code commits this stream to long-lived work, it must call + * __parsec_schedule_flush_private() before entering that work. A path that + * merely hands work to an existing manager and returns should retain the + * private task for this stream to execute next. */ struct parsec_task_s* next_task; diff --git a/parsec/interfaces/ptg/ptg-compiler/jdf2c.c b/parsec/interfaces/ptg/ptg-compiler/jdf2c.c index f63eb0e8e..70371a6ba 100644 --- a/parsec/interfaces/ptg/ptg-compiler/jdf2c.c +++ b/parsec/interfaces/ptg/ptg-compiler/jdf2c.c @@ -6748,22 +6748,6 @@ static void jdf_generate_code_hook_gpu(const jdf_t *jdf, jdf_generate_code_dry_run_before(jdf, f); jdf_coutput_prettycomment('-', "%s BODY", f->fname); - if( profile_on ) { - coutput("#if defined(PARSEC_PROF_TRACE)\n" - " if(gpu_stream->prof_event_track_enable) {\n" - " PARSEC_TASK_PROF_TRACE(gpu_stream->profiling,\n" - " PARSEC_PROF_FUNC_KEY_START(this_task->taskpool,\n" - " this_task->task_class->task_class_id),\n" - " (parsec_task_t*)this_task, 1);\n" - " gpu_task->prof_key_end = PARSEC_PROF_FUNC_KEY_END(this_task->taskpool,\n" - " this_task->task_class->task_class_id);\n" - " gpu_task->prof_event_id = this_task->task_class->key_functions->\n" - " key_hash(this_task->task_class->make_key(this_task->taskpool, ((parsec_task_t*)this_task)->locals), NULL);\n" - " gpu_task->prof_tp_id = this_task->taskpool->taskpool_id;\n" - " }\n" - "#endif /* PARSEC_PROF_TRACE */\n"); - } - if ( NULL != dyld ) { coutput(" /* Pointer to dynamic gpu function */\n" " {\n" @@ -6851,6 +6835,16 @@ static void jdf_generate_code_hook_gpu(const jdf_t *jdf, " gpu_task->task_type = PARSEC_GPU_TASK_TYPE_KERNEL;\n", dev_lower, jdf_basename, f->fname); + /* Execution profiling is emitted by the common GPU runtime after the + * submit hook has finalized a possible batch. Preserve the PTG body-level + * profiling property on the wrapper so that common path can honor it. + */ + coutput("#if defined(PARSEC_PROF_TRACE)\n" + " gpu_task->prof_exec_state = %s;\n" + "#endif\n", + profile_on ? "PARSEC_GPU_TASK_PROF_EXEC_PENDING" + : "PARSEC_GPU_TASK_PROF_EXEC_DISABLED"); + /* Set up stage in/out callbacks */ jdf_find_property(body->properties, "stage_in", &stage_in_property); coutput(" gpu_task->stage_in = %s;\n", (NULL == stage_in_property) ? "parsec_default_gpu_stage_in" diff --git a/parsec/mca/device/device_gpu.c b/parsec/mca/device/device_gpu.c index 67e7ca6c3..de0eebc07 100644 --- a/parsec/mca/device/device_gpu.c +++ b/parsec/mca/device/device_gpu.c @@ -172,6 +172,7 @@ static void parsec_device_release_gpu_task(parsec_gpu_task_t *gpu_task) static void parsec_device_task_t_constructor(parsec_gpu_task_t *gpu_task) { + gpu_task->priority = 0; gpu_task->task_type = PARSEC_GPU_TASK_TYPE_INVALID; /* need to be set later */ gpu_task->pushout = 0; gpu_task->last_status = 0; @@ -181,9 +182,10 @@ static void parsec_device_task_t_constructor(parsec_gpu_task_t *gpu_task) gpu_task->stage_out = NULL; gpu_task->release_device_task = NULL; #if defined(PARSEC_PROF_TRACE) - gpu_task->prof_key_end = 0; gpu_task->prof_event_id = 0; - gpu_task->prof_tp_id = 0; + gpu_task->prof_stage_key_end = -1; + gpu_task->prof_stage_object_id = 0; + gpu_task->prof_exec_state = PARSEC_GPU_TASK_PROF_EXEC_PENDING; #endif gpu_task->ec = NULL; gpu_task->last_data_check_epoch = UINT64_MAX; /* force at least one validation for the task */ @@ -2089,16 +2091,16 @@ parsec_device_data_stage_in( parsec_device_gpu_module_t* gpu_device, info.desc = (parsec_dc_t*)original; info.data_id = -1; } - gpu_task->prof_key_end = -1; + gpu_task->prof_stage_key_end = -1; if( PARSEC_GPU_TASK_TYPE_PREFETCH == gpu_task->task_type && (gpu_device->trackable_events & PARSEC_PROFILE_GPU_TRACK_PREFETCH) ) { - gpu_task->prof_key_end = parsec_gpu_prefetch_key_end; + gpu_task->prof_stage_key_end = parsec_gpu_prefetch_key_end; gpu_task->prof_event_id = (int64_t)gpu_elem->device_private; - gpu_task->prof_tp_id = gpu_device->super.device_index; + gpu_task->prof_stage_object_id = gpu_device->super.device_index; PARSEC_PROFILING_TRACE(gpu_stream->profiling, parsec_gpu_prefetch_key_start, gpu_task->prof_event_id, - gpu_task->prof_tp_id, + gpu_task->prof_stage_object_id, &info); } if(PARSEC_GPU_TASK_TYPE_PREFETCH != gpu_task->task_type && (gpu_device->trackable_events & PARSEC_PROFILE_GPU_TRACK_DATA_IN) ) { @@ -2164,19 +2166,6 @@ parsec_device_data_stage_in( parsec_device_gpu_module_t* gpu_device, return 1; /* positive returns have special meaning and are used for optimizations */ } -#if PARSEC_GPU_USE_PRIORITIES - -static inline parsec_list_item_t* parsec_device_push_task_ordered( parsec_list_t* list, - parsec_list_item_t* elem ) -{ - parsec_list_push_sorted(list, elem, parsec_execution_context_priority_comparator); - return elem; -} -#define PARSEC_PUSH_TASK parsec_device_push_task_ordered -#else -#define PARSEC_PUSH_TASK parsec_list_push_back -#endif - static inline int parsec_gpu_task_is_singleton(parsec_gpu_task_t *task) { @@ -2191,18 +2180,219 @@ parsec_gpu_task_is_singleton(parsec_gpu_task_t *task) return (item->list_next == item) && (item->list_prev == item); } +/* Merge a singleton or task ring into a stream using the stream's ordering + * policy. The generic list sorter needs an integer embedded in every list + * item, so refresh the wrapper priority snapshots before the merge instead of + * incorrectly applying parsec_task_t's priority offset to parsec_gpu_task_t. + */ +static inline void +parsec_gpu_stream_chain_pending(parsec_gpu_exec_stream_t *stream, + parsec_list_item_t *ring) +{ +#if PARSEC_GPU_USE_PRIORITIES + parsec_gpu_task_t *task = (parsec_gpu_task_t *)ring; + parsec_gpu_task_t *current = task; + + do { + assert(NULL != current->ec); + current->priority = current->ec->priority; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != task ); + + parsec_list_chain_sorted(stream->fifo_pending, ring, + offsetof(parsec_gpu_task_t, priority)); +#else + parsec_list_chain_back(stream->fifo_pending, ring); +#endif +} + static inline void parsec_gpu_stream_push_pending(parsec_gpu_exec_stream_t *stream, parsec_gpu_task_t *task) { - /* A completed batched kernel returns a proper task ring. Preserve that - * order when feeding the tasks to the next stream. + /* Singleton and batched tasks must obey the same stream policy; otherwise + * appending a successful batch can invalidate a priority-sorted FIFO. */ - if( !parsec_gpu_task_is_singleton(task) ) { - parsec_list_chain_back(stream->fifo_pending, &task->list_item); + parsec_gpu_stream_chain_pending(stream, &task->list_item); +} + +/* NEXT applies to the selected head rather than its tentatively collected + * followers. Merge those followers back under one lock using the stream's + * insertion policy, then leave the head as a singleton for normal NEXT handling. + */ +static inline void +parsec_gpu_stream_rollback_batch(parsec_gpu_exec_stream_t *stream, + parsec_gpu_task_t *batch_head) +{ + parsec_list_item_t *ring; + + ring = parsec_list_item_ring_chop(&batch_head->list_item); + PARSEC_LIST_ITEM_SINGLETON(&batch_head->list_item); + if( NULL != ring ) { + parsec_gpu_stream_chain_pending(stream, ring); + } +} + +#if defined(PARSEC_PROF_TRACE) +/* Return whether any member owns an open logical GPU execution interval. The + * state lives on each wrapper so it follows followers that are manually + * detached and returned to runtime scheduling. + */ +static inline int +parsec_gpu_profile_exec_ring_is_open(parsec_gpu_task_t *ring) +{ + _LIST_ITEM_ITERATOR(&ring->list_item, &ring->list_item, item, { + parsec_gpu_task_t *task = (parsec_gpu_task_t *)item; + + if( PARSEC_GPU_TASK_PROF_EXEC_OPEN == task->prof_exec_state ) { + return 1; + } + }); + return 0; +} + +/* Start each logical task at its first committed GPU submission. The user hook + * has returned at this point, so the runtime can see the complete finalized + * ring. Members already opened by an earlier AGAIN continuation are skipped. + */ +static inline void +parsec_gpu_profile_exec_ring_start(parsec_gpu_exec_stream_t *stream, + parsec_gpu_task_t *ring) +{ + parsec_gpu_task_t *task; + parsec_task_t *ec; + const parsec_task_class_t *tc; + int key_start, key_end; + + if( !stream->prof_event_track_enable || !parsec_profile_enabled ) { + return; + } + _LIST_ITEM_ITERATOR(&ring->list_item, &ring->list_item, item, { + task = (parsec_gpu_task_t *)item; + ec = task->ec; + tc = ec->task_class; + + if( (PARSEC_GPU_TASK_PROF_EXEC_PENDING == task->prof_exec_state) && + (NULL != ec->taskpool->profiling_array) ) { + key_start = PARSEC_PROF_FUNC_KEY_START(ec->taskpool, tc->task_class_id); + key_end = PARSEC_PROF_FUNC_KEY_END(ec->taskpool, tc->task_class_id); + if( (key_start >= 2) && (key_end >= 2) ) { + task->prof_event_id = tc->key_functions->key_hash( + tc->make_key(ec->taskpool, ec->locals), NULL); + PARSEC_PROFILING_TRACE_INFO_FN(stream->profiling, key_start, + task->prof_event_id, + ec->taskpool->taskpool_id, + tc->profile_info, (void *)ec); + task->prof_exec_state = PARSEC_GPU_TASK_PROF_EXEC_OPEN; + } + } + }); +} + +/* Close every open logical execution interval in a ring exactly once. This is + * called only for final completion or a terminal ownership transfer, never for + * an intermediate AGAIN event. + */ +static inline int +parsec_gpu_profile_exec_ring_end(parsec_gpu_exec_stream_t *stream, + parsec_gpu_task_t *ring) +{ + parsec_gpu_task_t *task = ring; + parsec_task_t *ec; + const parsec_task_class_t *tc; + int key_end; + int found = 0; + + do { + if( PARSEC_GPU_TASK_PROF_EXEC_OPEN == task->prof_exec_state ) { + found = 1; + if( stream->prof_event_track_enable ) { + ec = task->ec; + tc = ec->task_class; + key_end = PARSEC_PROF_FUNC_KEY_END(ec->taskpool, + tc->task_class_id); + PARSEC_PROFILING_TRACE(stream->profiling, key_end, + task->prof_event_id, + ec->taskpool->taskpool_id, NULL); + } + task->prof_exec_state = PARSEC_GPU_TASK_PROF_EXEC_PENDING; + } + task = (parsec_gpu_task_t *)task->list_item.list_next; + } while( task != ring ); + return found; +} + +/* Complete the profiling interval associated with one recorded stream event. + * Transfer stages use the legacy head-only key. Execution batches instead + * close every logical member, except when AGAIN retains the interval. + */ +static inline void +parsec_gpu_profile_event_complete(parsec_gpu_exec_stream_t *stream, + parsec_gpu_task_t *ring, + int keep_exec_open) +{ + if( keep_exec_open && parsec_gpu_profile_exec_ring_is_open(ring) ) { return; } - PARSEC_PUSH_TASK(stream->fifo_pending, &task->list_item); + if( !keep_exec_open && + parsec_gpu_profile_exec_ring_end(stream, ring) ) { + return; + } + if( stream->prof_event_track_enable && + (ring->prof_stage_key_end != -1) ) { + PARSEC_PROFILING_TRACE(stream->profiling, ring->prof_stage_key_end, + ring->prof_event_id, + ring->prof_stage_object_id, NULL); + } +} +#endif /* defined(PARSEC_PROF_TRACE) */ + +/* Release every wrapper in a task ring and return the number released. ASYNC + * transfers ownership of each underlying execution context to the submit hook, + * but the GPU engine remains responsible for returning all device wrappers to + * their allocator and removing each one from the manager's outstanding count. + */ +static int +parsec_gpu_task_ring_release(parsec_gpu_task_t *ring) +{ + int count = 0; + + while( NULL != ring ) { + parsec_gpu_task_t *task = ring; + + ring = (parsec_gpu_task_t *)parsec_list_item_ring_chop(&task->list_item); + PARSEC_LIST_ITEM_SINGLETON(&task->list_item); + task->release_device_task(task); + count++; + } + return count; +} + +/* A terminal submit error can be reported after the hook queued partial work. + * Record and wait for a stream event before cleaning the affected batch. The + * runtime currently treats this as fatal; draining the rest of the device is a + * separate recovery problem. + */ +static int +parsec_gpu_stream_quiesce_after_failure(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_exec_stream_t *stream) +{ + struct timespec delay = { .tv_sec = 0, .tv_nsec = 100 }; + int rc; + + rc = gpu_device->event_record(gpu_device, stream, stream->start); + if( PARSEC_SUCCESS != rc ) { + return PARSEC_HOOK_RETURN_ERROR; + } + + do { + rc = gpu_device->event_query(gpu_device, stream, stream->start); + if( 0 == rc ) { + nanosleep(&delay, NULL); + } + } while( 0 == rc ); + + return (1 == rc) ? PARSEC_SUCCESS : PARSEC_HOOK_RETURN_ERROR; } static inline int @@ -2242,6 +2432,21 @@ parsec_gpu_task_collect_batch(parsec_gpu_exec_stream_t *gpu_stream, assert(NULL != batch_head); assert(NULL != callback); + /* A batched AGAIN continuation already owns a committed ring. Let hooks + * call the collector on every coroutine resume without detaching followers + * or collecting unrelated pending tasks into the in-flight batch. + */ + if( !parsec_gpu_task_is_singleton(batch_head) ) { + parsec_gpu_task_t *current = + (parsec_gpu_task_t *)batch_head->list_item.list_next; + + while( current != batch_head ) { + nb_tasks++; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } + return nb_tasks; + } + parsec_list_item_singleton(&batch_head->list_item); head_task = batch_head->ec; @@ -2257,6 +2462,9 @@ parsec_gpu_task_collect_batch(parsec_gpu_exec_stream_t *gpu_stream, fifo_pending = gpu_stream->fifo_pending; assert(NULL != fifo_pending); + /* The collector deliberately has no scan bound. Submit-hook policy decides + * when enough work has been considered and returns STOP at that candidate. + */ parsec_list_lock(fifo_pending); for(item = (parsec_list_item_t *)fifo_pending->ghost_element.list_next; item != &fifo_pending->ghost_element; @@ -2269,14 +2477,18 @@ parsec_gpu_task_collect_batch(parsec_gpu_exec_stream_t *gpu_stream, continue; } rc = callback(candidate, batch_head, callback_data); - if( rc < 0 ) { - parsec_list_unlock(fifo_pending); - return rc; + if( PARSEC_GPU_TASK_BATCH_STOP == rc ) { + break; } - if( 0 == rc ) { + if( PARSEC_GPU_TASK_BATCH_ACCEPT == rc ) { (void)parsec_list_nolock_remove(fifo_pending, item); (void)parsec_list_item_ring_push(&batch_head->list_item, item); nb_tasks++; + continue; + } + if( PARSEC_GPU_TASK_BATCH_REJECT != rc ) { + parsec_list_unlock(fifo_pending); + return PARSEC_HOOK_RETURN_ERROR; } } parsec_list_unlock(fifo_pending); @@ -2343,7 +2555,7 @@ parsec_device_send_transfercomplete_cmd_to_device(parsec_data_copy_t *copy, gpu_task->ec->data[0].source_repo_entry = NULL; gpu_task->ec->data[0].source_repo = NULL; #if defined(PARSEC_PROF_TRACE) - gpu_task->prof_key_end = -1; /* D2D complete tasks are pure internal management, we do not trace them */ + gpu_task->prof_stage_key_end = -1; /* D2D complete tasks are pure internal management, we do not trace them */ #endif (void)current_dev; PARSEC_DEBUG_VERBOSE(3, parsec_gpu_output_stream, @@ -2625,16 +2837,16 @@ parsec_device_progress_stream( parsec_device_gpu_module_t* gpu_device, stream->end = (stream->end + 1) % stream->max_events; #if defined(PARSEC_PROF_TRACE) - if( stream->prof_event_track_enable ) { - if( task->prof_key_end != -1 ) { - PARSEC_PROFILING_TRACE(stream->profiling, task->prof_key_end, task->prof_event_id, task->prof_tp_id, NULL); - } - } + parsec_gpu_profile_event_complete( + stream, task, PARSEC_HOOK_RETURN_AGAIN == task->last_status); #endif /* (PARSEC_PROF_TRACE) */ if( PARSEC_HOOK_RETURN_AGAIN == task->last_status ) { - /* we can now reschedule the task on the same execution stream */ + /* AGAIN means the submit hook made progress on this exact task + * or batch. Keep the complete ring as continuation state and + * re-enter the same stage only after its event has completed. + */ PARSEC_DEBUG_VERBOSE(2, parsec_gpu_output_stream, - "GPU[%d:%s]: GPU task %p[%p] is ready to be rescheduled on the same GPU device and same stream", + "GPU[%d:%s]: GPU task ring %p[%p] is ready to continue on the same GPU device and same stream", gpu_device->super.device_index, gpu_device->super.name, (void*)task, (void*)task->ec); *out_task = NULL; goto schedule_task; @@ -2664,14 +2876,15 @@ parsec_device_progress_stream( parsec_device_gpu_module_t* gpu_device, assert( NULL == stream->tasks[stream->start] ); schedule_task: + /* New queue entries are singletons. An AGAIN continuation deliberately + * keeps the complete submitted ring together across event completions. + */ + assert(parsec_gpu_task_is_singleton(task) || + (PARSEC_HOOK_RETURN_AGAIN == task->last_status)); rc = progress_fct( gpu_device, es, task, stream ); if( 0 == rc && parsec_device_skip_empty_events ) { #if defined(PARSEC_PROF_TRACE) - if( stream->prof_event_track_enable ) { - if( task->prof_key_end != -1 ) { - PARSEC_PROFILING_TRACE(stream->profiling, task->prof_key_end, task->prof_event_id, task->prof_tp_id, NULL); - } - } + parsec_gpu_profile_event_complete(stream, task, 0); #endif /* If progress_fct added nothing on that stream, skip the GPU event. * Input stages with copies already queued on the input stream return a @@ -2685,15 +2898,39 @@ parsec_device_progress_stream( parsec_device_gpu_module_t* gpu_device, if( 0 > rc ) { if( PARSEC_HOOK_RETURN_AGAIN != rc ) { if( PARSEC_HOOK_RETURN_NEXT == rc ) { + /* NEXT applies only to the selected head. Restore tentative + * followers, then use the unchanged singleton NEXT path. + */ + parsec_gpu_stream_rollback_batch(stream, task); /* Don't reorder the push_back, we are running into physical constraints and need to delay * the resubmission of this task as much as possible, but without losing track of it * (aka. returning it to the upper level). */ parsec_gpu_stream_push_pending(stream, task); + } else if( PARSEC_HOOK_RETURN_ASYNC == rc ) { + /* A batch-aware hook transfers every execution context in the + * ring. The manager will release all device wrappers. + */ +#if defined(PARSEC_PROF_TRACE) + parsec_gpu_profile_exec_ring_end(stream, task); +#endif + *out_task = task; } else { - /* Something else is going on with this task, remove it from the stream queues - * and return it to the upper level for final decision on its fate. + /* ERROR and DISABLE are terminal. A hook may have queued work + * before failing, so quiesce the stream before returning the + * complete ring for cleanup. Unknown negative values are + * normalized to ERROR at this boundary. */ + rc = (PARSEC_HOOK_RETURN_DISABLE == rc) ? + PARSEC_HOOK_RETURN_DISABLE : PARSEC_HOOK_RETURN_ERROR; + task->last_status = rc; + if( PARSEC_SUCCESS != + parsec_gpu_stream_quiesce_after_failure(gpu_device, stream) ) { + rc = task->last_status = PARSEC_HOOK_RETURN_ERROR; + } +#if defined(PARSEC_PROF_TRACE) + parsec_gpu_profile_exec_ring_end(stream, task); +#endif *out_task = task; } return rc; @@ -2707,8 +2944,7 @@ parsec_device_progress_stream( parsec_device_gpu_module_t* gpu_device, * execution stream pending list (to be executed again). */ PARSEC_DEBUG_VERBOSE(10, parsec_gpu_output_stream, - "GPU[%d:%s]: GPU task %p has returned with ASYNC or AGAIN. Once the event " - "trigger the task will be handled accordingly", + "GPU[%d:%s]: GPU task ring %p returned AGAIN; continue it after the event completes", gpu_device->super.device_index, gpu_device->super.name, (void*)task); } task->last_status = rc; @@ -2858,16 +3094,16 @@ parsec_device_kernel_push( parsec_device_gpu_module_t *gpu_device, parsec_task_snprintf(tmp, MAX_TASK_STRLEN, this_task)); gpu_task->complete_stage = parsec_device_callback_complete_push; #if defined(PARSEC_PROF_TRACE) - gpu_task->prof_key_end = -1; /* We do not log that event as the completion of this task */ + gpu_task->prof_stage_key_end = -1; /* We do not log that event as the completion of this task */ #endif return input_stream_work; } /** - * @brief Prepare a task for execution on the GPU. Basically, does some upstream initialization, - * setup the profiling information and then calls directly into the task submission body. Upon - * return from the body handle the state machine of the task, taking care of the special cases - * such as AGAIN and ASYNC. + * @brief Prepare a task for execution on the GPU. Invoke the task submission + * body, then start execution profiling for every member of the finalized ring. + * Upon return from the body, handle the task state machine, including AGAIN + * continuations and ASYNC ownership transfer. * @returns An error if anything unexpected came out of the task submission body, otherwise */ static int @@ -2879,6 +3115,9 @@ parsec_device_kernel_exec( parsec_device_gpu_module_t *gpu_device, parsec_advance_task_function_t progress_fct = gpu_task->submit; parsec_task_t* this_task = gpu_task->ec; int rc; +#if defined(PARSEC_PROF_TRACE) + int continuing_batch; +#endif #if defined(PARSEC_DEBUG_NOISIER) char tmp[MAX_TASK_STRLEN]; @@ -2887,21 +3126,6 @@ parsec_device_kernel_exec( parsec_device_gpu_module_t *gpu_device, (parsec_task_t *) this_task), gpu_stream->name); #endif /* defined(PARSEC_DEBUG_NOISIER) */ (void)es; -#if defined(PARSEC_PROF_TRACE) - if (gpu_stream->prof_event_track_enable && - (0 == gpu_task->prof_key_end)) { - parsec_task_class_t* tc = (parsec_task_class_t*)this_task->task_class; - PARSEC_TASK_PROF_TRACE(gpu_stream->profiling, - PARSEC_PROF_FUNC_KEY_START(this_task->taskpool, - tc->task_class_id), - (parsec_task_t *) this_task, 1); - gpu_task->prof_key_end = PARSEC_PROF_FUNC_KEY_END(this_task->taskpool, tc->task_class_id); - gpu_task->prof_event_id = tc->key_functions->key_hash( - tc->make_key(this_task->taskpool, ((parsec_task_t *) this_task)->locals), NULL); - gpu_task->prof_tp_id = this_task->taskpool->taskpool_id; - } -#endif /* PARSEC_PROF_TRACE */ - #if defined(PARSEC_DEBUG_PARANOID) const parsec_flow_t *flow; for( uint i = 0; i < gpu_task->nb_flows /* this_task->task_class->nb_flows */; i++ ) { @@ -2915,14 +3139,32 @@ parsec_device_kernel_exec( parsec_device_gpu_module_t *gpu_device, } #endif /* defined(PARSEC_DEBUG_PARANOID) */ - /* The submit hook may turn gpu_task into a batch ring. Start from a clean - * singleton so stale list links left by release-mode list operations cannot - * be mistaken for a preexisting ring. + /* New submissions start from a clean singleton so stale release-mode list + * links cannot be mistaken for a batch. AGAIN is different: its ring is + * the submit hook's continuation state and must be passed back intact. */ - PARSEC_LIST_ITEM_SINGLETON(&gpu_task->list_item); +#if defined(PARSEC_PROF_TRACE) + /* A committed non-singleton AGAIN ring was fully accounted when it first + * yielded. Avoid rescanning every member on each coroutine progress step. + */ + continuing_batch = (PARSEC_HOOK_RETURN_AGAIN == gpu_task->last_status) && + !parsec_gpu_task_is_singleton(gpu_task); +#endif + if( PARSEC_HOOK_RETURN_AGAIN != gpu_task->last_status ) { + PARSEC_LIST_ITEM_SINGLETON(&gpu_task->list_item); + } (void)this_task; rc = progress_fct( gpu_device, gpu_task, gpu_stream ); +#if defined(PARSEC_PROF_TRACE) + if( !continuing_batch && + ((rc >= 0) || (PARSEC_HOOK_RETURN_AGAIN == rc)) ) { + /* The hook has finalized the submitted ring. Start every new logical + * member here so generated and user-defined hooks share one policy. + */ + parsec_gpu_profile_exec_ring_start(gpu_stream, gpu_task); + } +#endif gpu_task->last_status = rc; /* Empty-stage event skipping is only valid for input/output streams. * A non-negative kernel submit result means the execution stream needs an @@ -3120,16 +3362,16 @@ parsec_device_kernel_pop( parsec_device_gpu_module_t *gpu_device, info.desc = (parsec_dc_t*)original; info.data_id = -1; } - gpu_task->prof_key_end = parsec_gpu_moveout_key_end; - gpu_task->prof_tp_id = this_task->taskpool->taskpool_id; + gpu_task->prof_stage_key_end = parsec_gpu_moveout_key_end; + gpu_task->prof_stage_object_id = this_task->taskpool->taskpool_id; gpu_task->prof_event_id = this_task->task_class->key_functions->key_hash(this_task->task_class->make_key(this_task->taskpool, this_task->locals), NULL); PARSEC_PROFILING_TRACE(gpu_stream->profiling, parsec_gpu_moveout_key_start, gpu_task->prof_event_id, - gpu_task->prof_tp_id, + gpu_task->prof_stage_object_id, &info); } else { - gpu_task->prof_key_end = -1; + gpu_task->prof_stage_key_end = -1; } } #endif @@ -3364,6 +3606,24 @@ parsec_device_kernel_cleanout( parsec_device_gpu_module_t *gpu_device, return 0; } +/* Clean every kernel task in a terminally failed submit batch after its + * execution stream has been quiesced. Device-wide queue recovery is + * intentionally left to the failure path rather than hidden here. + */ +static void +parsec_device_kernel_cleanout_ring(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *ring) +{ + parsec_gpu_task_t *task = ring; + + do { + if( PARSEC_GPU_TASK_TYPE_KERNEL == task->task_type ) { + parsec_device_kernel_cleanout(gpu_device, task); + } + task = (parsec_gpu_task_t *)task->list_item.list_next; + } while( task != ring ); +} + /** * This version is based on 4 streams: one for transfers from the memory to * the GPU, 2 for kernel executions and one for transfers from the GPU into @@ -3378,9 +3638,11 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, void *_gpu_task ) { parsec_device_gpu_module_t* gpu_device = (parsec_device_gpu_module_t *)module; - int rc, exec_stream = 0; + int rc, exec_stream = 0, released_tasks; parsec_gpu_task_t *progress_task, *out_task_submit = NULL, *out_task_pop = NULL; parsec_gpu_task_t *gpu_task = (parsec_gpu_task_t*)_gpu_task; + parsec_gpu_task_t *failed_batch = NULL; + parsec_hook_return_t failure_status = PARSEC_HOOK_RETURN_DISABLE; #if defined(PARSEC_DEBUG_NOISIER) char tmp[MAX_TASK_STRLEN]; #endif @@ -3396,11 +3658,9 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, #endif /* defined(PARSEC_PROF_TRACE) */ /* Check the GPU status -- three kinds of values for rc: - * - rc < 0: somebody is doing a short atomic operation while there is no manager, - * so wait. - * - rc == 0: there is no manager, and at the exit of the while, this thread - * made rc go from 0 to 1, so it is the new manager of the GPU and - * needs to deal with gpu_task + * - rc < 0: somebody owns an exclusive no-manager transition, so wait. + * - rc == 0: there is no manager, and at the exit of the while this + * worker changed the mutex from 0 to 1 and became the manager. * - rc > 0: there is a manager, and at the exit of the while, this thread has * committed new work that the manager will need to do, but the work is * not in the queue yet. @@ -3422,6 +3682,18 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, parsec_fifo_push( &(gpu_device->pending), (parsec_list_item_t*)gpu_task ); return PARSEC_HOOK_RETURN_ASYNC; } + + /* Only the worker that changed mutex from 0 to 1 becomes the long-lived + * manager. Expose its private ready task before entering GPU progress; + * workers that merely enqueue behind this manager retain next_task. Other + * submitters may enqueue during the flush, but this manager has not started + * consuming the device queues yet. + */ + rc = __parsec_schedule_flush_private(es); + if( PARSEC_SUCCESS != rc ) { + return PARSEC_HOOK_RETURN_ERROR; + } + PARSEC_DEBUG_VERBOSE(5, parsec_gpu_output_stream, "GPU[%d:%s]: Entering GPU management", gpu_device->super.device_index, gpu_device->super.name); @@ -3448,8 +3720,12 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, parsec_device_kernel_push, gpu_task, &progress_task ); if( rc < 0 ) { /* In case of error progress_task is the task that raised it */ - if( PARSEC_HOOK_RETURN_ERROR == rc ) + if( PARSEC_HOOK_RETURN_ERROR == rc ) { + failure_status = PARSEC_HOOK_RETURN_ERROR; + failed_batch = progress_task; + progress_task = NULL; goto disable_gpu; + } /* We are in the early stages, and if there no room on the GPU for a task we need to * delay all retries for the same task for a little while. Meanwhile, put the task back * trigger a device flush, and keep executing tasks that have their data on the device. @@ -3481,8 +3757,12 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, parsec_device_kernel_exec, gpu_task, &progress_task ); if( rc < 0 ) { - if( (PARSEC_HOOK_RETURN_DISABLE == rc) || (PARSEC_HOOK_RETURN_ERROR == rc) ) + if( (PARSEC_HOOK_RETURN_DISABLE == rc) || (PARSEC_HOOK_RETURN_ERROR == rc) ) { + failure_status = rc; + failed_batch = progress_task; + progress_task = NULL; goto disable_gpu; + } if( PARSEC_HOOK_RETURN_ASYNC != rc ) { /* Reschedule the task. As the chore_id has been modified, another incarnation of the task will be executed. */ @@ -3519,8 +3799,12 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, parsec_device_kernel_pop, gpu_task, &progress_task ); if( rc < 0 ) { - if( (PARSEC_HOOK_RETURN_ERROR == rc) || (PARSEC_HOOK_RETURN_DISABLE == rc) ) + if( (PARSEC_HOOK_RETURN_ERROR == rc) || (PARSEC_HOOK_RETURN_DISABLE == rc) ) { + failure_status = rc; + failed_batch = progress_task; + progress_task = NULL; goto disable_gpu; + } } if( NULL != progress_task ) { /* We have a successfully completed task. However, it is not gpu_task, as @@ -3580,16 +3864,24 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, } parsec_device_kernel_epilog( gpu_device, gpu_task ); __parsec_complete_execution( es, gpu_task->ec ); + /* Completing a GPU task can reserve one newly enabled successor in this + * manager's private next_task slot. The manager does not return to normal + * task selection while GPU work remains, so make that successor stealable. + */ + rc = __parsec_schedule_flush_private(es); + assert(PARSEC_SUCCESS == rc); gpu_device->super.executed_tasks++; - remove_gpu_task: + remove_gpu_task: PARSEC_DEBUG_VERBOSE(10, parsec_gpu_output_stream, "GPU[%d:%s]: gpu_task %p freed", gpu_device->super.device_index, gpu_device->super.name, gpu_task); /* Release the GPU task */ - gpu_task->release_device_task(gpu_task); + released_tasks = parsec_gpu_task_ring_release(gpu_task); + assert(released_tasks > 0); - rc = parsec_atomic_fetch_dec_int32( &(gpu_device->mutex) ); - if( 1 == rc ) { /* I was the last one */ + rc = parsec_atomic_fetch_sub_int32(&(gpu_device->mutex), released_tasks); + assert(rc >= released_tasks); + if( released_tasks == rc ) { /* I released the last outstanding task(s) */ #if defined(PARSEC_PROF_TRACE) if( gpu_device->trackable_events & PARSEC_PROFILE_GPU_TRACK_OWN ) PARSEC_PROFILING_TRACE( es->es_profile, parsec_gpu_own_GPU_key_end, @@ -3604,10 +3896,19 @@ parsec_device_kernel_scheduler( parsec_device_module_t *module, goto fetch_task_from_shared_queue; disable_gpu: - /* Something wrong happened. Push all the pending tasks back on the - * cores, and disable the gpu. + /* The scheduler currently treats device failure as fatal. Clean the batch + * that observed the failure before propagating its terminal status. + */ + if( NULL != failed_batch ) { + parsec_device_kernel_cleanout_ring(gpu_device, failed_batch); + (void)parsec_gpu_task_ring_release(failed_batch); + } + /* TODO: Recover the tasks in every pending FIFO, recorded event slot, and + * gpu_device->pending before making DISABLE recoverable. The upper scheduler + * currently treats this return as fatal, so only the failed batch is + * quiesced and cleaned here. */ parsec_warning("GPU[%d:%s]: Critical issue related to the GPU discovered. Giving up", gpu_device->super.device_index, gpu_device->super.name); - return PARSEC_HOOK_RETURN_DISABLE; + return failure_status; } diff --git a/parsec/mca/device/device_gpu.h b/parsec/mca/device/device_gpu.h index b36a40718..36eff3a7b 100644 --- a/parsec/mca/device/device_gpu.h +++ b/parsec/mca/device/device_gpu.h @@ -50,14 +50,27 @@ typedef int (*parsec_advance_task_function_t)(parsec_device_gpu_module_t *gpu_d parsec_gpu_task_t *gpu_task, parsec_gpu_exec_stream_t *gpu_stream); +/* Actions returned by a GPU batch collector callback. Keep the callback's int + * return type and the existing ACCEPT/REJECT values, but reject values outside + * this enum so hook results and private sentinels cannot be mistaken for + * batching policy. + */ +typedef enum parsec_gpu_task_batch_action_e { + PARSEC_GPU_TASK_BATCH_ACCEPT = 0, + PARSEC_GPU_TASK_BATCH_REJECT = 1, + PARSEC_GPU_TASK_BATCH_STOP = 2 +} parsec_gpu_task_batch_action_t; + /* Callback used by parsec_gpu_task_collect_batch() to decide whether a - * pending task can be appended to the current batched task ring. + * pending task can be appended to the current batched task ring. It executes + * while gpu_stream->fifo_pending is locked and must therefore be short and + * nonblocking. It must not modify or acquire the same FIFO, call the collector + * recursively, or otherwise reenter pending-task operations on this stream. * - * Return values: - * < 0: stop iteration and return this error code to the caller. - * 0: extract candidate from the stream pending queue and append it to - * batch_head. - * > 0: leave candidate in the stream pending queue and continue. + * Return PARSEC_GPU_TASK_BATCH_ACCEPT to append the candidate, + * PARSEC_GPU_TASK_BATCH_REJECT to leave it pending and continue, or + * PARSEC_GPU_TASK_BATCH_STOP to end collection successfully. STOP leaves the + * current and all unvisited candidates pending. */ typedef int (*parsec_gpu_task_batch_cb_t)(parsec_gpu_task_t *candidate, parsec_gpu_task_t *batch_head, @@ -114,8 +127,21 @@ typedef struct parsec_gpu_flow_info_s { } parsec_gpu_flow_info_t; +#if defined(PARSEC_PROF_TRACE) +typedef enum parsec_gpu_task_prof_exec_state_e { + PARSEC_GPU_TASK_PROF_EXEC_DISABLED = 0, + PARSEC_GPU_TASK_PROF_EXEC_PENDING, + PARSEC_GPU_TASK_PROF_EXEC_OPEN +} parsec_gpu_task_prof_exec_state_t; +#endif + struct parsec_gpu_task_s { parsec_list_item_t list_item; + /* The stream queues sort wrappers directly and therefore cannot follow + * ec to read parsec_task_t::priority. Refresh this snapshot before each + * stream insertion. + */ + int32_t priority; uint16_t task_type; uint16_t pushout; int32_t last_status; @@ -125,9 +151,20 @@ struct parsec_gpu_task_s { parsec_stage_out_function_t *stage_out; parsec_release_device_task_function_t release_device_task; #if defined(PARSEC_PROF_TRACE) - int prof_key_end; + /* The event ID is shared by logical execution and deferred stage events. + * Keep this 64-bit member before the two 32-bit stage fields. + */ uint64_t prof_event_id; - uint32_t prof_tp_id; + /* Key and object ID saved for a deferred stage end. Prefetch events use the + * GPU device index, while move-out events use the taskpool ID. Logical + * execution derives both values directly from ec instead. + */ + int prof_stage_key_end; + uint32_t prof_stage_object_id; + /* Combine the body policy with whether an execution interval remains open + * across an AGAIN continuation. + */ + parsec_gpu_task_prof_exec_state_t prof_exec_state; #endif union { struct { @@ -340,10 +377,15 @@ int parsec_gpu_complete_w2r_task(parsec_device_gpu_module_t *gpu_device, parsec_ /** * Iterate over gpu_stream->fifo_pending and append accepted tasks to * batch_head. The callback receives each pending candidate, the task passed to - * the submit function, and user data. The callback should return 0 to append - * the candidate to batch_head's ring, a positive value to leave it pending, or - * a negative error code to stop the iteration. - * The callback must not modify gpu_stream->fifo_pending directly. + * the submit function, and user data. The scan is intentionally unbounded: the + * callback is responsible for returning PARSEC_GPU_TASK_BATCH_STOP when its + * policy has accepted enough work. STOP ends collection successfully and + * leaves the current and all unvisited candidates pending. + * + * The callback executes while gpu_stream->fifo_pending is locked. It must be + * short and nonblocking, and must not modify or acquire the same FIFO, call + * this collector recursively, or otherwise reenter pending-task operations on + * this stream. * If batching is disabled, unsupported by the head task's selected device, or * not enabled on batch_head's selected incarnation, no iteration is performed * and batch_head remains a singleton. Pending tasks whose selected incarnation @@ -351,9 +393,36 @@ int parsec_gpu_complete_w2r_task(parsec_device_gpu_module_t *gpu_device, parsec_ * calling the callback. * * Returns the number of additional tasks appended to batch_head's ring on - * success, or the negative callback error code. If an error is returned, tasks - * already accepted remain attached to batch_head and the remaining candidates - * stay in fifo_pending. + * success. Any callback result other than PARSEC_GPU_TASK_BATCH_ACCEPT, + * PARSEC_GPU_TASK_BATCH_REJECT, or PARSEC_GPU_TASK_BATCH_STOP is normalized to + * PARSEC_HOOK_RETURN_ERROR. If an error is returned, tasks already accepted + * remain attached to batch_head and the remaining candidates stay in + * fifo_pending. + * + * On an initial singleton submission, the collector scans fifo_pending and + * builds a tentative ring. On a batched PARSEC_HOOK_RETURN_AGAIN continuation, + * batch_head already owns the committed ring; the collector leaves it intact, + * does not scan fifo_pending, and returns its existing follower count. + * + * A submit result applies to the complete ring. AGAIN records an event and + * re-enters the submit hook with the same ring after the event completes; it + * therefore means progress was submitted for every member. NEXT restores the + * followers before applying the normal singleton NEXT handling to the head. + * ASYNC transfers every execution context in the ring to the submit hook; the + * hook must eventually complete or reschedule each one, while the GPU engine + * releases their device wrappers. ERROR and DISABLE quiesce the affected + * execution stream and clean the complete batch before the terminal status is + * propagated. Device-wide recovery after DISABLE is not currently supported. + * + * The runtime does not disband an AGAIN ring. A hook that wants to split it + * must detach the followers itself, restore valid singleton/ring linkage, and + * explicitly transfer every detached wrapper to a new owner, such as the + * stream pending FIFO using its configured priority policy. Returning AGAIN + * retains only the ring or singleton that remains attached to batch_head. + * Each wrapper also retains its open execution-profiling state when detached; + * returning it through the normal GPU path closes that interval at final + * completion. Releasing an open detached wrapper would leave an unmatched + * profiling start. */ int parsec_gpu_task_collect_batch(parsec_gpu_exec_stream_t *gpu_stream, parsec_gpu_task_t *batch_head, diff --git a/parsec/mca/sched/llp/sched_llp_module.c b/parsec/mca/sched/llp/sched_llp_module.c index 5c0f5e259..9371731af 100644 --- a/parsec/mca/sched/llp/sched_llp_module.c +++ b/parsec/mca/sched/llp/sched_llp_module.c @@ -340,6 +340,7 @@ parsec_list_item_t* lifo_merge_ring(parsec_list_item_t *next_in_lifo, /* insert a single element */ parsec_list_item_t* item = ring; ring = parsec_list_item_ring_chop(ring); + PARSEC_LIST_ITEM_SINGLETON(item); if (NULL != prev) { item->list_next = prev->list_next; prev->list_next = item; diff --git a/parsec/scheduling.c b/parsec/scheduling.c index 1d2625e16..1626d1efa 100644 --- a/parsec/scheduling.c +++ b/parsec/scheduling.c @@ -345,11 +345,18 @@ __parsec_schedule(parsec_execution_stream_t* es, /* * Schedule an array of rings of tasks with one entry per virtual process. - * If an execution stream is provided, this function will save the highest - * priority task (assuming the ring is ordered or the first task in the ring - * otherwise) on the current execution stream virtual process as the next - * task to be executed on the provided execution stream. Everything else gets - * pushed into the execution stream 0 of the corresponding virtual process. + * If an execution stream is provided, this function removes the highest + * priority task (assuming the ring is ordered, or the first task otherwise) + * from the local VP ring and stores it in submission_es->next_task. This + * one-element private queue is consumed before the scheduler is queried, so + * the task stays local and avoids a scheduler round trip. It is also invisible + * to every other execution stream until submission_es consumes or explicitly + * flushes it. Once code commits the stream to blocking or long-lived progress + * work, such as GPU management, it must flush this private slot to avoid + * delaying ready work. A path that hands work to an existing manager and + * returns to normal selection should retain the private task. + * Everything else gets pushed into the execution stream 0 of the corresponding + * virtual process. * If the provided execution stream is NULL, all tasks are delivered to their * respective vp. * @@ -397,8 +404,17 @@ int __parsec_schedule_vp(parsec_execution_stream_t* submission_es, if( vp == submission_es->virtual_process->vp_id ) { if( NULL == submission_es->next_task ) { + /* Reserve one local ready task outside the scheduler. The + * execution stream will consume it before scheduler selection, + * or expose it with __parsec_schedule_flush_private() before + * entering work that delays normal task selection. + */ submission_es->next_task = ring; ring = (parsec_task_t*)parsec_list_item_ring_chop(&ring->super); + /* next_task may be flushed through __parsec_schedule(), which + * requires even a single task to be a valid ring. + */ + PARSEC_LIST_ITEM_SINGLETON(submission_es->next_task); if( NULL == ring ) { task_rings[vp] = NULL; /* remove the tasks already scheduled */ continue; @@ -416,6 +432,10 @@ int __parsec_schedule_vp(parsec_execution_stream_t* submission_es, return ret; } +/* Move the task reserved in es->next_task back to the scheduler. The slot is + * private to es, and __parsec_schedule_vp() keeps its task as a singleton ring + * so it can be passed directly to __parsec_schedule(). + */ int __parsec_schedule_flush_private( parsec_execution_stream_t* es ) { parsec_task_t* task = es->next_task; diff --git a/tests/dsl/dtd/Testings.cmake b/tests/dsl/dtd/Testings.cmake index e24d4c54f..ee50fe3ef 100644 --- a/tests/dsl/dtd/Testings.cmake +++ b/tests/dsl/dtd/Testings.cmake @@ -25,6 +25,13 @@ parsec_addtest_cmd(dsl/dtd/untie ${SHM_TEST_CMD_LIST} dsl/dtd/dtd_test_untie) parsec_addtest_cmd(dsl/dtd/new_tile:cpu ${SHM_TEST_CMD_LIST} dsl/dtd/dtd_test_new_tile --mca device_cuda_enabled 0) if(PARSEC_HAVE_CUDA AND CMAKE_CUDA_COMPILER) parsec_addtest_cmd(dsl/dtd/new_tile:gpu ${SHM_TEST_CMD_LIST} ${CTEST_CUDA_LAUNCHER_OPTIONS} dsl/dtd/dtd_test_new_tile --mca device_cuda_enabled 1 --mca device cuda) + parsec_addtest_cmd(dsl/dtd/cuda_batch_status ${SHM_TEST_CMD_LIST} + ${CTEST_CUDA_LAUNCHER_OPTIONS} + dsl/dtd/dtd_test_cuda_again_async + --mca device_cuda_enabled 1 + --mca device_cuda_mask 1 + --mca device_enable_batching 1 + --mca device cuda) endif(PARSEC_HAVE_CUDA AND CMAKE_CUDA_COMPILER) if(PARSEC_HAVE_DEV_CAPABILITY_BATCH) parsec_addtest_cmd(dsl/dtd/batch_cpu ${SHM_TEST_CMD_LIST} dsl/dtd/dtd_test_batch_cpu) diff --git a/tests/dsl/dtd/dtd_test_cuda_again_async.c b/tests/dsl/dtd/dtd_test_cuda_again_async.c index a3a4e6c19..0ec4b5801 100644 --- a/tests/dsl/dtd/dtd_test_cuda_again_async.c +++ b/tests/dsl/dtd/dtd_test_cuda_again_async.c @@ -2,6 +2,7 @@ * Copyright (c) 2023 The University of Tennessee and The University * of Tennessee Research Foundation. All rights * reserved. + * Copyright (c) 2026 NVIDIA Corporation. All rights reserved. */ #include "parsec.h" @@ -11,115 +12,450 @@ #include "parsec/interfaces/dtd/insert_function_internal.h" #include "tests/tests_data.h" +#include + #if defined(PARSEC_HAVE_MPI) #include #endif /* defined(PARSEC_HAVE_MPI) */ -void parsec_dtd_pack_args( parsec_task_t *this_task, ... ) +/* Update this test's private PARSEC_VALUE arguments in place to model a + * coroutine continuation. DTD parameter descriptors are a counted array, not + * NULL-terminated, so never scan beyond the task class's declared parameters. + */ +static void +cuda_pack_value_args(parsec_task_t *this_task, ...) { parsec_dtd_task_t *current_task = (parsec_dtd_task_t *)this_task; + parsec_dtd_task_class_t *tc = + (parsec_dtd_task_class_t *)current_task->super.task_class; parsec_dtd_task_param_t *current_param = GET_HEAD_OF_PARAM_LIST(current_task); - int i = 0; void *tmp_val; - void **tmp_ref; va_list arguments; va_start(arguments, this_task); - while( current_param != NULL) { - if((current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_VALUE ) { - tmp_val = va_arg(arguments, void*); - memcpy(current_param->pointer_to_tile, tmp_val, current_param->arg_size); - } else if((current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_SCRATCH || - (current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_REF ) { - tmp_ref = va_arg(arguments, void**); - current_param->pointer_to_tile = *tmp_ref; - } else if((current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_INPUT || - (current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_INOUT || - (current_param->op_type & PARSEC_GET_OP_TYPE) == PARSEC_OUTPUT ) { - tmp_ref = va_arg(arguments, void**); - this_task->data[i].data_in = *tmp_ref; - assert(0); - i++; - } else { - parsec_warning("/!\\ Flag is not recognized in parsec_dtd_unpack_args /!\\.\n"); - assert(0); - } - current_param = current_param + 1; + for( int i = 0; i < tc->count_of_params; i++, current_param++ ) { + assert(PARSEC_VALUE == + (current_param->op_type & PARSEC_GET_OP_TYPE)); + tmp_val = va_arg(arguments, void *); + memcpy(current_param->pointer_to_tile, tmp_val, + current_param->arg_size); } va_end(arguments); } -static int max_repeat = 50; -static parsec_task_t* array_of_async_tasks[100]; +#define STATUS_TASKS 32 + +/* ASYNC transfers execution-context ownership to the submit hook. Controller + * tasks retrieve the contexts from this table and reschedule each one once; + * the completion arrays detect both lost and duplicate logical tasks. + */ +static parsec_task_t *array_of_async_tasks[STATUS_TASKS]; +static parsec_gpu_task_t *again_batch_tasks[STATUS_TASKS]; +static parsec_gpu_task_t *again_batch_order[STATUS_TASKS]; +static int async_completed[STATUS_TASKS]; +static int again_batch_completed[STATUS_TASKS]; +static int next_completed[STATUS_TASKS]; +static int async_batch_observed; +static int again_batch_observed; +static int again_batch_yields; +static int again_batch_release_count; +static int again_profile_state_valid = 1; +static int next_batch_observed; +static int next_returned; +static int priority_batch_observed; +static int priority_release_count; +static int priority_order_valid = 1; +static int priority_last_released = INT_MAX; +static parsec_gpu_task_t *stop_candidate; +static int stop_observed; +static int stop_candidate_completed; + +/* Verify that final event completion closes the logical execution interval + * before the device wrapper is released. + */ +static void +cuda_again_release(parsec_gpu_task_t *gpu_task) +{ +#if defined(PARSEC_PROF_TRACE) + if( PARSEC_GPU_TASK_PROF_EXEC_OPEN == gpu_task->prof_exec_state ) { + again_profile_state_valid = 0; + } +#endif + again_batch_release_count++; + PARSEC_OBJ_RELEASE(gpu_task); +} + +/* Only collect tasks from the same test phase and task class. Phase matching + * prevents resubmitted ASYNC contexts from joining their initial submission. + */ +static int +cuda_batch_match_same_phase(parsec_gpu_task_t *candidate, + parsec_gpu_task_t *batch_head, + void *callback_data) +{ + int candidate_id, candidate_phase; + int head_id, head_phase; + + (void)callback_data; + if( candidate->ec->task_class != batch_head->ec->task_class ) { + return PARSEC_GPU_TASK_BATCH_REJECT; + } + parsec_dtd_unpack_args(candidate->ec, &candidate_id, &candidate_phase); + parsec_dtd_unpack_args(batch_head->ec, &head_id, &head_phase); + (void)candidate_id; + (void)head_id; + return (candidate_phase == head_phase) ? PARSEC_GPU_TASK_BATCH_ACCEPT + : PARSEC_GPU_TASK_BATCH_REJECT; +} /** - * This test check the correct handling of the PARSEC_HOOK_RETURN_ASYNC and - * PARSEC_HOOK_RETURN_AGAIN. The cuda_task_async will atomically save the task - * onto a predefined array, and the cuda_task_again will repeat itself until - * the async task appears on the array. At that point it re-enables the async task - * and continue it's execution until a predefined number of iteration have been - * reached. If more iterations have been already done it will return asap. + * Exercise batched AGAIN, ASYNC, and NEXT returns, successful-ring priority + * ordering, and early collection STOP. AGAIN verifies that repeated + * coroutine-style yields preserve the exact ring. ASYNC hands every execution + * context to controller tasks, which reschedule them once. NEXT verifies that + * all restored followers complete exactly once. PRIORITY verifies ordering in + * the next GPU stream after a deliberately unsorted batch succeeds. STOP + * verifies that its current candidate remains pending and later executes. */ int cuda_task_async(parsec_device_gpu_module_t *gpu_device, parsec_gpu_task_t *gpu_task, parsec_gpu_exec_stream_t *gpu_stream) { - parsec_task_t *this_task = gpu_task->ec; - int i, first; + parsec_gpu_task_t *current; + int batch_count, i, first; - (void)gpu_device; (void)gpu_stream; - parsec_dtd_unpack_args(this_task, &i, &first); + (void)gpu_device; + batch_count = 1 + parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_match_same_phase, + NULL); + parsec_dtd_unpack_args(gpu_task->ec, &i, &first); if( 0 == first ) { - first += 1; /* mark the second call to this task */ - parsec_dtd_pack_args(this_task, &i, &first); - PARSEC_LIST_ITEM_SINGLETON(this_task); - fprintf(stdout, "Task %p preparing for async behavior\n", this_task); - parsec_atomic_cas_ptr(&array_of_async_tasks[i], NULL, this_task); + /* Fill the execution stream before handing tasks out asynchronously so + * at least one invocation exercises ASYNC with a real task ring. + */ + if( (1 == batch_count) && !async_batch_observed ) { + return PARSEC_HOOK_RETURN_AGAIN; + } + if( batch_count > 1 ) { + async_batch_observed = 1; + } + current = gpu_task; + do { + parsec_task_t *this_task = current->ec; + int stored; + + parsec_dtd_unpack_args(this_task, &i, &first); + assert(0 == first); + first = 1; + cuda_pack_value_args(this_task, &i, &first); + PARSEC_LIST_ITEM_SINGLETON(this_task); + assert(i >= 0 && i < STATUS_TASKS); + stored = parsec_atomic_cas_ptr(&array_of_async_tasks[i], NULL, this_task); + assert(stored); + if( !stored ) { + return PARSEC_HOOK_RETURN_ERROR; + } + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); return PARSEC_HOOK_RETURN_ASYNC; } - fprintf(stdout, "Task %p is back alive after an async. Complete and leave\n", this_task); + + current = gpu_task; + do { + parsec_dtd_unpack_args(current->ec, &i, &first); + assert(1 == first); + assert(i >= 0 && i < STATUS_TASKS); + assert(0 == async_completed[i]); + async_completed[i] = 1; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); return PARSEC_HOOK_RETURN_DONE; } +/* Wait until an ASYNC hook publishes the matching execution context, claim it + * atomically, and return it to normal runtime scheduling exactly once. + */ int cuda_task_again(parsec_device_gpu_module_t *gpu_device, parsec_gpu_task_t *gpu_task, parsec_gpu_exec_stream_t *gpu_stream) { parsec_task_t *this_task = gpu_task->ec; - int i, repeat, done; + parsec_task_t *async_task; + int i; (void)gpu_device; (void)gpu_stream; - parsec_dtd_unpack_args(this_task, &i, &repeat, &done); - if( !done ) { - repeat += 1; - if( NULL == array_of_async_tasks[i] ) { - fprintf(stdout, "Task [%d] %p waiting for the async task (repeat %d)\n", i, this_task, repeat - 1); - } else { - /* There is a small opportunity for race conditions between the insertion of the async task and - it's reschedule by the again task. - */ - sleep(1); - parsec_task_t* async_task = array_of_async_tasks[i]; - fprintf(stdout, "Async Task %p is reinserted into the runtime\n", async_task); - array_of_async_tasks[i] = NULL; - parsec_execution_stream_t* local_es = parsec_my_execution_stream(); - __parsec_reschedule(local_es, async_task); - done = 1; + parsec_dtd_unpack_args(this_task, &i); + assert(i >= 0 && i < STATUS_TASKS); + async_task = array_of_async_tasks[i]; + if( (NULL == async_task) || + !parsec_atomic_cas_ptr(&array_of_async_tasks[i], async_task, NULL) ) { + return PARSEC_HOOK_RETURN_AGAIN; + } + __parsec_reschedule(parsec_my_execution_stream(), async_task); + return PARSEC_HOOK_RETURN_DONE; +} + +/* Model a coroutine batch that submits progress twice before completing. The + * first batched AGAIN commits the ring; each continuation must receive exactly + * the same wrappers, while later phase-zero tasks remain independent work. + */ +int cuda_task_batch_again(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *gpu_task, + parsec_gpu_exec_stream_t *gpu_stream) +{ + parsec_gpu_task_t *current; + uint32_t seen = 0; + int batch_count, count = 0, id, phase; + + (void)gpu_device; + parsec_dtd_unpack_args(gpu_task->ec, &id, &phase); + if( 0 == phase ) { + if( again_batch_observed ) { + if( (id < 0) || (id >= STATUS_TASKS) || + (0 != again_batch_completed[id]) ) { + return PARSEC_HOOK_RETURN_ERROR; + } + again_batch_completed[id] = 1; + return PARSEC_HOOK_RETURN_DONE; + } + + batch_count = 1 + parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_match_same_phase, + NULL); + if( 1 == batch_count ) { + /* No work was collected, so this remains a normal singleton retry. */ + return PARSEC_HOOK_RETURN_AGAIN; + } + + again_batch_observed = batch_count; + current = gpu_task; + do { + parsec_dtd_unpack_args(current->ec, &id, &phase); + if( (id < 0) || (id >= STATUS_TASKS) || (0 != phase) || + (NULL != again_batch_tasks[id]) || (seen & (1U << id)) ) { + return PARSEC_HOOK_RETURN_ERROR; + } + seen |= (1U << id); + again_batch_tasks[id] = current; + again_batch_order[count] = current; + current->release_device_task = cuda_again_release; + phase = 1; + cuda_pack_value_args(current->ec, &id, &phase); + count++; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); + if( count != batch_count ) { + return PARSEC_HOOK_RETURN_ERROR; } - parsec_dtd_pack_args(this_task, &i, &repeat, &done); + again_batch_yields++; return PARSEC_HOOK_RETURN_AGAIN; } - if(repeat < max_repeat) { - fprintf(stdout, "Task [%d] %p is cycling while waiting for repeat (%d)\n", i, this_task, repeat); - repeat += 1; - parsec_dtd_pack_args(this_task, &i, &repeat, &done); + + /* Verify ring identity before calling the collector, so reconstructing a + * batch from followers returned to fifo_pending cannot satisfy the test. + */ + current = gpu_task; + do { + int member_phase; + + parsec_dtd_unpack_args(current->ec, &id, &member_phase); + if( (count >= STATUS_TASKS) || (id < 0) || (id >= STATUS_TASKS) || + (member_phase != phase) || + (again_batch_tasks[id] != current) || + (again_batch_order[count] != current) || (seen & (1U << id)) ) { + return PARSEC_HOOK_RETURN_ERROR; + } +#if defined(PARSEC_PROF_TRACE) + /* Every member starts once after the first finalized submission. The + * interval must remain open on each coroutine continuation. + */ + if( gpu_stream->prof_event_track_enable && parsec_profile_enabled && + (PARSEC_GPU_TASK_PROF_EXEC_OPEN != current->prof_exec_state) ) { + return PARSEC_HOOK_RETURN_ERROR; + } +#endif + seen |= (1U << id); + count++; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); + if( count != again_batch_observed ) { + return PARSEC_HOOK_RETURN_ERROR; + } + + /* Calling the collector on a continuation must report the existing ring + * without changing it or adding pending phase-zero tasks. + */ + batch_count = 1 + parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_match_same_phase, + NULL); + if( batch_count != again_batch_observed ) { + return PARSEC_HOOK_RETURN_ERROR; + } + + if( phase < 2 ) { + current = gpu_task; + do { + parsec_dtd_unpack_args(current->ec, &id, &phase); + phase++; + cuda_pack_value_args(current->ec, &id, &phase); + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); + again_batch_yields++; return PARSEC_HOOK_RETURN_AGAIN; } - fprintf(stdout, "Task [%d] %p is now officially complete\n", i, this_task); + + current = gpu_task; + do { + parsec_dtd_unpack_args(current->ec, &id, &phase); + if( (id < 0) || (id >= STATUS_TASKS) || (2 != phase) || + (0 != again_batch_completed[id]) ) { + return PARSEC_HOOK_RETURN_ERROR; + } + again_batch_completed[id] = 1; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); + return PARSEC_HOOK_RETURN_DONE; +} + +/* Force one multi-task NEXT result, then verify that the restored followers + * and delayed singleton head all return and complete exactly once. + */ +int cuda_task_next(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *gpu_task, + parsec_gpu_exec_stream_t *gpu_stream) +{ + parsec_gpu_task_t *current; + int batch_count, i, phase; + + (void)gpu_device; + batch_count = 1 + parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_match_same_phase, + NULL); + if( !next_returned ) { + /* AGAIN fills the stream until NEXT can be returned with followers. + * NEXT must restore those followers and requeue only the head. + */ + if( 1 == batch_count ) { + return PARSEC_HOOK_RETURN_AGAIN; + } + next_batch_observed = batch_count; + next_returned = 1; + return PARSEC_HOOK_RETURN_NEXT; + } + + current = gpu_task; + do { + parsec_dtd_unpack_args(current->ec, &i, &phase); + assert(0 == phase); + assert(i >= 0 && i < STATUS_TASKS); + assert(0 == next_completed[i]); + next_completed[i] = 1; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); return PARSEC_HOOK_RETURN_DONE; } +/* Record only wrappers from the deliberately unsorted priority batch. Other + * tasks may complete between them, but the marked wrappers must be released + * in non-increasing priority order after the ring enters the next stream. + */ +static void +cuda_priority_release(parsec_gpu_task_t *gpu_task) +{ + if( gpu_task->priority > priority_last_released ) { + priority_order_valid = 0; + } + priority_last_released = gpu_task->priority; + priority_release_count++; + PARSEC_OBJ_RELEASE(gpu_task); +} + +/* Build one successful batch, then make its linked order intentionally differ + * from its priority order. The next stream must sort the complete ring instead + * of appending it as-is. + */ +int cuda_task_priority(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *gpu_task, + parsec_gpu_exec_stream_t *gpu_stream) +{ + parsec_gpu_task_t *current; + int batch_count, position = 0; + + (void)gpu_device; + if( priority_batch_observed ) { + /* Leave later tasks as singletons so the next stream must merge a + * successful batch with independently submitted work. + */ + return PARSEC_HOOK_RETURN_DONE; + } + batch_count = 1 + parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_match_same_phase, + NULL); + if( 1 == batch_count ) { + return PARSEC_HOOK_RETURN_AGAIN; + } + priority_batch_observed = batch_count; + current = gpu_task; + do { + /* Put the lowest-priority task at the ring head and descending + * positive priorities behind it, guaranteeing an unsorted ring. + */ + current->ec->priority = (0 == position) ? 0 : batch_count - position + 1; + current->release_device_task = cuda_priority_release; + position++; + current = (parsec_gpu_task_t *)current->list_item.list_next; + } while( current != gpu_task ); + return PARSEC_HOOK_RETURN_DONE; +} + +/* Stop at the first compatible follower without accepting it. The collector + * must report success and leave this exact wrapper in the pending FIFO. + */ +static int +cuda_batch_stop_first(parsec_gpu_task_t *candidate, + parsec_gpu_task_t *batch_head, + void *callback_data) +{ + (void)callback_data; + if( candidate->ec->task_class != batch_head->ec->task_class ) { + return PARSEC_GPU_TASK_BATCH_REJECT; + } + stop_candidate = candidate; + stop_observed++; + return PARSEC_GPU_TASK_BATCH_STOP; +} + +/* Retry until a compatible pending task lets the callback exercise STOP. Once + * STOP succeeds, every task completes as a singleton; observing the saved + * wrapper later proves that STOP did not remove or lose its current candidate. + */ +int cuda_task_batch_stop(parsec_device_gpu_module_t *gpu_device, + parsec_gpu_task_t *gpu_task, + parsec_gpu_exec_stream_t *gpu_stream) +{ + int nb_batched; + + (void)gpu_device; + if( stop_observed ) { + if( gpu_task == stop_candidate ) { + stop_candidate_completed = 1; + } + return PARSEC_HOOK_RETURN_DONE; + } + + nb_batched = parsec_gpu_task_collect_batch(gpu_stream, gpu_task, + cuda_batch_stop_first, NULL); + if( nb_batched < 0 ) { + return nb_batched; + } + if( !stop_observed ) { + return PARSEC_HOOK_RETURN_AGAIN; + } + return (0 == nb_batched) ? PARSEC_HOOK_RETURN_DONE + : PARSEC_HOOK_RETURN_ERROR; +} + int main(int argc, char* argv[]) { int ret; @@ -137,12 +473,15 @@ int main(int argc, char* argv[]) world = 1; rank = 0; #endif + (void)rank; + (void)world; - parsec_context = parsec_init(-1, NULL, NULL); + parsec_context = parsec_init(-1, &argc, &argv); // Create new DTD taskpool parsec_taskpool_t *tp = parsec_dtd_taskpool_new(); - parsec_task_class_t *again_tc, *async_tc; + parsec_task_class_t *again_tc, *batch_again_tc, *async_tc, *next_tc; + parsec_task_class_t *priority_tc, *stop_tc; ret = parsec_context_start(parsec_context); PARSEC_CHECK_ERROR(ret, "parsec_context_start"); @@ -153,30 +492,86 @@ int main(int argc, char* argv[]) again_tc = parsec_dtd_create_task_class(tp, "AGAIN", sizeof(int), PARSEC_VALUE, /* i */ - sizeof(int), PARSEC_VALUE, /* repeat */ - sizeof(int), PARSEC_VALUE, /* done */ PARSEC_DTD_ARG_END); parsec_dtd_task_class_add_chore(tp, again_tc, PARSEC_DEV_CUDA, cuda_task_again); + batch_again_tc = parsec_dtd_create_task_class(tp, "BATCH AGAIN", + sizeof(int), PARSEC_VALUE, /* i */ + sizeof(int), PARSEC_VALUE, /* phase */ + PARSEC_DTD_ARG_END); + parsec_dtd_task_class_add_chore(tp, batch_again_tc, + PARSEC_DEV_CUDA | PARSEC_DEV_CHORE_ALLOW_BATCH, + cuda_task_batch_again); + async_tc = parsec_dtd_create_task_class(tp, "ASYNC", sizeof(int), PARSEC_VALUE, /* i */ - sizeof(int), PARSEC_VALUE, /* repeat */ + sizeof(int), PARSEC_VALUE, /* phase */ PARSEC_DTD_ARG_END); - parsec_dtd_task_class_add_chore(tp, async_tc, PARSEC_DEV_CUDA, cuda_task_async); + parsec_dtd_task_class_add_chore(tp, async_tc, + PARSEC_DEV_CUDA | PARSEC_DEV_CHORE_ALLOW_BATCH, + cuda_task_async); - int zero = 0, done = 0; - for( int i = 0; i < world; ++i ) { + next_tc = parsec_dtd_create_task_class(tp, "NEXT", + sizeof(int), PARSEC_VALUE, /* i */ + sizeof(int), PARSEC_VALUE, /* phase */ + PARSEC_DTD_ARG_END); + parsec_dtd_task_class_add_chore(tp, next_tc, + PARSEC_DEV_CUDA | PARSEC_DEV_CHORE_ALLOW_BATCH, + cuda_task_next); + + priority_tc = parsec_dtd_create_task_class(tp, "PRIORITY", + sizeof(int), PARSEC_VALUE, /* i */ + sizeof(int), PARSEC_VALUE, /* phase */ + PARSEC_DTD_ARG_END); + parsec_dtd_task_class_add_chore(tp, priority_tc, + PARSEC_DEV_CUDA | PARSEC_DEV_CHORE_ALLOW_BATCH, + cuda_task_priority); + + stop_tc = parsec_dtd_create_task_class(tp, "BATCH STOP", + sizeof(int), PARSEC_VALUE, /* i */ + PARSEC_DTD_ARG_END); + parsec_dtd_task_class_add_chore(tp, stop_tc, + PARSEC_DEV_CUDA | PARSEC_DEV_CHORE_ALLOW_BATCH, + cuda_task_batch_stop); + + int zero = 0; + for( int i = 0; i < STATUS_TASKS; ++i ) { parsec_dtd_insert_task_with_task_class(tp, async_tc, 0, PARSEC_DEV_ALL, - PARSEC_AFFINITY, &i, + PARSEC_VALUE, &i, PARSEC_VALUE, &zero, PARSEC_DTD_ARG_END); } - for( int i = 0; i < world; ++i ) { + for( int i = 0; i < STATUS_TASKS; ++i ) { parsec_dtd_insert_task_with_task_class(tp, again_tc, 0, PARSEC_DEV_ALL, - PARSEC_AFFINITY, &i, + PARSEC_VALUE, &i, + PARSEC_DTD_ARG_END); + } + + for( int i = 0; i < STATUS_TASKS; ++i ) { + parsec_dtd_insert_task_with_task_class(tp, batch_again_tc, 0, PARSEC_DEV_ALL, + PARSEC_VALUE, &i, + PARSEC_VALUE, &zero, + PARSEC_DTD_ARG_END); + } + + for( int i = 0; i < STATUS_TASKS; ++i ) { + parsec_dtd_insert_task_with_task_class(tp, next_tc, 0, PARSEC_DEV_ALL, + PARSEC_VALUE, &i, + PARSEC_VALUE, &zero, + PARSEC_DTD_ARG_END); + } + + for( int i = 0; i < STATUS_TASKS; ++i ) { + parsec_dtd_insert_task_with_task_class(tp, priority_tc, 0, PARSEC_DEV_ALL, + PARSEC_VALUE, &i, PARSEC_VALUE, &zero, - PARSEC_VALUE, &done, + PARSEC_DTD_ARG_END); + } + + for( int i = 0; i < STATUS_TASKS; ++i ) { + parsec_dtd_insert_task_with_task_class(tp, stop_tc, 0, PARSEC_DEV_ALL, + PARSEC_VALUE, &i, PARSEC_DTD_ARG_END); } @@ -187,8 +582,50 @@ int main(int argc, char* argv[]) ret = parsec_context_wait(parsec_context); PARSEC_CHECK_ERROR(ret, "parsec_context_wait"); + if( !async_batch_observed || (next_batch_observed <= 1) ) { + parsec_warning("GPU batch status test did not form ASYNC and NEXT task rings\n"); + ret = 1; + } + if( (again_batch_observed <= 1) || (2 != again_batch_yields) ) { + parsec_warning("GPU AGAIN batch size=%d yielded=%d times\n", + again_batch_observed, again_batch_yields); + ret = 1; + } + if( !again_profile_state_valid || + (again_batch_release_count != again_batch_observed) ) { + parsec_warning("GPU AGAIN profiling state valid=%d released=%d expected=%d\n", + again_profile_state_valid, again_batch_release_count, + again_batch_observed); + ret = 1; + } + if( (priority_batch_observed <= 1) || !priority_order_valid || + (priority_release_count != priority_batch_observed) ) { + parsec_warning("GPU priority batch size=%d released=%d ordered=%d\n", + priority_batch_observed, priority_release_count, + priority_order_valid); + ret = 1; + } + if( (1 != stop_observed) || !stop_candidate_completed ) { + parsec_warning("GPU batch STOP observed=%d candidate completed=%d\n", + stop_observed, stop_candidate_completed); + ret = 1; + } + for( int i = 0; i < STATUS_TASKS; i++ ) { + if( (1 != async_completed[i]) || (1 != again_batch_completed[i]) || + (1 != next_completed[i]) ) { + parsec_warning("GPU batch status task %d completed ASYNC=%d AGAIN=%d NEXT=%d times\n", + i, async_completed[i], again_batch_completed[i], + next_completed[i]); + ret = 1; + } + } + parsec_dtd_task_class_release(tp, again_tc); + parsec_dtd_task_class_release(tp, batch_again_tc); parsec_dtd_task_class_release(tp, async_tc); + parsec_dtd_task_class_release(tp, next_tc); + parsec_dtd_task_class_release(tp, priority_tc); + parsec_dtd_task_class_release(tp, stop_tc); parsec_taskpool_free(tp); @@ -197,4 +634,5 @@ int main(int argc, char* argv[]) #if defined(PARSEC_HAVE_MPI) MPI_Finalize(); #endif + return ret; } diff --git a/tests/dsl/dtd/dtd_test_simple_gemm.c b/tests/dsl/dtd/dtd_test_simple_gemm.c index 0b8ac8d76..026a911f7 100644 --- a/tests/dsl/dtd/dtd_test_simple_gemm.c +++ b/tests/dsl/dtd/dtd_test_simple_gemm.c @@ -99,7 +99,6 @@ typedef struct gemm_cuda_batch_match_data_s { #define RndF_Mul 5.4210108624275222e-20f #define RndD_Mul 5.4210108624275222e-20 #define NBELEM 1 -#define GEMM_CUDA_BATCH_LIMIT_REACHED (-1000000) static int gemm_cuda_batch_enabled(void) @@ -276,7 +275,7 @@ gemm_cuda_batch_match(parsec_gpu_task_t *candidate, if( (NULL != batch_data) && (batch_data->accepted >= batch_data->max_batch_size - 1) ) { - return GEMM_CUDA_BATCH_LIMIT_REACHED; + return PARSEC_GPU_TASK_BATCH_STOP; } if( (batch_head->ec->task_class == candidate->ec->task_class) && @@ -285,9 +284,9 @@ gemm_cuda_batch_match(parsec_gpu_task_t *candidate, if( NULL != batch_data ) { batch_data->accepted++; } - return 0; + return PARSEC_GPU_TASK_BATCH_ACCEPT; } - return 1; + return PARSEC_GPU_TASK_BATCH_REJECT; } static void @@ -567,9 +566,15 @@ int gemm_kernel_cuda(parsec_device_gpu_module_t *gpu_device, if( GEMM_CUDA_BATCH_CUBLAS == cuda_batch_mode ) { batch_pool = &stream_state->batch_pool; - if( !gemm_cuda_batch_pool_can_submit(batch_pool) ) { - return PARSEC_HOOK_RETURN_AGAIN; - } + } + + /* Once followers are collected, AGAIN preserves that exact submitted ring + * as continuation state. Predictable resource backpressure must therefore + * be handled before collection. + */ + if( (GEMM_CUDA_BATCH_CUBLAS == cuda_batch_mode) && + !gemm_cuda_batch_pool_can_submit(batch_pool) ) { + return PARSEC_HOOK_RETURN_AGAIN; } if( gemm_cuda_batch_enabled() && (cuda_max_batch_size > 1) ) { @@ -579,9 +584,7 @@ int gemm_kernel_cuda(parsec_device_gpu_module_t *gpu_device, }; int nb_batched = parsec_gpu_task_collect_batch(gpu_stream, gpu_task, gemm_cuda_batch_match, &batch_data); - if( GEMM_CUDA_BATCH_LIMIT_REACHED == nb_batched ) { - nb_batched = batch_data.accepted; - } else if( nb_batched < 0 ) { + if( nb_batched < 0 ) { return nb_batched; } batch_count += nb_batched; diff --git a/tests/runtime/cuda/stage_custom.jdf b/tests/runtime/cuda/stage_custom.jdf index 71b8fe2ee..87f27a335 100644 --- a/tests/runtime/cuda/stage_custom.jdf +++ b/tests/runtime/cuda/stage_custom.jdf @@ -127,15 +127,20 @@ stage_custom_batch_match(parsec_gpu_task_t *candidate, { int *how_many = (int *)callback_data; - if( (*how_many < 5) && - (batch_head->ec->task_class == candidate->ec->task_class) ) { + /* Collection runs under the pending FIFO lock. Stop once the batch is full + * instead of scanning and rejecting the rest of the pending tasks. + */ + if( *how_many >= 5 ) { + return PARSEC_GPU_TASK_BATCH_STOP; + } + if( batch_head->ec->task_class == candidate->ec->task_class ) { (*how_many)++; PARSEC_DEBUG_VERBOSE(10, parsec_debug_output, "Add task %p to the %p batch\n", candidate, batch_head); - return 0; + return PARSEC_GPU_TASK_BATCH_ACCEPT; } - return 1; + return PARSEC_GPU_TASK_BATCH_REJECT; } %}