20284524697

Committed 16 Dec 2025 10:18PM UTC coverage: 89.66% (+0.5%) from 89.184%

Build # 20284524697

Build Type

Pull #91

github

Committed by

web-flow

Commit Message

Merge 8dcc78458 into a73a760a4

Pull Request Pull Request #91: Implement multi-threading in NativeRunner

Run Details

424 of 468 new or added lines in 11 files covered. (90.6%)

3 existing lines in 2 files now uncovered.

6096 of 6799 relevant lines covered (89.66%)

7502415.88 hits per line

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

86.73

/coretrace/simulation_runner/native_runner/thread_manager.cpp

#include "thread_manager.hpp"

#include <chrono>
#include <future>
#include <sstream>
#include <thread>

namespace SolTrace::NativeRunner
{

    ThreadManager::ThreadManager()
    {
        this->initialize();
        return;
    }

    ThreadManager::~ThreadManager()
    {
        this->threads.clear();
        this->progress.clear();
        return;
    }

    unsigned int ThreadManager::manage(unsigned int id, future f)
    {
        assert(this->threads.find(id) == this->threads.end());

        this->threads[id] = std::move(f);
        this->progress[id] = 0.0;
        return id;
    }

    ThreadManager::ThreadStatus ThreadManager::monitor_until_completion()
    {
        ThreadStatus sts = ThreadStatus::SUCCESS;
        bool canceled = false;

        while (!this->threads.empty())
        {
            auto iter = this->threads.begin();
            while (iter != this->threads.end())
            {
                if (iter->second.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
                {
                    ThreadStatus thread_status = iter->second.get();
                    if (thread_status == ThreadStatus::SUCCESS)
                    {
                        ; // Intentional no-op
                    }
                    else if (thread_status == ThreadStatus::ERROR)
                    {
                        // Something went wrong -- shut everything down
                        this->cancel();
                        // Return ERROR status
                        sts = ThreadStatus::ERROR;
                        canceled = true;
                    }
                    else if (thread_status == ThreadStatus::CANCEL)
                    {
                        // Return CANCEL if threads were canceled
                        // by an external call to cancel()
                        sts = canceled ? sts : thread_status;
                    }
                    else
                    {
                        // Thread terminated with something other than
                        // SUCCESS, ERROR, or CANCEL. This is unexpected.
                        std::stringstream ss;
                        ss << "Thread " << iter->first
                           << " returned status "
                           << status_string(thread_status)
                           << ". This is an error.";
                        this->error_log(ss.str());

                        // Unexpected behavior so we terminate
                        this->cancel();
                        sts = ThreadStatus::ERROR;
                        canceled = true;
                    }
                    // Task is complete so we stop tracking it.
                    // Also increments to the next spot
                    iter = this->threads.erase(iter);
                }
                else
                {
                    ++iter;
                }
            }

            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }

        return sts;
    }

    void ThreadManager::error_log(const std::string &msg)
    {
        std::lock_guard<std::mutex> lk(this->message_mutex);
        this->messages.push_back(msg);
        return;
    }

    void ThreadManager::progress_update(unsigned int id, double prog)
    {
        // std::cout << "Update from thread " << id << " reporting "
        //           << prog << " complete." << std::endl;
        std::lock_guard<std::mutex> lk(this->progress_mutex);
        this->progress[id] = prog;
        return;
    }

    bool ThreadManager::terminate(unsigned int id)
    {
        std::lock_guard<std::mutex> lk(this->state_mutex);
        return this->state != ThreadStatus::RUNNING;
    }

    ThreadManager::ThreadStatus ThreadManager::status(double *progress) const
    {
        ThreadStatus sts = ThreadStatus::ERROR;
        // Create isolated scope for lock guard
        {
            std::lock_guard<std::mutex> lk(this->state_mutex);
            sts = this->state;
        }

        if (progress != nullptr)
        {
            int k = 0;
            double avg = 0.0;
            std::lock_guard<std::mutex> lk(this->progress_mutex);
            for (auto it = this->progress.cbegin();
                 it != this->progress.cend();
                 ++it)
            {
                // std::cout << "Thread " << k
                //           << " progress " << it->second
                //           << std::endl;
                ++k;
                avg += (it->second - avg) / k;
            }
            *progress = avg;
        }

        return sts;
    }

    void ThreadManager::cancel() const
    {
        std::lock_guard<std::mutex> lk(this->state_mutex);
        this->state = ThreadStatus::CANCEL;
        return;
    }

    void ThreadManager::print_log(std::ostream &os) const
    {
        std::lock_guard<std::mutex> lk(this->message_mutex);
        for (auto iter = this->messages.cbegin();
             iter != this->messages.cend();
             ++iter)
        {
            os << *iter;
        }
        return;
    }

    void ThreadManager::initialize()
    {
        // Must call prior to manage and not while running!!
        // this->next_id = 0;
        {
            std::lock_guard<std::mutex> lk(this->state_mutex);
            this->state = ThreadStatus::RUNNING;
        }
        {
            std::lock_guard<std::mutex> lk(this->progress_mutex);
            this->progress.clear();
            this->threads.clear();
        }
        {
            std::lock_guard<std::mutex> lk(this->message_mutex);
            this->messages.clear();
        }
        return;
    }

} // namespace SolTrace::NativeRunner

1	#include "thread_manager.hpp"
2
3	#include <chrono>
4	#include <future>
5	#include <sstream>
6	#include <thread>
7
8	namespace SolTrace::NativeRunner
9	{
10
11	ThreadManager::ThreadManager()	27✔
12	{
13	this->initialize();	27✔
14	return;	27✔
NEW 15	}	×
16
17	ThreadManager::~ThreadManager()	27✔
18	{
19	this->threads.clear();	27✔
20	this->progress.clear();	27✔
21	return;	27✔
22	}	27✔
23
24	unsigned int ThreadManager::manage(unsigned int id, future f)	25✔
25	{
26	assert(this->threads.find(id) == this->threads.end());	25✔
27
28	this->threads[id] = std::move(f);	25✔
29	this->progress[id] = 0.0;	25✔
30	return id;	25✔
31	}
32
33	ThreadManager::ThreadStatus ThreadManager::monitor_until_completion()	25✔
34	{
35	ThreadStatus sts = ThreadStatus::SUCCESS;	25✔
36	bool canceled = false;	25✔
37
38	while (!this->threads.empty())	7,172✔
39	{
40	auto iter = this->threads.begin();	7,147✔
41	while (iter != this->threads.end())	14,294✔
42	{
43	if (iter->second.wait_for(std::chrono::seconds(0)) == std::future_status::ready)	7,147✔
44	{
45	ThreadStatus thread_status = iter->second.get();	25✔
46	if (thread_status == ThreadStatus::SUCCESS)	25✔
47	{
48	; // Intentional no-op
49	}
50	else if (thread_status == ThreadStatus::ERROR)	1✔
51	{
52	// Something went wrong -- shut everything down
NEW 53	this->cancel();	×
54	// Return ERROR status
NEW 55	sts = ThreadStatus::ERROR;	×
NEW 56	canceled = true;	×
57	}
58	else if (thread_status == ThreadStatus::CANCEL)	1✔
59	{
60	// Return CANCEL if threads were canceled
61	// by an external call to cancel()
62	sts = canceled ? sts : thread_status;	1✔
63	}
64	else
65	{
66	// Thread terminated with something other than
67	// SUCCESS, ERROR, or CANCEL. This is unexpected.
NEW 68	std::stringstream ss;	×
NEW 69	ss << "Thread " << iter->first	×
70	<< " returned status "
NEW 71	<< status_string(thread_status)	×
NEW 72	<< ". This is an error.";	×
NEW 73	this->error_log(ss.str());	×
74
75	// Unexpected behavior so we terminate
NEW 76	this->cancel();	×
NEW 77	sts = ThreadStatus::ERROR;	×
NEW 78	canceled = true;	×
NEW 79	}	×
80	// Task is complete so we stop tracking it.
81	// Also increments to the next spot
82	iter = this->threads.erase(iter);	25✔
83	}
84	else
85	{
86	++iter;	7,122✔
87	}
88	}
89
90	std::this_thread::sleep_for(std::chrono::milliseconds(10));	7,147✔
91	}
92
93	return sts;	25✔
94	}
95
96	void ThreadManager::error_log(const std::string &msg)	1✔
97	{
98	std::lock_guard<std::mutex> lk(this->message_mutex);	1✔
99	this->messages.push_back(msg);	1✔
100	return;	2✔
101	}	1✔
102
103	void ThreadManager::progress_update(unsigned int id, double prog)	11,759✔
104	{
105	// std::cout << "Update from thread " << id << " reporting "
106	// << prog << " complete." << std::endl;
107	std::lock_guard<std::mutex> lk(this->progress_mutex);	11,759✔
108	this->progress[id] = prog;	11,759✔
109	return;	23,518✔
110	}	11,759✔
111
112	bool ThreadManager::terminate(unsigned int id)	11,759✔
113	{
114	std::lock_guard<std::mutex> lk(this->state_mutex);	11,759✔
115	return this->state != ThreadStatus::RUNNING;	11,759✔
116	}	11,759✔
117
118	ThreadManager::ThreadStatus ThreadManager::status(double *progress) const	3✔
119	{
120	ThreadStatus sts = ThreadStatus::ERROR;	3✔
121	// Create isolated scope for lock guard
122	{
123	std::lock_guard<std::mutex> lk(this->state_mutex);	3✔
124	sts = this->state;	3✔
125	}	3✔
126
127	if (progress != nullptr)	3✔
128	{
129	int k = 0;	1✔
130	double avg = 0.0;	1✔
131	std::lock_guard<std::mutex> lk(this->progress_mutex);	1✔
132	for (auto it = this->progress.cbegin();	1✔
133	it != this->progress.cend();	2✔
134	++it)	1✔
135	{
136	// std::cout << "Thread " << k
137	// << " progress " << it->second
138	// << std::endl;
139	++k;	1✔
140	avg += (it->second - avg) / k;	1✔
141	}
142	*progress = avg;	1✔
143	}	1✔
144
145	return sts;	3✔
146	}
147
148	void ThreadManager::cancel() const	1✔
149	{
150	std::lock_guard<std::mutex> lk(this->state_mutex);	1✔
151	this->state = ThreadStatus::CANCEL;	1✔
152	return;	2✔
153	}	1✔
154
155	void ThreadManager::print_log(std::ostream &os) const	1✔
156	{
157	std::lock_guard<std::mutex> lk(this->message_mutex);	1✔
158	for (auto iter = this->messages.cbegin();	1✔
159	iter != this->messages.cend();	2✔
160	++iter)	1✔
161	{
162	os << *iter;	1✔
163	}
164	return;	2✔
165	}	1✔
166
167	void ThreadManager::initialize()	27✔
168	{
169	// Must call prior to manage and not while running!!
170	// this->next_id = 0;
171	{
172	std::lock_guard<std::mutex> lk(this->state_mutex);	27✔
173	this->state = ThreadStatus::RUNNING;	27✔
174	}	27✔
175	{
176	std::lock_guard<std::mutex> lk(this->progress_mutex);	27✔
177	this->progress.clear();	27✔
178	this->threads.clear();	27✔
179	}	27✔
180	{
181	std::lock_guard<std::mutex> lk(this->message_mutex);	27✔
182	this->messages.clear();	27✔
183	}	27✔
184	return;	27✔
185	}
186
187	} // namespace SolTrace::NativeRunner

NREL / SolTrace / 20284524697

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