#882

Committed 31 Aug 2023 07:44PM UTC coverage: 41.798% (-44.7%) from 86.546%

Build # #882

Build Type

push

Committed by

Commit Message

Run Details

19442 of 46514 relevant lines covered (41.8%)

126375.38 hits per line

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

0.0

/examples/balancing/hpx_thread_phase.cpp

//  Copyright (c) 2011 Bryce Lelbach
//  Copyright (c) 2022 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/barrier.hpp>
#include <hpx/functional.hpp>
#include <hpx/init.hpp>
#include <hpx/modules/concurrency.hpp>
#include <hpx/mutex.hpp>
#include <hpx/thread.hpp>

#include <chrono>
#include <cstddef>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <utility>
#include <vector>

using hpx::lockfree::queue;

using hpx::program_options::options_description;
using hpx::program_options::value;
using hpx::program_options::variables_map;

using std::chrono::milliseconds;

using hpx::barrier;
using hpx::mutex;

using hpx::threads::make_thread_function_nullary;
using hpx::threads::register_thread;
using hpx::threads::thread_init_data;

using hpx::threads::get_self;
using hpx::threads::get_self_id;
using hpx::threads::get_thread_phase;
using hpx::threads::set_thread_state;
using hpx::threads::thread_id_ref_type;
using hpx::threads::thread_id_type;

typedef std::pair<thread_id_type, std::size_t> value_type;
typedef std::vector<value_type> fifo_type;

///////////////////////////////////////////////////////////////////////////////
void lock_and_wait(mutex& m, std::shared_ptr<barrier<>> b0,
    std::shared_ptr<barrier<>> b1, value_type& entry, std::size_t /* wait */)
{
    // Wait for all hpxthreads in this iteration to be created.
    b0->arrive_and_wait();

    // keep this thread alive while being suspended
    thread_id_ref_type this_ = get_self_id();

    while (true)
    {
        // Try to acquire the mutex.
        std::unique_lock<mutex> l(m, std::try_to_lock);

        if (l.owns_lock())
        {
            entry = value_type(this_.noref(), get_thread_phase(this_.noref()));
            break;
        }

        // Schedule a wakeup.
        set_thread_state(this_.noref(), milliseconds(30),
            hpx::threads::thread_schedule_state::pending);

        // Suspend this HPX thread.
        hpx::this_thread::suspend(
            hpx::threads::thread_schedule_state::suspended);
    }

    // Make hpx_main wait for us to finish.
    b1->arrive_and_wait();
}

///////////////////////////////////////////////////////////////////////////////
int hpx_main(variables_map& vm)
{
    std::size_t hpxthread_count = 0;

    if (vm.count("hpxthreads"))
        hpxthread_count = vm["hpxthreads"].as<std::size_t>();

    std::size_t mutex_count = 0;

    if (vm.count("mutexes"))
        mutex_count = vm["mutexes"].as<std::size_t>();

    std::size_t iterations = 0;

    if (vm.count("iterations"))
        iterations = vm["iterations"].as<std::size_t>();

    std::size_t wait = 0;

    if (vm.count("wait"))
        wait = vm["wait"].as<std::size_t>();

    for (std::size_t i = 0; i < iterations; ++i)
    {
        std::cout << "iteration: " << i << "\n";

        // Have the fifo preallocate storage.
        fifo_type hpxthreads(hpxthread_count);

        // Allocate the mutexes.
        std::vector<mutex> m(mutex_count);
        std::shared_ptr<barrier<>> b0 =
            std::make_shared<barrier<>>(hpxthread_count + 1);
        std::shared_ptr<barrier<>> b1 =
            std::make_shared<barrier<>>(hpxthread_count + 1);

        // keep created threads alive while they are suspended
        std::vector<thread_id_ref_type> ids;
        for (std::size_t j = 0; j < hpxthread_count; ++j)
        {
            // Compute the mutex to be used for this thread.
            std::size_t const index = j % mutex_count;

            thread_init_data data(
                make_thread_function_nullary(hpx::bind(&lock_and_wait,
                    std::ref(m[index]), b0, b1, std::ref(hpxthreads[j]), wait)),
                "lock_and_wait");
            ids.push_back(register_thread(data));
        }

        // Tell all hpxthreads that they can start running.
        b0->arrive_and_wait();

        // Wait for all hpxthreads to finish.
        b1->arrive_and_wait();

        // {{{ Print results for this iteration.
        for (value_type& entry : hpxthreads)
        {
            std::cout << "  " << entry.first << "," << entry.second << "\n";
        }
        // }}}
    }

    // Initiate shutdown of the runtime system.
    hpx::local::finalize();
    return 0;
}

