clusterizer : server and client can now communicate

includes/RT.hpp

@@ -6,7 +6,7 @@
 /*   By: ycontre <ycontre@student.42.fr>            +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2024/09/27 14:52:10 by TheRed            #+#    #+#             */
-/*   Updated: 2025/02/20 21:28:15 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/21 14:53:36 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
@@ -47,8 +47,9 @@
 
 # include <sys/socket.h>
 # include <netinet/in.h>
-# include <fcntl.h>
 # include <arpa/inet.h>
+# include <fcntl.h>
+# include <poll.h>
 
 struct Vertex {
     glm::vec2 position;

includes/RT/Clusterizer.hpp

@@ -6,7 +6,7 @@
 /*   By: tomoron <tomoron@student.42angouleme.fr>   +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2025/02/20 18:25:18 by tomoron           #+#    #+#             */
-/*   Updated: 2025/02/20 22:41:05 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/22 01:43:51 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
@@ -21,40 +21,72 @@ typedef struct s_job
 	glm::vec3	pos;
 	glm::vec2	dir;
 	size_t		samples;
+	size_t		id;
 } t_job;
 
+typedef struct s_client
+{
+	std::vector<uint8_t> buffer;
+	t_job				*curJob;
+	bool				ready;
+}	t_client;
+
+typedef enum e_msg
+{
+	RDY,
+	JOB,
+	JOB_RES_RQ,
+	ACK,
+	WAIT,
+	IMAGE,
+	ERR,
+	UNKNOWN
+} t_msg;
+
 class Clusterizer
 {
 	public:
-		Clusterizer(Arguments args);
+		Clusterizer(Arguments &args);
 		~Clusterizer();
 
-	private:
 		void update(void);
 		bool getError(void);
-
+	private:
 		bool				_isActive;
 		bool				_isServer;
 		bool				_error;
+
 		std::vector<t_job> _jobs;
 
 	private: //client
-		void initClient(void);
+		void initClient(std::string &dest);
+		void openClientConnection(const char *ip, int port);
+		void clientHandleBuffer(void);
 		void updateClient(void);
+		void clientGetJob(std::vector<uint8_t> &sendBuf);
+		void clientReceive(void);
 
-		std::string			_serverIp;
+		int						_serverFd;
+		std::string				_serverIp;
+		int						_serverPort;
+		std::vector<uint8_t>	_receiveBuffer;
+		t_job					_currentJob;
 
 	private: //server
 		void initServer(std::string port);
 		void updateServer(void);
 
-		void	initServerSocket(uint16_t port);
+		void	initServerSocket(int port);
-
+		void	acceptClients(void);
-		int		_serverSocket;
+		void	updatePollfds(void);
-
+		int		updateBuffer(int fd);
-
+		void	handleBuffer(int fd, std::vector<uint8_t> &buf);
-
+		void	deleteClient(int fd);
+		
 
+		int							_serverSocket;
+		struct pollfd				*_pollfds;
+		std::map<int, t_client>		_clients;
 };
 
 #endif

srcs/RT.cpp

@@ -6,7 +6,7 @@
 /*   By: ycontre <ycontre@student.42.fr>            +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2024/09/27 14:51:49 by TheRed            #+#    #+#             */
-/*   Updated: 2025/02/15 22:54:52 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/21 18:17:33 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
@@ -28,6 +28,10 @@ int main(int argc, char **argv)
 	if (args.error())
 		return (1);
 
+	Clusterizer clusterizer(args);
+	if (clusterizer.getError())
+		return(1);
+
 	Scene		scene(args.getSceneName());
 	if (scene.fail())
 		return (1);
@@ -67,6 +71,7 @@ int main(int argc, char **argv)
 
 	while (!window.shouldClose())
 	{
+		clusterizer.update();
 		window.updateDeltaTime();
 		
 		updateDataOnGPU(scene, buffers);

srcs/class/Clusterizer/Clusterizer.cpp

@@ -6,13 +6,13 @@
 /*   By: tomoron <tomoron@student.42angouleme.fr>   +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2025/02/20 18:24:39 by tomoron           #+#    #+#             */
