From fc83cb3fa5b15e9fb20a9180a3880297cf398de7 Mon Sep 17 00:00:00 2001 From: Mike Buland Date: Mon, 31 Jul 2023 16:12:54 -0700 Subject: Issue found with recycled fds. We don't always clean up instantly, but the system is reusing ids before we're ready. --- src/stable/client.cpp | 2 -- src/stable/queuebuf.cpp | 1 + src/stable/server.cpp | 57 +++++++++++++++++++++++-------------------------- src/stable/server.h | 2 +- 4 files changed, 29 insertions(+), 33 deletions(-) diff --git a/src/stable/client.cpp b/src/stable/client.cpp index 159d103..34dc798 100644 --- a/src/stable/client.cpp +++ b/src/stable/client.cpp @@ -13,8 +13,6 @@ #include "bu/clientlinkfactory.h" #include "bu/mutexlocker.h" -/** Read buffer size. */ -#define RBS (2000) // 1500 is the nominal MTU for ethernet, it's a good guess #ifdef PROFILE_BU_SERVER #define BU_PROFILE_START( x ) Bu::Profiler::getInstance().startEvent( x ) #define BU_PROFILE_END( x ) Bu::Profiler::getInstance().endEvent( x ) diff --git a/src/stable/queuebuf.cpp b/src/stable/queuebuf.cpp index f80f9d9..29da6ba 100644 --- a/src/stable/queuebuf.cpp +++ b/src/stable/queuebuf.cpp @@ -22,6 +22,7 @@ Bu::QueueBuf::~QueueBuf() { for( BlockList::iterator i = lBlocks.begin(); i; i++ ) delete[] *i; + lBlocks.clear(); } void Bu::QueueBuf::close() diff --git a/src/stable/server.cpp b/src/stable/server.cpp index cc89f64..160b3eb 100644 --- a/src/stable/server.cpp +++ b/src/stable/server.cpp @@ -15,7 +15,6 @@ #include "bu/socket.h" #include "bu/config.h" #include "bu/mutexlocker.h" -#include "bu/serverinterface.h" #include "bu/sio.h" @@ -150,8 +149,6 @@ void Bu::Server::scan() ); } - Bu::println("Cycle clear"); - for( int j = 0; j < iCount; j++ ) { if( hServers.has( ev[j].data.fd ) ) @@ -165,8 +162,6 @@ void Bu::Server::scan() } } - Bu::println("Waiting"); - Bu::List lDelete; // Now we just try to write all the pending data on all the sockets. // this could be done better eventually, if we care about the socket @@ -203,24 +198,32 @@ void Bu::Server::addClient( const Bu::ServerSocket *pSrv, Bu::Socket *pSocket ) throw Bu::ExceptionBase("No file descriptor?"); } - struct epoll_event ev; - ev.events = EPOLLIN | EPOLLONESHOT; - ev.data.fd = iFdCli; - epoll_ctl( pCore->fdRead, EPOLL_CTL_ADD, iFdCli, &ev ); - - ev.events = 0; - epoll_ctl( pCore->fdWrite, EPOLL_CTL_ADD, iFdCli, &ev ); - Client *pClient = new Client( iFdCli, ServerInterface( *this ) ); { Bu::MutexLocker l( mClients ); + if( hClients.has( iFdCli ) ) + { + Bu::println("!!!!!!!!! hClients has %1 already.").arg( iFdCli ); + } + if( hSockets.has( iFdCli ) ) + { + Bu::println("!!!!!!!!! hSockets has %1 already.").arg( iFdCli ); + } hClients.insert( iFdCli, pClient ); hSockets.insert( iFdCli, pSocket ); } + struct epoll_event ev; + ev.events = EPOLLIN | EPOLLONESHOT; + ev.data.fd = iFdCli; + epoll_ctl( pCore->fdRead, EPOLL_CTL_ADD, iFdCli, &ev ); + + ev.events = 0; + epoll_ctl( pCore->fdWrite, EPOLL_CTL_ADD, iFdCli, &ev ); + onNewConnection( pSrv, pClient, pSocket ); BU_PROFILE_END("addClient"); } @@ -333,6 +336,7 @@ void Bu::Server::shutdown() { closeClient( *i ); } + Bu::println("Cleaning up clients. Erased %1, remaining: %2.").arg( lClients.getSize() ).arg( hClients.getSize() ); hClients.clear(); } @@ -358,6 +362,11 @@ void Bu::Server::closeClient( fd iSocket ) { Bu::MutexLocker l( mClients ); BU_PROFILE_START("closeClient"); + + struct epoll_event ev; + epoll_ctl( pCore->fdRead, EPOLL_CTL_DEL, iSocket, &ev ); + epoll_ctl( pCore->fdWrite, EPOLL_CTL_DEL, iSocket, &ev ); + Bu::Client *pClient = hClients.get( iSocket ); Bu::Socket *pSocket = hSockets.get( iSocket ); onClosedConnection( pClient ); @@ -366,10 +375,6 @@ void Bu::Server::closeClient( fd iSocket ) pSocket->close(); hSockets.erase( iSocket ); - struct epoll_event ev; - epoll_ctl( pCore->fdRead, EPOLL_CTL_DEL, iSocket, &ev ); - epoll_ctl( pCore->fdWrite, EPOLL_CTL_DEL, iSocket, &ev ); - delete pClient; delete pSocket; BU_PROFILE_END("closeClient"); @@ -404,7 +409,7 @@ void Bu::Server::WriteMonitor::run() { if( ev[j].data.fd == rSrv.pCore->fdEvent ) { - Bu::println("Bu::Server::WriteMonitor -> got event on fdEvent, exiting..."); + // got event on fdEvent, exiting... return; } else @@ -498,8 +503,6 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket ) Bu::size iRead; Bu::size iTotal=0; - Bu::println("IoWorker::handleRead: starting"); - BU_PROFILE_START("client.read"); for(;;) { @@ -521,7 +524,6 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket ) } catch( Bu::ExceptionBase &e ) { - Bu::println("IoWorker::handleRead: exception, closing: %1").arg( e.what() ); pClient->disconnect(); //close( pSocket ); return; @@ -531,7 +533,6 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket ) if( iTotal == 0 ) { - Bu::println("IoWorker::handleRead: read nothing, closing"); pClient->disconnect(); //close( pSocket ); } @@ -547,9 +548,8 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket ) void Bu::Server::IoWorker::handleWrite( Client *pClient, Socket *pSocket ) { - Bu::println("IoWorker::handleWrite() "); char buf[RBS]; - if( pClient->hasOutput() > 0 ) + while( pClient->hasOutput() > 0 ) { int iAmnt = RBS; iAmnt = pClient->cbBuffer.server().peek( buf, iAmnt ); @@ -560,9 +560,9 @@ void Bu::Server::IoWorker::handleWrite( Client *pClient, Socket *pSocket ) void Bu::Server::IoWorker::close( Bu::Socket *pSocket ) { - Bu::Server::fd iFd; - pSocket->getFd( iFd ); - rSrv.closeClient( iFd ); +// Bu::Server::fd iFd; +// pSocket->getFd( iFd ); +// rSrv.closeClient( iFd ); } ///////// @@ -594,10 +594,7 @@ void Bu::Server::ClientWorker::run() continue; } - Bu::println("Processing input..."); pClient->processInput(); - Bu::println("Processing input complete."); - Bu::println("*** ClientBuf: input: %1b, output: %2b").arg( pClient->getInputSize() ).arg( pClient->getOutputSize() ); if( pClient->getOutputSize() > 0 ) { rSrv.clientWriteReady( pClient->getId() ); diff --git a/src/stable/server.h b/src/stable/server.h index 56ac29a..a839968 100644 --- a/src/stable/server.h +++ b/src/stable/server.h @@ -23,7 +23,7 @@ #include "bu/config.h" #ifndef PROFILE_BU_SERVER -// #define PROFILE_BU_SERVER 1 + #define PROFILE_BU_SERVER 1 #endif #ifdef PROFILE_BU_SERVER -- cgit v1.2.3