1 files changed, 151 insertions, 43 deletions

diff --git a/src/stable/server.cpp b/src/stable/server.cpp
index 7c44c11..cc89f64 100644
--- a/src/stable/server.cpp
+++ b/src/stable/server.cpp
@@ -8,11 +8,14 @@
 #include "bu/server.h"
 #include <errno.h>
 #include <unistd.h>
+#include <sys/epoll.h>
+#include <sys/eventfd.h>
 #include "bu/serversocket.h"
 #include "bu/client.h"
 #include "bu/socket.h"
 #include "bu/config.h"
 #include "bu/mutexlocker.h"
+#include "bu/serverinterface.h"
 
 #include "bu/sio.h"
 
@@ -25,14 +28,48 @@
 #endif
 
 #define RBS 1500
+#define MAX_EVENTS 20
+
+namespace Bu
+{
+    class __ServerCore
+    {
+    public:
+        __ServerCore()
+        {
+            fdRead = epoll_create( 1 );
+            fdWrite = epoll_create( 1 );
+            fdEvent = eventfd( 0, 0 );
+
+            struct epoll_event ev;
+            ev.events = EPOLLIN;
+            ev.data.fd = fdEvent;
+            epoll_ctl( fdRead, EPOLL_CTL_ADD, fdEvent, &ev );
+            epoll_ctl( fdWrite, EPOLL_CTL_ADD, fdEvent, &ev );
+        }
+
+        ~__ServerCore()
+        {
+            close( fdRead );
+            close( fdWrite );
+            close( fdEvent );
+        }
+
+        Server::fd fdRead;
+        Server::fd fdWrite;
+        Server::fd fdEvent;
+    };
+}
 
 Bu::Server::Server( int iIoWorkers, int iClientWorkers ) :
+    pCore( new Bu::__ServerCore() ),
     nTimeoutSec( 0 ),
     nTimeoutUSec( 0 ),
-    bAutoTick( false )
+    bAutoTick( false ),
+    tMonitorWrite( *this ),
+    bRunning( true )
 {
     BU_PROFILE_START("server");
-    FD_ZERO( &fdActive );
 
     if( iIoWorkers < 1 )
         iIoWorkers = 1;
@@ -52,11 +89,14 @@ Bu::Server::Server( int iIoWorkers, int iClientWorkers ) :
         lClientWorker.append( pWorker );
         pWorker->start();
     }
+    tMonitorWrite.start();
 }
 
 Bu::Server::~Server()
 {
     shutdown();
+    tMonitorWrite.join();
+    delete pCore;
     BU_PROFILE_START("server");
 }
 
@@ -68,7 +108,11 @@ void Bu::Server::addServerSocket( Bu::ServerSocket *pSocket )
         throw Bu::ExceptionBase("Cannot get file descriptor from "
                 "provided ServerSocket.");
     }
-    FD_SET( iFd, &fdActive );
+
+    struct epoll_event ev;
+    ev.events = EPOLLIN;
+    ev.data.fd = iFd;
+    epoll_ctl( pCore->fdRead, EPOLL_CTL_ADD, iFd, &ev );
     hServers.insert( iFd, pSocket );
 }
 
@@ -81,26 +125,22 @@ void Bu::Server::setTimeout( int nTimeoutSec, int nTimeoutUSec )
 void Bu::Server::scan()
 {
     BU_PROFILE_START("scan");
-    struct timeval xTimeout = { nTimeoutSec, nTimeoutUSec };
     
-    fd_set fdRead = fdActive;
-    fd_set fdWrite /* = fdActive*/;
-    fd_set fdException = fdActive;
-
-    FD_ZERO( &fdWrite );
     mClients.lock();
     for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
     {
         if( (*i)->hasOutput() )
         {
-            Bu::println("Socket has output...");
-            FD_SET( i.getKey(), &fdWrite );
+            // Queue it for output here?
         }
     }
     mClients.unlock();
 
-    if( TEMP_FAILURE_RETRY( select( FD_SETSIZE,
-            &fdRead, &fdWrite, &fdException, &xTimeout ) ) < 0 )
+    struct epoll_event ev[MAX_EVENTS];
+
+    int iCount = epoll_wait( pCore->fdRead, ev, MAX_EVENTS, nTimeoutSec*1000+nTimeoutUSec/1000 );
+
+    if( iCount < 0 )
     {
         char buf[1024];
         strerror_r( errno, buf, 1024 );
@@ -110,41 +150,29 @@ void Bu::Server::scan()
             );
     }
 
-    evIoCycle.clear();
     Bu::println("Cycle clear");
 
-    for( int j = 0; j < FD_SETSIZE; j++ )
+    for( int j = 0; j < iCount; j++ )
     {
-        if( FD_ISSET( j, &fdRead ) )
+        if( hServers.has( ev[j].data.fd ) )
         {
-            if( hServers.has( j ) )
-            {
-                Bu::ServerSocket *pSrv = hServers.get( j );
-                addClient( pSrv, pSrv->accept() );
-            }
-            else
-            {
-                evIoCycle.increment();
-                Bu::println("Increment (read)");
-                qIoEvent.enqueue( new Event( j, Event::Read ) );
-            }
+            Bu::ServerSocket *pSrv = hServers.get( ev[j].data.fd );
+            addClient( pSrv, pSrv->accept() );
         }
-        if( FD_ISSET( j, &fdWrite ) )
+        else
         {
-            evIoCycle.increment();
-            Bu::println("Increment (write)");
-            qIoEvent.enqueue( new Event( j, Event::Write ) );
+            qIoEvent.enqueue( new Event( ev[j].data.fd, Event::Read ) );
         }
     }
 
     Bu::println("Waiting");
