wirenboard / wb-mqtt-serial / 967

Committed 05 Oct 2026 09:41AM UTC coverage: 85.672% (+2.4%) from 83.268%

Build # 967

Build Type

push

github

Committed by

web-flow

Commit Message

Add config/Save RPC (#1273)

The RPC validates a configuration, writes it to the configuration file and
restarts the polling in-process. Port clients are stopped first, requests
arriving meanwhile get config-busy and queued ones get request-cancelled.

Coverage Stats

12445 of 14263 branches covered (87.25%)

1432 of 1623 new or added lines in 41 files covered. (88.23%)

8 existing lines in 4 files now uncovered.

21065 of 24588 relevant lines covered (85.67%)

1528.89 hits per line

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

96.69

/src/serial_client.cpp

#include "serial_client.h"
#include <chrono>
#include <iostream>
#include <unistd.h>

#include "modbus_ext_common.h"
#include "write_channel_serial_client_task.h"

using namespace std::chrono_literals;
using namespace std::chrono;

#define LOG(logger) logger.Log() << "[serial client] "

namespace
{
    const auto PORT_OPEN_ERROR_NOTIFICATION_INTERVAL = 5min;
    const auto CLOSED_PORT_CYCLE_TIME = 500ms;
    const auto MAX_POLL_TIME = 100ms;
    // const auto MAX_FLUSHES_WHEN_POLL_IS_DUE = 20;
    const auto BALANCING_THRESHOLD = 500ms;
    // const auto MIN_READ_EVENTS_TIME = 25ms;
    const size_t MAX_EVENT_READ_ERRORS = 10;

    std::chrono::milliseconds GetReadEventsPeriod(const TPort& port)
    {
        auto sendByteTime = port.GetSendTimeBytes(1);
        // >= 115200
        if (sendByteTime < 100us) {
            return 50ms;
        }
        // >= 38400
        if (sendByteTime < 300us) {
            return 100ms;
        }
        // < 38400
        return 200ms;
    }
};

void ISerialClientTask::Cancel()
{
    Cancelled = true;
}

bool ISerialClientTask::IsCancelled() const
{
    return Cancelled;
}

TSerialClient::TSerialClient(PFeaturePort port,
                             const TPortOpenCloseLogic::TSettings& openCloseSettings,
                             util::TGetNowFn nowFn,
                             size_t lowPriorityRateLimit)
    : Port(port),
      OpenCloseLogic(openCloseSettings, nowFn),
      ConnectLogger(PORT_OPEN_ERROR_NOTIFICATION_INTERVAL, "[serial client] "),
      NowFn(nowFn),
      LowPriorityRateLimit(lowPriorityRateLimit)
{}

TSerialClient::~TSerialClient()
{
    if (Port->IsOpen()) {
        Port->Close();
    }
}

void TSerialClient::AddDevice(PSerialDevice device)
{
    if (RegReader)
        throw TSerialDeviceException("can't add registers to the active client");
    for (const auto& reg: device->GetRegisters()) {
        if (Handlers.find(reg) != Handlers.end())
            throw TSerialDeviceException("duplicate register");
        auto handler = Handlers[reg] = std::make_shared<TRegisterHandler>(reg);
        RegList.push_back(reg);
        LOG(Debug) << "AddRegister: " << reg;
    }
    Devices.push_back(device);
}

void TSerialClient::Activate()
{
    std::lock_guard<std::mutex> lock(RegReaderMutex);
    if (!RegReader) {
        RegReader = std::make_unique<TSerialClientRegisterAndEventsReader>(Devices,
                                                                           GetReadEventsPeriod(*Port),
                                                                           NowFn,
                                                                           std::chrono::system_clock::now,
                                                                           LowPriorityRateLimit);
        LastAccessedDevice = std::make_unique<TSerialClientDeviceAccessHandler>(RegReader->GetEventsReader());
    }
}

void TSerialClient::WaitForPollAndFlush(steady_clock::time_point currentTime, steady_clock::time_point waitUntil)
{
    if (currentTime > waitUntil) {
        waitUntil = currentTime;
    }

    if (Debug.IsEnabled()) {
        LOG(Debug) << Port->GetDescription() << duration_cast<milliseconds>(currentTime.time_since_epoch()).count()
                   << ": Wait until " << duration_cast<milliseconds>(waitUntil.time_since_epoch()).count();
    }

    std::vector<PSerialClientTask> retryTasks;
    {
        std::unique_lock<std::mutex> lock(TasksMutex);

        while (TasksCv.wait_until(lock, waitUntil, [this]() { return !Tasks.empty() || (StopRequested && !Stopped); }))
        {
            // One by one, so RequestStop cancels every task which has not started
            while (!Tasks.empty()) {
                auto task = Tasks.front();
                Tasks.pop_front();
                lock.unlock();
                if (task->Run(Port, *LastAccessedDevice, Devices) == ISerialClientTask::TRunResult::RETRY) {
                    retryTasks.push_back(task);
                }
                lock.lock();
            }
            if (StopRequested) {
                Stopped = true;
                StoppedCv.notify_all();
            }
        }
    }
    for (auto& task: retryTasks) {
        AddTask(task);
    }
}

void TSerialClient::ProcessPolledRegister(PRegister reg)
{
    if (reg->GetErrorState().test(TRegister::ReadError) || reg->GetErrorState().test(TRegister::WriteError)) {
        if (RegisterErrorCallback) {
            RegisterErrorCallback(reg);
        }
    } else {
        if (RegisterReadCallback) {
            RegisterReadCallback(reg);
        }
    }
}

void TSerialClient::Cycle()
{
    Activate();
    if (IsStopRequested()) {
        auto currentTime = NowFn();
        WaitForPollAndFlush(currentTime, currentTime + CLOSED_PORT_CYCLE_TIME);
        return;
    }

    try {
        OpenCloseLogic.OpenIfAllowed(Port);
    } catch (const std::exception& e) {
        ConnectLogger.Log(e.what(), Debug, Error);
    }

    if (Port->IsOpen()) {
        ConnectLogger.DropTimeout();
        OpenPortCycle();
    } else {
        ClosedPortCycle();
    }
}

void TSerialClient::ClosedPortCycle()
{
    auto currentTime = NowFn();
    auto waitUntil = currentTime + CLOSED_PORT_CYCLE_TIME;
    WaitForPollAndFlush(currentTime, waitUntil);
    if (IsStopRequested()) {
        return;
    }

    RegReader->ClosedPortCycle(waitUntil, [this](PRegister reg) { ProcessPolledRegister(reg); });
}

void TSerialClient::SetTextValue(PRegister reg, const std::string& value)
{
    if (IsStopRequested()) {
        LOG(Debug) << reg->ToString() << " write is dropped, the client is stopped";
        return;
    }
    auto handler = GetHandler(reg);
    handler->SetTextValue(value);
    auto serialClientTask =
        std::make_shared<TWriteChannelSerialClientTask>(handler, RegisterReadCallback, RegisterErrorCallback);
    AddTask(serialClientTask);
}

void TSerialClient::SetReadCallback(const TSerialClient::TRegisterCallback& callback)
{
    RegisterReadCallback = callback;
}

void TSerialClient::SetErrorCallback(const TSerialClient::TRegisterCallback& callback)
{
    RegisterErrorCallback = callback;
}

PRegisterHandler TSerialClient::GetHandler(PRegister reg) const
{
    auto it = Handlers.find(reg);
    if (it == Handlers.end())
        throw TSerialDeviceException("register not found");
    return it->second;
}

void TSerialClient::OpenPortCycle()
{
    auto currentTime = NowFn();
    auto waitUntil = RegReader->GetDeadline(currentTime);
    // Limit waiting time to be responsive
    waitUntil = std::min(waitUntil, currentTime + MAX_POLL_TIME);
    WaitForPollAndFlush(currentTime, waitUntil);
    if (IsStopRequested()) {
        return;
    }

    auto device =
        RegReader->OpenPortCycle(*Port, [this](PRegister reg) { ProcessPolledRegister(reg); }, *LastAccessedDevice);

    if (device) {
        OpenCloseLogic.CloseIfNeeded(Port, device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED);
    }
}

PFeaturePort TSerialClient::GetPort()
{
    return Port;
}

std::list<PSerialDevice> TSerialClient::GetDevices()
{
    return Devices;
}

void TSerialClient::AddTask(PSerialClientTask task)
{
    {
        std::unique_lock<std::mutex> lock(TasksMutex);
        if (StopRequested) {
            task->Cancel();
        }
        Tasks.push_back(task);
    }
    TasksCv.notify_all();
}

void TSerialClient::RequestStop()
{
    {
        std::unique_lock<std::mutex> lock(TasksMutex);
        StopRequested = true;
        for (const auto& task: Tasks) {
            task->Cancel();
        }
    }
    TasksCv.notify_all();
}

bool TSerialClient::WaitStopped(std::chrono::steady_clock::time_point deadline)
{
    std::unique_lock<std::mutex> lock(TasksMutex);
    return StoppedCv.wait_until(lock, deadline, [this]() { return Stopped; });
}

void TSerialClient::Resume()
{
    {
        std::unique_lock<std::mutex> lock(TasksMutex);
        StopRequested = false;
        Stopped = false;
    }
    TasksCv.notify_all();
}

bool TSerialClient::IsStopRequested()
{
    std::unique_lock<std::mutex> lock(TasksMutex);
    return StopRequested;
}

void TSerialClient::SuspendPoll(PSerialDevice device, std::chrono::steady_clock::time_point currentTime)
{
    std::lock_guard<std::mutex> lock(RegReaderMutex);
    if (!RegReader) {
        throw std::runtime_error("the polling is not started");
    }
    RegReader->SuspendPoll(device, currentTime);
}

void TSerialClient::ResumePoll(PSerialDevice device)
{
    std::lock_guard<std::mutex> lock(RegReaderMutex);
    if (!RegReader) {
        throw std::runtime_error("the polling is not started");
    }
    RegReader->ResumePoll(device);
}

TSerialClientRegisterAndEventsReader::TSerialClientRegisterAndEventsReader(const std::list<PSerialDevice>& devices,
                                                                           std::chrono::milliseconds readEventsPeriod,
                                                                           util::TGetNowFn nowFn,
                                                                           util::TGetSystemTimeFn systemTimeFn,
                                                                           size_t lowPriorityRateLimit)
    : EventsReader(std::make_shared<TSerialClientEventsReader>(MAX_EVENT_READ_ERRORS)),
      RegisterPoller(lowPriorityRateLimit),
      TimeBalancer(BALANCING_THRESHOLD),
      ReadEventsPeriod(readEventsPeriod),
      SpentTime(nowFn),
      LastCycleWasTooSmallToPoll(false),
      NowFn(nowFn),
      SystemTimeFn(systemTimeFn)
{
    auto currentTime = NowFn();
    RegisterPoller.SetDevices(devices, currentTime);
    EventsReader->SetDevices(devices);
    TimeBalancer.AddEntry(TClientTaskType::POLLING, currentTime, TPriority::Low);
}

void TSerialClientRegisterAndEventsReader::ClosedPortCycle(std::chrono::steady_clock::time_point currentTime,
                                                           TRegisterCallback regCallback)
{
    EventsReader->SetReadErrors(regCallback);
    RegisterPoller.ClosedPortCycle(currentTime, regCallback);
}

class TSerialClientTaskHandler
{
public:
    TClientTaskType TaskType;
    TItemAccumulationPolicy Policy;
    milliseconds PollLimit;
    bool NotReady = true;

    bool operator()(TClientTaskType task, TItemAccumulationPolicy policy, milliseconds pollLimit)
    {
        PollLimit = pollLimit;
        TaskType = task;
        Policy = policy;
        NotReady = false;
        return true;
    }
};

PSerialDevice TSerialClientRegisterAndEventsReader::OpenPortCycle(TFeaturePort& port,
                                                                  TRegisterCallback regCallback,
                                                                  TSerialClientDeviceAccessHandler& lastAccessedDevice)
{
    // Count idle time as high priority task time to faster reach time balancing threshold
    if (LastCycleWasTooSmallToPoll) {
        TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::High);
    }

    SpentTime.Start();
    TSerialClientTaskHandler handler;
    TimeBalancer.AccumulateNext(SpentTime.GetStartTime(), handler, TItemSelectionPolicy::All);
    if (handler.NotReady) {
        return nullptr;
    }

    if (handler.TaskType == TClientTaskType::EVENTS) {
        if (EventsReader && EventsReader->HasDevicesWithEnabledEvents()) {
            lastAccessedDevice.PrepareToAccess(port, nullptr);
            EventsReader->ReadEvents(port, MAX_POLL_TIME, regCallback, NowFn);
            TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::High);
            TimeBalancer.AddEntry(TClientTaskType::EVENTS,
                                  SpentTime.GetStartTime() + ReadEventsPeriod,
                                  TPriority::High);
        }
        SpentTime.Start();

        // TODO: Need to notify port open/close logic about errors
        return nullptr;
    }

    // Some registers can have theoretical read time more than poll limit.
    // Define special cases when reading can exceed poll limit to read the registers:
    // 1. TimeBalancer can force reading of such registers.
    // 2. If there are not devices with enabled events, the only limiting timeout is MAX_POLL_TIME.
    //    We can miss it and read at least one register.
    const bool readAtLeastOneRegister =
        (handler.Policy == TItemAccumulationPolicy::Force) || !EventsReader->HasDevicesWithEnabledEvents();

    auto res = RegisterPoller.OpenPortCycle(port,
                                            SpentTime,
                                            std::min(handler.PollLimit, MAX_POLL_TIME),
                                            readAtLeastOneRegister,
                                            lastAccessedDevice,
                                            regCallback,
                                            SystemTimeFn);

    TimeBalancer.AddEntry(TClientTaskType::POLLING, res.Deadline, TPriority::Low);
    if (res.NotEnoughTime) {
        LastCycleWasTooSmallToPoll = true;
    } else {
        LastCycleWasTooSmallToPoll = false;
        TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::Low);
    }

    if (EventsReader->HasDevicesWithEnabledEvents() && !TimeBalancer.Contains(TClientTaskType::EVENTS)) {
        TimeBalancer.AddEntry(TClientTaskType::EVENTS, SpentTime.GetStartTime() + ReadEventsPeriod, TPriority::High);
    }

    SpentTime.Start();
    return res.Device;
}