-/*   Updated: 2025/02/20 22:42:42 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/21 21:45:24 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
 #include "RT.hpp"
 
-Clusterizer::Clusterizer(Arguments args)
+Clusterizer::Clusterizer(Arguments &args)
 {
 	_isActive = 1;
 	_isServer = 0;
@@ -27,8 +27,7 @@ Clusterizer::Clusterizer(Arguments args)
 	else if(args.getBoolean("client"))
 	{
 		_isServer = 0;
-		_serverIp = *args.getString("client");
+		initClient(*args.getString("client"));
-		initClient();
 	}
 	else
 		_isActive = 0;

srcs/class/Clusterizer/client.cpp

@@ -6,18 +6,109 @@
 /*   By: tomoron <tomoron@student.42angouleme.fr>   +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2025/02/20 21:08:38 by tomoron           #+#    #+#             */
-/*   Updated: 2025/02/20 21:21:45 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/22 02:02:32 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
 #include "RT.hpp"
 
-void Clusterizer::initClient(void)
+void Clusterizer::initClient(std::string &dest)
 {
+	_serverFd = 0;
+	if(dest.find(":") == std::string::npos)
+		std::cerr << "Client Initialisation error : invalid ip provided format must be <ip>:<port>" << std::endl;
+	_serverIp = dest.substr(0, dest.find(":"));
+	_serverPort = stoi(dest.substr(dest.find(":") + 1));
 
+	try
+	{
+		openClientConnection(_serverIp.c_str(), _serverPort);
+	}
+	catch(std::exception &e)
+	{
+		if(_error)
+			std::cerr << "\033[31mClient initialisation error : " << e.what() << std::endl;
+	}
+}
+
+void Clusterizer::openClientConnection(const char *ip, int port)
+{
+    struct sockaddr_in  serv_addr;
+	uint8_t sendBuffer;
+
+	if(port > 65535 || port < 0)
+	{
+		_error = 1;
+		throw std::runtime_error("invalid port provided");
+	}
+    _serverFd = socket(AF_INET, SOCK_STREAM, 0);
+    if (_serverFd < 0)
+	{
+		_error = 1;
+		throw std::runtime_error("can't create socket"); 
+	}
+    bzero(&serv_addr, sizeof(struct sockaddr_in));
+    serv_addr.sin_family = AF_INET;
+    serv_addr.sin_addr.s_addr = inet_addr(ip);
+    serv_addr.sin_port = htons(port);
+    if(connect(_serverFd, (struct sockaddr *)&serv_addr, sizeof(struct sockaddr_in)))
+	{
+		close(_serverFd);
+		_serverFd = 0;
+	}
+	sendBuffer = RDY;
+	(void)write(_serverFd, &sendBuffer, 1);
+}
+
+void Clusterizer::clientGetJob(std::vector<uint8_t> &sendBuf)
+{
+	if(_receiveBuffer.size() < sizeof(t_job) + 1)
+		return ;
+	
+	_currentJob = *(t_job *)(_receiveBuffer.data() + 1);
+	_receiveBuffer.erase(_receiveBuffer.begin(), _receiveBuffer.begin() + sizeof(t_job) + 1);
+	sendBuf.push_back(ACK);
+}
+
+void Clusterizer::clientHandleBuffer(void)
+{
+	std::vector<uint8_t> sendBuf;
+
+	if(_receiveBuffer[0] == JOB)
+		clientGetJob(sendBuf);
+
+	if(sendBuf.size())
+		(void)write(1, sendBuf.data(), sendBuf.size());
+}
+
+void Clusterizer::clientReceive(void)
+{
+	uint8_t	buffer[512];
+	size_t	ret;
+
+	ret = recv(_serverFd, buffer, 512, MSG_DONTWAIT);
+	if(ret == (size_t)-1)
+		return;
+	if(!ret)
+	{
+		close(_serverFd);
+		_serverFd = 0;
+		return ;
+	}
+	_receiveBuffer.insert(_receiveBuffer.end(), buffer, buffer + ret);
 }
 
 void Clusterizer::updateClient(void)
 {
+	if(!_serverFd)
+	{
+		std::cout << "server isn't connected, waiting for connection" << std::endl;
+		while(!_serverFd)
+			openClientConnection(_serverIp.c_str(), _serverPort);
+		std::cout << "server reconnected." << std::endl;
+	}
 
+	clientReceive();
+	if(_receiveBuffer.size())
+		clientHandleBuffer();
 }

srcs/class/Clusterizer/server.cpp

@@ -6,7 +6,7 @@
 /*   By: tomoron <tomoron@student.42angouleme.fr>   +#+  +:+       +#+        */
 /*                                                +#+#+#+#+#+   +#+           */
 /*   Created: 2025/02/20 21:08:38 by tomoron           #+#    #+#             */
