STEllAR-GROUP / hpx / #882

//  Copyright (c) 2011 Bryce Lelbach & Katelyn Kufahl

//  Copyright (c) 2007-2025 Hartmut Kaiser

//  Copyright (c) 2015 Anton Bikineev

//

//  SPDX-License-Identifier: BSL-1.0

//  Distributed under the Boost Software License, Version 1.0. (See accompanying

//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

#include <hpx/config.hpp>

#if defined(HPX_HAVE_NETWORKING)

#include <hpx/actions_base/actions_base_support.hpp>

#include <hpx/actions_base/plain_action.hpp>

#include <hpx/agas/addressing_service.hpp>

#include <hpx/agas_base/detail/hosted_component_namespace.hpp>

#include <hpx/agas_base/detail/hosted_locality_namespace.hpp>

#include <hpx/assert.hpp>

#include <hpx/async_distributed/put_parcel.hpp>

#include <hpx/components_base/agas_interface.hpp>

#include <hpx/components_base/server/managed_component_base.hpp>

#include <hpx/modules/agas_base.hpp>

#include <hpx/modules/errors.hpp>

#include <hpx/modules/execution_base.hpp>

#include <hpx/modules/format.hpp>

#include <hpx/modules/functional.hpp>

#include <hpx/modules/runtime_configuration.hpp>

#include <hpx/modules/serialization.hpp>

#include <hpx/modules/static_reinit.hpp>

#include <hpx/modules/timing.hpp>

#include <hpx/modules/topology.hpp>

#include <hpx/parcelset/detail/parcel_await.hpp>

#include <hpx/parcelset_base/parcel_interface.hpp>

#include <hpx/parcelset_base/parcelport.hpp>

#include <hpx/runtime_distributed.hpp>

#include <hpx/runtime_distributed/big_boot_barrier.hpp>

#include <hpx/runtime_distributed/runtime_fwd.hpp>

#include <cstddef>

#include <cstdint>

#include <cstdlib>

#include <functional>

#include <memory>

#include <mutex>

#include <random>

#include <string>

#include <type_traits>

#include <utility>

#include <vector>

#include <hpx/config/warnings_prefix.hpp>

namespace hpx::parcelset {

    // shortcut for get_runtime().get_parcel_handler()

    parcelhandler& get_parcel_handler();

}    // namespace hpx::parcelset

namespace hpx::agas::detail {

    void register_unassigned_typenames()

    {

        // supposed to be run on locality 0 locality communication

        hpx::serialization::detail::id_registry& serialization_registry =

        serialization_registry.fill_missing_typenames();

        hpx::actions::detail::action_registry& action_registry =

        action_registry.fill_missing_typenames();

    ///////////////////////////////////////////////////////////////////////////

        explicit unassigned_typename_sequence(bool /*dummy*/)

                hpx::serialization::detail::id_registry::instance()

                    .get_unassigned_typenames())

          , action_typenames(hpx::actions::detail::action_registry::instance()

        {

            // part running on worker node

            ar << serialization_typenames;

            ar << action_typenames;

        }

        void load(hpx::serialization::input_archive& ar, unsigned)

            ar >> serialization_typenames;

            ar >> action_typenames;

        }

