wirenboard / wb-mqtt-serial / 690

Committed 26 Aug 2025 01:28PM UTC coverage: 73.031%. Remained the same

Build # 690

Build Type

push

github

Committed by

web-flow

Commit Message

Add protocol parameter to device/Probe RPC (#988)

  * Add protocol to Modbus TCP port information in ports/Load RPC
  * Add protocol parameter to device/Probe RPC
  * Set TCP_NODELAY for TCP ports
  * Handle TCP port closing by remote

Run Details

6605 of 9406 branches covered (70.22%)

28 of 59 new or added lines in 11 files covered. (47.46%)

136 existing lines in 8 files now uncovered.

12554 of 17190 relevant lines covered (73.03%)

372.2 hits per line

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

54.62

/src/file_descriptor_port.cpp

#include "file_descriptor_port.h"
#include "common_utils.h"
#include "serial_exc.h"

#include <iomanip>
#include <iostream>
#include <sys/ioctl.h>
#include <sys/select.h>
#include <unistd.h>
#include <wblib/utils.h>

#include "log.h"

using namespace std;

#define LOG(logger) ::logger.Log() << "[port] "

namespace
{
    const chrono::milliseconds NoiseTimeout(1);
    const chrono::milliseconds ContinuousNoiseTimeout(100);
    const int ContinuousNoiseReopenNumber = 3;
}

TFileDescriptorPort::TFileDescriptorPort(): Fd(-1)
{}

TFileDescriptorPort::~TFileDescriptorPort()
{
    if (TFileDescriptorPort::IsOpen()) {
        close(Fd);
    }
}

void TFileDescriptorPort::Close()
{
    if (IsOpen()) {
        close(Fd);
        Fd = -1;
    }
}

bool TFileDescriptorPort::IsOpen() const
{
    return Fd >= 0;
}

void TFileDescriptorPort::CheckPortOpen() const
{
    if (!IsOpen()) {
        throw TSerialDeviceException("port not open");
    }
}

void TFileDescriptorPort::WriteBytes(const uint8_t* buf, int count)
{
    auto res = write(Fd, buf, count);
    if (res < count) {
        if (res < 0) {
            if (errno == EPIPE) {
                LOG(Warn) << GetDescription(false) << ": Port closed by remote side";
                Close();
            }
            throw TSerialDeviceErrnoException("serial write failed: ", errno);
        }
        stringstream ss;
        ss << "serial write failed: " << res << " bytes of " << count << " is written";
        throw TSerialDeviceException(ss.str());
    }

    LastInteraction = std::chrono::steady_clock::now();

    if (::Debug.IsEnabled()) {
        LOG(Debug) << GetDescription(false) << ": Write: " << WBMQTT::HexDump(buf, count);
    }
}

bool TFileDescriptorPort::Select(const chrono::microseconds& us)
{
    fd_set rfds;
    struct timeval tv, *tvp = 0;

    FD_ZERO(&rfds);
    FD_SET(Fd, &rfds);
    if (us.count() > 0) {
        tv.tv_sec = us.count() / 1000000;
        tv.tv_usec = us.count() % 1000000;
        tvp = &tv;
    }

    int r = select(Fd + 1, &rfds, NULL, NULL, tvp);
    if (r < 0) {
        throw TSerialDeviceErrnoException("TFileDescriptorPort::Select() failed: ", errno);
    }

    return r > 0;
}

void TFileDescriptorPort::OnReadyEmptyFd()
{}

uint8_t TFileDescriptorPort::ReadByte(const chrono::microseconds& timeout)
{
    CheckPortOpen();

    if (!Select(timeout)) {
        throw TSerialDeviceTransientErrorException("timeout");
    }

    uint8_t b;
    if (read(Fd, &b, 1) < 1) {
        throw TSerialDeviceException("read() failed");
    }

    LastInteraction = std::chrono::steady_clock::now();

    LOG(Debug) << GetDescription(false) << ": Read: " << hex << setw(2) << setfill('0') << int(b);

    return b;
}

size_t TFileDescriptorPort::ReadAvailableData(uint8_t* buf, size_t max_read)
{
    // We don't want to use non-blocking IO in general
    // (e.g. we want blocking writes), but we don't want
    // read() call below to block because actual frame
    // size is not known at this point. So we must
    // know how many bytes are available
    int nb = 0;
    if (ioctl(Fd, FIONREAD, &nb) < 0) {
        throw TSerialDeviceException("FIONREAD ioctl() failed");
    }

    // Got Fd as ready for read from select, but no actual data to read
    if (!nb) {
        OnReadyEmptyFd();
        return 0;
    }

    if (nb >= 0 && static_cast<size_t>(nb) > max_read) {
        nb = max_read;
    }

    int n = read(Fd, buf, nb);
    if (n < 0) {
        throw TSerialDeviceException("read() failed");
    }

    if (n < nb) { // may happen only due to a kernel/driver bug
        throw TSerialDeviceException("short read()");
    }

    return nb;
}

// Reading becomes unstable when using timeout less than default because of bufferization
TReadFrameResult TFileDescriptorPort::ReadFrame(uint8_t* buf,
                                                size_t size,
                                                const std::chrono::microseconds& responseTimeout,
                                                const std::chrono::microseconds& frameTimeout,
                                                TFrameCompletePred frame_complete)
{
    CheckPortOpen();
    TReadFrameResult res;

    if (!size) {
        return res;
    }

    util::TSpentTimeMeter spentTime(std::chrono::steady_clock::now);
    spentTime.Start();

    // Will wait first byte up to responseTimeout us
    auto selectTimeout = responseTimeout;
    while (res.Count < size) {
        if (frame_complete && frame_complete(buf, res.Count)) {
            break;
        }

        if (!Select(selectTimeout))
            break; // end of the frame

        size_t nb = ReadAvailableData(buf + res.Count, size - res.Count);

        // Got Fd as ready for read from select, but no actual data to read
        if (nb == 0) {
            continue;
        }

        // Got something, switch to frameTimeout to detect frame boundary
        // Delay between bytes in one message can't be more than frameTimeout
        selectTimeout = frameTimeout;
        if (res.Count == 0) {
            res.ResponseTime = spentTime.GetSpentTime();
        }
        res.Count += nb;
    }

    if (!res.Count) {
        throw TResponseTimeoutException();
    }

    LastInteraction = std::chrono::steady_clock::now();

    if (::Debug.IsEnabled()) {
        LOG(Debug) << GetDescription(false) << ": ReadFrame: " << WBMQTT::HexDump(buf, res.Count);
    }

    return res;
}

void TFileDescriptorPort::SkipNoise()
{
    uint8_t buf[255] = {};
    auto start = std::chrono::steady_clock::now();
    int ntries = 0;

    while (Select(NoiseTimeout)) {
        size_t nread = ReadAvailableData(buf, sizeof(buf) / sizeof(buf[0]));
        auto diff = std::chrono::steady_clock::now() - start;

        if (::Debug.IsEnabled()) {
            LOG(Debug) << "read noise: " << WBMQTT::HexDump(buf, nread);
        }

        // if we are still getting data for already "ContinuousNoiseTimeout" milliseconds
        if (nread > 0) {
            LastInteraction = std::chrono::steady_clock::now();

            if (diff > ContinuousNoiseTimeout) {
                if (ntries < ContinuousNoiseReopenNumber) {
                    LOG(Debug) << "continuous unsolicited data flow detected, reopen the port";
                    Reopen();
                    ntries += 1;
                    start = std::chrono::steady_clock::now();
                } else {
                    throw TSerialDeviceTransientErrorException("continuous unsolicited data flow");
                }
            }
        }
    }
}

void TFileDescriptorPort::SleepSinceLastInteraction(const chrono::microseconds& us)
{
    auto now = chrono::steady_clock::now();
    auto delta = chrono::duration_cast<chrono::microseconds>(now - LastInteraction);
    std::this_thread::sleep_for(us - delta);
    LOG(Debug) << GetDescription(false) << ": Sleep " << us.count() << " us";
}

1	#include "file_descriptor_port.h"
2	#include "common_utils.h"
3	#include "serial_exc.h"
4
5	#include <iomanip>
6	#include <iostream>
7	#include <sys/ioctl.h>
8	#include <sys/select.h>
9	#include <unistd.h>
10	#include <wblib/utils.h>
11
12	#include "log.h"
13
14	using namespace std;
15
16	#define LOG(logger) ::logger.Log() << "[port] "
17
18	namespace
19	{
20	const chrono::milliseconds NoiseTimeout(1);
21	const chrono::milliseconds ContinuousNoiseTimeout(100);
22	const int ContinuousNoiseReopenNumber = 3;
23	}
24
25	TFileDescriptorPort::TFileDescriptorPort(): Fd(-1)	25✔
26	{}	25✔
27
28	TFileDescriptorPort::~TFileDescriptorPort()	50✔
29	{
30	if (TFileDescriptorPort::IsOpen()) {	50✔
31	close(Fd);	×
32	}
33	}
34
35	void TFileDescriptorPort::Close()	2✔
36	{
37	if (IsOpen()) {	2✔
38	close(Fd);	2✔
39	Fd = -1;	2✔
40	}
41	}	2✔
42
43	bool TFileDescriptorPort::IsOpen() const	32✔
44	{
45	return Fd >= 0;	32✔
46	}
47
48	void TFileDescriptorPort::CheckPortOpen() const	1✔
49	{
50	if (!IsOpen()) {	1✔
51	throw TSerialDeviceException("port not open");	×
52	}
53	}	1✔
54
55	void TFileDescriptorPort::WriteBytes(const uint8_t* buf, int count)	2✔
56	{
57	auto res = write(Fd, buf, count);	2✔
58	if (res < count) {	2✔
59	if (res < 0) {	×
NEW 60	if (errno == EPIPE) {	×
NEW 61	LOG(Warn) << GetDescription(false) << ": Port closed by remote side";	×
NEW 62	Close();	×
63	}
64	throw TSerialDeviceErrnoException("serial write failed: ", errno);	×
65	}
66	stringstream ss;	×
UNCOV 67	ss << "serial write failed: " << res << " bytes of " << count << " is written";	×
68	throw TSerialDeviceException(ss.str());	×
69	}
70
71	LastInteraction = std::chrono::steady_clock::now();	2✔
72
73	if (::Debug.IsEnabled()) {	2✔
UNCOV 74	LOG(Debug) << GetDescription(false) << ": Write: " << WBMQTT::HexDump(buf, count);	×
75	}
76	}	2✔
77
78	bool TFileDescriptorPort::Select(const chrono::microseconds& us)	5✔
79	{
80	fd_set rfds;
81	struct timeval tv, *tvp = 0;	5✔
82
83	FD_ZERO(&rfds);	165✔
84	FD_SET(Fd, &rfds);	5✔
85	if (us.count() > 0) {	5✔
86	tv.tv_sec = us.count() / 1000000;	5✔
87	tv.tv_usec = us.count() % 1000000;	5✔
88	tvp = &tv;	5✔
89	}
90
91	int r = select(Fd + 1, &rfds, NULL, NULL, tvp);	5✔
92	if (r < 0) {	5✔
UNCOV 93	throw TSerialDeviceErrnoException("TFileDescriptorPort::Select() failed: ", errno);	×
94	}
95
96	return r > 0;	5✔
97	}
98
UNCOV 99	void TFileDescriptorPort::OnReadyEmptyFd()	×
100	{}
101
102	uint8_t TFileDescriptorPort::ReadByte(const chrono::microseconds& timeout)	1✔
103	{
104	CheckPortOpen();	1✔
105
106	if (!Select(timeout)) {	1✔
UNCOV 107	throw TSerialDeviceTransientErrorException("timeout");	×
108	}
109
110	uint8_t b;
111	if (read(Fd, &b, 1) < 1) {	1✔
UNCOV 112	throw TSerialDeviceException("read() failed");	×
113	}
114
115	LastInteraction = std::chrono::steady_clock::now();	1✔
116
117	LOG(Debug) << GetDescription(false) << ": Read: " << hex << setw(2) << setfill('0') << int(b);	1✔
118
119	return b;	1✔
120	}
121
122	size_t TFileDescriptorPort::ReadAvailableData(uint8_t* buf, size_t max_read)	1✔
123	{
124	// We don't want to use non-blocking IO in general
125	// (e.g. we want blocking writes), but we don't want
126	// read() call below to block because actual frame
127	// size is not known at this point. So we must
128	// know how many bytes are available
129	int nb = 0;	1✔
130	if (ioctl(Fd, FIONREAD, &nb) < 0) {	1✔
UNCOV 131	throw TSerialDeviceException("FIONREAD ioctl() failed");	×
132	}
133
134	// Got Fd as ready for read from select, but no actual data to read
135	if (!nb) {	1✔
UNCOV 136	OnReadyEmptyFd();	×
UNCOV 137	return 0;	×
138	}
139
140	if (nb >= 0 && static_cast<size_t>(nb) > max_read) {	1✔
141	nb = max_read;	×
142	}
143
144	int n = read(Fd, buf, nb);	1✔
145	if (n < 0) {	1✔
UNCOV 146	throw TSerialDeviceException("read() failed");	×
147	}
148
149	if (n < nb) { // may happen only due to a kernel/driver bug	1✔
150	throw TSerialDeviceException("short read()");	×
151	}
152
153	return nb;	1✔
154	}
155
156	// Reading becomes unstable when using timeout less than default because of bufferization
UNCOV 157	TReadFrameResult TFileDescriptorPort::ReadFrame(uint8_t* buf,	×
158	size_t size,
159	const std::chrono::microseconds& responseTimeout,
160	const std::chrono::microseconds& frameTimeout,
161	TFrameCompletePred frame_complete)
162	{
UNCOV 163	CheckPortOpen();	×
UNCOV 164	TReadFrameResult res;	×
165
UNCOV 166	if (!size) {	×
167	return res;	×
168	}
169
170	util::TSpentTimeMeter spentTime(std::chrono::steady_clock::now);	×
171	spentTime.Start();	×
172
173	// Will wait first byte up to responseTimeout us
174	auto selectTimeout = responseTimeout;	×
175	while (res.Count < size) {	×
UNCOV 176	if (frame_complete && frame_complete(buf, res.Count)) {	×
UNCOV 177	break;	×
178	}
179
180	if (!Select(selectTimeout))	×
181	break; // end of the frame	×
182
UNCOV 183	size_t nb = ReadAvailableData(buf + res.Count, size - res.Count);	×
184
185	// Got Fd as ready for read from select, but no actual data to read
UNCOV 186	if (nb == 0) {	×
187	continue;	×
188	}
189
190	// Got something, switch to frameTimeout to detect frame boundary
191	// Delay between bytes in one message can't be more than frameTimeout
UNCOV 192	selectTimeout = frameTimeout;	×
UNCOV 193	if (res.Count == 0) {	×
UNCOV 194	res.ResponseTime = spentTime.GetSpentTime();	×
195	}
196	res.Count += nb;	×
197	}
198
UNCOV 199	if (!res.Count) {	×
200	throw TResponseTimeoutException();	×
201	}
202
203	LastInteraction = std::chrono::steady_clock::now();	×
204
UNCOV 205	if (::Debug.IsEnabled()) {	×
UNCOV 206	LOG(Debug) << GetDescription(false) << ": ReadFrame: " << WBMQTT::HexDump(buf, res.Count);	×
207	}
208
209	return res;	×
210	}
211
212	void TFileDescriptorPort::SkipNoise()	3✔
213	{
214	uint8_t buf[255] = {};	3✔
215	auto start = std::chrono::steady_clock::now();	3✔
216	int ntries = 0;	3✔
217
218	while (Select(NoiseTimeout)) {	4✔
219	size_t nread = ReadAvailableData(buf, sizeof(buf) / sizeof(buf[0]));	1✔
220	auto diff = std::chrono::steady_clock::now() - start;	1✔
221
222	if (::Debug.IsEnabled()) {	1✔
UNCOV 223	LOG(Debug) << "read noise: " << WBMQTT::HexDump(buf, nread);	×
224	}
225
226	// if we are still getting data for already "ContinuousNoiseTimeout" milliseconds
227	if (nread > 0) {	1✔
228	LastInteraction = std::chrono::steady_clock::now();	1✔
229
230	if (diff > ContinuousNoiseTimeout) {	1✔
UNCOV 231	if (ntries < ContinuousNoiseReopenNumber) {	×
UNCOV 232	LOG(Debug) << "continuous unsolicited data flow detected, reopen the port";	×
UNCOV 233	Reopen();	×
UNCOV 234	ntries += 1;	×
235	start = std::chrono::steady_clock::now();	×
236	} else {
237	throw TSerialDeviceTransientErrorException("continuous unsolicited data flow");	×
238	}
239	}
240	}
241	}
242	}	3✔
243
244	void TFileDescriptorPort::SleepSinceLastInteraction(const chrono::microseconds& us)	2✔
245	{
246	auto now = chrono::steady_clock::now();	2✔
247	auto delta = chrono::duration_cast<chrono::microseconds>(now - LastInteraction);	2✔
248	std::this_thread::sleep_for(us - delta);	2✔
249	LOG(Debug) << GetDescription(false) << ": Sleep " << us.count() << " us";	2✔
250	}	2✔

wirenboard / wb-mqtt-serial / 690

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