wirenboard / wb-mqtt-smartweb / 2

Committed 31 Jul 2026 10:01AM UTC coverage: 33.897% (-0.4%) from 34.328%

Build # 2

Build Type

Pull #40

github

Specific Base 476708

Committed by

ekateluv

Commit Message

Lower coverage threshold to 33

Pull Request Pull Request #40: Use exported vars instead of MAKEFLAGS for coverage options

Coverage Stats

307 of 744 branches covered (41.26%)

501 of 1478 relevant lines covered (33.9%)

20.03 hits per line

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

0.0

/src/CanPort.cpp

#include "CanPort.h"

#include <algorithm>
#include <condition_variable>
#include <net/if.h>
#include <queue>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include <vector>
#include <wblib/utils.h>

#include "exceptions.h"
#include "log.h"

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

namespace
{
    const auto READ_TIMEOUT_MS = std::chrono::milliseconds(1000); // 1 sec for messages waiting
    const auto WRITE_TIMEOUT = std::chrono::seconds(5);           // 5 sec wait for port ready to write

    template<class TDuration> void setTimeval(timeval& tv, TDuration timeout)
    {
        tv.tv_sec = std::chrono::ceil<std::chrono::seconds>(timeout).count();
        tv.tv_usec = (std::chrono::ceil<std::chrono::microseconds>(timeout).count() % 1000) * 1000;
    }

    void initMsghdr(msghdr& msg, can_frame& frame, iovec& iov, uint8_t* ctrlmsg, size_t ctrlmsgSize)
    {
        iov.iov_base = &frame;
        iov.iov_len = sizeof(frame);
        msg.msg_iov = &iov;
        msg.msg_iovlen = 1;
        msg.msg_control = &ctrlmsg;
        msg.msg_controllen = ctrlmsgSize;
        msg.msg_flags = 0;
    }

    class TCanPort: public CAN::IPort
    {
        int Socket;
        std::thread Thread;
        std::atomic_bool Enabled;
        std::mutex HandlersMutex;
        std::mutex WriteMutex;
        std::vector<CAN::IFrameHandler*> Handlers;

        std::mutex WriteConfirmMutex;
        std::condition_variable WriteConfirmCv;
        bool WriteConfirmed = false;

        void SetWriteConfirmed()
        {
            std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);
            WriteConfirmed = true;
            WriteConfirmCv.notify_all();
        }

        void RunHandlers(const CAN::TFrame& frame)
        {
            std::unique_lock<std::mutex> lk(HandlersMutex);
            for (auto& handler: Handlers) {
                try {
                    if (handler->Handle(frame)) {
                        break;
                    }
                } catch (const std::exception& e) {
                    LOG(WBMQTT::Error) << e.what();
                }
            }
        }

        void ThreadFn()
        {
            while (Enabled.load()) {
                timeval tv;
                setTimeval(tv, READ_TIMEOUT_MS);
                fd_set rfds;
                FD_ZERO(&rfds);
                FD_SET(Socket, &rfds);
                int r = select(Socket + 1, &rfds, nullptr, nullptr, &tv);
                if (r < 0) {
                    LOG(WBMQTT::Error) << "select() failed " << strerror(errno);
                    exit(1);
                }
                if (r > 0) {
                    msghdr msg{0};
                    can_frame frame{0};
                    iovec iov{0};
                    uint8_t ctrlmsg[CMSG_SPACE(sizeof(struct timeval)) + CMSG_SPACE(sizeof(__u32))];
                    initMsghdr(msg, frame, iov, ctrlmsg, sizeof(ctrlmsg));

                    auto nread = recvmsg(Socket, &msg, 0);
                    if (nread == sizeof(CAN::TFrame)) {
                        if (msg.msg_flags & MSG_CONFIRM) {
                            SetWriteConfirmed();
                        } else {
                            RunHandlers(frame);
                        }
                    } else {
                        if (nread < 0) {
                            LOG(WBMQTT::Error) << "read() failed " << strerror(errno);
                            exit(1);
                        }
                        LOG(WBMQTT::Error) << "Got " << nread << " instead of " << sizeof(CAN::TFrame) << " bytes";
                    }
                }
            }
        }

