#868

Committed 16 Jan 2023 08:21PM UTC coverage: 86.487%. Remained the same

Build # #868

Build Type

push

Committed by StellarBot

Commit Message

Merge #6137

6137: Adding example of a simple master/slave distributed application r=hkaiser a=hkaiser

The purpose of this example is to demonstrate how HPX actions can be used to build a simple master-slave application. The master (locality 0) assigns work to the slaves (all other localities). Note that if this application is run on one locality only it uses the same locality for the master and the slave functionalities.

The slaves receive a message that encodes how many sub-tasks of a certain type they should spawn locally.


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

Run Details

72 of 72 new or added lines in 1 file covered. (100.0%)

174663 of 201952 relevant lines covered (86.49%)

1849169.69 hits per line

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

90.7

/libs/core/synchronization/tests/performance/channel_mpmc_throughput.cpp

//  Copyright (c) 2019 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)

//  This work is inspired by https://github.com/aprell/tasking-2.0

#include <hpx/local/future.hpp>
#include <hpx/local/init.hpp>
#include <hpx/local/thread.hpp>
#include <hpx/modules/timing.hpp>
#include <hpx/synchronization/channel_mpmc.hpp>

#include <cstddef>
#include <cstdint>
#include <functional>
#include <iostream>
#include <utility>

///////////////////////////////////////////////////////////////////////////////
struct data
{
    data() = default;

    explicit data(int d)
    {
        data_[0] = d;
    }

    int data_[8];
};

#if HPX_DEBUG
constexpr int NUM_TESTS = 1000000;
#else
constexpr int NUM_TESTS = 100000000;
#endif

///////////////////////////////////////////////////////////////////////////////
inline data channel_get(hpx::lcos::local::channel_mpmc<data> const& c)
{
    data result;
    while (!c.get(&result))
    {
        hpx::this_thread::yield();
    }
    return result;
}

inline void channel_set(hpx::lcos::local::channel_mpmc<data>& c, data&& val)
{
    while (!c.set(std::move(val)))    // NOLINT
    {
        hpx::this_thread::yield();
    }
}

///////////////////////////////////////////////////////////////////////////////
// Produce
double thread_func_0(hpx::lcos::local::channel_mpmc<data>& c)
{
    std::uint64_t start = hpx::chrono::high_resolution_clock::now();

    for (int i = 0; i != NUM_TESTS; ++i)
    {
        channel_set(c, data{i});
    }

    std::uint64_t end = hpx::chrono::high_resolution_clock::now();

    return static_cast<double>(end - start) / 1e9;
}

// Consume
double thread_func_1(hpx::lcos::local::channel_mpmc<data>& c)
{
    std::uint64_t start = hpx::chrono::high_resolution_clock::now();

    for (int i = 0; i != NUM_TESTS; ++i)
    {
        data d = channel_get(c);
        if (d.data_[0] != i)
        {
            std::cout << "Error!\n";
        }
    }

    std::uint64_t end = hpx::chrono::high_resolution_clock::now();

    return static_cast<double>(end - start) / 1e9;
}

int hpx_main()
{
    hpx::lcos::local::channel_mpmc<data> c(10000);

    hpx::future<double> producer = hpx::async(thread_func_0, std::ref(c));
    hpx::future<double> consumer = hpx::async(thread_func_1, std::ref(c));

    auto producer_time = producer.get();
    std::cout << "Producer throughput: " << (NUM_TESTS / producer_time)
              << " [op/s] (" << (producer_time / NUM_TESTS) << " [s/op])\n";

    auto consumer_time = consumer.get();
    std::cout << "Consumer throughput: " << (NUM_TESTS / consumer_time)
              << " [op/s] (" << (consumer_time / NUM_TESTS) << " [s/op])\n";

    return hpx::local::finalize();
}

int main(int argc, char* argv[])
{
    return hpx::local::init(hpx_main, argc, argv);
}

