Major changes to socket scheduling in EPoll.

Socket.cpp

@@ -10,15 +10,29 @@ namespace core {
       buffer = (char *)malloc(4096);
       length = 4096;
    }
+
+   Socket::Socket(EPoll &ePoll, std::string text) : ePoll(ePoll), text(text) {
+      coreutils::Log(coreutils::LOG_DEBUG_2) << "BMASocket object created [" << text << "].";
+      buffer = (char *)malloc(4096);
+      length = 4096;
+   }
    
    Socket::~Socket() {
-      ePoll.unregisterSocket(this);
-      close(descriptor);
       free(buffer);
+      if(descriptor == -1)
+	return;           
       onUnregister();
+      ePoll.unregisterSocket(this);
+      coreutils::Log(coreutils::LOG_DEBUG_3) << "Socket destroyed for socket " << descriptor << ".";
+      close(descriptor);
    }
    
    void Socket::setDescriptor(int descriptor) {
+      if((descriptor == -1) && (errno == 24)) {
+	 shutdown("Too many files open");
+	 coreutils::Exception("Too many files open. Refusing connection.");;
+      }
+      socketLock.lock();
       coreutils::Log(coreutils::LOG_DEBUG_3) << "Descriptor set to " << descriptor << " for Socket.";
       if(descriptor < 3)
         throw coreutils::Exception("Descriptor out of range", __FILE__, __LINE__);
@@ -26,6 +40,7 @@ namespace core {
       onRegister();
       ePoll.registerSocket(this);
       onRegistered();
+      socketLock.unlock();
    }
    
    int Socket::getDescriptor() {
@@ -43,43 +58,48 @@ namespace core {
    
    void Socket::onUnregister() {}
    
-   void Socket::eventReceived(struct epoll_event event) {
-               
+   bool Socket::eventReceived(struct epoll_event event) {
+
+      coreutils::Log(coreutils::LOG_DEBUG_1) << "eventReceived on " << descriptor << "; shutDown = " << shutDown << "; active = " << active << ";";
+	   
+      socketLock.lock();
+      
       if(event.events & EPOLLRDHUP) {
+	 coreutils::Log(coreutils::LOG_DEBUG_1) << "start EPOLLRDHUP " << descriptor;
          readHangup = true;
-         enable(false);
-         shutdown();
-         return;
+         shutdown("hangup received");
+         return false;
       }
 
-      if(event.events & EPOLLIN) 
-        receiveData(buffer, length);                  
-            
-      if(event.events & EPOLLOUT) 
-        writeSocket();
+      if(event.events & EPOLLIN) {
+	 coreutils::Log(coreutils::LOG_DEBUG_1) << "start EPOLLIN " << descriptor;	   
+	 receiveData(buffer, length);                  
+      }
+      
+      if(event.events & EPOLLWRNORM) {
+	 coreutils::Log(coreutils::LOG_DEBUG_1) << "start EPOLLOUT " << descriptor;	   
+	 writeSocket();
+      }
 
       if(event.events & EPOLLHUP) {
+	 coreutils::Log(coreutils::LOG_DEBUG_1) << "start EPOLLHUP " << descriptor;
+	 coreutils::Log(coreutils::LOG_DEBUG_1) << "end shutting down" << descriptor;
          shutdown();
-         return;
+	 return false;
       }
-         
-      enable(true);
+
+      coreutils::Log(coreutils::LOG_DEBUG_1) << "end with active = " << active << " on socket " << descriptor;
+      socketLock.unlock();
+
+      return active;
    }
    
-   void Socket::enable(bool mode) {
-      if(mode) {
-         if(fifo.empty()) {
-            if(!readHangup) {
-               active ? resetRead(): setRead();
-            }
-         }
-         else {
-            active ? resetReadWrite(): setReadWrite();
-         }
-         active = true;
-      }
-      else 
-        clear();
+   void Socket::onDataReceived(std::string data) {
+      throw coreutils::Exception("Need to override onDataReceived.", __FILE__, __LINE__, -1);
+   }
+      
+   void Socket::onDataReceived(char *buffer, int len) {
+      onDataReceived(std::string(buffer, len));                   
    }
    
    void Socket::receiveData(char *buffer, int bufferLength) {
@@ -91,7 +111,7 @@ namespace core {
       int error = -1;
       
       if((len = ::read(getDescriptor(), buffer, bufferLength)) >= 0) 
-        onDataReceived(std::string(buffer, len));       
+        onDataReceived(buffer, len);       
       else {
 	 
 	 error = errno;
@@ -114,78 +134,47 @@ namespace core {
       }
    }
    
-//   void Socket::onConnected() {
-//   }
-   
    void Socket::writeSocket() {
+      if(shutDown) 
+	return;
+      
       lock.lock();
       if(fifo.size() > 0) {
          ::write(descriptor, fifo.front().c_str(), fifo.front().length());
          fifo.pop();
-         enable(true);
+	 coreutils::Log(coreutils::LOG_DEBUG_4) << "reseSocket from writeSocket.";
+	 if(active)
+	   ePoll.resetSocket(this);
       }
       lock.unlock();
    }
    
-   void Socket::write(std::string data) {
+   int Socket::write(std::string data) {
+      if(!active) 
+	return -1;
+	
       lock.lock();
       fifo.emplace(data);   
-      enable(true);
+      coreutils::Log(coreutils::LOG_DEBUG_4) << "resetSocket from write. active is " << active;
+      if(active)
+	ePoll.resetSocket(this);
       lock.unlock();
+      return 1;
    }
    
    void Socket::output(std::stringstream &out) {
       out << "|" << descriptor << "|";
    }
+
+   bool Socket::needsToWrite() {
+      return fifo.size() > 0;
+   }   
    
-   void Socket::setRead() {
-      event.data.fd = descriptor;
-      event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;   
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, descriptor, NULL);  
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, event.data.fd, &event);      
-   }
-   
-   void Socket::setWrite() {
-      event.data.fd = descriptor;
-      event.events = EPOLLOUT | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;   
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, descriptor, NULL);  
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, descriptor, &event);      
-   }
-   
-   void Socket::setReadWrite() {
-      event.data.fd = descriptor;
-      event.events = EPOLLIN | EPOLLWRNORM | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;   
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, descriptor, NULL);  
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, descriptor, &event);      
-   }
-   
-   void Socket::resetRead() {
-      event.data.fd = descriptor;
-      event.events = EPOLLIN | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;   
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, descriptor, &event);      
-   }
-   
-   void Socket::resetWrite() {
-      event.data.fd = descriptor;
-      event.events = EPOLLOUT | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, descriptor, &event);      
-   }
-   
-   void Socket::resetReadWrite() {
-      event.data.fd = descriptor;
-      event.events = EPOLLIN | EPOLLOUT | EPOLLONESHOT | EPOLLRDHUP | EPOLLET;
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, descriptor, &event);      
-   }
-   
-   void Socket::clear() {
-      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, descriptor, NULL);
-   }
-   
-   void Socket::shutdown() {
-      coreutils::Log(coreutils::LOG_DEBUG_2) << "Shutdown requested on socket " << descriptor << ".";
+   void Socket::shutdown(std::string text) {
+      coreutils::Log(coreutils::LOG_DEBUG_2) << "Shutdown requested on socket " << descriptor << " with reason " << text << ".";
       shutDown = true;
-		enable(false);
-		delete this;
+      active = false;
+      delete this;
    }
    
 }