balser.cpp

/*
   CABAL - CAched BAlanced LAN
   Test for balance and priorities
   
   Tests also pseudo connections on a single UDP socket

   (single thread version)

   $Id: balser.cpp,v 1.3 2004/03/29 14:47:55 jonnymind Exp $
---------------------------------------------
   Begin      :
   Author     : Giancarlo Niccolai

   Last modified because:

*/

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


#include <map>
#include "balser.h"

using namespace std;
using namespace Cabal;


static ClientHash clients;
static unsigned long s_clientCount = 0;


bool DflTransmit::transmit( Connection *con )
{
   UDPSocket *skt = reinterpret_cast<UDPSocket *>(con->socket());
   PACKET pkt;
   unsigned long to_send;
   
   // pending resends?
   if( !m_resend.empty() ) 
   {
       to_send = * m_resend.begin();
       m_resend.pop_front();  
   }
   else if ( m_packetId < m_size ) {
      to_send = m_packetId;
      m_packetId ++;
   }
   else {
      // we are done: we've been called a last time to check for resends,
      // and we have none to do.
      cout << "Client " << m_userId <<"(" << m_dest.ip() <<":"<< m_dest.port() <<"): " <<
             "Transmission complete." << endl;
      
      // send a last goodbye transmission
      pkt.userId = m_userId;
      pkt.id = END_CONNECT; 
      pkt.timestamp = Net::timeOfDay(); 
      // for simplicity, we send all the packets of the same size.
      skt->sendTo( &pkt, sizeof( pkt ), m_dest );
   
      // remove client from active list.
      clients.erase( m_userId );
      return false;
   }
      
   pkt.userId = m_userId;
   pkt.id = to_send; 
   // in real app, we may have a global 
   // "tick" thread that provides all who needs with pre-cached
   // timestamp.
   pkt.timestamp = Net::timeOfDay(); 
   
   // just a data to be different from client to client.
   memset( pkt.data, static_cast<unsigned char>( to_send ), sizeof( pkt.data ) );
   
   // the real send operation; it (more or less) translates into a very fast inline.
   int iSent = skt->sendTo( &pkt, sizeof( pkt ), m_dest );
   
   if ( iSent > 0 ) 
   {
      con->account( iSent );
      
      // sent all packets?
      if ( m_packetId == m_size ) 
         // reschedule in timeout seconds to allow client to ask for last resends.
         con->schedule( 1500, m_prio );
      else
         // require fastest reschedulation possible; line and channel bandwidth will
         // limit us.
         con->schedule( 0, m_prio );
         
      return true;
   }
   else {
       // we had an error
      cout << "Client " << m_userId <<"(" << m_dest.ip() <<":"<< m_dest.port() <<"): " <<
           "lost on sendto error" << skt->osError() << ": "  << Net::errorDescription( skt->osError() ) << endl;
      return false;     
   }
}



         
bool DflReception::receive( Socket *skt )
{
   //we know we are fed with an UDP socket; 
   //recv() will put the remote address in its remote() field
   char buffer[1500];
   PACKET *pkt;
   
   // read all the data in the datagram indipendently of the final casting
   int len = skt->recv( buffer, 1500 );
   
   if ( len == 0 ) {
      cout << "Reception terminating: Socket closed." << endl;
      return false;
   }
   else if ( len < 0 ) {
      cout << "Reception terminating: Socket had error " << skt->osError() << " (" << 
         Net::errorDescription( skt->osError() ) << ")" << endl;
      return false;
   }
   // not conforming with datagram we want to have ?
   else if ( len != CLIENT_REP_SIZE ) { 
      cout << "Reception warning: Spurious data on socket: "<< len << "bytes" <<endl;
      // discard, but don't give up.
      return true;
   }
   
   // let's see who's the sender.
   pkt = reinterpret_cast< PACKET *>( buffer );
   if ( pkt->userId == START_CONNECT ) // connection request!
   {
      s_clientCount++;
      if ( s_clientCount == 0xf0000000 ) 
         s_clientCount = 1;
         
      // prepare to contact-back our newcomer.
      DflTransmit *trs = new DflTransmit( s_clientCount, 60, skt->remote(), medium_prio );
      Connection *con = new Connection( skt, trs, this ); // last field optional, only for reference.
      Line *line = new Line( LINE_BANDWITH );
      line->set( "filetransfer", con );
      con->schedule( 10 ); // prepare to start fast.
      // save our client.
      clients[ s_clientCount ] = new Client( s_clientCount, line );
   }
   else {
      // a real application may use also address/port to ckeck for
      // client identity.
      ClientHash::iterator incoming = clients.find( pkt->userId );
      if ( incoming == clients.end() ) {
         cout << "Reception warning: Invalid client ID: "<< pkt->userId << endl;
         // discard, but don't give up.
         return true;
      }
      
      Client *remote = incoming->second;
      //Now we have our client; a real app would account ping stats using the echoed timestamp.
      CLI_REQ *request = reinterpret_cast< CLI_REQ *>( pkt->data ); 
      
      switch( request->type ) 
      {
         case REQ_TYPE_RESEND:
         {
            // filetransfer connection must be open.
            Connection *remcon = remote->m_comLine->get( "filetransfer" ); 
            // we know what kind of transmission are dealing with what kind of transfers.
            DflTransmit *dfltrans = reinterpret_cast< DflTransmit *>( remcon->transmission() );
            dfltrans->resend( request->value );
            // anticipates resend;
            remcon->schedule( 0, high_prio );
            // do nothing; we are just happy, and we signal the client is alive.
         }
         // no break;
         
         case REQ_TYPE_ACK:
            // do nothing; we are just happy, and we signal the client is alive.
            remote->m_lastHail = Net::timeOfDay();
            remote->m_packetId = pkt->id;
            remote->m_rtt = remote->m_lastHail < pkt->timestamp ? 
                  CABAL_DAY_LENGTH - pkt->timestamp + remote->m_lastHail: remote->m_lastHail - pkt->timestamp;
      }
   }
   
   // continuing operations.
   return true;
}