1	// Copyright (c) 2019 Hartmut Kaiser
2	//
3	// SPDX-License-Identifier: BSL-1.0
4	// Distributed under the Boost Software License, Version 1.0. (See accompanying
5	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6
7	// This work is inspired by https://github.com/aprell/tasking-2.0
8
9	#include <hpx/local/future.hpp>
10	#include <hpx/local/init.hpp>
11	#include <hpx/local/thread.hpp>
12	#include <hpx/modules/timing.hpp>
13	#include <hpx/synchronization/channel_mpmc.hpp>
14
15	#include <cstddef>
16	#include <cstdint>
17	#include <functional>
18	#include <iostream>
19	#include <utility>
20
21	///////////////////////////////////////////////////////////////////////////////
22	struct data
23	{
24	data() = default;
25
26	explicit data(int d)	1,000,000✔
27	{
28	data_[0] = d;	1,000,000✔
29	}	1,000,000✔
30
31	int data_[8];
32	};
33
34	#if HPX_DEBUG
35	constexpr int NUM_TESTS = 1000000;
36	#else
37	constexpr int NUM_TESTS = 100000000;
38	#endif
39
40	///////////////////////////////////////////////////////////////////////////////
41	inline data channel_get(hpx::lcos::local::channel_mpmc<data> const& c)	1,000,000✔
42	{
43	data result;
44	while (!c.get(&result))	1,000,211✔
45	{
46	hpx::this_thread::yield();	211✔
47	}
48	return result;	1,000,000✔
49	}
50
51	inline void channel_set(hpx::lcos::local::channel_mpmc<data>& c, data&& val)	1,000,000✔
52	{
53	while (!c.set(std::move(val))) // NOLINT	1,000,000✔
54	{
55	hpx::this_thread::yield();	×
56	}
57	}	1,000,000✔
58
59	///////////////////////////////////////////////////////////////////////////////
60	// Produce
61	double thread_func_0(hpx::lcos::local::channel_mpmc<data>& c)	1✔
62	{
63	std::uint64_t start = hpx::chrono::high_resolution_clock::now();	1✔
64
65	for (int i = 0; i != NUM_TESTS; ++i)	1,000,001✔
66	{
67	channel_set(c, data{i});	1,000,000✔
68	}	1,000,000✔
69
70	std::uint64_t end = hpx::chrono::high_resolution_clock::now();	1✔
71
72	return static_cast<double>(end - start) / 1e9;	1✔
73	}
74
75	// Consume
76	double thread_func_1(hpx::lcos::local::channel_mpmc<data>& c)	1✔
77	{
78	std::uint64_t start = hpx::chrono::high_resolution_clock::now();	1✔
79
80	for (int i = 0; i != NUM_TESTS; ++i)	1,000,001✔
81	{
82	data d = channel_get(c);	1,000,000✔
83	if (d.data_[0] != i)	1,000,000✔
84	{
85	std::cout << "Error!\n";	×
86	}	×
87	}	1,000,000✔
88
89	std::uint64_t end = hpx::chrono::high_resolution_clock::now();	1✔
90
91	return static_cast<double>(end - start) / 1e9;	1✔
92	}
93
94	int hpx_main()	1✔
95	{
96	hpx::lcos::local::channel_mpmc<data> c(10000);	1✔
97
98	hpx::future<double> producer = hpx::async(thread_func_0, std::ref(c));	1✔
99	hpx::future<double> consumer = hpx::async(thread_func_1, std::ref(c));	1✔
100
101	auto producer_time = producer.get();	1✔
102	std::cout << "Producer throughput: " << (NUM_TESTS / producer_time)	1✔
103	<< " [op/s] (" << (producer_time / NUM_TESTS) << " [s/op])\n";	1✔
104
105	auto consumer_time = consumer.get();	1✔
106	std::cout << "Consumer throughput: " << (NUM_TESTS / consumer_time)	1✔
107	<< " [op/s] (" << (consumer_time / NUM_TESTS) << " [s/op])\n";	1✔
108
109	return hpx::local::finalize();	1✔
110	}	1✔
111
112	int main(int argc, char* argv[])	1✔
113	{
114	return hpx::local::init(hpx_main, argc, argv);	1✔
115	}	×

STEllAR-GROUP / hpx / #868

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