Issue found with recycled fds.

We don't always clean up instantly, but the system is reusing ids before
we're ready.

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<int> 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