/*******************************************************************
* Main procedure: create a server that accepts incoming connections
* and produces a cacher for each new socket.
* Cachers are destroyed as remote connections are closed.
********************************************************************/

int main ( int argc, char *argv[] )
{
   Net::init();
   int n_port;
   Channel *channel;

   cout << "Welcome from CABAL Balanced Server test." << endl;
   cout << "Channel Bandwith set to "<< CHANNEL_BANDWIDTH <<" bytes/sec." << endl;
   cout << "Client Bandwith set to "<< LINE_BANDWITH << " bytes/sec." << endl;
   cout << "Waiting loop set at 50ms." << endl;
   
   if ( argc > 1 ) {
      n_port = atoi( argv[1] );
   }
   else {
      n_port = DEFAULT_PORT;
   }

   // creates a channel for our server socket;
   channel = new Channel( CHANNEL_BANDWIDTH );
   
   try 
   {
      UDPSocket *usock = new UDPSocket( "0.0.0.0", n_port );
      cout << "Server waiting on port "<< n_port << endl;
      DflReception *rec = new DflReception();
      
      // assign the socket to a channel, with a given reception.
      usock->channel( channel, rec );
      // drop our reference for the socket.
      usock->decref();
      
      while( true )  // for now.
      {
         // sleep maximum 50ms
         bool res = channel->process( 50 );
         while( res ) res = channel->process( 0 );
         
         // now prune idle clients.
         ClientHash::iterator it = clients.begin();
         MSECS now = Net::timeOfDay();
         cout << "\r";
         while( it != clients.end() )
         {
            Client *cli = it->second;
            if( cli->m_lastHail < (now - 2000) && cli->m_lastHail > 2000 ) 
            {
               cout << cli->m_userId <<":K,";
               ClientHash::iterator old = it;
               it++;
               clients.erase( old );
               delete cli; // also deletes lines, connections and transmissions.
            }
            else {
               cout << cli->m_userId <<":" << cli->m_packetId <<  ":"<< cli->m_rtt << ",";
               it++;
            }
         }
         
         cout << "          " << flush;
      }
      
   }
   catch (Error e ) {
      cout << e <<"\n";
   }
   Net::exit();

   return 0;
}

---