wirenboard / wb-mqtt-smartweb / 47

Committed 31 Jul 2026 10:42AM UTC coverage: 33.897% (-0.8%) from 34.703%

Build # 47

Build Type

push

github

Committed by

web-flow

Commit Message

Use exported vars instead of MAKEFLAGS for coverage options (#40)

Coverage Stats

307 of 744 branches covered (41.26%)

501 of 1478 relevant lines covered (33.9%)

10.02 hits per line

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

0.0

/src/scheduler.cpp

#include "scheduler.h"

#include <algorithm>
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <vector>

#include <wblib/utils.h>

namespace
{

    struct TTaskDescription
    {
        std::chrono::steady_clock::time_point NextRun;
        std::shared_ptr<ITask> Task;

        TTaskDescription(std::chrono::steady_clock::time_point nextRun, std::shared_ptr<ITask> task)
            : NextRun(nextRun),
              Task(task)
        {}
    };

    class TSimpleThreadedScheduler: public IScheduler
    {
        std::thread Thread;

        std::atomic_bool Enabled;
        std::mutex TasksMutex;
        std::vector<TTaskDescription> Tasks;

        std::mutex WaitMutex;
        std::condition_variable ConditionVariable;

        std::string ThreadName;

        void MakeIteration()
        {
            auto now = std::chrono::steady_clock::now();
            std::unique_lock<std::mutex> tasksLock(TasksMutex);
            if (Tasks.empty()) {
                tasksLock.unlock();
                std::unique_lock<std::mutex> waitLock(WaitMutex);
                ConditionVariable.wait(waitLock);
                return;
            }
            if (Tasks.begin()->NextRun <= now) {
                TTaskDescription td = *Tasks.begin();
                Tasks.erase(Tasks.begin());
                tasksLock.unlock();
                auto res = td.Task->Run();
                tasksLock.lock();
                for (auto& t: res) {
                    Tasks.emplace_back(now + t->GetPeriod(), t);
                }
                return;
            }
            std::stable_sort(Tasks.begin(), Tasks.end(), [](const auto& t1, const auto& t2) {
                return t1.NextRun < t2.NextRun;
            });
            auto nextRun = Tasks.begin()->NextRun;
            tasksLock.unlock();
            std::unique_lock<std::mutex> waitLock(WaitMutex);
            ConditionVariable.wait_until(waitLock, nextRun);
        }

    public:
        TSimpleThreadedScheduler(const std::string& threadName): Enabled(true), ThreadName(threadName)
        {
            Thread = std::thread([&]() {
                WBMQTT::SetThreadName(ThreadName);
                while (Enabled.load()) {
                    MakeIteration();
                }
            });
        }

        ~TSimpleThreadedScheduler()
        {
            Enabled.store(false);
            ConditionVariable.notify_one();
            Thread.join();
        }

        void AddTask(std::shared_ptr<ITask> task) override
        {
            std::unique_lock<std::mutex> lk(TasksMutex);
            Tasks.emplace(Tasks.begin(), std::chrono::steady_clock::time_point(), task);
            ConditionVariable.notify_one();
        }
    };

    class TPeriodicTask: public ITask
    {
        std::function<void()> Fn;
        std::chrono::microseconds Period;
        std::string Name;

    public:
        TPeriodicTask(const std::chrono::microseconds& period, std::function<void()> fn, const std::string& name)
            : Fn(fn),
              Period(period),
              Name(name)
        {}

        std::vector<std::shared_ptr<ITask>> Run() override
        {
            Fn();
            return {shared_from_this()};
        }

        const std::string& GetName() const override
        {
            return Name;
        }

        const std::chrono::microseconds& GetPeriod() const override
        {
            return Period;
        }
    };
}

IScheduler* MakeSimpleThreadedScheduler(const std::string& threadName)
{
    return new TSimpleThreadedScheduler(threadName);
}

std::shared_ptr<ITask> MakePeriodicTask(const std::chrono::microseconds& period,
                                        std::function<void()> fn,
                                        const std::string& name)
{
    return std::make_shared<TPeriodicTask>(period, fn, name);
}

1	#include "scheduler.h"
2
3	#include <algorithm>
4	#include <atomic>
5	#include <condition_variable>
6	#include <mutex>
7	#include <thread>
8	#include <vector>
9
10	#include <wblib/utils.h>
11
12	namespace
13	{
14
15	struct TTaskDescription
16	{
17	std::chrono::steady_clock::time_point NextRun;
18	std::shared_ptr<ITask> Task;
19
20	TTaskDescription(std::chrono::steady_clock::time_point nextRun, std::shared_ptr<ITask> task)	×
21	: NextRun(nextRun),	×
22	Task(task)	×
23	{}	×
24	};
25
26	class TSimpleThreadedScheduler: public IScheduler
27	{
28	std::thread Thread;
29
30	std::atomic_bool Enabled;
31	std::mutex TasksMutex;
32	std::vector<TTaskDescription> Tasks;
33
34	std::mutex WaitMutex;
35	std::condition_variable ConditionVariable;
36
37	std::string ThreadName;
38
39	void MakeIteration()	×
40	{
41	auto now = std::chrono::steady_clock::now();	×
42	std::unique_lock<std::mutex> tasksLock(TasksMutex);	×
43	if (Tasks.empty()) {	×
44	tasksLock.unlock();	×
45	std::unique_lock<std::mutex> waitLock(WaitMutex);	×
46	ConditionVariable.wait(waitLock);	×
47	return;	×
48	}	×
49	if (Tasks.begin()->NextRun <= now) {	×
50	TTaskDescription td = *Tasks.begin();	×
51	Tasks.erase(Tasks.begin());	×
52	tasksLock.unlock();	×
53	auto res = td.Task->Run();	×
54	tasksLock.lock();	×
55	for (auto& t: res) {	×
56	Tasks.emplace_back(now + t->GetPeriod(), t);	×
57	}
58	return;	×
59	}	×
60	std::stable_sort(Tasks.begin(), Tasks.end(), [](const auto& t1, const auto& t2) {	×
61	return t1.NextRun < t2.NextRun;	×
62	});
63	auto nextRun = Tasks.begin()->NextRun;	×
64	tasksLock.unlock();	×
65	std::unique_lock<std::mutex> waitLock(WaitMutex);	×
66	ConditionVariable.wait_until(waitLock, nextRun);	×
67	}	×
68
69	public:
70	TSimpleThreadedScheduler(const std::string& threadName): Enabled(true), ThreadName(threadName)	×
71	{
72	Thread = std::thread([&]() {	×
73	WBMQTT::SetThreadName(ThreadName);	×
74	while (Enabled.load()) {	×
75	MakeIteration();	×
76	}
77	});	×
78	}	×
79
80	~TSimpleThreadedScheduler()	×
81	{	×
82	Enabled.store(false);	×
83	ConditionVariable.notify_one();	×
84	Thread.join();	×
85	}	×
86
87	void AddTask(std::shared_ptr<ITask> task) override	×
88	{
89	std::unique_lock<std::mutex> lk(TasksMutex);	×
90	Tasks.emplace(Tasks.begin(), std::chrono::steady_clock::time_point(), task);	×
91	ConditionVariable.notify_one();	×
92	}	×
93	};
94
95	class TPeriodicTask: public ITask
96	{
97	std::function<void()> Fn;
98	std::chrono::microseconds Period;
99	std::string Name;
100
101	public:
102	TPeriodicTask(const std::chrono::microseconds& period, std::function<void()> fn, const std::string& name)	×
103	: Fn(fn),	×
104	Period(period),	×
105	Name(name)	×
106	{}	×
107
108	std::vector<std::shared_ptr<ITask>> Run() override	×
109	{
110	Fn();	×
111	return {shared_from_this()};	×
112	}	×
113
114	const std::string& GetName() const override	×
115	{
116	return Name;	×
117	}
118
119	const std::chrono::microseconds& GetPeriod() const override	×
120	{
121	return Period;	×
122	}
123	};
124	}
125
126	IScheduler* MakeSimpleThreadedScheduler(const std::string& threadName)	×
127	{
128	return new TSimpleThreadedScheduler(threadName);	×
129	}
130
131	std::shared_ptr<ITask> MakePeriodicTask(const std::chrono::microseconds& period,	×
132	std::function<void()> fn,
133	const std::string& name)
134	{
135	return std::make_shared<TPeriodicTask>(period, fn, name);	×
136	}

wirenboard / wb-mqtt-smartweb / 47

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