#874

Committed 25 Jan 2023 02:55PM UTC coverage: 85.911% (+0.03%) from 85.883%

Build # #874

Build Type

push

Committed by StellarBot

Commit Message

Merge #6153

6153: Assign global ids to managed component instances that are locally resolvable r=hkaiser a=hkaiser

This will reduce contention while resolving global ids.

Co-authored-by: Hartmut Kaiser <hartmut.kaiser@gmail.com>

Run Details

61 of 61 new or added lines in 5 files covered. (100.0%)

173513 of 201968 relevant lines covered (85.91%)

2065249.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.15

/libs/full/agas/src/route.cpp

//  Copyright (c) 2011 Vinay C Amatya
//  Copyright (c) 2007-2021 Hartmut Kaiser
//
//  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/plain_action.hpp>
#include <hpx/agas/addressing_service.hpp>
#include <hpx/agas_base/server/primary_namespace.hpp>
#include <hpx/assert.hpp>
#include <hpx/async_distributed/continuation.hpp>
#include <hpx/async_distributed/detail/post.hpp>
#include <hpx/components_base/agas_interface.hpp>
#include <hpx/parcelset/parcel.hpp>
#include <hpx/parcelset_base/detail/parcel_route_handler.hpp>
#include <hpx/runtime_local/runtime_local.hpp>
#include <hpx/timing/scoped_timer.hpp>

#include <atomic>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <utility>

namespace hpx::detail {

    void update_agas_cache(hpx::naming::gid_type const& gid,
        hpx::naming::address const& addr, std::uint64_t count,
        std::uint64_t offset)
    {
        hpx::agas::update_cache_entry(gid, addr, count, offset);
    }
}    // namespace hpx::detail

HPX_PLAIN_ACTION_ID(hpx::detail::update_agas_cache, update_agas_cache_action,
    hpx::actions::update_agas_cache_action_id)

namespace hpx::agas::server {

    void primary_namespace::route(parcelset::parcel&& p)
    {
        util::scoped_timer<std::atomic<std::int64_t>> update(
            counter_data_.route_.time_, counter_data_.route_.enabled_);
        counter_data_.increment_route_count();

        naming::gid_type const& gid = p.destination();
        naming::address& addr = p.addr();
        resolved_type cache_address;

        // resolve destination addresses, we should be able to resolve all of
        // them, otherwise it's an error
        {
            std::unique_lock<mutex_type> l(mutex_);

            error_code& ec = throws;

            // wait for any migration to be completed
            if (naming::detail::is_migratable(gid))
            {
                wait_for_migration_locked(l, gid, ec);
            }

            cache_address = resolve_gid_locked(l, gid, ec);

            if (ec || hpx::get<0>(cache_address) == naming::invalid_gid)
            {
                l.unlock();

                HPX_THROWS_IF(ec, hpx::error::no_success,
                    "primary_namespace::route",
                    "can't route parcel to unknown gid: {}", gid);

                return;
            }

            // retain don't store in cache flag
            if (!naming::detail::store_in_cache(gid))
            {
                naming::detail::set_dont_store_in_cache(
                    hpx::get<0>(cache_address));
            }

            gva const g = hpx::get<1>(cache_address)
                              .resolve(gid, hpx::get<0>(cache_address));

            addr.locality_ = g.prefix;
            addr.type_ = g.type;
            addr.address_ = g.lva();
        }

        hpx::id_type source = p.source_id();

        // either send the parcel on its way or execute actions locally
        if (naming::get_locality_id_from_gid(addr.locality_) ==
            agas::get_locality_id())
        {
            // destination is local
            if (p.schedule_action())
            {
                // object was migrated, route again
                agas::route(HPX_MOVE(p),
                    &hpx::parcelset::detail::parcel_route_handler,
                    threads::thread_priority::normal);
            }
        }
        else
        {
            // destination is remote
            hpx::parcelset::put_parcel(HPX_MOVE(p));
        }

        runtime& rt = get_runtime();
        if (rt.get_state() < hpx::state::pre_shutdown)
        {
            // asynchronously update cache on source locality
            // update remote cache if the id is not flagged otherwise
            naming::gid_type const& id = hpx::get<0>(cache_address);
            if (id && naming::detail::store_in_cache(id))
            {
                gva const& g = hpx::get<1>(cache_address);
                naming::address addr(g.prefix, g.type, g.lva());

                HPX_ASSERT(naming::is_locality(source));

                hpx::post<update_agas_cache_action>(
                    source, id, addr, g.count, g.offset);
            }
        }
    }
}    // namespace hpx::agas::server

#endif

