The TCP server in the code snippet below is designed to:
- Read data using one thread while writing data with another thread.
- Allow only one client to be connected at a time.
- Automatically close the connection with the client and re-establish it if the connection is lost.
std::mutex
is used to protected the socket fd which is read/written by two threads. Is it a better way to achieve this goal without ASIO or other third party libraries?
@Guntram Blohm said it's perfectly ok to read and write a socket from two threads at the same time.
However, m_socket_fd
is used by threads which both read and write it(i.e. TcpServer::send()
may be still called while the value of m_socket_fd
is modified after accept() sucessfully returns and TcpServer::close
may be called at the same time by either of the two threads without the mutex).
Here is the code snippet:
#include <arpa/inet.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include <array>
#include <chrono>
#include <thread>
namespace Network {
bool is_valid_ip(const std::string& input_ip);
int set_nonblock(int sockfd);
int enb_reuse_addr_port(int listen_fd);
int set_send_timeout(int socket_fd, struct timeval timeout);
int set_recv_timeout(int socket_fd, struct timeval timeout);
int disable_nagle(int socket_fd);
} // namespace Network
bool Network::is_valid_ip(const std::string& input_ip) {
int num_point = 3;
int temp_num = -1;
if (input_ip.empty()) {
return false;
}
for (auto it = input_ip.begin(); it != input_ip.end(); it++) {
if (*it == '.') {
num_point--;
if (temp_num < 0 || temp_num > 255) {
return false;
}
temp_num = -1;
if (num_point >= 0) {
continue;
} else {
return false;
}
}
if (*it < '0' || *it > '9') {
return false;
}
temp_num == -1 ? temp_num = *it - '0' : temp_num = *it - '0' + temp_num * 10;
}
if (num_point != 0 || (temp_num < 0 || temp_num > 255)) {
return false;
}
return true;
}
int Network::set_nonblock(int sockfd) {
int flags = fcntl(sockfd, F_GETFL, 0);
fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
return 0;
}
int Network::enb_reuse_addr_port(int listen_fd) {
int opt = 1;
if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt)) < 0) {
return -1;
}
return 0;
}
int Network::set_send_timeout(int socket_fd, struct timeval timeout) {
return setsockopt(socket_fd, SOL_SOCKET, SO_SNDTIMEO, (char*)&timeout, sizeof(struct timeval));
}
int Network::set_recv_timeout(int socket_fd, struct timeval timeout) {
return setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, (char*)&timeout, sizeof(struct timeval));
}
int Network::disable_nagle(int socket_fd) {
int nagle_status = 1;
return setsockopt(socket_fd, IPPROTO_TCP, TCP_NODELAY, (char*)&nagle_status, sizeof(int));
}
class TcpServer {
public:
TcpServer(const std::string& ip,
unsigned short port,
unsigned int send_timeout, //microseconds
unsigned int recv_timeout); //microseconds
AlgTcpServer(const AlgTcpServer&) = delete;
AlgTcpServer& operator=(const AlgTcpServer&) = delete;
int start();
int process();
int send(const std::vector<char>& data);
int close();
private:
const std::string m_ip;
const unsigned short m_port;
const unsigned m_recv_timeout;
const unsigned m_send_timeout;
int m_listen_fd;
int m_socket_fd;
std::vector<char> m_data;
std::mutex m_socket_mtx;
};
TcpServer::TcpServer(const std::string& ip,
unsigned short port,
unsigned int send_timeout,
unsigned int recv_timeout)
: m_ip(ip), m_port(port), m_recv_timeout(recv_timeout), m_send_timeout(send_timeout),
m_listen_fd(-1), m_socket_fd(-1) {}
int TcpServer::start() {
if (!Network::is_valid_ip(m_ip)) {
return -1;
}
while (true) {
m_listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (m_listen_fd < 0) {
return -2;
}
Network::enb_reuse_addr_port(m_listen_fd);
struct sockaddr_in server_sockaddr;
memset(&server_sockaddr, 0, sizeof(server_sockaddr));
server_sockaddr.sin_family = AF_INET;
in_addr_t serviceIp = inet_addr(m_ip.c_str());
server_sockaddr.sin_addr.s_addr = serviceIp;
server_sockaddr.sin_port = htons(m_port);
if (bind(m_listen_fd, (struct sockaddr*)&server_sockaddr, sizeof(server_sockaddr)) < 0) {
perror("bind failed");
return -3;
}
if (listen(m_listen_fd, 1) < 0) {
perror("listen failed");
return -4;
}
while (true) {
struct sockaddr_in client_address;
socklen_t client_address_len = sizeof(client_address);
int socket_fd =
accept(m_listen_fd, (struct sockaddr*)&client_address, &client_address_len);
if (socket_fd == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
continue;
} else {
perror("accept failed");
::close(m_listen_fd);
return -4;
}
} else {
Network::disable_nagle(socket_fd);
struct timeval timeout_recv;
timeout_recv.tv_sec = 0;
timeout_recv.tv_usec = m_recv_timeout;
Network::set_recv_timeout(socket_fd, timeout_recv);
struct timeval timeout_send;
timeout_send.tv_sec = 0;
timeout_send.tv_usec = m_send_timeout;
Network::set_send_timeout(socket_fd, timeout_send);
::close(m_listen_fd);
m_listen_fd = -1;
{
//m_socket_fd also used by another thread to send out data
std::lock_guard<std::mutex> lock(m_socket_mtx);
m_socket_fd = socket_fd;
}
break;
}
}
process();
}
return 0;
}
int TcpServer::process() {
int ret = 0;
std::cout << "enter " << __PRETTY_FUNCTION__ << std::endl;
std::array<char, 100> tmp;
while (true) {
tmp.fill('\0');
int ret = 0;
{
std::lock_guard<std::mutex> lock(m_socket_mtx);
if (m_socket_fd < 0) {
std::cout << "read: invalid fd" << std::endl;
return -1;
}
ret = recv(m_socket_fd, tmp.data(), tmp.size(), 0);
}
if (ret < 0) {
if (errno != EAGAIN || errno != EWOULDBLOCK) {
perror("recv failed");
ret = -2;
break;
} else {
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
} else if (ret == 0) {
{
std::lock_guard<std::mutex> lock(m_socket_mtx);
this->close();
}
std::cout << "other side has been closed" << std::endl;
ret = -3;
break;
} else {
std::cout << tmp.data() << std::endl;
}
}
return ret;
}
int TcpServer::close() {
if (m_socket_fd > 0) {
::close(m_socket_fd);
shutdown(m_socket_fd, SHUT_RDWR);
m_socket_fd = -1;
}
return 0;
}
int TcpServer::send(const std::vector<char>& data) {
if (data.empty()) {
return -1;
}
const size_t len = data.size();
size_t num_sended = 0;
std::lock_guard<std::mutex> lock(m_socket_mtx);
if (m_socket_fd < 0) {
std::cout << "m_socket_fd is not valid" << std::endl;
return -2;
}
while (num_sended < len) {
ssize_t res = ::send(m_socket_fd, data.data() + num_sended, len - num_sended, MSG_NOSIGNAL);
if (res < 0) {
perror("send failed");
this->close();
break;
}
num_sended += res;
}
return 0;
}
int main() {
TcpServer server("127.0.0.1", 6666, 100, 50);
std::thread manager_and_reciver([&server]() { server.start(); });
std::thread sender([&server]() {
while (true) {
std::string str("hello, world!");
std::vector<char> data{str.begin(), str.end()};
server.send(data);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
};
});
manager_and_reciver.join();
sender.join();
}