    public:
        TCanPort(const std::string& ifname)
        {
            struct sockaddr_can addr;
            struct ifreq ifr;

            if ((Socket = socket(PF_CAN, SOCK_RAW, CAN_RAW)) < 0) {
                throw std::runtime_error(std::string("Error while opening CAN socket: ") + strerror(errno));
            }

            int enable = 1;
            setsockopt(Socket, SOL_CAN_RAW, CAN_RAW_LOOPBACK, &enable, sizeof(enable));
            setsockopt(Socket, SOL_CAN_RAW, CAN_RAW_RECV_OWN_MSGS, &enable, sizeof(enable));

            strncpy(ifr.ifr_name, ifname.c_str(), IFNAMSIZ - 1);
            ifr.ifr_name[IFNAMSIZ - 1] = '\0';
            ifr.ifr_ifindex = if_nametoindex(ifr.ifr_name);
            if (!ifr.ifr_ifindex) {
                throw TInterfaceNotFoundError(std::string("if_nametoindex failed for interface '") + ifr.ifr_name +
                                              "', " + strerror(errno));
            }

            addr.can_family = AF_CAN;
            addr.can_ifindex = ifr.ifr_ifindex;

            LOG(WBMQTT::Info) << ifname.c_str() << " at index " << ifr.ifr_ifindex;

            if (bind(Socket, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
                throw std::runtime_error(std::string("Error in CAN socket bind: ") + strerror(errno));
            }

            Enabled.store(true);

            Thread = std::thread([this]() {
                WBMQTT::SetThreadName("CAN listener");
                ThreadFn();
            });
        }

        ~TCanPort()
        {
            Enabled.store(false);
            if (Thread.joinable()) {
                Thread.join();
            }
            close(Socket);
        }

        void AddHandler(CAN::IFrameHandler* handler)
        {
            std::unique_lock<std::mutex> lk(HandlersMutex);
            Handlers.push_back(handler);
        }

        void RemoveHandler(CAN::IFrameHandler* handler)
        {
            std::unique_lock<std::mutex> lk(HandlersMutex);
            Handlers.erase(std::remove(Handlers.begin(), Handlers.end(), handler), Handlers.end());
        }

        void Send(const CAN::TFrame& frame)
        {
            std::unique_lock<std::mutex> lk(WriteMutex);
            {
                std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);
                WriteConfirmed = false;
            }

            auto nbytes = write(Socket, &frame, CAN_MTU);

            if (nbytes < static_cast<int>(CAN_MTU)) {
                throw std::runtime_error(std::string("CAN write error: ") + strerror(errno));
            }
            std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);
            if (WriteConfirmed) {
                return;
            }
            if (std::cv_status::timeout == WriteConfirmCv.wait_for(waitLock, WRITE_TIMEOUT)) {
                throw std::runtime_error("CAN write timeout");
            }
        }
    };
}

std::shared_ptr<CAN::IPort> CAN::MakePort(const std::string& ifname)
{
    return std::make_shared<TCanPort>(ifname);
}