1	// Copyright (c) 2011 Vinay C Amatya
2	// Copyright (c) 2007-2021 Hartmut Kaiser
3	//
4	// SPDX-License-Identifier: BSL-1.0
5	// Distributed under the Boost Software License, Version 1.0. (See accompanying
6	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7
8	#include <hpx/config.hpp>
9
10	#if defined(HPX_HAVE_NETWORKING)
11	#include <hpx/actions_base/plain_action.hpp>
12	#include <hpx/agas/addressing_service.hpp>
13	#include <hpx/agas_base/server/primary_namespace.hpp>
14	#include <hpx/assert.hpp>
15	#include <hpx/async_distributed/continuation.hpp>
16	#include <hpx/async_distributed/detail/post.hpp>
17	#include <hpx/components_base/agas_interface.hpp>
18	#include <hpx/parcelset/parcel.hpp>
19	#include <hpx/parcelset_base/detail/parcel_route_handler.hpp>
20	#include <hpx/runtime_local/runtime_local.hpp>
21	#include <hpx/timing/scoped_timer.hpp>
22
23	#include <atomic>
24	#include <cstddef>
25	#include <cstdint>
26	#include <mutex>
27	#include <utility>
28
29	namespace hpx::detail {
30
31	void update_agas_cache(hpx::naming::gid_type const& gid,	×
32	hpx::naming::address const& addr, std::uint64_t count,
33	std::uint64_t offset)
34	{
35	hpx::agas::update_cache_entry(gid, addr, count, offset);	×
36	}	×
37	} // namespace hpx::detail
38
39	HPX_PLAIN_ACTION_ID(hpx::detail::update_agas_cache, update_agas_cache_action,	4,648✔
40	hpx::actions::update_agas_cache_action_id)
41
42	namespace hpx::agas::server {
43
44	void primary_namespace::route(parcelset::parcel&& p)	3,142✔
45	{
46	util::scoped_timer<std::atomic<std::int64_t>> update(	3,142✔
47	counter_data_.route_.time_, counter_data_.route_.enabled_);	3,142✔
48	counter_data_.increment_route_count();	3,142✔
49
50	naming::gid_type const& gid = p.destination();	3,142✔
51	naming::address& addr = p.addr();	3,142✔
52	resolved_type cache_address;	3,142✔
53
54	// resolve destination addresses, we should be able to resolve all of
55	// them, otherwise it's an error
56	{
57	std::unique_lock<mutex_type> l(mutex_);	3,142✔
58
59	error_code& ec = throws;	3,142✔
60
61	// wait for any migration to be completed
62	if (naming::detail::is_migratable(gid))	3,142✔
63	{
64	wait_for_migration_locked(l, gid, ec);	3,142✔
65	}	3,142✔
66
67	cache_address = resolve_gid_locked(l, gid, ec);	3,142✔
68
69	if (ec \|\| hpx::get<0>(cache_address) == naming::invalid_gid)	3,142✔
70	{
71	l.unlock();	×
72
73	HPX_THROWS_IF(ec, hpx::error::no_success,	×
74	"primary_namespace::route",
75	"can't route parcel to unknown gid: {}", gid);
76
77	return;	×
78	}
79
80	// retain don't store in cache flag
81	if (!naming::detail::store_in_cache(gid))	3,142✔
82	{
83	naming::detail::set_dont_store_in_cache(	3,142✔
84	hpx::get<0>(cache_address));	3,142✔
85	}	3,142✔
86
87	gva const g = hpx::get<1>(cache_address)	3,142✔
88	.resolve(gid, hpx::get<0>(cache_address));	3,142✔
89
90	addr.locality_ = g.prefix;	3,142✔
91	addr.type_ = g.type;	3,142✔
92	addr.address_ = g.lva();	3,142✔
93	}	3,142✔
94
95	hpx::id_type source = p.source_id();	3,142✔
96
97	// either send the parcel on its way or execute actions locally
98	if (naming::get_locality_id_from_gid(addr.locality_) ==	3,142✔
99	agas::get_locality_id())	3,142✔
100	{
101	// destination is local
102	if (p.schedule_action())	1,268✔
103	{
104	// object was migrated, route again
105	agas::route(HPX_MOVE(p),	×
106	&hpx::parcelset::detail::parcel_route_handler,	×
107	threads::thread_priority::normal);
108	}	×
109	}	1,268✔
110	else
111	{
112	// destination is remote
113	hpx::parcelset::put_parcel(HPX_MOVE(p));	1,874✔
114	}
115
116	runtime& rt = get_runtime();	3,142✔
117	if (rt.get_state() < hpx::state::pre_shutdown)	3,142✔
118	{
119	// asynchronously update cache on source locality
120	// update remote cache if the id is not flagged otherwise
121	naming::gid_type const& id = hpx::get<0>(cache_address);	3,142✔
122	if (id && naming::detail::store_in_cache(id))	3,142✔
123	{
124	gva const& g = hpx::get<1>(cache_address);	×
125	naming::address addr(g.prefix, g.type, g.lva());	×
126
127	HPX_ASSERT(naming::is_locality(source));	×
128
129	hpx::post<update_agas_cache_action>(	×
130	source, id, addr, g.count, g.offset);	×
131	}	×
132	}	3,142✔
133	}	3,142✔
134	} // namespace hpx::agas::server
135
136	#endif

STEllAR-GROUP / hpx / #874

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous