ノンブロッキングなファイルディスクリプタを用いて「I/O多重化」を施したエコーサーバーを作成してみた。

# はじめに
どうも。こんにちは。
[42tokyo](https://42tokyo.jp/)といったところで、ソケットプログラミング？？を学んでいます。
前回は、[「I/O多重化（システムコールpoll();を使用）」を施したサーバを作成してみた。](https://qiita.com/gu-chi/items/243fa63e17617bb9ef77)を書きました。

今回は、ノンブロッキングなファイルディスクリプタを用いて「I/O多重化」を施したエコーサーバーを作成してみます。

このサーバでは、クライアントがメッセージを送信すると、
サーバ側で受け取ったメッセージを出力し、再度クライアントに同様のメッセージを返送します。
クライアントでは、サーバーから返送されたメッセージを受信し、出力するようになっています。

使用した言語は、C++98です。
「I/O多重化」は、システムコール[poll();](https://ja.manpages.org/poll/2)を使って実現しました。

間違い等があれば、ご指摘ください。
# コード
:::note warn
注意：このコードは、バグを多分に含んでおります。
:::

```c++:Server.hpp
#ifndef SERVER_HPP
# define SERVER_HPP

#include <fcntl.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <poll.h>
#include <iostream>
#include <string>
#include <cstring>

class Server {
	private:
		int							socketFd_;
		socklen_t					socketAddressLen_;  // socklen_tはコンパイラによってエラー。intの場合もある。
		const int					maxClients_;
		struct sockaddr_in			socketAddress_;
		struct pollfd				fds_[7];

		void	initializeServerSocket(unsigned short port);
		void	initializeClientSockets();
		void	handleServerSocket();
		void	handleStandardInput();
		void	handleClientSocket(int clientIndex);
		void	handleReceivedData(int clientIndex);
		ssize_t	recvNonBlocking(int clientIndex, char* buffer, size_t bufferSize);
		ssize_t	sendNonBlocking(int clientIndex, const char* buffer, size_t dataSize);
		void	handleClientDisconnect(int clientIndex);

	public:
		explicit Server(unsigned short port);
		~Server();
		void	run(void);
};

#endif  // SERVER_HPP
```
```c++:Server.cpp
#include "./Server.hpp"

static void fatalError(const std::string& message) {
	std::perror(message.c_str());
    exit(EXIT_FAILURE);
}

static void	setFdFlags(const int fd, const int setFlags) {
	int	flags = 0;

	if ((flags = fcntl(fd, F_GETFL, 0)) < 0) {
		fatalError("fcntl");
	}
	flags |= setFlags;
	if (fcntl(fd, F_SETFL, flags) < 0) {
		fatalError("fcntl");
	}
}

Server::Server(unsigned short port) : \
	socketFd_(0), socketAddressLen_(sizeof(socketAddress_)), maxClients_(5) {
		initializeServerSocket(port);
		initializeClientSockets();
}

void	Server::initializeServerSocket(unsigned short port) {
	socketFd_ = socket(AF_INET, SOCK_STREAM, 0);
	if (socketFd_ < 0) {
		fatalError("socket");
	}

	setFdFlags(socketFd_, O_NONBLOCK);

	memset(&socketAddress_, 0, sizeof(socketAddress_));
	socketAddress_.sin_family = AF_INET;
	socketAddress_.sin_addr.s_addr = INADDR_ANY;
	socketAddress_.sin_port = htons(port);

	if (bind(socketFd_, reinterpret_cast<struct sockaddr*>(&socketAddress_), \
				sizeof(socketAddress_)) < 0) {
		fatalError("bind");
	}
	if (listen(socketFd_, 3) < 0) {
		fatalError("listen");
	}
	std::cout << "Server is listening on port " << port << "..." << std::endl;
}

void	Server::initializeClientSockets() {
	fds_[0].fd = socketFd_;
	for (int i = 1; i <= maxClients_; ++i) {
		fds_[i].fd = -1;
	}
	fds_[maxClients_].fd = STDIN_FILENO;
	for (int i = 0; i <= maxClients_; ++i) {
		fds_[i].events = POLLIN;
		fds_[i].revents = 0;
	}
}

Server::~Server() {
	close(socketFd_);
}

void	Server::run() {
	while (1) {
		int result = poll(fds_, maxClients_ + 1, 3 * 1000);

		if (result == -1) {
			fatalError("poll");
		}
		if (result == 0) {
			std::cout << "poll: Timeout 3 seconds..." << std::endl;
			continue;
		}
		handleServerSocket();
		handleStandardInput();
		for (int i = 1; i <= maxClients_; ++i) {
			handleClientSocket(i);
		}
	}
}

void	Server::handleServerSocket() {
	if (!(fds_[0].revents & POLLIN)) {
		return;
	}
	int newSocket = accept(socketFd_, \
			reinterpret_cast<struct sockaddr*>(&socketAddress_), &socketAddressLen_);
	if (newSocket < 0) {
		close(newSocket);
		fatalError("accept");
	}
	setFdFlags(newSocket, O_NONBLOCK);
	for (int i = 1; i <= maxClients_; ++i) {
		if (fds_[i].fd == -1) {
			fds_[i].fd = newSocket;
			std::cout << "New client connected. Socket: " \
				<< newSocket << std::endl;
			break;
		}
	}
}

void	Server::handleStandardInput() {
	if (fds_[maxClients_].revents & POLLIN) {
		std::string	input;
		std::getline(std::cin, input);
		if (input == "exit") {
			std::cout << "See You..." << std::endl;
			exit(0);
		}
	}
}

void	Server::handleClientSocket(int clientIndex) {
	if (fds_[clientIndex].fd != -1 && (fds_[clientIndex].revents & POLLIN)) {
		handleReceivedData(clientIndex);
	}
}

void	Server::handleReceivedData(int clientIndex) {
	char	buffer[1024] = {0};
	ssize_t	sendMsgSize = 0;
	ssize_t	recvMsgSize = 0;

	recvMsgSize = recvNonBlocking(clientIndex, buffer, sizeof(buffer) - 1);
	if (recvMsgSize <= 0) {
		handleClientDisconnect(clientIndex);
		return;
	}
	std::cout << "Client socket " << fds_[clientIndex].fd \
		<< " message: " << buffer << std::endl;
	sendMsgSize = sendNonBlocking(clientIndex, buffer, recvMsgSize);
	if (sendMsgSize <= 0) {
		handleClientDisconnect(clientIndex);
		return;
	}
	if (recvMsgSize != sendMsgSize) {
		fatalError("send");
	}
}

ssize_t Server::recvNonBlocking(int clientIndex, char* buffer, \
		size_t bufferSize) {
	ssize_t	recvMsgSize = 0;

	while (1) {
		recvMsgSize = recv(fds_[clientIndex].fd, buffer, \
				bufferSize, MSG_DONTWAIT);

		if (recvMsgSize >= 0) {
			break;
		}
		if (errno == EAGAIN || errno == EWOULDBLOCK) {
			errno = 0;
			recvMsgSize = 0;
			continue;
		} else if (errno == ECONNRESET) {
			recvMsgSize = -1;
			break;
		} else {
			fatalError("recv");
		}
	}
	return (recvMsgSize);
}

ssize_t	Server::sendNonBlocking(int clientIndex, const char* buffer, \
		size_t dataSize) {
	ssize_t sendMsgSize = 0;

	while (1) {
		sendMsgSize = send(fds_[clientIndex].fd, buffer, \
				dataSize, MSG_DONTWAIT);

		if (sendMsgSize >= 0) {
			break;
		}
		if (errno == EAGAIN || errno == EWOULDBLOCK) {
			std::cout << "No data sent." << std::endl;
			errno = 0;
			sendMsgSize = 0;
			continue;
		} else if (errno == ECONNRESET) {
			sendMsgSize = -1;
			break;
		} else {
			fatalError("send");
		}
	}
	return (sendMsgSize);
}

void	Server::handleClientDisconnect(int clientIndex) {
	std::cout << "Client socket " << fds_[clientIndex].fd \
		<< " disconnected." << std::endl;
	close(fds_[clientIndex].fd);
	fds_[clientIndex].fd = -1;
}

int	main() {
	Server	Server(8080);

	Server.run();
	return (0);
}
```
```c++:Client.hpp
#ifndef CLIENT_HPP
# define CLIENT_HPP

#include <fcntl.h>
#include <poll.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>

#include <iostream>
#include <string>
#include <cstring>
#include <cstdlib>
#include <cstdio>

class Client {
 private:
	 int					socketFd_;
	 struct sockaddr_in		serverSocketAddress_;

 public:
	 Client(const std::string& serverIP, unsigned short serverPort);
	 ~Client();

	 void		connectToServer();
	 ssize_t	sendMessage(const std::string& message);
	 ssize_t	receiveMessage();
};

#endif  // CLIENT_HPP
```
```c++:Client.cpp
#include "./Client.hpp"

static void fatalError(const std::string& message) {
	std::perror(message.c_str());
    exit(EXIT_FAILURE);
}

static void	setFdFlags(const int fd, const int setFlags) {
	int	flags = 0;

	if ((flags = fcntl(fd, F_GETFL, 0)) < 0) {
		fatalError("fcntl");
	}
	flags |= setFlags;
	if (fcntl(fd, F_SETFL, flags) < 0) {
		fatalError("fcntl");
	}
}


Client::Client(const std::string& serverIP, unsigned short serverPort) :
	socketFd_(0) {
	this->socketFd_ = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);

	if (this->socketFd_ < 0) {
		fatalError("socket");
	}
	setFdFlags(this->socketFd_, O_NONBLOCK);
	memset(&this->serverSocketAddress_, 0, sizeof(this->serverSocketAddress_));
	this->serverSocketAddress_.sin_family = AF_INET;
	this->serverSocketAddress_.sin_addr.s_addr = inet_addr(serverIP.c_str());
	this->serverSocketAddress_.sin_port = htons(serverPort);
}

Client::~Client() {
	close(this->socketFd_);
}

void Client::connectToServer() {
	struct pollfd	pollFd;
	int				ret = 0;

	pollFd.fd = this->socketFd_;
	pollFd.events = POLLOUT;

	ret = connect(this->socketFd_, \
				reinterpret_cast<struct sockaddr *>(&this->serverSocketAddress_), \
				sizeof(this->serverSocketAddress_));
	if (ret == 0) {
		std::cout << "Connection established!" << std::endl;
		return;
	}
	if (ret == -1 && errno != EINPROGRESS) {
		fatalError("connect");
	}
	std::cout << "Connecting..." << std::endl;
	while (1) {
		int	pollRet = 0;

		pollRet = poll(&pollFd, 1, 3 * 1000);
		if (pollRet < 0) {
			fatalError("poll");
		}
        if (pollRet == 0) {
  			std::cout << "poll: Timeout 3 seconds..." << std::endl;
			continue;
		}
		if (pollFd.revents & POLLOUT) {
			std::cout << "Connection established!" << std::endl;
			break;
		}
	}
}

ssize_t	Client::sendMessage(const std::string& message) {
	ssize_t	sendMsgSize = 0;

	while (1) {
		sendMsgSize = send(this->socketFd_, message.c_str(), \
				message.size(), MSG_DONTWAIT);

		if (sendMsgSize >= 0) {
			break;
		}
		if (errno == EAGAIN || errno == EWOULDBLOCK) {
			std::cout << "No data sent." << std::endl;
			errno = 0;
			sendMsgSize = 0;
			continue;
		} else if (errno == ECONNRESET) {
			sendMsgSize = -1;
			break;
		} else {
			fatalError("send");
		}
	}
	return (sendMsgSize);
}

ssize_t	Client::receiveMessage() {
	char	buffer[1024] = {0};
	int		recvMsgSize = 0;

	std::cout << "Received: " << std::flush;
	while (1) {
		recvMsgSize = recv(this->socketFd_, buffer, sizeof(buffer) - 1, MSG_DONTWAIT);
		if (recvMsgSize >= 0) {
			break;
		}
		if (errno == EAGAIN || errno == EWOULDBLOCK) {
			// std::cout << "No data received." << std::endl;
			errno = 0;
			recvMsgSize = 0;
			continue;
		} else if (errno == ECONNRESET) {
			recvMsgSize = -1;
			break;
		} else {
			fatalError("recv");
		}
	}
	buffer[recvMsgSize] = '\0';
	std::cout << buffer << std::endl;
	return (recvMsgSize);
}

int main(int argc, char *argv[]) {
	if (argc != 2) {
		std::cerr << "Usage: " << argv[0] << " <Echo Word>\n";
		exit(1);
	}

	const std::string	message = argv[1];

	Client	client("127.0.0.1", 8080);

	client.connectToServer();
	while (1) {
		ssize_t	sendMsgSize = 0;
		ssize_t	recvMsgSize = 0;

		sendMsgSize = client.sendMessage(message);
		if (sendMsgSize < 0) {
			break;
		}
		recvMsgSize = client.receiveMessage();
		if (recvMsgSize < 0) {
			break;
		}
		if (sendMsgSize != recvMsgSize) {
			fatalError("recv");
		}
		sleep(5);
	}
	return (0);
}
```

# 確認
実行ファイルの作成
```shell:shell
~$ c++ -Wall -Wextra -Werror -std=c++98 -pedantic-errors ./Server.cpp -o server
~$ c++ -Wall -Wextra -Werror -std=c++98 -pedantic-errors ./Client.cpp -o client
```

複数のプロセスまたは複数のターミナルを開き以下のコマンドを実行

サーバーの起動
```shell:shell
~$ ./server
```

クライアントを実行
```shell:shell
~$ ./client "message"
```

実行結果の例
![スクリーンショット 2024-01-24 16.47.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/282897/f1ae8b50-6394-f34e-3a38-cfa2fda2ec58.png)


# 最後に
今回はコードが長くなったので、ChatGPTにリファクタリングをお願いしてみました。

実装していくなかで、エラーメッセージが表示されると楽しい気持ちになりますね。
ただ、errnoはすぐわかるのに、定数名がわからないことが多々あり、少し困ったのが印象的でした。
（定数名を把握することは、必要ないのかもしれませんが。。。）

やっと基本的なことを一通り学べた思います。（エラー処理などの細かいところがまだまだ残っていますが。）
次回は、RFCを読み、IRCクライアントから送られてくるリクエストを受け取りレスポンスを返す機能を実装していきたいと思います。
ありがとうございました！