10059975553

Committed 23 Jul 2024 01:33PM UTC coverage: 37.622% (+0.09%) from 37.537%

Build # 10059975553

Build Type

Pull #2357

github

Committed by

web-flow

Commit Message

Merge 0b6ca6e1b into f1aff89a6

Pull Request Pull Request #2357: core: Correctly close sockets

Run Details

43 of 46 new or added lines in 5 files covered. (93.48%)

17 existing lines in 2 files now uncovered.

11469 of 30485 relevant lines covered (37.62%)

260.35 hits per line

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

78.89

/src/mavsdk/core/tcp_server_connection.cpp

#include "tcp_server_connection.h"
#include "log.h"

#include <cassert>
#include <fcntl.h>

#ifdef WINDOWS
#ifndef MINGW
#pragma comment(lib, "Ws2_32.lib") // Without this, Ws2_32.lib is not included in static library.
#endif
#include <Windows.h>
#include <Winsock2.h>
#include <ws2tcpip.h>
#else
#include <netinet/in.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <netdb.h>
#endif

#ifndef WINDOWS
#define GET_ERROR(_x) strerror(_x)
#else
#define GET_ERROR(_x) WSAGetLastError()
#endif

namespace mavsdk {
TcpServerConnection::TcpServerConnection(
    Connection::ReceiverCallback receiver_callback,
    std::string local_ip,
    int local_port,
    ForwardingOption forwarding_option) :
    Connection(std::move(receiver_callback), forwarding_option),
    _local_ip(std::move(local_ip)),
    _local_port(local_port)
{}

TcpServerConnection::~TcpServerConnection()
{
    stop();
}

ConnectionResult TcpServerConnection::start()
{
    if (!start_mavlink_receiver()) {
        return ConnectionResult::ConnectionsExhausted;
    }

#ifdef WINDOWS
    WSADATA wsa;
    if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
        LogErr() << "Error: Winsock failed, error: " << WSAGetLastError();
        return ConnectionResult::SocketError;
    }
#endif

    _server_socket_fd.reset(socket(AF_INET, SOCK_STREAM, 0));
    if (_server_socket_fd.empty()) {
        LogErr() << "socket error: " << GET_ERROR(errno);
        return ConnectionResult::SocketError;
    }

    sockaddr_in server_addr{};
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(_local_port);

    if (bind(_server_socket_fd.get(), reinterpret_cast<sockaddr*>(&server_addr), sizeof(server_addr)) <
        0) {
        LogErr() << "bind error: " << GET_ERROR(errno);
        return ConnectionResult::SocketError;
    }

    if (listen(_server_socket_fd.get(), 3) < 0) {
        LogErr() << "listen error: " << GET_ERROR(errno);
        return ConnectionResult::SocketError;
    }

    _accept_receive_thread =
        std::make_unique<std::thread>(&TcpServerConnection::accept_client, this);

    return ConnectionResult::Success;
}

ConnectionResult TcpServerConnection::stop()
{
    _should_exit = true;

    _client_socket_fd.close();
    _server_socket_fd.close();

    if (_accept_receive_thread && _accept_receive_thread->joinable()) {
        _accept_receive_thread->join();
        _accept_receive_thread.reset();
    }

    // We need to stop this after stopping the receive thread, otherwise
    // it can happen that we interfere with the parsing of a message.
    stop_mavlink_receiver();

    return ConnectionResult::Success;
}

bool TcpServerConnection::send_message(const mavlink_message_t& message)
{
    uint8_t buffer[MAVLINK_MAX_PACKET_LEN];
    uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);

    assert(buffer_len <= MAVLINK_MAX_PACKET_LEN);

#if !defined(MSG_NOSIGNAL)
    auto flags = 0;
#else
    auto flags = MSG_NOSIGNAL;
#endif

    const auto send_len =
        send(_client_socket_fd.get(), reinterpret_cast<const char*>(buffer), buffer_len, flags);

    if (send_len != buffer_len) {
        LogErr() << "send failure: " << GET_ERROR(errno);
        return false;
    }
    return true;
}