-    while( evIoCycle.wait() > 0 ) { }
 
     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
     // wanting to accept writes (using a select).
     mClients.lock();
+
     for( ClientHash::iterator i = hClients.begin(); i != hClients.end(); i++ )
     {
         if( (*i)->wantsDisconnect() && !(*i)->hasOutput() )
@@ -174,10 +202,18 @@ void Bu::Server::addClient( const Bu::ServerSocket *pSrv, Bu::Socket *pSocket )
     {
         throw Bu::ExceptionBase("No file descriptor?");
     }
-    FD_SET( iFdCli, &fdActive );
+
+    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(
-        new SrvClientLinkFactory()
+        iFdCli,
+        ServerInterface( *this )
         );
     {
         Bu::MutexLocker l( mClients );
@@ -252,6 +288,11 @@ void Bu::Server::tick()
 void Bu::Server::shutdown()
 {
     {
+        int64_t iVal = 1;
+        write( pCore->fdEvent, &iVal, sizeof(int64_t) );
+    }
+
+    {
         qIoEvent.stop();
         qClientEvent.stop();
         Bu::Server::Event *pEv;
@@ -294,6 +335,24 @@ void Bu::Server::shutdown()
     }
     hClients.clear();
 }
+        
+void Bu::Server::clientReadReady( fd iFd )
+{
+    // Re-arm
+    struct epoll_event ev;
+    ev.events = EPOLLIN | EPOLLONESHOT;
+    ev.data.fd = iFd;
+    epoll_ctl( pCore->fdRead, EPOLL_CTL_MOD, iFd, &ev );
+}
+
+void Bu::Server::clientWriteReady( fd iFd )
+{
+    // Re-arm
+    struct epoll_event ev;
+    ev.events = EPOLLOUT | EPOLLONESHOT;
+    ev.data.fd = iFd;
+    epoll_ctl( pCore->fdWrite, EPOLL_CTL_MOD, iFd, &ev );
+}
 
 void Bu::Server::closeClient( fd iSocket )
 {
@@ -306,13 +365,60 @@ void Bu::Server::closeClient( fd iSocket )
     hClients.erase( iSocket );
     pSocket->close();
     hSockets.erase( iSocket );
-    FD_CLR( iSocket, &fdActive );
+
+    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");
 }
 
 ////////
+// WriteMonitor
+////
+Bu::Server::WriteMonitor::WriteMonitor( Server &rSrv ) :
+    rSrv( rSrv )
+{
+}
+
+Bu::Server::WriteMonitor::~WriteMonitor()
+{
+}
+
+void Bu::Server::WriteMonitor::run()
+{
+    setName("busrv-writeMon");
+    struct epoll_event ev[MAX_EVENTS];
+    for(;;)
+    {
+        int iCount = epoll_wait( rSrv.pCore->fdWrite, ev, MAX_EVENTS, 5000 );
+        if( iCount < 0 )
+        {
+            // Bad error?
+            return;
+        }
+
+        for( int j = 0; j < iCount; j++ )
+        {
+            if( ev[j].data.fd == rSrv.pCore->fdEvent )
+            {
+                Bu::println("Bu::Server::WriteMonitor -> got event on fdEvent, exiting...");
+                return;
+            }
+            else
+            {
+                // Queue the write
+                rSrv.qIoEvent.enqueue(
+                    new Event( ev[j].data.fd, Event::Write )
+                    );
+            }
+        }
+    }
+}
+
+////////
 // Event
 ////
 
@@ -365,7 +471,6 @@ void Bu::Server::IoWorker::run()
         if( !rSrv.getClientAndSocket( pEv->getId(), pClient, pSocket ) )
         {
             delete pEv;
-            rSrv.evIoCycle.decrement();
             continue;
         }
 
@@ -384,9 +489,6 @@ void Bu::Server::IoWorker::run()
         }
         
         delete pEv;
-
-        Bu::println("decrement");
-        rSrv.evIoCycle.decrement();
     }
 }
 
@@ -404,7 +506,6 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket )
         try
         {
             iRead = pSocket->read( buf, RBS );
-            Bu::println("IoWorker::handleRead: read() -> %1").arg( iRead );
 
             if( iRead == 0 )
             {
@@ -421,7 +522,8 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket )
         catch( Bu::ExceptionBase &e )
         {
             Bu::println("IoWorker::handleRead: exception, closing: %1").arg( e.what() );
-            close( pSocket );
+            pClient->disconnect();
+            //close( pSocket );
             return;
         }
     }
@@ -430,13 +532,16 @@ void Bu::Server::IoWorker::handleRead( Client *pClient, Socket *pSocket )
     if( iTotal == 0 )
     {
         Bu::println("IoWorker::handleRead: read nothing, closing");
-        close( pSocket );
+        pClient->disconnect();
+        //close( pSocket );
     }
     else
     {
         Bu::Server::fd iFd;
         pSocket->getFd( iFd );
         rSrv.qClientEvent.enqueue( new Event( iFd, Event::Process ) );
+
+        rSrv.clientReadReady( iFd );
     }
 }
 
@@ -449,7 +554,6 @@ void Bu::Server::IoWorker::handleWrite( Client *pClient, Socket *pSocket )
         int iAmnt = RBS;
         iAmnt = pClient->cbBuffer.server().peek( buf, iAmnt );
         int iReal = pSocket->write( buf, iAmnt );
-        Bu::println("IoWorker::handleWrite: Copy out: iAmnt=%1, iReal=%2\n>%3<").arg( iAmnt ).arg( iReal ).arg( Bu::String( buf, iReal ) );
         pClient->cbBuffer.server().seek( iReal );
     }
 }
@@ -494,6 +598,10 @@ void Bu::Server::ClientWorker::run()
         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() );
+        }
         delete pEv;
     }
 }