socketeventshandler.cpp

#include "socketeventshandler.h"

#include "socketevents.h"

namespace n2nc {
namespace net {

SocketEventsHandler::SocketEventsHandler(){
  ::pipe(this->m_wakeup);
  ::fcntl(this->m_wakeup[0],F_SETFL,O_NONBLOCK);
  ::fcntl(this->m_wakeup[1],F_SETFL,O_NONBLOCK);
/*  this->m_max_fd = std::max(this->m_max_fd, this->m_wakeup[0]);
  this->m_max_fd = std::max(this->m_max_fd, this->m_wakeup[1]);*/
}
SocketEventsHandler::~SocketEventsHandler(){
  ::close(this->m_wakeup[0]);
  ::close(this->m_wakeup[1]);

}


int SocketEventsHandler::add(Socket &sock,check_for_t checkfor, SocketEvents& se){
//   std::map<Socket*,SocketEvents*>::iterator it;
//   std::map<Socket::fd_t,socket_extra_t*>::iterator it ;
  
//   this->m_map.insert(std::pair<Socket*,SocketEvents*>(&sock,se));
  socket_extra_t *sockinfo = new socket_extra_t();
  ::bzero(sockinfo,sizeof(socket_extra_t));
  sockinfo->check = checkfor ;
  sockinfo->sock = &sock ;
  sockinfo->se = &se ;
  sockinfo->wait = NULL ;
  
  this->m_lock.lock();
//   this->tryAddMaxFD(sock.getFD() );

  this->m_map_new.insert( std::pair<Socket::fd_t,socket_extra_t*>(sock.getFD(),sockinfo) );
  this->m_lock.unlock();
  char t_dummy[4] ;
  ::write(this->m_wakeup[0],t_dummy,4); /* wakeup select to refresh pollings fds */
}

int SocketEventsHandler::del(Socket &sock){
  /* check it the caller is the same thread of socket event hanfler main_loop */
  if(pthread_equal(pthread_self(),this->m_thread)){
    this->m_map_new.find(sock.getFD())->second = NULL ; /* main_loop will accordly delete this socket */
//     this->computeMaxFD();
  }else{
    this->m_lock.lock();
    this->m_map_new.erase(sock.getFD());
    this->m_lock.unlock();
  }
//   this->m_map_new.erase(sock.getFD()); FIXME
  char t_dummy[4] ;
  ::write(this->m_wakeup[0],t_dummy,4);
}


void* SocketEventsHandler::entry_point(){
  this->setName("SKEV");
  this->_main_loop();
  return NULL ;
}

int SocketEventsHandler::waitForEvents(){
//   n2nc::sync::Thread::init(); /* FIXME call this my main */
  this->run(NULL);

}

int SocketEventsHandler::_main_loop(){
  int maxfd ,retval ;
  char t_dummy[1024];
  fd_set read_fds, write_fds, except_fds ;
  Socket *soc ;
  std::map<Socket::fd_t,socket_extra_t*>::iterator it ;
  
  while(true){
reinithack:
  
    /* zero variables */
    FD_ZERO(&read_fds);
    FD_ZERO(&write_fds);
    FD_ZERO(&except_fds);
    this->m_max_fd = 0 ;
    
    
    this->m_lock.lock();
    it = this->m_map_new.begin() ;
    while(it != this->m_map_new.end()){
      if(! it->second){std::cerr << "come aspettavo" << std::endl; this->m_map_new.erase(it);this->m_lock.unlock();goto reinithack; }
      
      if(it->second->check & SocketEventsHandler::NOTHING ){
        it++;
//         this->m_lock.unlock();
        continue ;
      }

      this->tryAddMaxFD(it->first);
      if(it->second->check & SocketEventsHandler::READ )    FD_SET(it->first,&read_fds);
      if(it->second->check & SocketEventsHandler::WRITE )   FD_SET(it->first,&write_fds);
      if(it->second->check & SocketEventsHandler::EXCEPT )  FD_SET(it->first,&except_fds);
      
      it++;
    }
    this->m_lock.unlock();
    
    FD_SET(this->m_wakeup[0],&read_fds);
    this->m_max_fd = std::max(this->m_max_fd, this->m_wakeup[0]);
//     this->m_max_fd = std::max(this->m_max_fd, this->m_wakeup[1]);

    /* SELECT */
    retval = ::select(this->m_max_fd+1, &read_fds, &write_fds, &except_fds, NULL);
    if(retval == -1)  std::cerr << "select error" << std::endl ;
    
    int a;
    this->m_lock.lock();
//     this->m_lock.unlock();
    it = this->m_map_new.begin() ;
    while(it != this->m_map_new.end()){
      if(! it->second){std::cerr << "come aspettavo" << std::endl; this->m_map_new.erase(it);this->m_lock.unlock();goto reinithack; }
      a = (READ * FD_ISSET(it->first,&read_fds)) + (WRITE * FD_ISSET(it->first,&write_fds)) + (EXCEPT * FD_ISSET(it->first,&except_fds) )   ;
      if( a ){
        if (it->second->wait) it->second->wait->signal();
        if (it->second->se)   it->second->se->onSelectEvent(it->second,(check_for_t)a );  /* FIXME by this function removing a socket cause deadlock */
      }
      it++;
    }
    this->m_lock.unlock();
  
    ::read(this->m_wakeup[0],t_dummy,1024);
  
  }
}

int SocketEventsHandler::computeMaxFD(){
//   std::for_each(this->m_map_new.begin(),this->m_map_new.end(), &SocketEventsHandler::tryAddMaxFD );
  std::map<Socket::fd_t,socket_extra_t*>::iterator it ;
  this->m_max_fd = 0 ;
  this->tryAddMaxFD( this->m_wakeup[0] );
  this->tryAddMaxFD( this->m_wakeup[1] );
  it = this->m_map_new.begin() ;
  while(it != this->m_map_new.end()){
    this->tryAddMaxFD(it->first);
    it++;
  }
  return this->m_max_fd ;
}

int SocketEventsHandler::tryAddMaxFD(Socket::fd_t fd){
  return this->m_max_fd = std::max(this->m_max_fd, fd);
}
/* FIXME remove me */
int SocketEventsHandler::tryAddMaxFD(const std::pair<Socket::fd_t,socket_extra_t*>& mypair){
  return this->m_max_fd = std::max(this->m_max_fd, mypair.first );
}


bool SocketEventsHandler::enableSocket(const Socket& sock){
  std::map<Socket::fd_t,socket_extra_t*>::iterator tmp_res ;
  this->m_lock.lock();
  tmp_res = this->m_map_new.find(sock.getFD());
  if (tmp_res == this->m_map_new.end() ) {
  /* Not found */
    this->m_lock.unlock();
    return false ;
  }
  tmp_res->second->check = (check_for_t)(tmp_res->second->check & ~NOTHING) ;
  this->m_lock.unlock();
  return true ;
}

bool SocketEventsHandler::disableSocket(const Socket& sock){
  std::map<Socket::fd_t,socket_extra_t*>::iterator tmp_res ;
  this->m_lock.lock();
  tmp_res = this->m_map_new.find(sock.getFD());
  if (tmp_res == this->m_map_new.end() ) {
  /* Not found */
    this->m_lock.unlock();
    return false ;
  }
  tmp_res->second->check = (check_for_t)(tmp_res->second->check | NOTHING) ;
  this->m_lock.unlock();
  return true ;
}



}
}

---