Two parties communicate over a network sending back and forth up to 100 messages. Each message sent is a response to the last received message and depends on it. This means that both parties have to wait idly for the response to arrive before they can carry on with computations. Speed is quite critical and deteriorates very quickly when the network has moderate latency (due to the many messages and unavoidable waiting for the response). The messages are not very large, hence bandwidth does not seem to matter much.
Probably what I want to do can be achieved by using some library. If so, please point this out to me, preferably together with a demo or link to sources on how to use it. For lack of better alternatives I'm using (POSIX-API) TCP sockets here. I decided to use TCP rather than UDP because all data must be received in the correct order and the packet header size is not a relevant overhead, especially since latency is the issue.
This was my first time using sockets and I surely made many mistakes, both specific to sockets as well as pure C++ (I'm using C++17 although the code also compiles with C++11). My problem seems very standard and most of the code is puzzled together from some tutorials on sockets but I was struggling to find detailed sources on best practices.
Below is a simplified demo code that illustrates the way I handle the TCP logic. I tried to shorten it as much as possible but it is still quite long. Some comments:
tcp_helpers.hdeclares (AND defines for brevity of this post) functions containing all the TCP logic. The other two code files are an example application (main methods to run server and client). In my real code I encapsulate the TCP logic in classes, which internally call the functions shown here.- My messages can be variable size and have custom-defined headers specifying the length. The contents of a message is an array of custom-defined C-structs. Aside from these structs only having fixed-size primitive-type fields and no further structure, I would like my network code to work with any such user-defined struct-type. This leads to a big problem with portability: my code will probably not work if the two communicating systems use different byte order or different struct alignment. I am currently postponing this issue unless there is a straightforward way to take care of it.
- I disable Nagle's algorithm to ensure the TCP packets are sent as soon as the message is ready. I learned about this from asking a Stackoverflow question.
Some questions I have already:
- The first version of my
send_full_messagefunction (see linked Stackoverflow question) was making two sys-calls tosend, once for the (custom) header (8 byte struct) and once for the actual message (array of structs). In this version I reduced it to a single sys-call by copying header and data into a buffer (using perhaps ugly C-style memory manipulation). I did not notice a difference in performance compared to the original (sending the header as a separate packet). Which method is perferable? Can this be achieved more elegantly? - The pointer arithmetic in the
receive_structsfunction seems ugly. What would be the best-practice solution here? - Is there anything else I could do to make this faster (just as I had not known about Nagle's algorithm before asking)?
// tcp_helpers.h.
// NOTE: Requires C++11, tested also with C++17. Using this code in the present form may be ill-advised.
// This is not a true header file (contains all method definitions for brevity).
#include <vector>
#include <iostream>
#include <string>
#include <sstream>
#include <cerrno> // for checking socket error messages
#include <cstdint> // for fixed length integer types
#include <cstring> // for memcpy
#include <unistd.h> // POSIX specific
#include <sys/socket.h> // POSIX specific
#include <netinet/in.h> // POSIX specific
#include <netinet/tcp.h> // POSIX specific
#include <arpa/inet.h> // POSIX specific
//////////////////// PROFILING ///////////////////
#include <chrono>
static auto start = std::chrono::high_resolution_clock::now();
// print a message with timestamp for rudimentary profiling. (I don't actually use this in my code)
void print_now(const std::string &message) {
auto t2 = std::chrono::high_resolution_clock::now();
std::chrono::duration<double> time_span = t2 - start;
std::cout << time_span.count() << ": " << message << std::endl;
}
//////////////////// PROFILING ///////////////////
struct TCPMessageHeader { // Header for each message (I really use this).
uint8_t protocol_name[4];
uint32_t message_bytes;
};
struct ServerSends { // The server sends messages that are arrays of this struct (just an example).
uint16_t a;
uint32_t b;
uint32_t c;
};
typedef uint8_t ClientSends; // The client sends messages that are arrays of this (just an example).
namespace TCP_Helpers {
void disable_nagles_algorithm(int socket_fd) {
const int enable_no_delay = 1; // Disable Nagle's algorithm for TCP socket to improve performance
if (setsockopt(socket_fd, IPPROTO_TCP, TCP_NODELAY, &enable_no_delay, sizeof(enable_no_delay))) {
throw std::runtime_error("Failed to disble Nagle's algorithm by setting socket options");
}
}
int init_client(const std::string &ip_address, int port) {
int sock_fd;
struct sockaddr_in serv_addr{};
if ((sock_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
throw std::runtime_error("TCP Socket creation failed\n");
}
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(port);
// Convert IPv4 address from text to binary form
if (inet_pton(AF_INET, ip_address.c_str(), &serv_addr.sin_addr) <= 0) {
throw std::runtime_error("Invalid address/ Address not supported for TCP connection\n");
}
if (connect(sock_fd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
throw std::runtime_error("Failed to connect to server.\n");
}
disable_nagles_algorithm(sock_fd);
return sock_fd;
}
int init_server(int port) {
int server_fd;
int new_socket;
struct sockaddr_in address{};
int opt = 1;
int addrlen = sizeof(address);
// Creating socket file descriptor
if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
throw std::runtime_error("socket creation failed\n");
}
if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {
throw std::runtime_error("failed to set socket options");
}
address.sin_family = AF_INET;
address.sin_addr.s_addr = INADDR_ANY;
address.sin_port = htons(port);
// Forcefully attaching socket to the port
if (bind(server_fd, (struct sockaddr *) &address, sizeof(address)) < 0) {
throw std::runtime_error("bind failed");
}
if (listen(server_fd, 3) < 0) {
throw std::runtime_error("listen failed");
}
if ((new_socket = accept(server_fd, (struct sockaddr *) &address, (socklen_t *) &addrlen)) < 0) {
throw std::runtime_error("accept failed");
}
if (close(server_fd)) // don't need to listen for any more tcp connections (PvP connection).
throw std::runtime_error("closing server socket failed");
disable_nagles_algorithm(new_socket);
return new_socket;
}
template<typename NakedStruct>
void send_full_message(int fd, TCPMessageHeader header_to_send, const std::vector<NakedStruct> &structs_to_send) {
const size_t num_message_bytes = sizeof(NakedStruct) * structs_to_send.size();
if (header_to_send.message_bytes != num_message_bytes) {
throw std::runtime_error("Trying to send struct vector whose size does not"
" match the size claimed by message header");
}
print_now("Begin send_full_message");
// copy header and structs vector contents to new vector (buffer) of bytes and sent via TCP.
// Does not seem to be faster than sending two separate packets for header/message. Can the copy be avoided?
std::vector<uint8_t> full_msg_packet(sizeof(header_to_send) + num_message_bytes);
memcpy(full_msg_packet.data(), &header_to_send, sizeof(header_to_send));
memcpy(full_msg_packet.data() + sizeof(header_to_send), structs_to_send.data(), num_message_bytes);
// maybe need timeout and more error handling?
size_t bytes_to_send = full_msg_packet.size() * sizeof(uint8_t);
int send_retval;
while (bytes_to_send != 0) {
send_retval = send(fd, full_msg_packet.data(), sizeof(uint8_t) * full_msg_packet.size(), 0);
if (send_retval == -1) {
int errsv = errno; // from errno.h
std::stringstream s;
s << "Sending data failed (locally). Errno:" << errsv
<< " while sending header of size" << sizeof(header_to_send)
<< " and data of size " << header_to_send.message_bytes << ".";
throw std::runtime_error(s.str());
}
bytes_to_send -= send_retval;
}
print_now("end send_full_message.");
}
template<typename NakedStruct>
std::vector<NakedStruct> receive_structs(int fd, uint32_t bytes_to_read) {
print_now("Begin receive_structs");
unsigned long num_structs_to_read;
// ensure expected message is non-zero length and a multiple of the SingleBlockParityRequest struct
if (bytes_to_read > 0 && bytes_to_read % sizeof(NakedStruct) == 0) {
num_structs_to_read = bytes_to_read / sizeof(NakedStruct);
} else {
std::stringstream s;
s << "Message length (bytes_to_read = " << bytes_to_read <<
" ) specified in header does not divide into required stuct size (" << sizeof(NakedStruct) << ").";
throw std::runtime_error(s.str());
}
// vector must have size > 0 for the following pointer arithmetic to work
// (this method must check this in above code).
std::vector<NakedStruct> received_data(num_structs_to_read);
int valread;
while (bytes_to_read > 0) // need to include some sort of timeout?!
{
valread = read(fd,
((uint8_t *) (&received_data[0])) +
(num_structs_to_read * sizeof(NakedStruct) - bytes_to_read),
bytes_to_read);
if (valread == -1) {
throw std::runtime_error("Reading from socket file descriptor failed");
} else {
bytes_to_read -= valread;
}
}
print_now("End receive_structs");
return received_data;
}
void send_header(int fd, TCPMessageHeader header_to_send) {
print_now("Start send_header");
int bytes_to_send = sizeof(header_to_send);
int send_retval;
while (bytes_to_send != 0) {
send_retval = send(fd, &header_to_send, sizeof(header_to_send), 0);
if (send_retval == -1) {
int errsv = errno; // from errno.h
std::stringstream s;
s << "Sending data failed (locally). Errno:" << errsv << " while sending (lone) header.";
throw std::runtime_error(s.str());
}
bytes_to_send -= send_retval;
}
print_now("End send_header");
}
TCPMessageHeader receive_header(int fd) {
print_now("Start receive_header (calls receive_structs)");
TCPMessageHeader retval = receive_structs<TCPMessageHeader>(fd, sizeof(TCPMessageHeader)).at(0);
print_now("End receive_header (calls receive_structs)");
return retval;
}
}
// main_server.cpp
#include "tcp_helpers.h"
int main() {
int port = 20000;
int socket_fd = TCP_Helpers::init_server(port);
while (true) { // server main loop
TCPMessageHeader rcv_header = TCP_Helpers::receive_header(socket_fd);
if (rcv_header.protocol_name[0] == 0) // using first byte of header name as signal to end
break;
// receive message
auto rcv_message = TCP_Helpers::receive_structs<ClientSends>(socket_fd, rcv_header.message_bytes);
// for (ClientSends ex : rcv_message) // example "use" of the received data that takes a bit of time.
// std::cout << static_cast<int>(ex) << " ";
// std::cout << std::endl << std::endl;
auto bunch_of_zeros = std::vector<ServerSends>(1000); // send a "response" containing 1000 structs of zeros
TCPMessageHeader send_header{"abc", 1000 * sizeof(ServerSends)};
TCP_Helpers::send_full_message(socket_fd, send_header, bunch_of_zeros);
}
exit(EXIT_SUCCESS);
}
// main_client.cpp
#include "tcp_helpers.h"
int main() {
// establish connection to server and get socket file descriptor.
int port = 20000;
auto ip = "127.0.0.1";
int socket1_fd = TCP_Helpers::init_client(ip, port);
std::cout << "connected." << std::endl;
for (int i = 0; i < 20; ++i) { // repeat (for runtime statistics) sending and receiving arbitrary data
// send a message containing 500 structs of zeros
auto bunch_of_zeros = std::vector<ClientSends>(500);
TCPMessageHeader send_header{"abc", 500 * sizeof(ClientSends)};
TCP_Helpers::send_full_message(socket1_fd, send_header, bunch_of_zeros);
// receive response
TCPMessageHeader rcv_header = TCP_Helpers::receive_header(socket1_fd);
auto rcv_message = TCP_Helpers::receive_structs<ServerSends>(socket1_fd, rcv_header.message_bytes);
// for (ServerSends ex : rcv_message) // example "use" of the received data that takes a bit of time.
// std::cout << ex.a << ex.b << ex.c << " ";
// std::cout << std::endl << std::endl;
}
auto end_header = TCPMessageHeader{}; // initialized all fields to zero. "end" signal in this demonstration.
TCP_Helpers::send_header(socket1_fd, end_header);
exit(EXIT_SUCCESS);
}