        HPX_SERIALIZATION_SPLIT_MEMBER();

    };

    ///////////////////////////////////////////////////////////////////////////

    struct assigned_id_sequence

    {

        assigned_id_sequence() = default;

        explicit assigned_id_sequence(

            unassigned_typename_sequence const& typenames)

        {

            register_ids_on_main_loc(typenames);

        }

        void save(hpx::serialization::output_archive& ar, unsigned) const

            ar << action_ids;

        }

        void load(hpx::serialization::input_archive& ar, unsigned)

            ar >> action_ids;

        }

        HPX_SERIALIZATION_SPLIT_MEMBER();

            unassigned_typename_sequence const& unassigned_ids)

        {

            {

                hpx::serialization::detail::id_registry& registry =

                std::uint32_t max_id = registry.get_max_registered_id();

                for (std::string const& s :

                    unassigned_ids.serialization_typenames)

                    std::uint32_t id = registry.try_get_id(s);

                    if (id ==

                        // this id is not registered yet

                    }

                    serialization_ids.push_back(id);

                }

                hpx::actions::detail::action_registry& registry =

                std::uint32_t max_id = registry.max_id_;

                for (std::string const& s : unassigned_ids.action_typenames)

                {

                    {

                        id = ++max_id;

                    action_ids.push_back(id);

                }

        void register_ids_on_worker_loc() const

        {

                hpx::serialization::detail::id_registry& registry =

                    hpx::serialization::detail::id_registry::instance();

                // Yes, we look up the unassigned typenames twice, but this allows

                // to avoid using globals and protects from race conditions during

                // de-serialization.

                    registry.get_unassigned_typenames();

                // we should have received as many ids as we have unassigned names

                HPX_ASSERT(typenames.size() == serialization_ids.size());

                {

                    registry.register_typename(

                        typenames[k], serialization_ids[k]);

                }

                // fill in holes which might have been caused by initialization

                registry.fill_missing_typenames();

            }

            {

                hpx::actions::detail::action_registry& registry =

                    hpx::actions::detail::action_registry::instance();

                // to avoid using globals and protects from race conditions during

                // de-serialization.

                    registry.get_unassigned_typenames();

                // we should have received as many ids as we have unassigned names

                HPX_ASSERT(typenames.size() == action_ids.size());

                {

                    registry.register_typename(typenames[k], action_ids[k]);

                }

                // order problems

            }

        }

        std::vector<std::uint32_t> serialization_ids;

namespace hpx::agas {

    template <typename Action, typename... Args>

    void big_boot_barrier::apply(std::uint32_t source_locality_id,

        std::uint32_t target_locality_id, parcelset::locality dest, Action act,

        Args&&... args)

    {    // {{{

        HPX_ASSERT(pp);

            naming::get_gid_from_locality_id(target_locality_id));

        parcelset::parcel p(parcelset::detail::create_parcel::call(

            naming::get_gid_from_locality_id(target_locality_id),

            HPX_MOVE(addr), act, HPX_FORWARD(Args, args)...));

#if defined(HPX_HAVE_PARCEL_PROFILING)

        if (!p.parcel_id())

            p.parcel_id() =

                parcelset::parcel::generate_unique_id(source_locality_id);

        }

#else

        HPX_UNUSED(source_locality_id);

#endif

        parcelset::detail::parcel_await_apply(HPX_MOVE(p),

            parcelset::write_handler_type(), 0,

            [this, dest](

                parcelset::parcel&& p, parcelset::write_handler_type&&) {

                pp->send_early_parcel(dest, HPX_MOVE(p));

    template <typename Action, typename... Args>

        ,

        ,

        Action act, Args&&... args)

        naming::address addr(

            naming::get_gid_from_locality_id(target_locality_id));

        parcelset::put_parcel(

            hpx::id_type(naming::get_gid_from_locality_id(target_locality_id),

                hpx::id_type::management_type::unmanaged),

            HPX_MOVE(addr), act, HPX_FORWARD(Args, args)...);

    }    // }}}

    //    lcos::detail::promise<

    //        response

    //    >

    //  , components::managed_component<

    //        lcos::detail::promise<

    //            response

    //          , response

    //        >

    //    >

    //> response_heap_type;

    // This structure is used when a locality registers with node zero

    // (first round trip)

    struct registration_header

    {

        registration_header()

          : primary_ns_ptr(nullptr)

          , symbol_ns_ptr(nullptr)

          , cores_needed(0)

          , num_threads(0)

        {

            naming::address_type primary_ns_ptr_,

            naming::address_type symbol_ns_ptr_, std::uint32_t cores_needed_,

            std::uint32_t num_threads_, std::string const& hostname_,

            detail::unassigned_typename_sequence const& typenames_,

          : endpoints(endpoints_)

          , primary_ns_ptr(primary_ns_ptr_)

          , symbol_ns_ptr(symbol_ns_ptr_)

          , cores_needed(cores_needed_)

          , num_threads(num_threads_)

        naming::address_type primary_ns_ptr;

        naming::address_type symbol_ns_ptr;

        std::uint32_t num_threads;

        std::string hostname;    // hostname of locality

        detail::unassigned_typename_sequence typenames;

        naming::gid_type prefix;    // suggested prefix (optional)

        template <typename Archive>

        void serialize(Archive& ar, unsigned int const)

        {

            // clang-format off

            ar & endpoints;

            ar & address;

            primary_ns_ptr = reinterpret_cast<naming::address_type>(address);

            address = reinterpret_cast<std::size_t>(symbol_ns_ptr);

            symbol_ns_ptr = reinterpret_cast<naming::address_type>(address);

            ar & cores_needed;

            ar & hostname;

            ar & prefix;

            // clang-format on

        }

    // is trying to register (first roundtrip).

    {

        notification_header()

          : num_localities(0)

          , used_cores(0)

        {

        }

            naming::address const& locality_ns_address_,

            naming::address const& component_ns_address_,

            std::uint32_t num_localities_, std::uint32_t used_cores_,

            parcelset::endpoints_type const& agas_endpoints_,

            detail::assigned_id_sequence const& ids_)

          : prefix(prefix_)

          , agas_locality(agas_locality_)

          , locality_ns_address(locality_ns_address_)

          , primary_ns_address(primary_ns_address_)

          , component_ns_address(component_ns_address_)

          , used_cores(used_cores_)

          , agas_endpoints(agas_endpoints_)

          , ids(ids_)

        {

        naming::gid_type prefix;

        naming::address locality_ns_address;

        naming::address component_ns_address;

        naming::address symbol_ns_address;

        std::uint32_t num_localities;

        std::uint32_t used_cores;

        parcelset::endpoints_type agas_endpoints;

        detail::assigned_id_sequence ids;

        std::vector<parcelset::endpoints_type> endpoints;

        template <typename Archive>

        void serialize(Archive& ar, unsigned int const)

        {

            // clang-format off

            ar & prefix;

            ar & agas_locality;

            ar & primary_ns_address;

            ar & component_ns_address;

            // clang-format on

        }

    void register_worker(registration_header const& header);

    // }}}

    // {{{ early action types

    using register_worker_action =

        actions::direct_action<void (*)(registration_header const&),

            register_worker>;

    using notify_worker_action =

        actions::direct_action<void (*)(notification_header const&),

            notify_worker>;

    // }}}

}    // namespace hpx::agas

using hpx::agas::notify_worker_action;

using hpx::agas::register_worker_action;

HPX_ACTION_HAS_CRITICAL_PRIORITY(register_worker_action)

HPX_ACTION_HAS_CRITICAL_PRIORITY(notify_worker_action)

HPX_REGISTER_ACTION_ID(register_worker_action, register_worker_action,

    hpx::actions::register_worker_action_id)

HPX_REGISTER_ACTION_ID(notify_worker_action, notify_worker_action,

    hpx::actions::notify_worker_action_id)

    void register_worker(registration_header const& header)

    {

        // This lock acquires the bbb mutex on creation. When it goes out of scope,

        // its dtor calls big_boot_barrier::notify().

        big_boot_barrier::scoped_lock lock(get_big_boot_barrier());

        agas::addressing_service& agas_client = naming::get_agas_client();

        if (HPX_UNLIKELY(agas_client.is_connecting()))

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "a locality in connect mode cannot be an AGAS server.");

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "agas::register_worker",

                "registration parcel received by non-bootstrap locality.");

        if (prefix != naming::invalid_gid &&

            naming::get_locality_id_from_gid(prefix) == 0)

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "agas::register_worker",

                "this is reserved for the console",

                header.endpoints);

        }

        if (!agas_client.register_locality(

                header.endpoints, prefix, header.num_threads))

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "agas::register_worker",

        }

        naming::address locality_addr(agas::get_locality(),

            to_int(components::component_enum_type::agas_locality_namespace),

            agas_client.locality_ns_->ptr());

        naming::address primary_addr(agas::get_locality(),

            to_int(components::component_enum_type::agas_primary_namespace),

            agas_client.primary_ns_.ptr());

        naming::address component_addr(agas::get_locality(),

            to_int(components::component_enum_type::agas_component_namespace),

        naming::address symbol_addr(agas::get_locality(),

            to_int(components::component_enum_type::agas_symbol_namespace),

        // assign cores to the new locality

        std::uint32_t first_core =

            rt.assign_cores(header.hostname, header.cores_needed);

        big_boot_barrier& bbb = get_big_boot_barrier();

        detail::assigned_id_sequence assigned_ids(header.typenames);

        notification_header hdr(prefix, bbb.here(), locality_addr, primary_addr,

            first_core, bbb.get_endpoints(), assigned_ids);

        parcelset::locality here = bbb.here();

            if (loc.second.type() == here.type())

                dest = loc.second;

        }

        // collect endpoints from all registering localities

            naming::get_locality_id_from_gid(prefix), header.endpoints);

        // TODO: Handle cases where localities try to connect to AGAS while it's

        // shutting down.

        if (agas_client.get_status() != hpx::state::starting)

            // We can just send the parcel now, the connecting locality isn't a part

            // of startup synchronization.

            get_big_boot_barrier().apply_late(0,

                naming::get_locality_id_from_gid(prefix), dest,

        }

        else

            // AGAS is starting up; this locality is participating in startup

            // synchronization.

            // delay the final response until the runtime system is up and running

            hpx::move_only_function<void()>* thunk =

                new hpx::move_only_function<void()>(util::one_shot(

                    hpx::bind_front(&big_boot_barrier::apply_notification,

                        &get_big_boot_barrier(), 0,

                        naming::get_locality_id_from_gid(prefix), dest,

                        HPX_MOVE(hdr))));

            get_big_boot_barrier().add_thunk(thunk);

    // AGAS callback to client (first round trip response)

        // This lock acquires the bbb mutex on creation. When it goes out of scope,

        big_boot_barrier::scoped_lock lock(get_big_boot_barrier());

        header.ids.register_ids_on_worker_loc();

        runtime_distributed& rt = get_runtime_distributed();

        if (HPX_UNLIKELY(agas_client.get_status() != hpx::state::starting))

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

        }

        util::runtime_configuration& cfg = rt.get_config();

        // set our prefix

        agas_client.set_local_locality(header.prefix);

        agas_client.register_console(header.agas_endpoints);

        cfg.parse("assigned locality",

                naming::get_locality_id_from_gid(header.prefix)));

        naming::gid_type const& here = agas::get_locality();

        // register runtime support component

            header.prefix.get_msb(), rt.get_runtime_support_lva());

            components::get_component_type<

                components::server::runtime_support>(),

            rt.get_runtime_support_lva());

        agas_client.bind_local(runtime_support_gid, runtime_support_address);

        runtime_support_gid.set_lsb(std::uint64_t(0));

        agas_client.bind_local(runtime_support_gid, runtime_support_address);

        rt.init_id_pool_range();

        // store number of initial localities

        cfg.set_num_localities(header.num_localities);

        // store number of used cores by other localities

        rt.assign_cores();

        agas_client.pre_cache_endpoints(header.endpoints);

    }

    void big_boot_barrier::apply_notification(std::uint32_t source_locality_id,

        std::uint32_t target_locality_id, parcelset::locality const& dest,

        hdr.endpoints = localities;

        apply(source_locality_id, target_locality_id, dest,

    }

    void big_boot_barrier::add_locality_endpoints(std::uint32_t locality_id,

        if (localities.size() < static_cast<std::size_t>(locality_id) + 1)

    }

    {

        std::unique_lock<std::mutex> lock(mtx);

        {

            cond.wait(lock);    //-V1089

        {

            agas_client.pre_cache_endpoints(localities);

        }

    }

    inline std::size_t get_number_of_bootstrap_connections(

        service_mode service_type = ini.get_agas_service_mode();

        {

            std::size_t num_localities =

            result = num_localities ? num_localities - 1 : 0;

        }

        return result;

    }

        parcelset::endpoints_type const& endpoints_,

        util::runtime_configuration const& ini_)

      , endpoints(endpoints_)

      , service_type(ini_.get_agas_service_mode())

      , cond()

            add_locality_endpoints(0, get_endpoints());

        }

    {    // {{{

        spin();

    namespace detail {

        std::uint32_t get_number_of_pus_in_cores(std::uint32_t num_cores)

            std::uint32_t num_pus = 0;

            for (std::uint32_t i = 0; i != num_cores; ++i)

                std::uint32_t num_pus_core = static_cast<std::uint32_t>(

                if (num_pus_core == ~std::uint32_t(0))

                    return num_cores;    // assume one pu per core

                num_pus += num_pus_core;

            }

        }

    }    // namespace detail

    void big_boot_barrier::wait_hosted(std::string const& locality_name,

        naming::address::address_type primary_ns_server,

        naming::address::address_type symbol_ns_server)

    {    // {{{

        HPX_ASSERT(service_mode::bootstrap != service_type);

        HPX_ASSERT(nullptr != primary_ns_server);

        HPX_ASSERT(nullptr != symbol_ns_server);

        runtime& rt = get_runtime_distributed();

        // get the number of cores we need for our locality. This respects the

        // affinity description. Cores that are partially used are counted as well

        std::uint32_t cores_needed = rt.assign_cores();

        std::uint32_t num_threads =

        naming::gid_type suggested_prefix;

            rt.get_config().get_entry("hpx.locality", "-1");

        {

            suggested_prefix = naming::get_gid_from_locality_id(

                util::from_string<std::uint32_t>(locality_str, -1));

        }

        detail::unassigned_typename_sequence unassigned(true);

        registration_header hdr(parcelset::get_parcel_handler().endpoints(),

            primary_ns_server, symbol_ns_server, cores_needed, num_threads,

            locality_name, unassigned, suggested_prefix);

        // random first parcel id

        apply(static_cast<std::uint32_t>(std::random_device{}()), 0,

        spin();

    }    // }}}

#if defined(HPX_MSVC)

#pragma warning(push)

    void big_boot_barrier::notify()

        agas::addressing_service& agas_client = naming::get_agas_client();

        bool notify = false;

        {

            std::lock_guard<std::mutex> lk(mtx, std::adopt_lock);

                --connected;

            }

        }

        if (notify)