-/*   Updated: 2025/02/20 22:43:00 by tomoron          ###   ########.fr       */
+/*   Updated: 2025/02/21 22:19:20 by tomoron          ###   ########.fr       */
 /*                                                                            */
 /* ************************************************************************** */
 
@@ -14,9 +14,12 @@
 
 void Clusterizer::initServer(std::string port)
 {
+	_pollfds = 0;
+
 	try
 	{
 		initServerSocket(stoi(port));
+		std::cout << "server initialized with socket " << _serverSocket << std::endl;
 	}
 	catch(std::exception &e)
 	{
@@ -25,24 +28,107 @@ void Clusterizer::initServer(std::string port)
 	}
 }
 
-void	Clusterizer::initServerSocket(uint16_t port)
+void	Clusterizer::initServerSocket(int port)
 {
 	struct sockaddr_in	s_addr;
 
+	if(port > 65535 || port < 0)
+		throw std::runtime_error("invalid port");
 	_serverSocket = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
 	if (_serverSocket < 0)
 		throw std::runtime_error("can't create socket");
 	s_addr.sin_family = AF_INET;
 	s_addr.sin_addr.s_addr = INADDR_ANY;
-	s_addr.sin_port = port >> 8 | port << 8;
+	s_addr.sin_port = htons(port);
 	if (bind(_serverSocket, (struct sockaddr *)&s_addr, \
 	sizeof(struct sockaddr_in)) < 0)
 		throw std::runtime_error("can't bind socket");
 	if (::listen(_serverSocket, 50) < 0)
 		throw std::runtime_error("can't listen on socket");
+	if(_serverSocket == -1)
+		throw std::runtime_error("unknown error");
+}
+
+void	Clusterizer::updatePollfds(void)
+{
+	if(_pollfds)
+		delete[] _pollfds;
+	_pollfds = new struct pollfd[_clients.size()];
+	for(auto it = _clients.begin(); it != _clients.end(); it++)
+	{
+		_pollfds[std::distance(it, _clients.begin())].fd = it->first;
+		_pollfds[std::distance(it, _clients.begin())].events = POLLIN;
+		_pollfds[std::distance(it, _clients.begin())].revents = 0;
+	}
+}
+
+void Clusterizer::acceptClients(void)
+{
+	int fd;
+
+	fd = accept(_serverSocket, 0, 0);
+	if (fd != -1) {
+		std::cout << "new client :" << fd << std::endl;
+		_clients[fd].ready = 0;
+		_clients[fd].curJob = 0;
+		updatePollfds();
+	}
+}
+
+void Clusterizer::deleteClient(int fd)
+{
+	std::map<int, t_client>::iterator it;
+
+	std::cout << "client disconnected" << std::endl;
+	it = _clients.find(fd);
+	if(it == _clients.end())
+		return;
+	_clients.erase(it);
+	updatePollfds();
+	close(fd);
+}
+
+void Clusterizer::handleBuffer(int fd, std::vector<uint8_t> &buf)
+{
+	std::vector<uint8_t> sendBuffer;
+
+	if(buf[0] == RDY)
+	{
+		_clients[fd].ready = 1;
+		sendBuffer.push_back(ACK);
+	}
+
+	if(sendBuffer.size())
+		(void)write(fd, sendBuffer.data(), sendBuffer.size());
+}
+
+int Clusterizer::updateBuffer(int fd)
+{
+	uint8_t buffer[512];
+	size_t ret;
+
+	ret = recv(fd, buffer, 512, 0);
+	if(!ret)
+		return(1);
+	_clients[fd].buffer.insert(_clients[fd].buffer.end(), buffer, buffer + ret);
+	handleBuffer(fd, _clients[fd].buffer);
+	return(0);
 }
 
 void Clusterizer::updateServer(void)
 {
+	int recv;
 
+	acceptClients();
+	recv = poll(_pollfds, _clients.size(), 1);
+	if(!recv)
+		return ;
+	for(auto it = _clients.begin(); it != _clients.end(); it++)
+	{
+		if(_pollfds[std::distance(it, _clients.begin())].revents & POLLIN)
+		{
+			if (updateBuffer(it->first))
+				deleteClient(it->first);
+		}
+	}
 }