void TcpServerConnection::accept_client()
{
#ifdef WINDOWS
    // Set server socket to non-blocking
    u_long iMode = 1;
    int iResult = ioctlsocket(_server_socket_fd.get(), FIONBIO, &iMode);
    if (iResult != 0) {
        LogErr() << "ioctlsocket failed with error: " << WSAGetLastError();
    }
#else
    // Set server socket to non-blocking
    int flags = fcntl(_server_socket_fd.get(), F_GETFL, 0);
    fcntl(_server_socket_fd.get(), F_SETFL, flags | O_NONBLOCK);
#endif

    while (!_should_exit) {
        fd_set readfds;
        FD_ZERO(&readfds);
        FD_SET(_server_socket_fd.get(), &readfds);

        // Set timeout to 1 second
        timeval timeout;
        timeout.tv_sec = 1;
        timeout.tv_usec = 0;

        const int activity = select(_server_socket_fd.get() + 1, &readfds, nullptr, nullptr, &timeout);

        if (activity < 0 && errno != EINTR) {
            LogErr() << "select error: " << GET_ERROR(errno);
            continue;
        }

        if (activity == 0) {
            // Timeout, no incoming connection
            continue;
        }

        if (FD_ISSET(_server_socket_fd.get(), &readfds)) {
            sockaddr_in client_addr{};
            socklen_t client_addr_len = sizeof(client_addr);

            _client_socket_fd.reset(accept(
                _server_socket_fd.get(), reinterpret_cast<sockaddr*>(&client_addr), &client_addr_len));
            if (_client_socket_fd.empty()) {
                if (_should_exit) {
                    return;
                }
                LogErr() << "accept error: " << GET_ERROR(errno);
                continue;
            }

            receive();
        }
    }
}

void TcpServerConnection::receive()
{
    std::array<char, 2048> buffer{};

    bool dataReceived = false;
    while (!dataReceived && !_should_exit) {
        const auto recv_len = recv(_client_socket_fd.get(), buffer.data(), buffer.size(), 0);

#ifdef WINDOWS
        if (recv_len == SOCKET_ERROR) {
            // On Windows, on the first try, select says there is something
            // but recv doesn't succeed yet, and we just need to try again.
            if (WSAGetLastError() == WSAEWOULDBLOCK) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                continue;
            }
            // And at the end, we get an abort that we can silently ignore.
            if (WSAGetLastError() == WSAECONNABORTED) {
                return;
            }
        }
#else
        if (recv_len < 0) {
            // On macOS we presumably see the same thing, and have to try again.
            if (errno == EAGAIN) {
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
                continue;
            }

            LogErr() << "recv failed: " << GET_ERROR(errno);
            return;
        }
#endif

        if (recv_len == 0) {
            continue;
        }

        _mavlink_receiver->set_new_datagram(buffer.data(), static_cast<int>(recv_len));

        // Parse all mavlink messages in one data packet. Once exhausted, we'll exit while.
        while (_mavlink_receiver->parse_message()) {
            receive_message(_mavlink_receiver->get_last_message(), this);
        }
    }
}

} // namespace mavsdk

