tom/RT_GPU
1
0
Fork 0
mirror of https://github.com/TheRedShip/RT_GPU.git synced 2025-09-27 18:48:36 +02:00
Code Issues Packages Projects Releases Wiki Activity

server now receives images and create final video. server now handles clients disconnect

Browse Source
This commit is contained in:
tomoron
2025-02-25 22:33:18 +01:00
parent 29f00cf9b2
commit 5ba33c4f69
11 changed files with 320 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
srcs/class/Clusterizer/server.cpp
View File

@ -6,7 +6,7 @@
/* By: tomoron <tomoron@student.42angouleme.fr> +#+ +:+ +#+ */
/* +#+#+#+#+#+ +#+ */
/* Created: 2025/02/20 21:08:38 by tomoron #+# #+# */
/* Updated: 2025/02/25 00:53:02 by tomoron ### ########.fr */
/* Updated: 2025/02/25 22:06:58 by tomoron ### ########.fr */
/* */
/* ************************************************************************** */
@ -15,7 +15,7 @@
void Clusterizer::initServer(std::string port)
{
_pollfds = 0;
_curId = 0;
_curFrame = 0;
try
{
@ -77,6 +77,47 @@ int Clusterizer::acceptClients(void)
return(1);
}
void Clusterizer::redistributeJob(t_job *job)
{
size_t highestInProgress = 0;
auto clientHighest = _clients.begin();
t_job *replaced;
std::vector<t_job *>::iterator found;
if(!_clients.size())
{
found = std::find(_jobs[IN_PROGRESS].begin(), _jobs[IN_PROGRESS].end(), job);
_jobs[IN_PROGRESS].erase(found);
_jobs[WAITING].insert(_jobs[WAITING].begin(), job);
}
for(auto it = _clients.begin(); it != _clients.end(); it++)
{
if(!it->second.curJob)
{
highestInProgress = (size_t) -1;
clientHighest = it;
}
else if(it->second.curJob->frameNb > highestInProgress)
{
highestInProgress = it->second.curJob->frameNb;
clientHighest = it;
}
}
if(clientHighest->second.curJob)
{
(void)write(clientHighest->first, (uint8_t []){ABORT}, 1);
replaced = clientHighest->second.curJob;
found = std::find(_jobs[IN_PROGRESS].begin(), _jobs[IN_PROGRESS].end(), replaced);
_jobs[IN_PROGRESS].erase(found);
_jobs[WAITING].insert(_jobs[WAITING].begin(), clientHighest->second.curJob);
}
clientHighest->second.curJob = job;
(void)write(clientHighest->first, (uint8_t []){JOB}, 1);
(void)write(clientHighest->first, job, sizeof(t_job));
}
void Clusterizer::deleteClient(int fd)
{
std::map<int, t_client>::iterator it;
@ -85,23 +126,88 @@ void Clusterizer::deleteClient(int fd)
it = _clients.find(fd);
if(it == _clients.end())
return;
if(_clients[fd].curJob)
redistributeJob(_clients[fd].curJob);
_clients.erase(it);
updatePollfds();
close(fd);
}
void Clusterizer::handleBuffer(int fd, std::vector<uint8_t> &buf)
void Clusterizer::getImageFromClient(int fd, std::vector<uint8_t> &buf)
{
std::vector<uint8_t> sendBuffer;
std::cout << "client sent an image" << std::endl;
buf.erase(buf.begin());
if(_clients[fd].gotGo && _clients[fd].curJob)
_renderer->addImageToRender(buf);
else
{
std::cout << "client sent an image before receiving a ready signal, dropping image" << std::endl;
buf.erase(buf.begin(), buf.begin() + (WIDTH * HEIGHT * 3));
return;
}
buf.erase(buf.begin(), buf.begin() + (WIDTH * HEIGHT * 3));
_clients[fd].gotGo = 0;
_clients[fd].curJob = 0;
_clients[fd].readyRespond = 0;
_jobs[IN_PROGRESS].erase(std::find(_jobs[IN_PROGRESS].begin(), _jobs[IN_PROGRESS].end(), _clients[fd].curJob));
_jobs[DONE].push_back(_clients[fd].curJob);
_clients[fd].curJob = 0;
_curFrame++;
}
bool Clusterizer::handleBuffer(int fd, std::vector<uint8_t> &buf)
{
std::vector<uint8_t> sendBuffer;
uint8_t tmp;
if(buf[0] == RDY)
{
_clients[fd].ready = 1;
buf.erase(buf.begin());
std::cout << "client " << fd << " is ready !" << std::endl;
}
else if(buf[0] == IMG_SEND_RQ)
{
std::cout << "client " << fd << " is ready to send its image" << std::endl;
_clients[fd].readyRespond = 1;
buf.erase(buf.begin());
}
else if(buf[0] == PROGRESS_UPDATE)
{
if(buf.size() < 2)
return (0);
tmp = buf[1];
std::cout << "client " << fd << " sent a progress update, new progress : " << (int)tmp << std::endl;
buf.erase(buf.begin(), buf.begin() + 2);
if (tmp <= 100)
_clients[fd].progress = tmp;
}
else if(buf[0] == IMG)
{
if(buf.size() < ((WIDTH * HEIGHT * 3) + 1))
{
std::cout << "incomplete IMG request, " << buf.size() << "/"<< (WIDTH * HEIGHT * 3) << std::endl;
return (0);
}
getImageFromClient(fd, buf);
}
else
{
std::cout << "client sent an unknown request, ignoring" << std::endl;
buf.erase(buf.begin());
}
if(sendBuffer.size())
(void)write(fd, sendBuffer.data(), sendBuffer.size());
return(buf.size() != 0);
}
int Clusterizer::updateBuffer(int fd)
@ -113,14 +219,14 @@ int Clusterizer::updateBuffer(int fd)
if(!ret || ret == (size_t)-1)
return(1);
_clients[fd].buffer.insert(_clients[fd].buffer.end(), buffer, buffer + ret);
handleBuffer(fd, _clients[fd].buffer);
while(handleBuffer(fd, _clients[fd].buffer))
;;
return(0);
}
int Clusterizer::dispatchJobs(void)
{
t_job *tmp;
uint8_t sendBuf;
int dispatched;
dispatched = 0;
@ -129,17 +235,17 @@ int Clusterizer::dispatchJobs(void)
for(auto it = _clients.begin(); it != _clients.end(); it++)
{
if(it->second.ready)
if(it->second.ready && !it->second.curJob)
{
tmp = _jobs[WAITING].front();
_jobs[WAITING].erase(_jobs[WAITING].begin());
_jobs[IN_PROGRESS].push_back(tmp);
sendBuf = JOB;
(void)write(it->first, &sendBuf, 1);
(void)write(it->first, &tmp, sizeof(t_job));
it->second.ready = 0;
(void)write(it->first, (uint8_t []){JOB}, 1);
(void)write(it->first, tmp, sizeof(t_job));
it->second.progress = 0;
it->second.curJob = tmp;
it->second.readyRespond = 0;
it->second.gotGo = 0;
dispatched = 1;
}
if(!_jobs[WAITING].size())
@ -148,6 +254,23 @@ int Clusterizer::dispatchJobs(void)
return (dispatched);
}
int Clusterizer::handleSendRequests(void)
{
int action;
action = 0;
for(auto it = _clients.begin(); it != _clients.end(); it++)
{
if(it->second.curJob && it->second.curJob->frameNb == _curFrame + 1 && it->second.readyRespond && !it->second.gotGo)
{
action = 1;
(void)write(it->first, (uint8_t []){RDY}, 1);
it->second.gotGo = 1;
}
}
return(action);
}
void Clusterizer::addJob(glm::vec3 pos, glm::vec2 dir, size_t samples, size_t frame, GPUDenoise &denoise)
{
t_job *tmp;
@ -159,6 +282,7 @@ void Clusterizer::addJob(glm::vec3 pos, glm::vec2 dir, size_t samples, size_t fr
tmp->frameNb = frame;
tmp->denoise = denoise;
_jobs[WAITING].push_back(tmp);
_curFrame = 0;
std::cout << "new job added : " << std::endl;
std::cout << " - pos : " << glm::to_string(pos) << std::endl;
@ -166,15 +290,45 @@ void Clusterizer::addJob(glm::vec3 pos, glm::vec2 dir, size_t samples, size_t fr
std::cout << std::endl;
}
void Clusterizer::abortJobs(void)
{
for(auto it = _jobs[WAITING].begin(); it != _jobs[WAITING].end(); it++)
delete *it;
_jobs[WAITING].clear();
for(auto it = _clients.begin(); it != _clients.end(); it++)
{
if(it->second.curJob)
{
(void)write(it->first, (uint8_t []){ABORT}, 1);
delete it->second.curJob;
it->second.curJob = 0;
it->second.progress = 0;
}
}
_jobs[IN_PROGRESS].clear();
for(auto it = _jobs[DONE].begin(); it != _jobs[DONE].end(); it++)
delete *it;
_jobs[DONE].clear();
}
void Clusterizer::updateServer(void)
{
int recv;
int didSomething;
if(!_jobs[WAITING].size() && !_jobs[IN_PROGRESS].size() && _jobs[DONE].size())
{
std::cout << "clusterized render done, closing output video" << std::endl;
_renderer->endRender(this);
}
didSomething = 1;
while(didSomething)
{
didSomething = acceptClients();
if(handleSendRequests())
didSomething = 1;
if(dispatchJobs())
didSomething = 1;
recv = poll(_pollfds, _clients.size(), 1);