std::chrono::steady_clock::time_point TSerialClientRegisterAndEventsReader::GetDeadline(
    std::chrono::steady_clock::time_point currentTime) const
{
    return TimeBalancer.GetDeadline();
}

PSerialClientEventsReader TSerialClientRegisterAndEventsReader::GetEventsReader() const
{
    return EventsReader;
}

void TSerialClientRegisterAndEventsReader::SuspendPoll(PSerialDevice device,
                                                       std::chrono::steady_clock::time_point currentTime)
{
    RegisterPoller.SuspendPoll(device, currentTime);
}

void TSerialClientRegisterAndEventsReader::ResumePoll(PSerialDevice device)
{
    RegisterPoller.ResumePoll(device);
}

1	#include "serial_client.h"
2	#include <chrono>
3	#include <iostream>
4	#include <unistd.h>
5
6	#include "modbus_ext_common.h"
7	#include "write_channel_serial_client_task.h"
8
9	using namespace std::chrono_literals;
10	using namespace std::chrono;
11
12	#define LOG(logger) logger.Log() << "[serial client] "
13
14	namespace
15	{
16	const auto PORT_OPEN_ERROR_NOTIFICATION_INTERVAL = 5min;
17	const auto CLOSED_PORT_CYCLE_TIME = 500ms;
18	const auto MAX_POLL_TIME = 100ms;
19	// const auto MAX_FLUSHES_WHEN_POLL_IS_DUE = 20;
20	const auto BALANCING_THRESHOLD = 500ms;
21	// const auto MIN_READ_EVENTS_TIME = 25ms;
22	const size_t MAX_EVENT_READ_ERRORS = 10;
23
24	std::chrono::milliseconds GetReadEventsPeriod(const TPort& port)	131 ✔
25	{
26	auto sendByteTime = port.GetSendTimeBytes(1);	131 ✔
27	// >= 115200
28	if (sendByteTime < 100us) {	131 ✔
29	return 50ms;	2 ✔
30	}
31	// >= 38400
32	if (sendByteTime < 300us) {	129 ✔
33	return 100ms;	×
34	}
35	// < 38400
36	return 200ms;	129 ✔
37	}
38	};
39
40	void ISerialClientTask::Cancel()	13 ✔
41	{
42	Cancelled = true;	13 ✔
43	}	13 ✔
44
45	bool ISerialClientTask::IsCancelled() const	260 ✔
46	{
47	return Cancelled;	260 ✔
48	}
49
50	TSerialClient::TSerialClient(PFeaturePort port,	157 ✔
51	const TPortOpenCloseLogic::TSettings& openCloseSettings,
52	util::TGetNowFn nowFn,
53	size_t lowPriorityRateLimit)	157 ✔
54	: Port(port),	157 ✔
55	OpenCloseLogic(openCloseSettings, nowFn),	157 ✔
56	ConnectLogger(PORT_OPEN_ERROR_NOTIFICATION_INTERVAL, "[serial client] "),	314 ✔
57	NowFn(nowFn),	157 ✔
58	LowPriorityRateLimit(lowPriorityRateLimit)	628 ✔
59	{}	157 ✔
60
61	TSerialClient::~TSerialClient()	157 ✔
62	{
63	if (Port->IsOpen()) {	157 ✔
64	Port->Close();	58 ✔
65	}
66	}	157 ✔
67
68	void TSerialClient::AddDevice(PSerialDevice device)	149 ✔
69	{
70	if (RegReader)	149 ✔
71	throw TSerialDeviceException("can't add registers to the active client");	×
72	for (const auto& reg: device->GetRegisters()) {	764 ✔
73	if (Handlers.find(reg) != Handlers.end())	615 ✔
74	throw TSerialDeviceException("duplicate register");	×
75	auto handler = Handlers[reg] = std::make_shared<TRegisterHandler>(reg);	615 ✔
76	RegList.push_back(reg);	615 ✔
77	LOG(Debug) << "AddRegister: " << reg;	615 ✔
78	}	615 ✔
79	Devices.push_back(device);	149 ✔
80	}	149 ✔
81
82	void TSerialClient::Activate()	704 ✔
83	{
84	std::lock_guard<std::mutex> lock(RegReaderMutex);	704 ✔
85	if (!RegReader) {	704 ✔
86	RegReader = std::make_unique<TSerialClientRegisterAndEventsReader>(Devices,	131 ✔
87	GetReadEventsPeriod(*Port),	131 ✔
88	NowFn,	131 ✔
89	std::chrono::system_clock::now,
90	LowPriorityRateLimit);	131 ✔
91	LastAccessedDevice = std::make_unique<TSerialClientDeviceAccessHandler>(RegReader->GetEventsReader());	131 ✔
92	}
93	}	704 ✔
94
95	void TSerialClient::WaitForPollAndFlush(steady_clock::time_point currentTime, steady_clock::time_point waitUntil)	704 ✔
96	{
97	if (currentTime > waitUntil) {	704 ✔
98	waitUntil = currentTime;	642 ✔
99	}
100
101	if (Debug.IsEnabled()) {	704 ✔
102	LOG(Debug) << Port->GetDescription() << duration_cast<milliseconds>(currentTime.time_since_epoch()).count()	6 ✔
103	<< ": Wait until " << duration_cast<milliseconds>(waitUntil.time_since_epoch()).count();	3 ✔
104	}
105
106	std::vector<PSerialClientTask> retryTasks;	704 ✔
107	{
108	std::unique_lock<std::mutex> lock(TasksMutex);	704 ✔
109
110	while (TasksCv.wait_until(lock, waitUntil, [this]() { return !Tasks.empty() \|\| (StopRequested && !Stopped); }))	3,051 ✔
111	{
112	// One by one, so RequestStop cancels every task which has not started
113	while (!Tasks.empty()) {	266 ✔
114	auto task = Tasks.front();	163 ✔
115	Tasks.pop_front();	163 ✔
116	lock.unlock();	163 ✔
117	if (task->Run(Port, *LastAccessedDevice, Devices) == ISerialClientTask::TRunResult::RETRY) {	163 ✔
118	retryTasks.push_back(task);	13 ✔
119	}
120	lock.lock();	163 ✔
121	}	163 ✔
122	if (StopRequested) {	103 ✔
123	Stopped = true;	24 ✔
124	StoppedCv.notify_all();	24 ✔
125	}
126	}
127	}	704 ✔
128	for (auto& task: retryTasks) {	717 ✔
129	AddTask(task);	13 ✔
130	}
131	}	704 ✔
132
133	void TSerialClient::ProcessPolledRegister(PRegister reg)	1,116 ✔
134	{
135	if (reg->GetErrorState().test(TRegister::ReadError) \|\| reg->GetErrorState().test(TRegister::WriteError)) {	1,116 ✔
136	if (RegisterErrorCallback) {	187 ✔
137	RegisterErrorCallback(reg);	187 ✔
138	}
139	} else {
140	if (RegisterReadCallback) {	929 ✔
141	RegisterReadCallback(reg);	929 ✔
142	}
143	}
144	}	1,116 ✔
145
146	void TSerialClient::Cycle()	704 ✔
147	{
148	Activate();	704 ✔
149	if (IsStopRequested()) {	704 ✔
150	auto currentTime = NowFn();	1 ✔
151	WaitForPollAndFlush(currentTime, currentTime + CLOSED_PORT_CYCLE_TIME);	1 ✔
152	return;	1 ✔
153	}
154
155	try {
156	OpenCloseLogic.OpenIfAllowed(Port);	751 ✔
157	} catch (const std::exception& e) {	48 ✔
158	ConnectLogger.Log(e.what(), Debug, Error);	96 ✔
159	}	48 ✔
160
161	if (Port->IsOpen()) {	703 ✔
162	ConnectLogger.DropTimeout();	654 ✔
163	OpenPortCycle();	654 ✔
164	} else {
165	ClosedPortCycle();	49 ✔
166	}
167	}
168
169	void TSerialClient::ClosedPortCycle()	49 ✔
170	{
171	auto currentTime = NowFn();	49 ✔
172	auto waitUntil = currentTime + CLOSED_PORT_CYCLE_TIME;	49 ✔
173	WaitForPollAndFlush(currentTime, waitUntil);	49 ✔
174	if (IsStopRequested()) {	49 ✔
175	return;	20 ✔
176	}
177
178	RegReader->ClosedPortCycle(waitUntil, [this](PRegister reg) { ProcessPolledRegister(reg); });	66 ✔
179	}
180
181	void TSerialClient::SetTextValue(PRegister reg, const std::string& value)	121 ✔
182	{
183	if (IsStopRequested()) {	121 ✔
184	LOG(Debug) << reg->ToString() << " write is dropped, the client is stopped";	1 ✔
185	return;	1 ✔
186	}
187	auto handler = GetHandler(reg);	120 ✔
188	handler->SetTextValue(value);	120 ✔
189	auto serialClientTask =
190	std::make_shared<TWriteChannelSerialClientTask>(handler, RegisterReadCallback, RegisterErrorCallback);	120 ✔
191	AddTask(serialClientTask);	120 ✔
192	}	120 ✔
193
194	void TSerialClient::SetReadCallback(const TSerialClient::TRegisterCallback& callback)	157 ✔
195	{
196	RegisterReadCallback = callback;	157 ✔
197	}	157 ✔
198
199	void TSerialClient::SetErrorCallback(const TSerialClient::TRegisterCallback& callback)	157 ✔
200	{
201	RegisterErrorCallback = callback;	157 ✔
202	}	157 ✔
203
204	PRegisterHandler TSerialClient::GetHandler(PRegister reg) const	120 ✔
205	{
206	auto it = Handlers.find(reg);	120 ✔
207	if (it == Handlers.end())	120 ✔
208	throw TSerialDeviceException("register not found");	×
209	return it->second;	240 ✔
210	}
211
212	void TSerialClient::OpenPortCycle()	654 ✔
213	{
214	auto currentTime = NowFn();	654 ✔
215	auto waitUntil = RegReader->GetDeadline(currentTime);	654 ✔
216	// Limit waiting time to be responsive
217	waitUntil = std::min(waitUntil, currentTime + MAX_POLL_TIME);	654 ✔
218	WaitForPollAndFlush(currentTime, waitUntil);	654 ✔
219	if (IsStopRequested()) {	654 ✔
NEW 220	return;	×
221	}
222
223	auto device =
224	RegReader->OpenPortCycle(Port, [this](PRegister reg) { ProcessPolledRegister(reg); }, LastAccessedDevice);	1,733 ✔
225
226	if (device) {	654 ✔
227	OpenCloseLogic.CloseIfNeeded(Port, device->GetConnectionState() == TDeviceConnectionState::DISCONNECTED);	641 ✔
228	}
229	}	654 ✔
230
231	PFeaturePort TSerialClient::GetPort()	26 ✔
232	{
233	return Port;	26 ✔
234	}
235
236	std::list<PSerialDevice> TSerialClient::GetDevices()	34 ✔
237	{
238	return Devices;	34 ✔
239	}
240
241	void TSerialClient::AddTask(PSerialClientTask task)	164 ✔
242	{
243	{
244	std::unique_lock<std::mutex> lock(TasksMutex);	164 ✔
245	if (StopRequested) {	164 ✔
NEW 246	task->Cancel();	×
247	}
248	Tasks.push_back(task);	164 ✔
249	}	164 ✔
250	TasksCv.notify_all();	164 ✔
251	}	164 ✔
252
253	void TSerialClient::RequestStop()	25 ✔
254	{
255	{
256	std::unique_lock<std::mutex> lock(TasksMutex);	25 ✔
257	StopRequested = true;	25 ✔
258	for (const auto& task: Tasks) {	35 ✔
259	task->Cancel();	10 ✔
260	}
261	}	25 ✔
262	TasksCv.notify_all();	25 ✔
263	}	25 ✔
264
265	bool TSerialClient::WaitStopped(std::chrono::steady_clock::time_point deadline)	25 ✔
266	{
267	std::unique_lock<std::mutex> lock(TasksMutex);	25 ✔
268	return StoppedCv.wait_until(lock, deadline, [this]() { return Stopped; });	96 ✔
269	}	25 ✔
270
271	void TSerialClient::Resume()	5 ✔
272	{
273	{
274	std::unique_lock<std::mutex> lock(TasksMutex);	5 ✔
275	StopRequested = false;	5 ✔
276	Stopped = false;	5 ✔
277	}	5 ✔
278	TasksCv.notify_all();	5 ✔
279	}	5 ✔
280
281	bool TSerialClient::IsStopRequested()	1,528 ✔
282	{
283	std::unique_lock<std::mutex> lock(TasksMutex);	1,528 ✔
284	return StopRequested;	1,528 ✔
285	}	1,528 ✔
286
287	void TSerialClient::SuspendPoll(PSerialDevice device, std::chrono::steady_clock::time_point currentTime)	1 ✔
288	{
289	std::lock_guard<std::mutex> lock(RegReaderMutex);	1 ✔
290	if (!RegReader) {	1 ✔
NEW 291	throw std::runtime_error("the polling is not started");	×
292	}
293	RegReader->SuspendPoll(device, currentTime);	1 ✔
294	}	1 ✔
295
296	void TSerialClient::ResumePoll(PSerialDevice device)	1 ✔
297	{
298	std::lock_guard<std::mutex> lock(RegReaderMutex);	1 ✔
299	if (!RegReader) {	1 ✔
NEW 300	throw std::runtime_error("the polling is not started");	×
301	}
302	RegReader->ResumePoll(device);	2 ✔
303	}	1 ✔
304
305	TSerialClientRegisterAndEventsReader::TSerialClientRegisterAndEventsReader(const std::list<PSerialDevice>& devices,	156 ✔
306	std::chrono::milliseconds readEventsPeriod,
307	util::TGetNowFn nowFn,
308	util::TGetSystemTimeFn systemTimeFn,
309	size_t lowPriorityRateLimit)	156 ✔
310	: EventsReader(std::make_shared<TSerialClientEventsReader>(MAX_EVENT_READ_ERRORS)),	156 ✔
311	RegisterPoller(lowPriorityRateLimit),	156 ✔
312	TimeBalancer(BALANCING_THRESHOLD),	156 ✔
313	ReadEventsPeriod(readEventsPeriod),	156 ✔
314	SpentTime(nowFn),	156 ✔
315	LastCycleWasTooSmallToPoll(false),	156 ✔
316	NowFn(nowFn),	156 ✔
317	SystemTimeFn(systemTimeFn)	156 ✔
318	{
319	auto currentTime = NowFn();	156 ✔
320	RegisterPoller.SetDevices(devices, currentTime);	156 ✔
321	EventsReader->SetDevices(devices);	156 ✔
322	TimeBalancer.AddEntry(TClientTaskType::POLLING, currentTime, TPriority::Low);	156 ✔
323	}	156 ✔
324
325	void TSerialClientRegisterAndEventsReader::ClosedPortCycle(std::chrono::steady_clock::time_point currentTime,	29 ✔
326	TRegisterCallback regCallback)
327	{
328	EventsReader->SetReadErrors(regCallback);	29 ✔
329	RegisterPoller.ClosedPortCycle(currentTime, regCallback);	29 ✔
330	}	29 ✔
331
332	class TSerialClientTaskHandler
333	{
334	public:
335	TClientTaskType TaskType;
336	TItemAccumulationPolicy Policy;
337	milliseconds PollLimit;
338	bool NotReady = true;
339
340	bool operator()(TClientTaskType task, TItemAccumulationPolicy policy, milliseconds pollLimit)	957 ✔
341	{
342	PollLimit = pollLimit;	957 ✔
343	TaskType = task;	957 ✔
344	Policy = policy;	957 ✔
345	NotReady = false;	957 ✔
346	return true;	957 ✔
347	}
348	};
349
350	PSerialDevice TSerialClientRegisterAndEventsReader::OpenPortCycle(TFeaturePort& port,	968 ✔
351	TRegisterCallback regCallback,
352	TSerialClientDeviceAccessHandler& lastAccessedDevice)
353	{
354	// Count idle time as high priority task time to faster reach time balancing threshold
355	if (LastCycleWasTooSmallToPoll) {	968 ✔
356	TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::High);	58 ✔
357	}
358
359	SpentTime.Start();	968 ✔
360	TSerialClientTaskHandler handler;	968 ✔
361	TimeBalancer.AccumulateNext(SpentTime.GetStartTime(), handler, TItemSelectionPolicy::All);	968 ✔
362	if (handler.NotReady) {	968 ✔
363	return nullptr;	11 ✔
364	}
365
366	if (handler.TaskType == TClientTaskType::EVENTS) {	957 ✔
367	if (EventsReader && EventsReader->HasDevicesWithEnabledEvents()) {	90 ✔
368	lastAccessedDevice.PrepareToAccess(port, nullptr);	87 ✔
369	EventsReader->ReadEvents(port, MAX_POLL_TIME, regCallback, NowFn);	87 ✔
370	TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::High);	87 ✔
371	TimeBalancer.AddEntry(TClientTaskType::EVENTS,	87 ✔
372	SpentTime.GetStartTime() + ReadEventsPeriod,	174 ✔
373	TPriority::High);
374	}
375	SpentTime.Start();	90 ✔
376
377	// TODO: Need to notify port open/close logic about errors
378	return nullptr;	90 ✔
379	}
380
381	// Some registers can have theoretical read time more than poll limit.
382	// Define special cases when reading can exceed poll limit to read the registers:
383	// 1. TimeBalancer can force reading of such registers.
384	// 2. If there are not devices with enabled events, the only limiting timeout is MAX_POLL_TIME.
385	// We can miss it and read at least one register.
386	const bool readAtLeastOneRegister =
387	(handler.Policy == TItemAccumulationPolicy::Force) \|\| !EventsReader->HasDevicesWithEnabledEvents();	867 ✔
388
389	auto res = RegisterPoller.OpenPortCycle(port,	1,734 ✔
390	SpentTime,	867 ✔
391	std::min(handler.PollLimit, MAX_POLL_TIME),	867 ✔
392	readAtLeastOneRegister,
393	lastAccessedDevice,
394	regCallback,
395	SystemTimeFn);	1,734 ✔
396
397	TimeBalancer.AddEntry(TClientTaskType::POLLING, res.Deadline, TPriority::Low);	867 ✔
398	if (res.NotEnoughTime) {	867 ✔
399	LastCycleWasTooSmallToPoll = true;	37 ✔
400	} else {
401	LastCycleWasTooSmallToPoll = false;	830 ✔
402	TimeBalancer.UpdateSelectionTime(ceil<milliseconds>(SpentTime.GetSpentTime()), TPriority::Low);	830 ✔
403	}
404
405	if (EventsReader->HasDevicesWithEnabledEvents() && !TimeBalancer.Contains(TClientTaskType::EVENTS)) {	867 ✔
406	TimeBalancer.AddEntry(TClientTaskType::EVENTS, SpentTime.GetStartTime() + ReadEventsPeriod, TPriority::High);	15 ✔
407	}
408
409	SpentTime.Start();	867 ✔
410	return res.Device;	867 ✔
411	}	867 ✔
412
413	std::chrono::steady_clock::time_point TSerialClientRegisterAndEventsReader::GetDeadline(	968 ✔
414	std::chrono::steady_clock::time_point currentTime) const
415	{
416	return TimeBalancer.GetDeadline();	968 ✔
417	}
418
419	PSerialClientEventsReader TSerialClientRegisterAndEventsReader::GetEventsReader() const	156 ✔
420	{
421	return EventsReader;	156 ✔
422	}
423
424	void TSerialClientRegisterAndEventsReader::SuspendPoll(PSerialDevice device,	4 ✔
425	std::chrono::steady_clock::time_point currentTime)
426	{
427	RegisterPoller.SuspendPoll(device, currentTime);	4 ✔
428	}	4 ✔
429
430	void TSerialClientRegisterAndEventsReader::ResumePoll(PSerialDevice device)	3 ✔
431	{
432	RegisterPoller.ResumePoll(device);	3 ✔
433	}	2 ✔

wirenboard / wb-mqtt-serial / 967

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