1	#include "tcp_server_connection.h"
2	#include "log.h"
3
4	#include <cassert>
5	#include <fcntl.h>
6
7	#ifdef WINDOWS
8	#ifndef MINGW
9	#pragma comment(lib, "Ws2_32.lib") // Without this, Ws2_32.lib is not included in static library.
10	#endif
11	#include <Windows.h>
12	#include <Winsock2.h>
13	#include <ws2tcpip.h>
14	#else
15	#include <netinet/in.h>
16	#include <sys/select.h>
17	#include <sys/socket.h>
18	#include <arpa/inet.h>
19	#include <errno.h>
20	#include <netdb.h>
21	#endif
22
23	#ifndef WINDOWS
24	#define GET_ERROR(_x) strerror(_x)
25	#else
26	#define GET_ERROR(_x) WSAGetLastError()
27	#endif
28
29	namespace mavsdk {
30	TcpServerConnection::TcpServerConnection(	1✔
31	Connection::ReceiverCallback receiver_callback,
32	std::string local_ip,
33	int local_port,
34	ForwardingOption forwarding_option) :	1✔
35	Connection(std::move(receiver_callback), forwarding_option),	1✔
36	_local_ip(std::move(local_ip)),	1✔
37	_local_port(local_port)	2✔
38	{}	1✔
39
40	TcpServerConnection::~TcpServerConnection()	1✔
41	{
42	stop();	1✔
43	}	1✔
44
45	ConnectionResult TcpServerConnection::start()	1✔
46	{
47	if (!start_mavlink_receiver()) {	1✔
48	return ConnectionResult::ConnectionsExhausted;	×
49	}
50
51	#ifdef WINDOWS
52	WSADATA wsa;
53	if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) {
54	LogErr() << "Error: Winsock failed, error: " << WSAGetLastError();
55	return ConnectionResult::SocketError;
56	}
57	#endif
58
59	_server_socket_fd.reset(socket(AF_INET, SOCK_STREAM, 0));	1✔
60	if (_server_socket_fd.empty()) {	1✔
61	LogErr() << "socket error: " << GET_ERROR(errno);	×
62	return ConnectionResult::SocketError;	×
63	}
64
65	sockaddr_in server_addr{};	1✔
66	server_addr.sin_family = AF_INET;	1✔
67	server_addr.sin_addr.s_addr = INADDR_ANY;	1✔
68	server_addr.sin_port = htons(_local_port);	1✔
69
70	if (bind(_server_socket_fd.get(), reinterpret_cast<sockaddr*>(&server_addr), sizeof(server_addr)) <	1✔
71	0) {
72	LogErr() << "bind error: " << GET_ERROR(errno);	×
73	return ConnectionResult::SocketError;	×
74	}
75
76	if (listen(_server_socket_fd.get(), 3) < 0) {	1✔
UNCOV 77	LogErr() << "listen error: " << GET_ERROR(errno);	×
78	return ConnectionResult::SocketError;	×
79	}
80
81	_accept_receive_thread =	1✔
82	std::make_unique<std::thread>(&TcpServerConnection::accept_client, this);	1✔
83
84	return ConnectionResult::Success;	1✔
85	}
86
87	ConnectionResult TcpServerConnection::stop()	1✔
88	{
89	_should_exit = true;	1✔
90
91	_client_socket_fd.close();	1✔
92	_server_socket_fd.close();	1✔
93
94	if (_accept_receive_thread && _accept_receive_thread->joinable()) {	1✔
95	_accept_receive_thread->join();	1✔
96	_accept_receive_thread.reset();	1✔
97	}
98
99	// We need to stop this after stopping the receive thread, otherwise
100	// it can happen that we interfere with the parsing of a message.
101	stop_mavlink_receiver();	1✔
102
103	return ConnectionResult::Success;	1✔
104	}
105
106	bool TcpServerConnection::send_message(const mavlink_message_t& message)	12✔
107	{
108	uint8_t buffer[MAVLINK_MAX_PACKET_LEN];	12✔
109	uint16_t buffer_len = mavlink_msg_to_send_buffer(buffer, &message);	12✔
110
111	assert(buffer_len <= MAVLINK_MAX_PACKET_LEN);	12✔
112
113	#if !defined(MSG_NOSIGNAL)
114	auto flags = 0;
115	#else
116	auto flags = MSG_NOSIGNAL;	12✔
117	#endif
118
119	const auto send_len =
120	send(_client_socket_fd.get(), reinterpret_cast<const char*>(buffer), buffer_len, flags);	12✔
121
122	if (send_len != buffer_len) {	12✔
123	LogErr() << "send failure: " << GET_ERROR(errno);	5✔
124	return false;	5✔
125	}
126	return true;	7✔
127	}
128
129	void TcpServerConnection::accept_client()	1✔
130	{
131	#ifdef WINDOWS
132	// Set server socket to non-blocking
133	u_long iMode = 1;
134	int iResult = ioctlsocket(_server_socket_fd.get(), FIONBIO, &iMode);
135	if (iResult != 0) {
136	LogErr() << "ioctlsocket failed with error: " << WSAGetLastError();
137	}
138	#else
139	// Set server socket to non-blocking
140	int flags = fcntl(_server_socket_fd.get(), F_GETFL, 0);	1✔
141	fcntl(_server_socket_fd.get(), F_SETFL, flags \| O_NONBLOCK);	1✔
142	#endif
143
144	while (!_should_exit) {	2✔
145	fd_set readfds;	1✔
146	FD_ZERO(&readfds);	1✔
147	FD_SET(_server_socket_fd.get(), &readfds);	1✔
148
149	// Set timeout to 1 second
150	timeval timeout;	1✔
151	timeout.tv_sec = 1;	1✔
152	timeout.tv_usec = 0;	1✔
153
154	const int activity = select(_server_socket_fd.get() + 1, &readfds, nullptr, nullptr, &timeout);	1✔
155
156	if (activity < 0 && errno != EINTR) {	1✔
157	LogErr() << "select error: " << GET_ERROR(errno);	×
UNCOV 158	continue;	×
159	}
160
161	if (activity == 0) {	1✔
162	// Timeout, no incoming connection
UNCOV 163	continue;	×
164	}
165
166	if (FD_ISSET(_server_socket_fd.get(), &readfds)) {	1✔
167	sockaddr_in client_addr{};	1✔
168	socklen_t client_addr_len = sizeof(client_addr);	1✔
169
170	_client_socket_fd.reset(accept(	1✔
171	_server_socket_fd.get(), reinterpret_cast<sockaddr*>(&client_addr), &client_addr_len));
172	if (_client_socket_fd.empty()) {	1✔
173	if (_should_exit) {	×
174	return;	×
175	}
176	LogErr() << "accept error: " << GET_ERROR(errno);	×
177	continue;	×
178	}
179
180	receive();	1✔
181	}
182	}
183	}
184
185	void TcpServerConnection::receive()	1✔
186	{
187	std::array<char, 2048> buffer{};	1✔
188
189	bool dataReceived = false;	1✔
190	while (!dataReceived && !_should_exit) {	38,620✔
191	const auto recv_len = recv(_client_socket_fd.get(), buffer.data(), buffer.size(), 0);	38,619✔
192
193	#ifdef WINDOWS
194	if (recv_len == SOCKET_ERROR) {
195	// On Windows, on the first try, select says there is something
196	// but recv doesn't succeed yet, and we just need to try again.
197	if (WSAGetLastError() == WSAEWOULDBLOCK) {
198	std::this_thread::sleep_for(std::chrono::milliseconds(10));
199	continue;
200	}
201	// And at the end, we get an abort that we can silently ignore.
202	if (WSAGetLastError() == WSAECONNABORTED) {
203	return;
204	}
205	}
206	#else
207	if (recv_len < 0) {	38,619✔
208	// On macOS we presumably see the same thing, and have to try again.
UNCOV 209	if (errno == EAGAIN) {	×
UNCOV 210	std::this_thread::sleep_for(std::chrono::milliseconds(10));	×
UNCOV 211	continue;	×
212	}
213
214	LogErr() << "recv failed: " << GET_ERROR(errno);	×
215	return;	×
216	}
217	#endif
218
219	if (recv_len == 0) {	38,619✔
220	continue;	38,614✔
221	}
222
223	_mavlink_receiver->set_new_datagram(buffer.data(), static_cast<int>(recv_len));	5✔
224
225	// Parse all mavlink messages in one data packet. Once exhausted, we'll exit while.
226	while (_mavlink_receiver->parse_message()) {	10✔
227	receive_message(_mavlink_receiver->get_last_message(), this);	5✔
228	}
229	}
230	}
231
232	} // namespace mavsdk

mavlink / MAVSDK / 10059975553

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