1	#include "CanPort.h"
2
3	#include <algorithm>
4	#include <condition_variable>
5	#include <net/if.h>
6	#include <queue>
7	#include <string.h>
8	#include <sys/ioctl.h>
9	#include <sys/socket.h>
10	#include <sys/types.h>
11	#include <unistd.h>
12	#include <vector>
13	#include <wblib/utils.h>
14
15	#include "exceptions.h"
16	#include "log.h"
17
18	#define LOG(logger) ::logger.Log() << "[CAN] "
19
20	namespace
21	{
22	const auto READ_TIMEOUT_MS = std::chrono::milliseconds(1000); // 1 sec for messages waiting
23	const auto WRITE_TIMEOUT = std::chrono::seconds(5); // 5 sec wait for port ready to write
24
25	template<class TDuration> void setTimeval(timeval& tv, TDuration timeout)	×
26	{
27	tv.tv_sec = std::chrono::ceil<std::chrono::seconds>(timeout).count();	×
28	tv.tv_usec = (std::chrono::ceil<std::chrono::microseconds>(timeout).count() % 1000) * 1000;	×
29	}	×
30
31	void initMsghdr(msghdr& msg, can_frame& frame, iovec& iov, uint8_t* ctrlmsg, size_t ctrlmsgSize)	×
32	{
33	iov.iov_base = &frame;	×
34	iov.iov_len = sizeof(frame);	×
35	msg.msg_iov = &iov;	×
36	msg.msg_iovlen = 1;	×
37	msg.msg_control = &ctrlmsg;	×
38	msg.msg_controllen = ctrlmsgSize;	×
39	msg.msg_flags = 0;	×
40	}	×
41
42	class TCanPort: public CAN::IPort
43	{
44	int Socket;
45	std::thread Thread;
46	std::atomic_bool Enabled;
47	std::mutex HandlersMutex;
48	std::mutex WriteMutex;
49	std::vector<CAN::IFrameHandler*> Handlers;
50
51	std::mutex WriteConfirmMutex;
52	std::condition_variable WriteConfirmCv;
53	bool WriteConfirmed = false;
54
55	void SetWriteConfirmed()	×
56	{
57	std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);	×
58	WriteConfirmed = true;	×
59	WriteConfirmCv.notify_all();	×
60	}	×
61
62	void RunHandlers(const CAN::TFrame& frame)	×
63	{
64	std::unique_lock<std::mutex> lk(HandlersMutex);	×
65	for (auto& handler: Handlers) {	×
66	try {
67	if (handler->Handle(frame)) {	×
68	break;	×
69	}
70	} catch (const std::exception& e) {	×
71	LOG(WBMQTT::Error) << e.what();	×
72	}	×
73	}
74	}	×
75
76	void ThreadFn()	×
77	{
78	while (Enabled.load()) {	×
79	timeval tv;
80	setTimeval(tv, READ_TIMEOUT_MS);	×
81	fd_set rfds;
82	FD_ZERO(&rfds);	×
83	FD_SET(Socket, &rfds);	×
84	int r = select(Socket + 1, &rfds, nullptr, nullptr, &tv);	×
85	if (r < 0) {	×
86	LOG(WBMQTT::Error) << "select() failed " << strerror(errno);	×
87	exit(1);	×
88	}
89	if (r > 0) {	×
90	msghdr msg{0};	×
91	can_frame frame{0};	×
92	iovec iov{0};	×
93	uint8_t ctrlmsg[CMSG_SPACE(sizeof(struct timeval)) + CMSG_SPACE(sizeof(__u32))];
94	initMsghdr(msg, frame, iov, ctrlmsg, sizeof(ctrlmsg));	×
95
96	auto nread = recvmsg(Socket, &msg, 0);	×
97	if (nread == sizeof(CAN::TFrame)) {	×
98	if (msg.msg_flags & MSG_CONFIRM) {	×
99	SetWriteConfirmed();	×
100	} else {
101	RunHandlers(frame);	×
102	}
103	} else {
104	if (nread < 0) {	×
105	LOG(WBMQTT::Error) << "read() failed " << strerror(errno);	×
106	exit(1);	×
107	}
108	LOG(WBMQTT::Error) << "Got " << nread << " instead of " << sizeof(CAN::TFrame) << " bytes";	×
109	}
110	}
111	}
112	}	×
113
114	public:
115	TCanPort(const std::string& ifname)	×
116	{	×
117	struct sockaddr_can addr;
118	struct ifreq ifr;
119
120	if ((Socket = socket(PF_CAN, SOCK_RAW, CAN_RAW)) < 0) {	×
121	throw std::runtime_error(std::string("Error while opening CAN socket: ") + strerror(errno));	×
122	}
123
124	int enable = 1;	×
125	setsockopt(Socket, SOL_CAN_RAW, CAN_RAW_LOOPBACK, &enable, sizeof(enable));	×
126	setsockopt(Socket, SOL_CAN_RAW, CAN_RAW_RECV_OWN_MSGS, &enable, sizeof(enable));	×
127
128	strncpy(ifr.ifr_name, ifname.c_str(), IFNAMSIZ - 1);	×
129	ifr.ifr_name[IFNAMSIZ - 1] = '\0';	×
130	ifr.ifr_ifindex = if_nametoindex(ifr.ifr_name);	×
131	if (!ifr.ifr_ifindex) {	×
132	throw TInterfaceNotFoundError(std::string("if_nametoindex failed for interface '") + ifr.ifr_name +	×
133	"', " + strerror(errno));	×
134	}
135
136	addr.can_family = AF_CAN;	×
137	addr.can_ifindex = ifr.ifr_ifindex;	×
138
139	LOG(WBMQTT::Info) << ifname.c_str() << " at index " << ifr.ifr_ifindex;	×
140
141	if (bind(Socket, (struct sockaddr*)&addr, sizeof(addr)) < 0) {	×
142	throw std::runtime_error(std::string("Error in CAN socket bind: ") + strerror(errno));	×
143	}
144
145	Enabled.store(true);	×
146
147	Thread = std::thread([this]() {	×
148	WBMQTT::SetThreadName("CAN listener");	×
149	ThreadFn();	×
150	});	×
151	}	×
152
153	~TCanPort()	×
154	{	×
155	Enabled.store(false);	×
156	if (Thread.joinable()) {	×
157	Thread.join();	×
158	}
159	close(Socket);	×
160	}	×
161
162	void AddHandler(CAN::IFrameHandler* handler)	×
163	{
164	std::unique_lock<std::mutex> lk(HandlersMutex);	×
165	Handlers.push_back(handler);	×
166	}	×
167
168	void RemoveHandler(CAN::IFrameHandler* handler)	×
169	{
170	std::unique_lock<std::mutex> lk(HandlersMutex);	×
171	Handlers.erase(std::remove(Handlers.begin(), Handlers.end(), handler), Handlers.end());	×
172	}	×
173
174	void Send(const CAN::TFrame& frame)	×
175	{
176	std::unique_lock<std::mutex> lk(WriteMutex);	×
177	{
178	std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);	×
179	WriteConfirmed = false;	×
180	}	×
181
182	auto nbytes = write(Socket, &frame, CAN_MTU);	×
183
184	if (nbytes < static_cast<int>(CAN_MTU)) {	×
185	throw std::runtime_error(std::string("CAN write error: ") + strerror(errno));	×
186	}
187	std::unique_lock<std::mutex> waitLock(WriteConfirmMutex);	×
188	if (WriteConfirmed) {	×
189	return;	×
190	}
191	if (std::cv_status::timeout == WriteConfirmCv.wait_for(waitLock, WRITE_TIMEOUT)) {	×
192	throw std::runtime_error("CAN write timeout");	×
193	}
194	}	×
195	};
196	}
197
198	std::shared_ptr<CAN::IPort> CAN::MakePort(const std::string& ifname)	×
199	{
200	return std::make_shared<TCanPort>(ifname);	×
201	}

wirenboard / wb-mqtt-smartweb / 2

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