From a80b9878822455f85c99550f4e69e4b0ff713894 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Fri, 26 May 2023 14:22:51 -0400 Subject: [PATCH 01/10] TTG parsec data copy: add back-pointer to self This is needed to work-around missing support for offsetof in virtual classes. Instead we store the pointer to the derived class inside the non-virtual base class. This allows us to store data copies in parsec lifo or lists and get the virtual derived object back. Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg_data_copy.h | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/ttg/ttg/parsec/ttg_data_copy.h b/ttg/ttg/parsec/ttg_data_copy.h index 461984e3d2..17ac18c0f4 100644 --- a/ttg/ttg/parsec/ttg_data_copy.h +++ b/ttg/ttg/parsec/ttg_data_copy.h @@ -15,7 +15,22 @@ namespace ttg_parsec { * to facilitate the ref-counting of the data copy. * TODO: create abstractions for all fields in parsec_data_copy_t that we access. */ - struct ttg_data_copy_t : public parsec_data_copy_t { + + /* fwd decl */ + struct ttg_data_copy_t; + + /* special type: stores a pointer to the ttg_data_copy_t. This is necessary + * because ttg_data_copy_t has virtual functions so we cannot cast from parsec_data_copy_t + * to ttg_data_copy_t (offsetof is not supported for virtual classes). + * The self pointer is a back-pointer to the ttg_data_copy_t. */ + struct ttg_data_copy_self_t : public parsec_data_copy_t { + ttg_data_copy_t *self; + ttg_data_copy_self_t(ttg_data_copy_t* dc) + : self(dc) + { } + }; + + struct ttg_data_copy_t : public ttg_data_copy_self_t { #if defined(PARSEC_PROF_TRACE) && defined(PARSEC_TTG_PROFILE_BACKEND) int64_t size; int64_t uid; @@ -73,6 +88,7 @@ namespace ttg_parsec { } ttg_data_copy_t() + : ttg_data_copy_self_t(this) { /* TODO: do we need this construction? */ PARSEC_OBJ_CONSTRUCT(this, parsec_data_copy_t); From 124e34415938ea38cc96ad544c7048b0a1365054 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Fri, 26 May 2023 14:25:03 -0400 Subject: [PATCH 02/10] PaRSEC backend: Restore priorities Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg.h | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index b88874b0e7..bcc137e9bf 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -568,7 +568,6 @@ namespace ttg_parsec { : data_count(data_count) , defer_writer(defer_writer) , release_task_cb(release_fn) { - int32_t p = priority; PARSEC_LIST_ITEM_SINGLETON(&parsec_task.super); parsec_task.mempool_owner = mempool; parsec_task.task_class = task_class; @@ -774,10 +773,6 @@ namespace ttg_parsec { return PARSEC_HOOK_RETURN_DONE; } - static parsec_key_fn_t parsec_tasks_hash_fcts = {.key_equal = parsec_hash_table_generic_64bits_key_equal, - .key_print = parsec_hash_table_generic_64bits_key_print, - .key_hash = parsec_hash_table_generic_64bits_key_hash}; - template class rma_delayed_activate { std::vector _keylist; @@ -1668,11 +1663,11 @@ namespace ttg_parsec { char *taskobj = (char *)parsec_thread_mempool_allocate(mempool); int32_t priority = 0; if constexpr (!keyT_is_Void) { - //priority = priomap(key); + priority = priomap(key); /* placement-new the task */ newtask = new (taskobj) task_t(key, mempool, &this->self, world_impl.taskpool(), this, priority); } else { - //priority = priomap(); + priority = priomap(); /* placement-new the task */ newtask = new (taskobj) task_t(mempool, &this->self, world_impl.taskpool(), this, priority); } From cd678fe85e4761777172f4c9ce09f830bae9b4f9 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Fri, 26 May 2023 14:28:56 -0400 Subject: [PATCH 03/10] PaRSEC backend: offload reductions to dedicated task When we encounter a streaming terminal we schedule a dedicated reducer task. Subsequent inputs are put into a LIFO. Once the reducer task runs it will process all available inputs for that terminal but will not block waiting for all inputs. In the worst case, we create N tasks (one for each input), likely less. Signed-off-by: Joseph Schuchart --- tests/unit/fibonacci.cc | 3 +- ttg/ttg/parsec/ttg.h | 359 +++++++++++++++++++++++++++++++++++----- 2 files changed, 317 insertions(+), 45 deletions(-) diff --git a/tests/unit/fibonacci.cc b/tests/unit/fibonacci.cc index 6434aaf555..574661b338 100644 --- a/tests/unit/fibonacci.cc +++ b/tests/unit/fibonacci.cc @@ -49,8 +49,9 @@ TEST_CASE("Fibonacci", "[fib][core]") { const auto F_n_plus_2 = F_n_plus_1 + F_n; ttg::sendv<1>(F_n_plus_1, outs); ttg::send<0>(F_n_plus_2, F_n_plus_1, outs); - } else + } else { ttg::finalize<1>(outs); + } }, ttg::edges(F2F), ttg::edges(F2F, F2P)); auto print_op = ttg::make_tt( diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index bcc137e9bf..e24e93b71a 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -526,10 +526,36 @@ namespace ttg_parsec { typedef void (release_task_fn)(parsec_ttg_task_base_t*); - typedef struct { + struct stream_info_t { std::size_t goal; std::size_t size; - } size_goal_t; + parsec_lifo_t reduce_copies; + std::atomic reduce_count; + }; + + protected: + template + void init_stream_info_impl(TT *tt, std::array& streams) { + if constexpr (TT::numins > i) { + if (std::get(tt->input_reducers)) { + streams[i].goal = tt->static_stream_goal[i]; + streams[i].size = 0; + PARSEC_OBJ_CONSTRUCT(&streams[i].reduce_copies, parsec_lifo_t); + streams[i].reduce_count.store(0, std::memory_order_relaxed); + } + /* recursion */ + if constexpr((i + 1) < TT::numins) { + init_stream_info_impl(tt, streams); + } + } + } + + template + void init_stream_info(TT *tt, std::array& streams) { + init_stream_info_impl<0>(tt, streams); + } + + public: /* Poor-mans virtual function * We cannot use virtual inheritance or private visibility because we @@ -588,7 +614,7 @@ namespace ttg_parsec { static constexpr size_t num_streams = TT::numins; TT* tt; key_type key; - size_goal_t stream[num_streams] = {}; + std::array streams; parsec_ttg_task_t(parsec_thread_mempool_t *mempool, parsec_task_class_t *task_class) : parsec_ttg_task_base_t(mempool, task_class, num_streams) { @@ -615,6 +641,8 @@ namespace ttg_parsec { parsec_task.data[i].data_in = nullptr; } + init_stream_info(tt, streams); + // We store the hash of the key and the address where it can be found in locals considered as a scratchpad uint64_t hv = ttg::hash>{}(key); *(uintptr_t*)&(parsec_task.locals[0]) = hv; @@ -634,7 +662,7 @@ namespace ttg_parsec { struct parsec_ttg_task_t : public parsec_ttg_task_base_t { static constexpr size_t num_streams = TT::numins; TT* tt; - size_goal_t stream[num_streams] = {}; + std::array streams; parsec_ttg_task_t(parsec_thread_mempool_t *mempool, parsec_task_class_t *task_class) : parsec_ttg_task_base_t(mempool, task_class, num_streams) { @@ -655,6 +683,8 @@ namespace ttg_parsec { for (int i = 0; i < num_streams; ++i) { parsec_task.data[i].data_in = nullptr; } + + init_stream_info(tt, streams); } static void release_task(parsec_ttg_task_base_t* task_base) { @@ -1144,6 +1174,7 @@ namespace ttg_parsec { private: using task_t = detail::parsec_ttg_task_t; + friend detail::parsec_ttg_task_base_t; friend task_t; /* the offset of the key placed after the task structure in the memory from mempool */ @@ -1194,7 +1225,8 @@ namespace ttg_parsec { // For now use same type for unary/streaming input terminals, and stream reducers assigned at runtime ttg::meta::detail::input_reducers_t input_reducers; //!< Reducers for the input terminals (empty = expect single value) - std::array static_stream_goal; + std::array inpute_reducers_taskclass = { nullptr }; + std::array static_stream_goal = { std::numeric_limits::max() }; int num_pullins = 0; bool m_defer_writer = TTG_PARSEC_DEFER_WRITER; @@ -1313,6 +1345,95 @@ namespace ttg_parsec { parsec_ttg_caller = NULL; } + struct reducer_task_t { + parsec_task_t parsec_task; + task_t *task; + + reducer_task_t(task_t* task, parsec_thread_mempool_t *mempool, + parsec_task_class_t *task_class, parsec_taskpool_t *taskpool, + int32_t priority) + : task(task) + { + PARSEC_LIST_ITEM_SINGLETON(&parsec_task.super); + parsec_task.mempool_owner = mempool; + parsec_task.task_class = task_class; + parsec_task.status = PARSEC_TASK_STATUS_HOOK; + parsec_task.taskpool = taskpool; + parsec_task.priority = priority; + parsec_task.chore_mask = 1<<0; + } + }; + + template + static parsec_hook_return_t static_reducer_op(parsec_execution_stream_s *es, parsec_task_t *parsec_task) { + using rtask_t = reducer_task_t; + using value_t = std::tuple_element_t; + constexpr const bool val_is_void = ttg::meta::is_void_v; + rtask_t *rtask = (rtask_t*)parsec_task; + task_t *task = rtask->task; + ttT *baseobj = task->tt; + derivedT *obj = static_cast(baseobj); + + auto& reducer = std::get(baseobj->input_reducers); + + assert(parsec_ttg_caller == NULL); + parsec_ttg_caller = static_cast(task); + + if (obj->tracing()) { + if constexpr (!ttg::meta::is_void_v) + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", task->key, ": reducer executing"); + else + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : reducer executing"); + } + + /* the copy to reduce into */ + detail::ttg_data_copy_t *target_copy; + target_copy = static_cast(task->parsec_task.data[i].data_in); + assert(val_is_void || nullptr != target_copy); + /* once we hit 0 we have to stop since another thread might enqueue a new reduction task */ + std::size_t c; + do { + if constexpr(!val_is_void) { + /* the copies to reduce out of */ + detail::ttg_data_copy_t *source_copy; + parsec_list_item_t *item; + item = parsec_lifo_pop(&task->streams[i].reduce_copies); + if (nullptr == item) { + break; // maybe someone is changing the goal right now + } + source_copy = ((detail::ttg_data_copy_self_t *)(item))->self; + reducer(*reinterpret_cast *>(target_copy->device_private), + *reinterpret_cast *>(source_copy->device_private)); + detail::release_data_copy(source_copy); + } else if constexpr(val_is_void) { + reducer(); // invoke control reducer + } + // there is only one task working on this stream, so no need to be atomic here + task->streams[i].size++; + } while ((c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)) > 1); + + /* finalize_argstream sets goal to 1, so size may be larger than goal */ + bool complete = (task->streams[i].size >= task->streams[i].goal); + + if (complete && c == 1) { + /* task is still in the hash table, have release_task remove it */ + task->remove_from_hash = true; + task->release_task(task); + } + + parsec_ttg_caller = NULL; + + if (obj->tracing()) { + if constexpr (!ttg::meta::is_void_v) + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", task->key, ": done executing"); + else + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : done executing"); + } + + return PARSEC_HOOK_RETURN_DONE; + } + + protected: template uint64_t unpack(T &obj, void *_bytes, uint64_t pos) { @@ -1678,14 +1799,37 @@ namespace ttg_parsec { newtask->function_template_class_ptr[static_cast(ttg::ExecutionSpace::CUDA)] = reinterpret_cast(&TT::static_op); - for (int i = 0; i < static_stream_goal.size(); ++i) { - newtask->stream[i].goal = static_stream_goal[i]; + ttg::trace(world.rank(), ":", get_name(), " : ", key, ": creating task"); + return newtask; + } + + + template + reducer_task_t *create_new_reducer_task(task_t *task) { + /* make sure we can reuse the existing memory pool and don't have to create a new one */ + static_assert(sizeof(task_t) >= sizeof(reducer_task_t)); + constexpr const bool keyT_is_Void = ttg::meta::is_void_v; + auto &world_impl = world.impl(); + reducer_task_t *newtask; + parsec_thread_mempool_t *mempool = get_task_mempool(); + char *taskobj = (char *)parsec_thread_mempool_allocate(mempool); + // use the priority of the task we stream into + int32_t priority = 0; + if constexpr (!keyT_is_Void) { + priority = priomap(task->key); + ttg::trace(world.rank(), ":", get_name(), " : ", task->key, ": creating reducer task"); + } else { + priority = priomap(); + ttg::trace(world.rank(), ":", get_name(), ": creating reducer task"); } + /* placement-new the task */ + newtask = new (taskobj) reducer_task_t(task, mempool, inpute_reducers_taskclass[i], + world_impl.taskpool(), priority); - ttg::trace(world.rank(), ":", get_name(), " : ", key, ": creating task"); return newtask; } + // Used to set the i'th argument template void set_arg_local_impl(const Key &key, Value &&value, detail::ttg_data_copy_t *copy_in = nullptr, @@ -1711,7 +1855,7 @@ namespace ttg_parsec { task_t *task; auto &world_impl = world.impl(); auto &reducer = std::get(input_reducers); - bool release = true; + bool release = false; bool remove_from_hash = true; bool discover_task = true; bool get_pull_data = false; @@ -1771,7 +1915,20 @@ namespace ttg_parsec { if (reducer) { // is this a streaming input? reduce the received value // N.B. Right now reductions are done eagerly, without spawning tasks // this means we must lock - parsec_hash_table_lock_bucket(&tasks_table, hk); + //parsec_hash_table_lock_bucket(&tasks_table, hk); + + auto submit_reducer_task = [&](auto *task){ + /* check if we need to create a task */ + std::size_t c = task->streams[i].reduce_count.fetch_add(1, std::memory_order_release); + if (0 == c) { + /* we are responsible for creating the reduction task */ + reducer_task_t *reduce_task; + reduce_task = create_new_reducer_task(task); + parsec_task_t *vp_task_rings[1] = { &reduce_task->parsec_task }; + parsec_execution_stream_t *es = world_impl.execution_stream(); + __parsec_schedule_vp(es, vp_task_rings, 0); + } + }; if constexpr (!ttg::meta::is_void_v) { // for data values // have a value already? if not, set, otherwise reduce @@ -1782,19 +1939,35 @@ namespace ttg_parsec { * mechanism (it would release the task, not the reduction value). */ copy = detail::create_new_datacopy(std::forward(value)); task->parsec_task.data[i].data_in = copy; + task->streams[i].size++; + if (task->streams[i].size == task->streams[i].goal) { + release = true; + } } else { - reducer(*reinterpret_cast *>(copy->device_private), value); + if (nullptr != parsec_ttg_caller) { + copy = detail::find_copy_in_task(parsec_ttg_caller, &value); + } + if (nullptr != copy) { + /* retain the data copy */ + copy = detail::register_data_copy(copy, task, input_is_const); + } else { + /* create a new copy */ + copy = detail::create_new_datacopy(std::forward(value)); + } + /* enqueue the data copy to be reduced */ + parsec_lifo_push(&task->streams[i].reduce_copies, ©->super); + submit_reducer_task(task); + //reducer(*reinterpret_cast *>(copy->device_private), value); } } else { - reducer(); // even if this was a control input, must execute the reducer for possible side effects + /* submit reducer for void values to handle side effects */ + submit_reducer_task(task); } - task->stream[i].size++; - release = (task->stream[i].size == task->stream[i].goal); - if (release) { - parsec_hash_table_nolock_remove(&tasks_table, hk); - remove_from_hash = false; - } - parsec_hash_table_unlock_bucket(&tasks_table, hk); + //if (release) { + // parsec_hash_table_nolock_remove(&tasks_table, hk); + // remove_from_hash = false; + //} + //parsec_hash_table_unlock_bucket(&tasks_table, hk); } else { /* whether the task needs to be deferred or not */ if constexpr (!valueT_is_Void) { @@ -1819,6 +1992,8 @@ namespace ttg_parsec { * make a copy of the original data */ release = (copy->push_task != &task->parsec_task); task->parsec_task.data[i].data_in = copy; + } else { + release = true; } } task->remove_from_hash = remove_from_hash; @@ -2342,7 +2517,7 @@ namespace ttg_parsec { this->trace(world.rank(), ":", get_name(), ": setting global stream size for terminal ", i); // Check if stream is already bounded - if (static_stream_goal[i] > 0) { + if (static_stream_goal[i] > std::numeric_limits::max()) { ttg::print_error(world.rank(), ":", get_name(), " : error stream is already bounded : ", i); throw std::runtime_error("TT::set_static_argstream_size called for a bounded stream"); } @@ -2393,16 +2568,22 @@ namespace ttg_parsec { #endif } } + parsec_hash_table_unlock_bucket(&tasks_table, hk); // TODO: Unfriendly implementation, cannot check if stream is already bounded // TODO: Unfriendly implementation, cannot check if stream has been finalized already // commit changes - task->stream[i].goal = size; - bool release = (task->stream[i].size == task->stream[i].goal); - parsec_hash_table_unlock_bucket(&tasks_table, hk); - - if (release) release_task(task); + // 1) "lock" the stream by incrementing the reduce_count + // 2) set the goal + // 3) "unlock" the stream + // only one thread will see the reduce_count be zero and the goal match the size + task->streams[i].reduce_count.fetch_add(1, std::memory_order_acquire); + task->streams[i].goal = size; + auto c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_release); + if (1 == c && (task->streams[i].size >= size)) { + release_task(task); + } } } @@ -2447,16 +2628,22 @@ namespace ttg_parsec { #endif } } + parsec_hash_table_unlock_bucket(&tasks_table, hk); // TODO: Unfriendly implementation, cannot check if stream is already bounded // TODO: Unfriendly implementation, cannot check if stream has been finalized already // commit changes - task->stream[i].goal = size; - bool release = (task->stream[i].size == task->stream[i].goal); - parsec_hash_table_unlock_bucket(&tasks_table, hk); - - if (release) release_task(task); + // 1) "lock" the stream by incrementing the reduce_count + // 2) set the goal + // 3) "unlock" the stream + // only one thread will see the reduce_count be zero and the goal match the size + task->streams[i].reduce_count.fetch_add(1, std::memory_order_acquire); + task->streams[i].goal = size; + auto c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_release); + if (1 == c && (task->streams[i].size >= size)) { + release_task(task); + } } } @@ -2488,8 +2675,8 @@ namespace ttg_parsec { auto hk = reinterpret_cast(&key); task_t *task = nullptr; - parsec_hash_table_lock_bucket(&tasks_table, hk); - if (nullptr == (task = (task_t *)parsec_hash_table_nolock_find(&tasks_table, hk))) { + //parsec_hash_table_lock_bucket(&tasks_table, hk); + if (nullptr == (task = (task_t *)parsec_hash_table_find(&tasks_table, hk))) { ttg::print_error(world.rank(), ":", get_name(), ":", key, " : error finalize called on stream that never received an input data: ", i); throw std::runtime_error("TT::finalize called on stream that never received an input data"); @@ -2499,10 +2686,16 @@ namespace ttg_parsec { // TODO: Unfriendly implementation, cannot check if stream has been finalized already // commit changes - task->stream[i].size = 1; - parsec_hash_table_unlock_bucket(&tasks_table, hk); - - release_task(task); + // 1) "lock" the stream by incrementing the reduce_count + // 2) set the goal + // 3) "unlock" the stream + // only one thread will see the reduce_count be zero and the goal match the size + task->streams[i].reduce_count.fetch_add(1, std::memory_order_acquire); + task->streams[i].goal = 1; + auto c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_release); + if (1 == c && (task->streams[i].size >= 1)) { + release_task(task); + } } } @@ -2532,8 +2725,7 @@ namespace ttg_parsec { auto hk = static_cast(0); task_t *task = nullptr; - parsec_hash_table_lock_bucket(&tasks_table, hk); - if (nullptr == (task = (task_t *)parsec_hash_table_nolock_find(&tasks_table, hk))) { + if (nullptr == (task = (task_t *)parsec_hash_table_find(&tasks_table, hk))) { ttg::print_error(world.rank(), ":", get_name(), " : error finalize called on stream that never received an input data: ", i); throw std::runtime_error("TT::finalize called on stream that never received an input data"); @@ -2543,10 +2735,16 @@ namespace ttg_parsec { // TODO: Unfriendly implementation, cannot check if stream has been finalized already // commit changes - task->stream[i].size = 1; - parsec_hash_table_unlock_bucket(&tasks_table, hk); - - release_task(task); + // 1) "lock" the stream by incrementing the reduce_count + // 2) set the goal + // 3) "unlock" the stream + // only one thread will see the reduce_count be zero and the goal match the size + task->streams[i].reduce_count.fetch_add(1, std::memory_order_acquire); + task->streams[i].goal = 1; + auto c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_release); + if (1 == c && (task->streams[i].size >= 1)) { + release_task(task); + } } } @@ -2781,8 +2979,7 @@ namespace ttg_parsec { , keymap(std::is_same>::value ? decltype(keymap)(ttg::detail::default_keymap(world)) : decltype(keymap)(std::forward(keymap_))) - , priomap(decltype(keymap)(std::forward(priomap_))) - , static_stream_goal() { + , priomap(decltype(keymap)(std::forward(priomap_))) { // Cannot call these in base constructor since terminals not yet constructed if (innames.size() != numinedges) throw std::logic_error("ttg_parsec::TT: #input names != #input terminals"); if (outnames.size() != numouts) throw std::logic_error("ttg_parsec::TT: #output names != #output terminals"); @@ -2937,6 +3134,12 @@ namespace ttg_parsec { free((void*)self.name); self.name = nullptr; } + for (std::size_t i = 0; i < numins; ++i) { + if (inpute_reducers_taskclass[i] != nullptr) { + std::free(inpute_reducers_taskclass[i]); + inpute_reducers_taskclass[i] = nullptr; + } + } release(); } @@ -2988,6 +3191,74 @@ namespace ttg_parsec { void set_input_reducer(Reducer &&reducer) { ttg::trace(world.rank(), ":", get_name(), " : setting reducer for terminal ", i); std::get(input_reducers) = reducer; + + parsec_task_class_t *tc = inpute_reducers_taskclass[i]; + if (nullptr == tc) { + tc = (parsec_task_class_t *)std::calloc(1, sizeof(*tc)); + inpute_reducers_taskclass[i] = tc; + + tc->name = strdup((get_name() + std::string(" reducer ") + std::to_string(i)).c_str()); + tc->task_class_id = get_instance_id(); + tc->nb_parameters = 0; + tc->nb_locals = 0; + tc->nb_flows = numflows; + + auto &world_impl = world.impl(); + + if( world_impl.profiling() ) { + // first two ints are used to store the hash of the key. + tc->nb_parameters = (sizeof(void*)+sizeof(int)-1)/sizeof(int); + // seconds two ints are used to store a pointer to the key of the task. + tc->nb_locals = self.nb_parameters + (sizeof(void*)+sizeof(int)-1)/sizeof(int); + + // If we have parameters and locals, we need to define the corresponding dereference arrays + tc->params[0] = &detail::parsec_taskclass_param0; + tc->params[1] = &detail::parsec_taskclass_param1; + + tc->locals[0] = &detail::parsec_taskclass_param0; + tc->locals[1] = &detail::parsec_taskclass_param1; + tc->locals[2] = &detail::parsec_taskclass_param2; + tc->locals[3] = &detail::parsec_taskclass_param3; + } + tc->make_key = make_key; + tc->key_functions = &tasks_hash_fcts; + tc->task_snprintf = parsec_ttg_task_snprintf; + +#if defined(PARSEC_PROF_TRACE) + tc->profile_info = &parsec_ttg_task_info; +#endif + + world_impl.taskpool()->nb_task_classes = std::max(world_impl.taskpool()->nb_task_classes, static_castnb_task_classes)>(self.task_class_id+1)); + +#if 0 + // FIXME: currently only support reduction on the host + if constexpr (derived_has_cuda_op()) { + self.incarnations = (__parsec_chore_t *)malloc(3 * sizeof(__parsec_chore_t)); + ((__parsec_chore_t *)self.incarnations)[0].type = PARSEC_DEV_CUDA; + ((__parsec_chore_t *)self.incarnations)[0].evaluate = NULL; + ((__parsec_chore_t *)self.incarnations)[0].hook = detail::hook_cuda; + ((__parsec_chore_t *)self.incarnations)[1].type = PARSEC_DEV_CPU; + ((__parsec_chore_t *)self.incarnations)[1].evaluate = NULL; + ((__parsec_chore_t *)self.incarnations)[1].hook = detail::hook; + ((__parsec_chore_t *)self.incarnations)[2].type = PARSEC_DEV_NONE; + ((__parsec_chore_t *)self.incarnations)[2].evaluate = NULL; + ((__parsec_chore_t *)self.incarnations)[2].hook = NULL; + } else +#endif // 0 + { + tc->incarnations = (__parsec_chore_t *)malloc(2 * sizeof(__parsec_chore_t)); + ((__parsec_chore_t *)tc->incarnations)[0].type = PARSEC_DEV_CPU; + ((__parsec_chore_t *)tc->incarnations)[0].evaluate = NULL; + ((__parsec_chore_t *)tc->incarnations)[0].hook = &static_reducer_op; + ((__parsec_chore_t *)tc->incarnations)[1].type = PARSEC_DEV_NONE; + ((__parsec_chore_t *)tc->incarnations)[1].evaluate = NULL; + ((__parsec_chore_t *)tc->incarnations)[1].hook = NULL; + } + + /* the reduction task does not alter the termination detection because the target task will execute */ + tc->release_task = &parsec_release_task_to_mempool; + tc->complete_execution = NULL; + } } /// define the reducer function to be called when additional inputs are From cdf2aa6501e85011369969e4258aec2b34e8b229 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Tue, 6 Jun 2023 15:43:17 -0400 Subject: [PATCH 04/10] PaRSEC: replace thread-local es with parsec_my_execution_stream() The thread-local variable was a hack that is not needed anymore, and apparently didn't work properly. Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index e24e93b71a..6aef75ca63 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -2958,7 +2958,7 @@ namespace ttg_parsec { parsec_key_fn_t tasks_hash_fcts = {key_equal, key_print, key_hash}; static parsec_hook_return_t complete_task_and_release(parsec_execution_stream_t *es, parsec_task_t *t) { - auto *task = (detail::parsec_ttg_task_base_t *)t; + task_t *task = (task_t *)t; for (int i = 0; i < task->data_count; i++) { detail::ttg_data_copy_t *copy = static_cast(task->parsec_task.data[i].data_in); if (nullptr == copy) continue; From 70432ae209b206d09db99d190acc2e3d7959697b Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Tue, 6 Jun 2023 15:45:15 -0400 Subject: [PATCH 05/10] PaRSEC: Fix check for bounded stream in static_argstream_size Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index 6aef75ca63..1048e8109e 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -2517,7 +2517,7 @@ namespace ttg_parsec { this->trace(world.rank(), ":", get_name(), ": setting global stream size for terminal ", i); // Check if stream is already bounded - if (static_stream_goal[i] > std::numeric_limits::max()) { + if (static_stream_goal[i] < std::numeric_limits::max()) { ttg::print_error(world.rank(), ":", get_name(), " : error stream is already bounded : ", i); throw std::runtime_error("TT::set_static_argstream_size called for a bounded stream"); } From 36fb8442264e4fabf28cdb15c6119e4e8ea02f9c Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Tue, 6 Jun 2023 16:11:25 -0400 Subject: [PATCH 06/10] Add stream unit test Signed-off-by: Joseph Schuchart --- tests/unit/CMakeLists.txt | 2 +- tests/unit/streams.cc | 74 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) create mode 100644 tests/unit/streams.cc diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 2016872e25..879972aa4c 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -1,7 +1,7 @@ include(AddTTGExecutable) # TT unit test: core TTG ops -add_ttg_executable(core-unittests-ttg "fibonacci.cc;ranges.cc;tt.cc;unit_main.cpp" LINK_LIBRARIES "Catch2::Catch2") +add_ttg_executable(core-unittests-ttg "fibonacci.cc;ranges.cc;tt.cc;unit_main.cpp;streams.cc" LINK_LIBRARIES "Catch2::Catch2") # serialization test: probes serialization via all supported serialization methods (MADNESS, Boost::serialization, cereal) that are available add_executable(serialization "serialization.cc;unit_main.cpp") diff --git a/tests/unit/streams.cc b/tests/unit/streams.cc new file mode 100644 index 0000000000..bef3a2837e --- /dev/null +++ b/tests/unit/streams.cc @@ -0,0 +1,74 @@ +#include +#include + +#include "ttg.h" + +#include "ttg/serialization/std/pair.h" +#include "ttg/util/hash/std/pair.h" + + + +TEST_CASE("streams", "[streams][core]") { + // in distributed memory we must count how many messages the reducer will receive + SECTION("concurrent-stream-size") { + ttg::Edge I2O; + ttg::Edge O2S; + const auto nranks = ttg::default_execution_context().size(); + + constexpr std::size_t N = 10000; + constexpr std::size_t SLICE = 500; + constexpr const timespec ts = { .tv_sec = 0, .tv_nsec = 10000 }; + constexpr int VALUE = 1; + std::atomic reduce_ops = 0; + + auto op = ttg::make_tt( + [&](const int &n, int&& i, + std::tuple> &outs) { + int key = n/SLICE; + nanosleep(&ts, nullptr); + if (n < N-1) { + ttg::send<0>(key, std::forward(i), outs); + //ttg::print("sent to sink ", key); + } else { + // set the size of the last reducer + if (N%SLICE > 0) { + ttg::set_size<0>(key, N%SLICE, outs); + std::cout << "set_size key " << key << " size " << N%SLICE << std::endl; + } + // forward the value + ttg::send<0>(key, std::forward(i), outs); + //ttg::print("finalized last sink ", key); + } + }, + ttg::edges(I2O), ttg::edges(O2S)); + + auto sink_op = ttg::make_tt( + [&](const int key, const int &value) { + std::cout << "sink " << key << std::endl; + if (!(value == SLICE || key == (N/SLICE))) { + std::cout << "SINK ERROR: key " << key << " value " << value << " SLICE " << SLICE << " N " << N << std::endl; + } + CHECK((value == SLICE || key == (N/SLICE))); + reduce_ops++; + }, + ttg::edges(O2S), ttg::edges()); + + op->set_keymap([=](const auto &key) { return nranks - 1; }); + op->set_trace_instance(true); + sink_op->set_input_reducer<0>([&](int &a, const int &b) { + a += 1; // we count invocations + CHECK(b == VALUE); + reduce_ops++; + }, SLICE); + make_graph_executable(op); + ttg::ttg_fence(ttg::default_execution_context()); + if (ttg::default_execution_context().rank() == 0) { + for (std::size_t i = 0; i < N; ++i) { + op->invoke(i, VALUE); + } + } + auto &sink_op_real = *sink_op; + ttg::ttg_fence(ttg::default_execution_context()); + CHECK(reduce_ops == N); + } +} // TEST_CASE("streams") \ No newline at end of file From 6d41c937eb89b7d729a718e6b0a1bc738be2030b Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Tue, 6 Jun 2023 16:14:33 -0400 Subject: [PATCH 07/10] PaRSEC: misc fixes to the reducer offload task Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg.h | 39 ++++++++++++++++++++++++++------------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index 1048e8109e..3b18b203fc 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -1391,7 +1391,9 @@ namespace ttg_parsec { target_copy = static_cast(task->parsec_task.data[i].data_in); assert(val_is_void || nullptr != target_copy); /* once we hit 0 we have to stop since another thread might enqueue a new reduction task */ - std::size_t c; + std::size_t c = 0; + std::size_t size = 0; + assert(task->streams[i].reduce_count > 0); do { if constexpr(!val_is_void) { /* the copies to reduce out of */ @@ -1399,7 +1401,8 @@ namespace ttg_parsec { parsec_list_item_t *item; item = parsec_lifo_pop(&task->streams[i].reduce_copies); if (nullptr == item) { - break; // maybe someone is changing the goal right now + // maybe someone is changing the goal right now + break; } source_copy = ((detail::ttg_data_copy_self_t *)(item))->self; reducer(*reinterpret_cast *>(target_copy->device_private), @@ -1409,13 +1412,16 @@ namespace ttg_parsec { reducer(); // invoke control reducer } // there is only one task working on this stream, so no need to be atomic here - task->streams[i].size++; - } while ((c = task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)) > 1); + size = ++task->streams[i].size; + //std::cout << "static_reducer_op key " << task->key << " size " << size << " of " << task->streams[i].goal << std::endl; + } while ((c = (task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)-1)) > 0); + //} while ((c = (--task->streams[i].reduce_count)) > 0); /* finalize_argstream sets goal to 1, so size may be larger than goal */ - bool complete = (task->streams[i].size >= task->streams[i].goal); + bool complete = (size >= task->streams[i].goal); - if (complete && c == 1) { + + if (complete && c == 0) { /* task is still in the hash table, have release_task remove it */ task->remove_from_hash = true; task->release_task(task); @@ -1878,7 +1884,10 @@ namespace ttg_parsec { parsec_hash_table_nolock_remove(&tasks_table, hk); remove_from_hash = false; } - parsec_hash_table_unlock_bucket(&tasks_table, hk); + /* if we have a reducer, we need to hold on to the lock for just a little longer */ + if (!reducer) { + parsec_hash_table_unlock_bucket(&tasks_table, hk); + } } else { task = create_new_task(key); world_impl.increment_created(); @@ -1913,14 +1922,13 @@ namespace ttg_parsec { } if (reducer) { // is this a streaming input? reduce the received value - // N.B. Right now reductions are done eagerly, without spawning tasks - // this means we must lock - //parsec_hash_table_lock_bucket(&tasks_table, hk); - auto submit_reducer_task = [&](auto *task){ /* check if we need to create a task */ std::size_t c = task->streams[i].reduce_count.fetch_add(1, std::memory_order_release); + //std::size_t c = task->streams[i].reduce_count++; if (0 == c) { + //std::cout << "submit reducer task for task " << task->key + // << " size " << task->streams[i].size << " of " << task->streams[i].goal << std::endl; /* we are responsible for creating the reduction task */ reducer_task_t *reduce_task; reduce_task = create_new_reducer_task(task); @@ -1943,13 +1951,17 @@ namespace ttg_parsec { if (task->streams[i].size == task->streams[i].goal) { release = true; } + /* now we can unlock the bucket */ + parsec_hash_table_unlock_bucket(&tasks_table, hk); } else { + /* unlock the bucket, the lock is not needed anymore */ + parsec_hash_table_unlock_bucket(&tasks_table, hk); if (nullptr != parsec_ttg_caller) { copy = detail::find_copy_in_task(parsec_ttg_caller, &value); } if (nullptr != copy) { /* retain the data copy */ - copy = detail::register_data_copy(copy, task, input_is_const); + copy = detail::register_data_copy(copy, task, true); } else { /* create a new copy */ copy = detail::create_new_datacopy(std::forward(value)); @@ -1957,9 +1969,10 @@ namespace ttg_parsec { /* enqueue the data copy to be reduced */ parsec_lifo_push(&task->streams[i].reduce_copies, ©->super); submit_reducer_task(task); - //reducer(*reinterpret_cast *>(copy->device_private), value); } } else { + /* unlock the bucket, the lock is not needed anymore */ + parsec_hash_table_unlock_bucket(&tasks_table, hk); /* submit reducer for void values to handle side effects */ submit_reducer_task(task); } From 7670a0837a68d4d31d870806df6013a57c472011 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Fri, 9 Jun 2023 10:35:33 -0400 Subject: [PATCH 08/10] PaRSEC: fix copy handling for streaming terminals Signed-off-by: Joseph Schuchart --- ttg/ttg/parsec/ttg.h | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index 3b18b203fc..3846e66fa5 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -1940,13 +1940,11 @@ namespace ttg_parsec { if constexpr (!ttg::meta::is_void_v) { // for data values // have a value already? if not, set, otherwise reduce - detail::ttg_data_copy_t *copy = nullptr; - if (nullptr == (copy = static_cast(task->parsec_task.data[i].data_in))) { + if (nullptr == static_cast(task->parsec_task.data[i].data_in)) { using decay_valueT = std::decay_t; /* For now, we always create a copy because we cannot rely on the task_release * mechanism (it would release the task, not the reduction value). */ - copy = detail::create_new_datacopy(std::forward(value)); - task->parsec_task.data[i].data_in = copy; + task->parsec_task.data[i].data_in = detail::create_new_datacopy(std::forward(value)); task->streams[i].size++; if (task->streams[i].size == task->streams[i].goal) { release = true; @@ -1954,6 +1952,7 @@ namespace ttg_parsec { /* now we can unlock the bucket */ parsec_hash_table_unlock_bucket(&tasks_table, hk); } else { + detail::ttg_data_copy_t *copy = nullptr; /* unlock the bucket, the lock is not needed anymore */ parsec_hash_table_unlock_bucket(&tasks_table, hk); if (nullptr != parsec_ttg_caller) { @@ -2524,7 +2523,7 @@ namespace ttg_parsec { /// \param size positive integer that specifies the default stream size template void set_static_argstream_size(std::size_t size) { - assert(std::get(input_reducers) && "TT::set_argstream_size called on nonstreaming input terminal"); + assert(std::get(input_reducers) && "TT::set_static_argstream_size called on nonstreaming input terminal"); assert(size > 0 && "TT::set_static_argstream_size(key,size) called with size=0"); this->trace(world.rank(), ":", get_name(), ": setting global stream size for terminal ", i); From 67d8520067fbfebb0e06a2b73cae22f827366eb3 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Thu, 6 Jul 2023 19:30:23 -0400 Subject: [PATCH 09/10] PaRSEC: Allow reducer tasks to be deferred to avoid copying stream inputs In the PaRSEC backend, reducer tasks will be treated as any other task and be deferred if there are readers on their first input. All other inputs are read-only and will be handled that way. This avoids extra copies if there are no other writers on these copies. Signed-off-by: Joseph Schuchart --- tests/unit/streams.cc | 3 +- ttg/ttg/parsec/ttg.h | 194 +++++++++++++++++++++++++----------------- 2 files changed, 116 insertions(+), 81 deletions(-) diff --git a/tests/unit/streams.cc b/tests/unit/streams.cc index bef3a2837e..108c8610b9 100644 --- a/tests/unit/streams.cc +++ b/tests/unit/streams.cc @@ -60,6 +60,7 @@ TEST_CASE("streams", "[streams][core]") { CHECK(b == VALUE); reduce_ops++; }, SLICE); + make_graph_executable(op); ttg::ttg_fence(ttg::default_execution_context()); if (ttg::default_execution_context().rank() == 0) { @@ -67,7 +68,7 @@ TEST_CASE("streams", "[streams][core]") { op->invoke(i, VALUE); } } - auto &sink_op_real = *sink_op; + ttg::ttg_fence(ttg::default_execution_context()); CHECK(reduce_ops == N); } diff --git a/ttg/ttg/parsec/ttg.h b/ttg/ttg/parsec/ttg.h index 3846e66fa5..57a8ebf2c2 100644 --- a/ttg/ttg/parsec/ttg.h +++ b/ttg/ttg/parsec/ttg.h @@ -696,6 +696,36 @@ namespace ttg_parsec { parsec_key_t pkey() { return 0; } }; + /** + * Reducer task representing one or more stream reductions. + * A reducer task may be deferred on its first input (the object into which + * all other inputs are folded). Once that input becomes available the task + * is submitted and reduces all available inputs. Additional reducer tasks may + * be submitted until all required inputs have been processed. + */ + struct reducer_task_t : public parsec_ttg_task_base_t { + parsec_ttg_task_base_t *parent_task; + bool is_first; + + reducer_task_t(parsec_ttg_task_base_t* task, parsec_thread_mempool_t *mempool, + parsec_task_class_t *task_class, parsec_taskpool_t *taskpool, + int32_t priority, bool is_first) + : parsec_ttg_task_base_t(mempool, task_class, taskpool, priority, + 0, &release_task, + true /* deferred until other readers have completed */) + , parent_task(task) + , is_first(is_first) + { } + + static void release_task(parsec_ttg_task_base_t* task_base) { + /* reducer tasks have one mutable input so the task can be submitted on the first release */ + parsec_task_t *vp_task_rings[1] = { &task_base->parsec_task }; + parsec_execution_stream_t *es = ttg::default_execution_context().impl().execution_stream(); + __parsec_schedule_vp(es, vp_task_rings, 0); + } + }; + + inline ttg_data_copy_t *find_copy_in_task(parsec_ttg_task_base_t *task, const void *ptr) { ttg_data_copy_t *res = nullptr; if (task == nullptr || ptr == nullptr) { @@ -1345,61 +1375,56 @@ namespace ttg_parsec { parsec_ttg_caller = NULL; } - struct reducer_task_t { - parsec_task_t parsec_task; - task_t *task; - - reducer_task_t(task_t* task, parsec_thread_mempool_t *mempool, - parsec_task_class_t *task_class, parsec_taskpool_t *taskpool, - int32_t priority) - : task(task) - { - PARSEC_LIST_ITEM_SINGLETON(&parsec_task.super); - parsec_task.mempool_owner = mempool; - parsec_task.task_class = task_class; - parsec_task.status = PARSEC_TASK_STATUS_HOOK; - parsec_task.taskpool = taskpool; - parsec_task.priority = priority; - parsec_task.chore_mask = 1<<0; - } - }; - template static parsec_hook_return_t static_reducer_op(parsec_execution_stream_s *es, parsec_task_t *parsec_task) { - using rtask_t = reducer_task_t; + using rtask_t = detail::reducer_task_t; using value_t = std::tuple_element_t; constexpr const bool val_is_void = ttg::meta::is_void_v; rtask_t *rtask = (rtask_t*)parsec_task; - task_t *task = rtask->task; - ttT *baseobj = task->tt; + task_t *parent_task = static_cast(rtask->parent_task); + ttT *baseobj = parent_task->tt; derivedT *obj = static_cast(baseobj); auto& reducer = std::get(baseobj->input_reducers); - assert(parsec_ttg_caller == NULL); - parsec_ttg_caller = static_cast(task); - if (obj->tracing()) { if constexpr (!ttg::meta::is_void_v) - ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", task->key, ": reducer executing"); + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", parent_task->key, ": reducer executing"); else ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : reducer executing"); } /* the copy to reduce into */ detail::ttg_data_copy_t *target_copy; - target_copy = static_cast(task->parsec_task.data[i].data_in); + target_copy = static_cast(parent_task->parsec_task.data[i].data_in); assert(val_is_void || nullptr != target_copy); /* once we hit 0 we have to stop since another thread might enqueue a new reduction task */ std::size_t c = 0; std::size_t size = 0; - assert(task->streams[i].reduce_count > 0); + assert(parent_task->streams[i].reduce_count > 0); + if (rtask->is_first) { + if (0 == (parent_task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)-1)) { + /* we were the first and there is nothing to be done */ + if (obj->tracing()) { + if constexpr (!ttg::meta::is_void_v) + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", parent_task->key, ": first reducer empty"); + else + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : first reducer empty"); + } + + return PARSEC_HOOK_RETURN_DONE; + } + } + + assert(parsec_ttg_caller == NULL); + parsec_ttg_caller = rtask->parent_task; + do { if constexpr(!val_is_void) { /* the copies to reduce out of */ detail::ttg_data_copy_t *source_copy; parsec_list_item_t *item; - item = parsec_lifo_pop(&task->streams[i].reduce_copies); + item = parsec_lifo_pop(&parent_task->streams[i].reduce_copies); if (nullptr == item) { // maybe someone is changing the goal right now break; @@ -1412,26 +1437,28 @@ namespace ttg_parsec { reducer(); // invoke control reducer } // there is only one task working on this stream, so no need to be atomic here - size = ++task->streams[i].size; - //std::cout << "static_reducer_op key " << task->key << " size " << size << " of " << task->streams[i].goal << std::endl; - } while ((c = (task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)-1)) > 0); + size = ++parent_task->streams[i].size; + //std::cout << "static_reducer_op size " << size << " of " << parent_task->streams[i].goal << std::endl; + } while ((c = (parent_task->streams[i].reduce_count.fetch_sub(1, std::memory_order_acq_rel)-1)) > 0); //} while ((c = (--task->streams[i].reduce_count)) > 0); /* finalize_argstream sets goal to 1, so size may be larger than goal */ - bool complete = (size >= task->streams[i].goal); - + bool complete = (size >= parent_task->streams[i].goal); + //std::cout << "static_reducer_op size " << size + // << " of " << parent_task->streams[i].goal << " complete " << complete + // << " c " << c << std::endl; if (complete && c == 0) { /* task is still in the hash table, have release_task remove it */ - task->remove_from_hash = true; - task->release_task(task); + parent_task->remove_from_hash = true; + parent_task->release_task(parent_task); } parsec_ttg_caller = NULL; if (obj->tracing()) { if constexpr (!ttg::meta::is_void_v) - ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", task->key, ": done executing"); + ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : ", parent_task->key, ": done executing"); else ttg::trace(obj->get_world().rank(), ":", obj->get_name(), " : done executing"); } @@ -1811,12 +1838,12 @@ namespace ttg_parsec { template - reducer_task_t *create_new_reducer_task(task_t *task) { + detail::reducer_task_t *create_new_reducer_task(task_t *task, bool is_first) { /* make sure we can reuse the existing memory pool and don't have to create a new one */ - static_assert(sizeof(task_t) >= sizeof(reducer_task_t)); + static_assert(sizeof(task_t) >= sizeof(detail::reducer_task_t)); constexpr const bool keyT_is_Void = ttg::meta::is_void_v; auto &world_impl = world.impl(); - reducer_task_t *newtask; + detail::reducer_task_t *newtask; parsec_thread_mempool_t *mempool = get_task_mempool(); char *taskobj = (char *)parsec_thread_mempool_allocate(mempool); // use the priority of the task we stream into @@ -1829,8 +1856,8 @@ namespace ttg_parsec { ttg::trace(world.rank(), ":", get_name(), ": creating reducer task"); } /* placement-new the task */ - newtask = new (taskobj) reducer_task_t(task, mempool, inpute_reducers_taskclass[i], - world_impl.taskpool(), priority); + newtask = new (taskobj) detail::reducer_task_t(task, mempool, inpute_reducers_taskclass[i], + world_impl.taskpool(), priority, is_first); return newtask; } @@ -1921,20 +1948,30 @@ namespace ttg_parsec { #endif } - if (reducer) { // is this a streaming input? reduce the received value - auto submit_reducer_task = [&](auto *task){ + auto get_copy_fn = [&](detail::parsec_ttg_task_base_t *task, auto&& value, bool is_const){ + detail::ttg_data_copy_t *copy = copy_in; + if (nullptr != parsec_ttg_caller) { + copy = detail::find_copy_in_task(parsec_ttg_caller, &value); + } + if (nullptr != copy) { + /* retain the data copy */ + copy = detail::register_data_copy(copy, task, true); + } else { + /* create a new copy */ + copy = detail::create_new_datacopy(std::forward(value)); + } + return copy; + }; + + if (reducer && 1 != task->streams[i].goal) { // is this a streaming input? reduce the received value + auto submit_reducer_task = [&](auto *parent_task){ /* check if we need to create a task */ - std::size_t c = task->streams[i].reduce_count.fetch_add(1, std::memory_order_release); - //std::size_t c = task->streams[i].reduce_count++; + std::size_t c = parent_task->streams[i].reduce_count.fetch_add(1, std::memory_order_release); if (0 == c) { - //std::cout << "submit reducer task for task " << task->key - // << " size " << task->streams[i].size << " of " << task->streams[i].goal << std::endl; /* we are responsible for creating the reduction task */ - reducer_task_t *reduce_task; - reduce_task = create_new_reducer_task(task); - parsec_task_t *vp_task_rings[1] = { &reduce_task->parsec_task }; - parsec_execution_stream_t *es = world_impl.execution_stream(); - __parsec_schedule_vp(es, vp_task_rings, 0); + detail::reducer_task_t *reduce_task; + reduce_task = create_new_reducer_task(parent_task, false); + reduce_task->release_task(reduce_task); // release immediately } }; @@ -1942,29 +1979,34 @@ namespace ttg_parsec { // have a value already? if not, set, otherwise reduce if (nullptr == static_cast(task->parsec_task.data[i].data_in)) { using decay_valueT = std::decay_t; - /* For now, we always create a copy because we cannot rely on the task_release - * mechanism (it would release the task, not the reduction value). */ - task->parsec_task.data[i].data_in = detail::create_new_datacopy(std::forward(value)); - task->streams[i].size++; - if (task->streams[i].size == task->streams[i].goal) { - release = true; + + /* first input value, create a task and bind it to the copy */ + detail::reducer_task_t *reduce_task; + reduce_task = create_new_reducer_task(task, true); + + /* get the copy to use as input for this task */ + detail::ttg_data_copy_t *copy = get_copy_fn(reduce_task, std::forward(value), false); + + /* put the copy into the task */ + task->parsec_task.data[i].data_in = copy; + + /* protected by the bucket lock */ + task->streams[i].size = 1; + task->streams[i].reduce_count.store(1, std::memory_order_relaxed); + + if (copy->push_task != &reduce_task->parsec_task) { + reduce_task->release_task(reduce_task); } + /* now we can unlock the bucket */ parsec_hash_table_unlock_bucket(&tasks_table, hk); } else { - detail::ttg_data_copy_t *copy = nullptr; /* unlock the bucket, the lock is not needed anymore */ parsec_hash_table_unlock_bucket(&tasks_table, hk); - if (nullptr != parsec_ttg_caller) { - copy = detail::find_copy_in_task(parsec_ttg_caller, &value); - } - if (nullptr != copy) { - /* retain the data copy */ - copy = detail::register_data_copy(copy, task, true); - } else { - /* create a new copy */ - copy = detail::create_new_datacopy(std::forward(value)); - } + + /* get the copy to use as input for this task */ + detail::ttg_data_copy_t *copy = get_copy_fn(task, std::forward(value), true); + /* enqueue the data copy to be reduced */ parsec_lifo_push(&task->streams[i].reduce_copies, ©->super); submit_reducer_task(task); @@ -1988,17 +2030,9 @@ namespace ttg_parsec { throw std::logic_error("bad set arg"); } - detail::ttg_data_copy_t *copy = copy_in; - if (nullptr == copy_in && nullptr != parsec_ttg_caller) { - copy = detail::find_copy_in_task(parsec_ttg_caller, &value); - } + /* get the copy to use as input for this task */ + detail::ttg_data_copy_t *copy = get_copy_fn(task, std::forward(value), input_is_const); - if (nullptr != copy) { - /* register_data_copy might provide us with a different copy if !input_is_const */ - copy = detail::register_data_copy(copy, task, input_is_const); - } else { - copy = detail::create_new_datacopy(std::forward(value)); - } /* if we registered as a writer and were the first to register with this copy * we need to defer the release of this task to give other tasks a chance to * make a copy of the original data */ From 68a985211e62c9a847f09fc43f6265141be64775 Mon Sep 17 00:00:00 2001 From: Joseph Schuchart Date: Fri, 7 Jul 2023 12:10:47 -0400 Subject: [PATCH 10/10] Add ttg::execute calls to streams/fib tests Signed-off-by: Joseph Schuchart --- tests/unit/fibonacci.cc | 16 ++++++++++------ tests/unit/streams.cc | 17 +++++++++-------- 2 files changed, 19 insertions(+), 14 deletions(-) diff --git a/tests/unit/fibonacci.cc b/tests/unit/fibonacci.cc index 574661b338..3d335c3d90 100644 --- a/tests/unit/fibonacci.cc +++ b/tests/unit/fibonacci.cc @@ -30,6 +30,7 @@ TEST_CASE("Fibonacci", "[fib][core]") { if (ttg::default_execution_context().size() == 1) { ttg::Edge F2F; ttg::Edge F2P; + auto world = ttg::default_execution_context(); auto fib_op = ttg::make_tt( // computes next value: F_{n+2} = F_{n+1} + F_{n}, seeded by F_1 = 1, F_0 = 0 @@ -62,8 +63,9 @@ TEST_CASE("Fibonacci", "[fib][core]") { ttg::edges(F2P), ttg::edges()); print_op->set_input_reducer<0>([](int &a, const int &b) { a = a + b; }); make_graph_executable(fib_op); - if (ttg::default_execution_context().rank() == 0) fib_op->invoke(1, 0); - ttg::ttg_fence(ttg::default_execution_context()); + if (world.rank() == 0) fib_op->invoke(1, 0); + ttg::execute(world); + ttg::ttg_fence(world); } } @@ -71,7 +73,8 @@ TEST_CASE("Fibonacci", "[fib][core]") { SECTION("distributed-memory") { ttg::Edge> F2F; ttg::Edge F2P; - const auto nranks = ttg::default_execution_context().size(); + auto world = ttg::default_execution_context(); + const auto nranks = world.size(); auto fib_op = ttg::make_tt( // computes next value: F_{n+2} = F_{n+1} + F_{n}, seeded by F_1 = 1, F_0 = 0 @@ -104,8 +107,9 @@ TEST_CASE("Fibonacci", "[fib][core]") { a = a + b; }); make_graph_executable(fib_op); - ttg::ttg_fence(ttg::default_execution_context()); - if (ttg::default_execution_context().rank() == 0) fib_op->invoke(0, std::make_pair(1, 0)); - ttg::ttg_fence(ttg::default_execution_context()); + ttg::ttg_fence(world); + if (world.rank() == 0) fib_op->invoke(0, std::make_pair(1, 0)); + ttg::execute(world); + ttg::ttg_fence(world); } } // TEST_CAST("Fibonacci") diff --git a/tests/unit/streams.cc b/tests/unit/streams.cc index 108c8610b9..355276cedb 100644 --- a/tests/unit/streams.cc +++ b/tests/unit/streams.cc @@ -13,10 +13,11 @@ TEST_CASE("streams", "[streams][core]") { SECTION("concurrent-stream-size") { ttg::Edge I2O; ttg::Edge O2S; - const auto nranks = ttg::default_execution_context().size(); + auto world = ttg::default_execution_context(); + const auto nranks = world.size(); - constexpr std::size_t N = 10000; - constexpr std::size_t SLICE = 500; + constexpr std::size_t SLICE = 20; + std::size_t N = SLICE * 2 * world.size(); constexpr const timespec ts = { .tv_sec = 0, .tv_nsec = 10000 }; constexpr int VALUE = 1; std::atomic reduce_ops = 0; @@ -33,7 +34,6 @@ TEST_CASE("streams", "[streams][core]") { // set the size of the last reducer if (N%SLICE > 0) { ttg::set_size<0>(key, N%SLICE, outs); - std::cout << "set_size key " << key << " size " << N%SLICE << std::endl; } // forward the value ttg::send<0>(key, std::forward(i), outs); @@ -62,14 +62,15 @@ TEST_CASE("streams", "[streams][core]") { }, SLICE); make_graph_executable(op); - ttg::ttg_fence(ttg::default_execution_context()); - if (ttg::default_execution_context().rank() == 0) { + ttg::ttg_fence(world); + if (world.rank() == 0) { for (std::size_t i = 0; i < N; ++i) { op->invoke(i, VALUE); } } - ttg::ttg_fence(ttg::default_execution_context()); - CHECK(reduce_ops == N); + ttg::execute(world); + ttg::ttg_fence(world); + CHECK(reduce_ops == (N/world.size())); } } // TEST_CASE("streams") \ No newline at end of file