///////////////////////////////////////////////////////////////////////////////
int main(int argc, char* argv[])
{
    // Configure application-specific options.
    options_description desc_commandline(
        "Usage: " HPX_APPLICATION_STRING " [options]");

    // clang-format off
    desc_commandline.add_options()
        ("hpxthreads,T", value<std::size_t>()->default_value(128),
            "the number of PX threads to invoke")
        ("mutexes,M", value<std::size_t>()->default_value(1),
            "the number of mutexes to use")
        ("wait", value<std::size_t>()->default_value(30),
            "the number of milliseconds to wait between each lock attempt")
        ("iterations", value<std::size_t>()->default_value(1),
            "the number of times to repeat the test")
        ;
    // clang-format on

    // Initialize and run HPX.
    hpx::local::init_params init_args;
    init_args.desc_cmdline = desc_commandline;

    return hpx::local::init(hpx_main, argc, argv, init_args);
}

1	// Copyright (c) 2011 Bryce Lelbach
2	// Copyright (c) 2022 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/barrier.hpp>
9	#include <hpx/functional.hpp>
10	#include <hpx/init.hpp>
11	#include <hpx/modules/concurrency.hpp>
12	#include <hpx/mutex.hpp>
13	#include <hpx/thread.hpp>
14
15	#include <chrono>
16	#include <cstddef>
17	#include <functional>
18	#include <iostream>
19	#include <memory>
20	#include <mutex>
21	#include <utility>
22	#include <vector>
23
24	using hpx::lockfree::queue;
25
26	using hpx::program_options::options_description;
27	using hpx::program_options::value;
28	using hpx::program_options::variables_map;
29
30	using std::chrono::milliseconds;
31
32	using hpx::barrier;
33	using hpx::mutex;
34
35	using hpx::threads::make_thread_function_nullary;
36	using hpx::threads::register_thread;
37	using hpx::threads::thread_init_data;
38
39	using hpx::threads::get_self;
40	using hpx::threads::get_self_id;
41	using hpx::threads::get_thread_phase;
42	using hpx::threads::set_thread_state;
43	using hpx::threads::thread_id_ref_type;
44	using hpx::threads::thread_id_type;
45
46	typedef std::pair<thread_id_type, std::size_t> value_type;
47	typedef std::vector<value_type> fifo_type;
48
49	///////////////////////////////////////////////////////////////////////////////
50	void lock_and_wait(mutex& m, std::shared_ptr<barrier<>> b0,	×
51	std::shared_ptr<barrier<>> b1, value_type& entry, std::size_t /* wait */)
52	{
53	// Wait for all hpxthreads in this iteration to be created.
54	b0->arrive_and_wait();	×
55
56	// keep this thread alive while being suspended
57	thread_id_ref_type this_ = get_self_id();	×
58
59	while (true)
60	{
61	// Try to acquire the mutex.
62	std::unique_lock<mutex> l(m, std::try_to_lock);
63
64	if (l.owns_lock())	×
65	{
66	entry = value_type(this_.noref(), get_thread_phase(this_.noref()));	×
67	break;
68	}
69
70	// Schedule a wakeup.
71	set_thread_state(this_.noref(), milliseconds(30),	×
72	hpx::threads::thread_schedule_state::pending);
73
74	// Suspend this HPX thread.
75	hpx::this_thread::suspend(	×
76	hpx::threads::thread_schedule_state::suspended);
77	}
78
79	// Make hpx_main wait for us to finish.
80	b1->arrive_and_wait();	×
81	}	×
82
83	///////////////////////////////////////////////////////////////////////////////
84	int hpx_main(variables_map& vm)	×
85	{
86	std::size_t hpxthread_count = 0;
87
88	if (vm.count("hpxthreads"))	×
89	hpxthread_count = vm["hpxthreads"].as<std::size_t>();	×
90
91	std::size_t mutex_count = 0;
92
93	if (vm.count("mutexes"))	×
94	mutex_count = vm["mutexes"].as<std::size_t>();	×
95
96	std::size_t iterations = 0;
97
98	if (vm.count("iterations"))	×
99	iterations = vm["iterations"].as<std::size_t>();	×
100
101	std::size_t wait = 0;
102
103	if (vm.count("wait"))	×
104	wait = vm["wait"].as<std::size_t>();	×
105
106	for (std::size_t i = 0; i < iterations; ++i)	×
107	{
108	std::cout << "iteration: " << i << "\n";	×
109
110	// Have the fifo preallocate storage.
111	fifo_type hpxthreads(hpxthread_count);	×
112
113	// Allocate the mutexes.
114	std::vector<mutex> m(mutex_count);	×
115	std::shared_ptr<barrier<>> b0 =
116	std::make_shared<barrier<>>(hpxthread_count + 1);	×
117	std::shared_ptr<barrier<>> b1 =
118	std::make_shared<barrier<>>(hpxthread_count + 1);	×
119
120	// keep created threads alive while they are suspended
121	std::vector<thread_id_ref_type> ids;	×
122	for (std::size_t j = 0; j < hpxthread_count; ++j)	×
123	{
124	// Compute the mutex to be used for this thread.
125	std::size_t const index = j % mutex_count;	×
126
127	thread_init_data data(
128	make_thread_function_nullary(hpx::bind(&lock_and_wait,	×
129	std::ref(m[index]), b0, b1, std::ref(hpxthreads[j]), wait)),
130	"lock_and_wait");
131	ids.push_back(register_thread(data));	×
132	}
133
134	// Tell all hpxthreads that they can start running.
135	b0->arrive_and_wait();	×
136
137	// Wait for all hpxthreads to finish.
138	b1->arrive_and_wait();	×
139
140	// {{{ Print results for this iteration.
141	for (value_type& entry : hpxthreads)	×
142	{
143	std::cout << " " << entry.first << "," << entry.second << "\n";	×
144	}
145	// }}}
146	}	×
147
148	// Initiate shutdown of the runtime system.
149	hpx::local::finalize();	×
150	return 0;	×
151	}
152
153	///////////////////////////////////////////////////////////////////////////////
154	int main(int argc, char* argv[])	×
155	{
156	// Configure application-specific options.
157	options_description desc_commandline(
158	"Usage: " HPX_APPLICATION_STRING " [options]");	×
159
160	// clang-format off
161	desc_commandline.add_options()	×
162	("hpxthreads,T", value<std::size_t>()->default_value(128),	×
163	"the number of PX threads to invoke")
164	("mutexes,M", value<std::size_t>()->default_value(1),	×
165	"the number of mutexes to use")
166	("wait", value<std::size_t>()->default_value(30),	×
167	"the number of milliseconds to wait between each lock attempt")
168	("iterations", value<std::size_t>()->default_value(1),	×
169	"the number of times to repeat the test")
170	;
171	// clang-format on
172
173	// Initialize and run HPX.
174	hpx::local::init_params init_args;	×
175	init_args.desc_cmdline = desc_commandline;	×
176
177	return hpx::local::init(hpx_main, argc, argv, init_args);	×
178	}	×

STEllAR-GROUP / hpx / #882

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