cabal_channel.cpp

/*
   CABAL - CAched BAlanced LAN

   Channel class.

   $Id: cabal_channel.cpp,v 1.7 2004/04/05 21:54:18 jonnymind Exp $
---------------------------------------------
   Begin      : 2004-01-08 19:53 UTC+0100
   Author     : Giancarlo Niccolai

   Last modified because:

*/

#ifdef __MSC_VER
   #pragma warning ( disable : 4786 )
#endif

#include <cassert>
#include <cabal_channel.h>
#include <cabal_socket.h>
#include <cabal_conn.h>
#include <cabal_trans.h>
#include <cabal_recpt.h>

#include <iostream>
using namespace std;

namespace Cabal
{

void Channel::addSocket( Socket *skt )
{
   skt->incref();
   m_sockets.push_back( skt );
}
   
void Channel::removeSocket( Socket *skt ) 
{
   SocketList::iterator it = m_sockets.begin();
   while( it != m_sockets.end() ){
      if ( *it == skt ) {
         m_sockets.erase( it );
         skt->decref();
         return;
      }
      it++;
   }
   //oops, it wasent ours
}

void Channel::schedule( Connection *con, const Priority prio )
{
   ConList *shedList = &m_cons[static_cast<int>(prio)];
   
   ConList::reverse_iterator rit = shedList->rbegin();
   MSECS conSched = con->nextSchedule();
   
   while( rit != shedList->rend() )
   {
      if ( (*rit)->nextSchedule() <= conSched ) 
      {
         shedList->insert( rit.base(), con );
         break;
      }
      rit ++;
   }
   
   // it's the most urgent! reprioritize ourself.
   if ( rit == shedList->rend() ) {
      shedList->insert( shedList->begin(), con );
      m_nextSchedule = conSched;
   }
}
   
void Channel::deschedule( const Connection *con, const Priority prio, const bool complete )
{
   // a descheduled connection is more likely near the bottom.
   ConList *lst = &m_cons[static_cast<int>(prio)];
   ConList::iterator it = lst->begin();
   
   while( it != lst->end() )
   {
      if ( *it == con ) {
         it = lst->erase( it );
         if ( ! complete )
            return;  
      }
      else
         it++;
   }
}
   

bool Channel::process( MSECS msecs )
{
   RawSocket maxfd;
   fd_set wr_set, rd_set;
   fd_set *wr_set_point, *rd_set_point;
   struct timeval tv, *wait_tv;
   ConList *clist_used;
   SocketList::iterator sit;
   
   // temporal references.
   MSECS curTime = Net::timeOfDay();
   MSECS spanTime = curTime + msecs;
   MSECS cleanTime;
   
   if ( m_bandwidth > 0 ) {
      cleanTime = m_lastSentTime + (m_lastSentSize * 1000 / m_bandwidth);
   }
   else 
      cleanTime = -1;
      
   // get the maximum time that writers must wait.
   
   MSECS readTime;

   // have we a schedule?
   if ( m_nextSchedule > 0 ) 
   {
      // is the first next schedule in this time slice? 
      // Have we enough bandwidth for this time slice ?
      if ( m_nextSchedule < spanTime && ( cleanTime < m_nextSchedule || cleanTime < curTime ) ) {
         // then, wait for the highest time between next schedule and clean (bandwidth) time.
         readTime = ( m_nextSchedule > cleanTime ? m_nextSchedule : cleanTime ) - curTime;
         // readTime <= 0 means immediately go to read/write phase.
         /*MSECS minWait = minimalWait(); // else, temper with minimal wait.
         if ( readTime < minWait ) {
            readTime = msecs < minWait ? msecs : minWait;
         }*/
      }
      else
         // use all the timeslice
         readTime = msecs == 0 ? 1: msecs; // set a fake read idle wait
   }
   else {
      // dedicates all the time for reading.
      readTime = msecs;
   }
   
   if ( readTime > 0 )
   {           
      bool readDone = false;      
      // reads all subscribed sockets.
      if ( readTime > msecs ) readTime = msecs;
      FD_ZERO( &rd_set );
      maxfd = 0;
      sit = m_sockets.begin();
      rd_set_point = 0;
      while( sit != m_sockets.end() )
      {
         Socket *skt = (*sit);
         // remove closed sockets in the meanwhile
         if ( skt->closed() ) {
            sit = m_sockets.erase( sit );
            skt->decref();
            continue;
         }
         
         if ( skt->reception() ) 
         {
            RawSocket fd = skt->rawSocket();
            FD_SET( fd, &rd_set );
            rd_set_point = &rd_set;
            if ( maxfd < fd ) maxfd = fd;
         }
         sit++;
      }
      
      // WAIT EVEN if there are no things to wait for, unless timeout is zero.
      if ( rd_set_point != 0 || msecs > 0 )
      {      
   
         // this time is completely dedicated to reading; so start reading.
         // Wait for the read time.
         tv.tv_sec = readTime/1000;
         tv.tv_usec = (readTime % 1000) * 1000;
         
         int result;
         Net::resetError();
         result = select( maxfd + 1, rd_set_point, 0, 0, &tv );
         
         // now, parse incoming traffic, if present.
         if ( result > 0 ) 
         {
            sit = m_sockets.begin();
            readDone = true;
            
            while( result > 0 && sit != m_sockets.end() )
            {
               Socket *skt = (*sit);
               if ( skt->reception() != 0 && FD_ISSET( skt->rawSocket(), rd_set_point ) )
               {
                  bool cont = skt->reception()->receive( skt );
                  result--;
                  // in case of error on the socket, remove it.
                  if ( skt->osError() != 0 || skt->closed() || ! cont )
                  {
                     sit = m_sockets.erase( sit );
                     skt->decref();
                     // important; if you don't do this, it's possible that all
                     // the receivers are just killed.
                     Net::resetError();
                     continue;
                  }
               }
               sit ++;
            }
            curTime = Net::timeOfDay();
         
            // Too late? Next time will be better; and if we have received early reads,
            // return if the next schedule was not available in current timeslice (msecs)      
            if ( curTime > spanTime ) { 
               //TRICK: round-robinize the receiver list.
               recptRoundRobin();
               return readDone; // we did something?
            }
         }
         else if ( msecs == readTime ) // all timeslice elapsed in waiting for nothing; return.
            return false;
      }
      else if( msecs == readTime )
         return false;
   }
   
   // ok, now it's time for read/writing.
   // if we are here, part (or all) the schedule is dedicated to writing
   // Again (even if already done), resubscribe for reading.
   // Prepare sockets with readers to subscribe the select
   
   FD_ZERO( &rd_set );
   maxfd = 0;
   sit = m_sockets.begin();
   rd_set_point = 0;
   while( sit != m_sockets.end() )
   {
      Socket *skt = (*sit);
      // remove closed sockets in the meanwhile
      if ( skt->closed() ) {
         sit = m_sockets.erase( sit );
         skt->decref();
         continue;
      }
      
      if ( skt->reception() ) 
      {
         RawSocket fd = skt->rawSocket();
         FD_SET( fd, &rd_set );
         // uses as a flag; it's a 32bit copy, as a bool...
         rd_set_point = &rd_set;
         if ( maxfd < fd ) maxfd = fd;
      }
      sit++;
   }
      
   // prepare the connection willing to send to enter the select for writing.
   
   FD_ZERO( &wr_set );   
   // we know we must write, or we would have dropped the function before.
   //... unless some writer scheduled to write has been removed.
   wr_set_point = 0;
      
   // Get the first non empty priority queue
   for( int queueId = 0; queueId < CABAL_PRIORITIES ; queueId++ )
   {
      if ( ! m_cons[ queueId ].empty() ) 
      {
         ConList::iterator it = m_cons[queueId].begin();
         // if the object is scheduled to go, put it in the select set.
         MSECS sched = (*it)->nextSchedule();
                     
         if ( sched > 0 && sched <= curTime ) 
         {  
            // get at least one.
            do {
               // ignore closed sockets in the meanwhile
               Socket *skt = (*it)->socket();
               if ( ! skt->closed() ) {
                  RawSocket fd = skt->rawSocket();
                  //FD_Setting a socker more than one time has no effect... so we do it!
                  FD_SET( fd, &wr_set );
                  wr_set_point = &wr_set;  // using the pointer as a flag.
                  if ( maxfd < fd ) maxfd = fd;
               }
               it++;
            } while ( it != m_cons[ queueId ].end() && (*it)->nextSchedule() <= curTime );
            
            // get only the first queue that has scheduled things
            clist_used = & m_cons[ queueId ];
            
            // reprioritize next schedule
            if ( it != clist_used->end() ) 
            {
               m_nextSchedule = (*it)->nextSchedule();
            }
            else {
               m_nextSchedule = -1;
               for ( int qid = queueId+1; qid < CABAL_PRIORITIES; qid ++ ) 
               {
                  if ( ! m_cons[ qid ].empty() ) 
                  {
                     m_nextSchedule = m_cons[ qid ].front()->nextSchedule();
                     break;
                  }
               }
            }
                     
            break;  
         }
         // else, adequate the future schedule.
         else if ( m_nextSchedule > sched || m_nextSchedule == -1 )
            m_nextSchedule = sched;
      }
   }
   
   
   if( msecs >= 0 ) 
   {
      // how much span time remained?   
      spanTime -= curTime; // curtime is updated by the reader only loop, if it happens..
      if ( spanTime < 0 ) spanTime = 0;

      tv.tv_sec = spanTime/1000;
      tv.tv_usec = (spanTime % 1000) * 1000;
      wait_tv = &tv;
   }
   else
   {
      // prepare for infinite wait
      wait_tv = 0;
   }

   
   // manage a special case. If we don't have to wait and don't have any set,
   // return immediately
   if ( (wait_tv == 0 || ( wait_tv != 0 && wait_tv->tv_sec == 0 && wait_tv->tv_usec == 0 ) )
         && rd_set_point == 0 && wr_set_point == 0 )
   {
      return false;
   }

   int result;
   Net::resetError();
   result = select( maxfd + 1, rd_set_point, wr_set_point, 0, wait_tv );

   // if there aren't news, return immediately
   //TODO: error management;
   if ( result <= 0 ) 
      return false;
      
   //parse first the receivers.
   if ( rd_set_point ) 
   {
      sit = m_sockets.begin();
      
      while( result > 0 && sit != m_sockets.end() )
      {
         Socket *skt = (*sit);
         if ( skt->reception() != 0 && FD_ISSET( skt->rawSocket(), rd_set_point ) )
         {
            bool cont = skt->reception()->receive( skt );
            result--;
            // in case of error on the socket, remove it.
            if ( skt->osError() != 0 || skt->closed() || ! cont )
            {
               sit = m_sockets.erase( sit );
               skt->decref();
               Net::resetError();
               continue;
            }
         }
         sit ++;
      }
   }
   
   //TRICK: round-robinize the receiver list.
   recptRoundRobin();
   
   // now for writing.
   if ( wr_set_point && result )
   {
     // reset also schedule, to allow other writers to set their own.
     
     ConList::iterator clit = clist_used->begin();
      
      while( result > 0 && clit != clist_used->end() )
      {
         Connection *con = (*clit);
         RawSocket fd = con->socket()->rawSocket();
         
         if ( FD_ISSET( fd, wr_set_point ) )
         {
            bool cont = con->transmission()->transmit( con );
            result--;
            // always remove the transmission from scheduled list.
            clit = clist_used->erase( clit );
            // and always reset the socket status: only one write per socket.
            FD_CLR( fd, wr_set_point );
            
            // requested to terminate?
            if ( ! cont )
               // a funny grammar, but it's right.
               con->line()->remove( con ); 
               
            Net::resetError();
            // do not erase in case of error on socket; this is an application side
            // business we are not interested in. If the transmission wants to
            // be removed it just return false, the rest it's its business.
         }
         else
            clit++;
      }
   }
   
   // signal that we had some processing
   return true;
}

} // namespace

/* End of cabal_channel.cpp */

---