fixed SIGPIPE killing program

This commit is contained in:
2020-02-08 21:53:09 +01:00
parent eb6a202c71
commit bc40ddbd9f
2 changed files with 43 additions and 107 deletions

View File

@ -10,6 +10,6 @@ add_executable(${PROJECT_NAME} ${SRC_FILES})
target_link_libraries( ${PROJECT_NAME} ${LIBS}) target_link_libraries( ${PROJECT_NAME} ${LIBS})
set_target_properties( ${PROJECT_NAME} PROPERTIES COMPILE_FLAGS -m64 LINK_FLAGS -m64) set_target_properties( ${PROJECT_NAME} PROPERTIES COMPILE_FLAGS -m64 LINK_FLAGS -m64)
add_definitions(" -std=c++11 -Wall ") add_definitions(" -std=c++11 -Wall -O2")
install(TARGETS serialmultiplexer RUNTIME DESTINATION bin) install(TARGETS serialmultiplexer RUNTIME DESTINATION bin)

148
main.cpp
View File

@ -14,25 +14,18 @@
#include "serial_io.h" #include "serial_io.h"
#include "Socket.h" #include "Socket.h"
#include "clienthandler.h"
#define VERSION "v0.5" #define VERSION "v0.6"
volatile bool stop = false; volatile bool stop = false;
volatile bool resettSerialPort = false; volatile bool resettSerialPort = false;
struct epoll_event ev;
void intHandler(int sig) void intHandler(int sig)
{ {
stop = true; stop = true;
} }
void alarmHandler(int sig)
{
resettSerialPort = true;
signal(sig, alarmHandler);
alarm(600);
}
static void printUsage() static void printUsage()
{ {
std::cout<<"usage mulitplexer [option]\n\ std::cout<<"usage mulitplexer [option]\n\
@ -103,18 +96,22 @@ static int parseCmdArgs(int argc, char** argv, Config *config)
return 0; return 0;
} }
void acceptThreadFunction( TCPServerSocket* servSock, std::vector<TCPSocket*>* clientSockets, std::mutex* clientSocketsMutex, int pollQue ) void acceptThreadFunction( TCPServerSocket* servSock, std::vector<ClientHandler>* clients, std::mutex* clientsMutex, int pollQue )
{ {
while(!stop) while(!stop)
{ {
TCPSocket* newSock = servSock->accept(); TCPSocket* newSock = servSock->accept();
if(newSock != nullptr) if(newSock != nullptr)
{ {
clientSocketsMutex->lock(); clientsMutex->lock();
clientSockets->push_back(newSock); // Wait for a client to connect clients->push_back(ClientHandler(newSock)); // Wait for a client to connect
clientSocketsMutex->unlock(); struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = clients->size()-1;
clientsMutex->unlock();
epoll_ctl(pollQue, EPOLL_CTL_ADD, newSock->getFD(), &ev); epoll_ctl(pollQue, EPOLL_CTL_ADD, newSock->getFD(), &ev);
clientSockets->back()->send("UVOS serial multiplexer " VERSION "\n", sizeof("UVOS serial multiplexer " VERSION "\n")-1); char welcomeMesg[] = "UVOS serial multiplexer " VERSION "\n";
clients->back().write(welcomeMesg, sizeof(welcomeMesg)-1);
std::cout<<"got client\n"; std::cout<<"got client\n";
} }
std::this_thread::sleep_for(std::chrono::seconds(1)); std::this_thread::sleep_for(std::chrono::seconds(1));
@ -148,7 +145,6 @@ int main(int argc, char* argv[])
std::cout<<"UVOS serial mulitplexer "<<VERSION<<'\n'; std::cout<<"UVOS serial mulitplexer "<<VERSION<<'\n';
int pollQue = epoll_create1(0); int pollQue = epoll_create1(0);
ev.events = EPOLLIN;
int serial = openSerialPort(config); int serial = openSerialPort(config);
if(!config.noSerial) if(!config.noSerial)
@ -158,15 +154,17 @@ int main(int argc, char* argv[])
if(serial == -1) return 1; if(serial == -1) return 1;
else else
{ {
struct epoll_event ev; struct epoll_event ev = {};
ev.events = EPOLLIN;
ev.data.fd = -1;
epoll_ctl(pollQue, EPOLL_CTL_ADD, serial, &ev); epoll_ctl(pollQue, EPOLL_CTL_ADD, serial, &ev);
} }
tcflush(serial, TCIOFLUSH); tcflush(serial, TCIOFLUSH);
} }
else std::cout<<"Sinkless mode\n"; else std::cout<<"Sinkless mode\n";
std::mutex clientSocketsMutex; std::mutex clientsMutex;
std::vector<TCPSocket*> clientSockets; std::vector<ClientHandler> clients;
std::thread* acceptThread; std::thread* acceptThread;
TCPServerSocket* servSock; TCPServerSocket* servSock;
@ -176,7 +174,7 @@ int main(int argc, char* argv[])
{ {
servSock = new TCPServerSocket(config.port, 5, true); // Server Socket object servSock = new TCPServerSocket(config.port, 5, true); // Server Socket object
servSock->setBlocking(false); servSock->setBlocking(false);
acceptThread = new std::thread(acceptThreadFunction, servSock, &clientSockets, &clientSocketsMutex, pollQue); acceptThread = new std::thread(acceptThreadFunction, servSock, &clients, &clientsMutex, pollQue);
} }
catch(SocketException &e) catch(SocketException &e)
{ {
@ -184,114 +182,52 @@ int main(int argc, char* argv[])
return 1; return 1;
} }
char buffer[4096];
signal(SIGINT, intHandler); signal(SIGINT, intHandler);
signal(SIGTERM, intHandler); signal(SIGTERM, intHandler);
signal(SIGALRM, alarmHandler); signal(SIGPIPE, SIG_IGN); //ignore SIGPIPE
if(config.reinit) alarm(600);
std::cout<<"starting loop\n"; std::cout<<"starting loop\n";
while(!stop) while(!stop)
{ {
epoll_wait(pollQue, &ev, 1, 2000); struct epoll_event ev;
int readlen = sRead(serial, buffer, 4096); if( epoll_wait(pollQue, &ev, 1, 2000) == 1)
for(unsigned int i = 0; i < clientSockets.size(); i++)
{ {
clientSocketsMutex.lock(); if(ev.data.fd != -1)
try
{ {
if(readlen > 0) int i = ev.data.fd;
std::cout<<"client poll\n";
clientsMutex.lock();
if(ev.events & EPOLLIN) clients[i].run(&clients, serial, config.verbose);
if((ev.events & (EPOLLHUP | EPOLLERR)) || clients[i].isDisconnected())
{ {
if(config.verbose) clients[i].cleanUp();
{ clients.erase(clients.begin()+i);
std::cout<<"bcst: "; std::cout<<"client "<<i<<" disconnected\n";
for( int j = 0; j < readlen; j++ )std::cout<<buffer[j];
std::cout<<std::endl;
}
clientSockets[i]->send(buffer, readlen);
} }
clientsMutex.unlock();
char inBuffer[4096]; }
int reclen = clientSockets[i]->recv(inBuffer, 4096); else
if( reclen > 0 ) {
char buffer[4096];
int readlen = sRead(serial, buffer, 4096);
if(config.verbose)
{ {
if(config.verbose) std::cout<<"Sending \"";
{ for(size_t i = 0; i < readlen; ++i) std::cout<<buffer[i];
std::cout<<"rec: "; std::cout<<"\" to clients from serial\n";
for( int j = 0; j < reclen; j++ )std::cout<<inBuffer[j];
}
if(strncmp( inBuffer, "bcst:", 5) == 0)
{
if(config.verbose)
{
std::cout<<"bcst: ";
for( int j = 0; j < reclen; j++ )std::cout<<inBuffer[j];
std::cout<<std::endl;
}
for(unsigned int j = 0; j < clientSockets.size(); j++) if(i != j) clientSockets[j]->send(inBuffer, reclen);
}
else if(strncmp( inBuffer, "reinitsp", 8) == 0 && serial != -1)
{
if(serialport_set_config(serial, config.baud) == 0)
{
char response[] = "reinit serial port succsesfull\n";
std::cerr<<response;
clientSockets[i]->send(response, sizeof(response));
}
else
{
char response[] = "reinit serial port failed!\n";
std::cerr<<response;
clientSockets[i]->send(response, sizeof(response));
}
}
else
{
if(config.verbose)
{
std::cout<<"wrote \"";
for( int j = 0; j < reclen; j++ )std::cout<<inBuffer[j];
std::cout<<"\" to serial \n";
}
sWrite(serial, inBuffer, reclen);
}
if(config.verbose) std::cout<<std::endl;
} }
else if(reclen == 0) for(ClientHandler& client : clients)
{ {
std::cout<<"client "<<i<<" disconnected"<<'\n'; client.write(buffer, readlen);
clientSockets[i]->cleanUp();
clientSockets.erase(clientSockets.begin()+i);
i--;
if(i < 0) i=0;
} }
} }
catch(SocketException &e)
{
std::cout<<e.what()<<std::endl;
clientSockets[i]->cleanUp();
clientSockets.erase(clientSockets.begin()+i);
i--;
if(i < 0) i=0;
}
clientSocketsMutex.unlock();
}
if(resettSerialPort)
{
close(serial);
serial = openSerialPort(config);
resettSerialPort = false;
} }
} }
acceptThread->join(); acceptThread->join();
delete acceptThread; delete acceptThread;
for(unsigned int i = 0; i < clientSockets.size(); i++) clientSockets[i]->cleanUp(); for(ClientHandler& client : clients) client.cleanUp();
servSock->cleanUp(); servSock->cleanUp();
delete servSock; delete servSock;
return 0; return 0;