#pragma warning(pop)

#endif

    // This is triggered in runtime_impl::start, after the early action handler

    // until this point so that the AGAS locality can come up.

    {

        if (service_mode::bootstrap == service_type)

        {

            while (thunks.pop(p))

            {

                {

                }

                }

                delete p;

        }

    void big_boot_barrier::add_thunk(hpx::move_only_function<void()>* f)

        std::size_t k = 0;

        while (!thunks.push(f))

        {

            // Wait until successfully pushed ...

                k, "hpx::agas::big_boot_barrier::add_thunk");

        }

    }

    ///////////////////////////////////////////////////////////////////////////////

    void create_big_boot_barrier(parcelset::parcelport* pp_,

        parcelset::endpoints_type const& endpoints_,

        util::runtime_configuration const& ini_)

    {

            bbb;

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "create_big_boot_barrier",

        bbb.get().reset(new big_boot_barrier(pp_, endpoints_, ini_));

    void destroy_big_boot_barrier()

    {

            bbb;

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "destroy_big_boot_barrier",

                "big_boot_barrier has not been created yet");

        bbb.get().reset();

    big_boot_barrier& get_big_boot_barrier()

    {

            bbb;

        {

            HPX_THROW_EXCEPTION(hpx::error::internal_server_error,

                "get_big_boot_barrier",

        }

        return *(bbb.get());

    }

}    // namespace hpx::agas

#endif