---

StaticByzantineQuorumSystem.h

/*
 * Copyright 2003 Michael A. Marsh, Cornell University. All rights reserved.
 * This software is released under the modified BSD license.
 * See the file LICENSE in the top-level directory for details.
 */
//
// $Id: StaticByzantineQuorumSystem.h,v 1.3 2004/05/19 15:56:56 mmarsh Exp $
//
// $Log: StaticByzantineQuorumSystem.h,v $
// Revision 1.3  2004/05/19 15:56:56  mmarsh
// *** empty log message ***
//
// Revision 1.2  2003/11/04 22:31:50  mmarsh
// *** empty log message ***
//
//

#ifndef __CODEX_QUORUM_STATICBYZANTINEQUORUMSYSTEM_H__
#define __CODEX_QUORUM_STATICBYZANTINEQUORUMSYSTEM_H__

#include <vector>
#include <errno.h>
#include <sys/time.h>

#include "QuorumSystem.h"
#include "RemoteServer.h"
#include "Socket.h"
#include "CODEX_Exceptions/Check.h"

namespace CODEX_Quorum
{
   
   template< int N, int T >
   class StaticByzantineQuorumSystem : public ByzantineQuorumSystem
   {
      public :
         StaticByzantineQuorumSystem( ResponseTracker* aRT );
         virtual ~StaticByzantineQuorumSystem();

         unsigned int quorumSystemSize();
         unsigned int quorumSize();
         unsigned int faultsTolerated();

         bool broadcastMessage( const Message& msg,
                                ReturnVector& rets,
                                ResponseCallback* cb );
         bool unicastMessage( const MessageVector& msgs,
                              ReturnVector& rets,
                              ResponseCallback* cb = 0 );
         void poll( MessageDispatcher& requestDispatcher,
                    QuorumDispatcher& responseDispatcher );

         void setServer( unsigned int i , RemoteServer* server );

         const RemoteServer* server(unsigned int i) const;

      protected :

      private :
         RemoteServer*          m_servers[N];
         pair<size_t,Message*>  m_buffers[N];
   };

   class QSESBQSReturnsBad : public QSExceptionBase
   {
      public :
         QSESBQSReturnsBad( const string& fname,
                            int line,
                            int got,
                            int needed) :
            QSExceptionBase( fname, line ),
            m_got( got ),
            m_needed( needed )
         {}

         virtual ~QSESBQSReturnsBad() {}

      protected :
         void derivedMsg() const;

      private :
         int m_got;
         int m_needed;
   };

   template<int N, int T>
   StaticByzantineQuorumSystem<N,T>::StaticByzantineQuorumSystem(
      ResponseTracker* aRT) :
      ByzantineQuorumSystem( aRT )
   {
      // Link-time check of template parameters
      CODEX_Exceptions::Check< ( N > 3*T ) >();
      for ( int i = 0 ; i < N ; ++i )
      {
         m_servers[i] = 0;
         m_buffers[i].first = 0;
         m_buffers[i].second = 0;
      }
   }

   template<int N, int T>
   StaticByzantineQuorumSystem<N,T>::~StaticByzantineQuorumSystem()
   {
      for ( int i = 0 ; i < N ; ++i )
      {
         if ( 0 != m_buffers[i].second )
         {
            delete m_buffers[i].second;
         }
         if ( 0 != m_servers[i] )
         {
            delete m_servers[i];
         }
      }
   }

   template<int N, int T>
   unsigned int
   StaticByzantineQuorumSystem<N,T>::quorumSystemSize()
   {
      return N;
   }

   template<int N, int T>
   unsigned int
   StaticByzantineQuorumSystem<N,T>::quorumSize()
   {
      int num = N + T + 1;
      return ( num/2 + num%2 );
   }

   template<int N, int T>
   unsigned int
   StaticByzantineQuorumSystem<N,T>::faultsTolerated()
   {
      return T;
   }

   template<int N, int T>
   bool
   StaticByzantineQuorumSystem<N,T>::broadcastMessage(
      const Message& msg, ReturnVector& rets, ResponseCallback* cb )
   {
      if ( rets.size() < N )
      {
         throw QSESBQSReturnsBad( __FILE__ , __LINE__ , rets.size() , N );
      }

      unsigned char* key = extractKey( msg.buffer() );
      bool ownKey = true;

      if ( 0 != cb )
      {
         // Register this message with the ResponseTracker
         ResponseInfo* pRI = new ResponseInfo( cb );
//         unsigned char* key = extractKey( msg.buffer() );
         m_responseTracker->insert( key, pRI );
         ownKey = false;
      }

      struct timeval tv;
      gettimeofday( &tv, 0 );

      unsigned int nerror = 0;
      ResponseInfo* pRI = (*m_responseTracker)( key );
      for ( int i = 0 ; i < N ; ++i )
      {
         if ( rets[i].returnCode() == RemoteServerReturn::kFailure )
         {
            if ( 0 != pRI )
            {
               // This relies on the internal map, because ResponseInfo doesn't
               // in principle know how many servers to keep track of.
               typedef ResponseInfo::ServerMap ServerMap;
               const ServerMap& sMap = pRI->map();
               ServerMap::const_iterator smItr = sMap.find( i );
               if ( smItr != sMap.end() )
               {
                  // This server has already been contacted successfully.
                  rets[i].setReturnCode( RemoteServerReturn::kSuccess );
                  continue;
               }
            }
            rets[i].updateTime( tv );
         }
         else
         {
            // This server has already been contacted successfully.
            continue;
         }
         if ( 0 != m_servers[i] )
         {
            try
            {
               m_servers[i]->sendTo(msg, rets[i]);
               if ( rets[i].errorCode() != RemoteServerReturn::kNone )
               {
                  ++nerror;
               }
               if ( 0 != pRI )
               {
                  // We only count success when a response is received.
                  rets[i].setReturnCode(RemoteServerReturn::kFailure);
               }
            }
            catch ( ... )
            {
               ++nerror;
               delete m_servers[i];
               m_servers[i] = 0;
            }
         }
         else
         {
            // We don't have a connection to this server, but do we
            // actually need one at this point?
            if ( ( rets[i].returnCode() != RemoteServerReturn::kSuccess ) &&
                 ( rets[i].errorCode() != RemoteServerReturn::kNone ) )
            {
               ++nerror;
            }
         }
      }
      if ( ownKey ) delete [] key;
      // Old semantics
      //return ( nerror <= faultsTolerated() );
      return ( 0 == nerror );
   }

   template<int N, int T>
   bool
   StaticByzantineQuorumSystem<N,T>::unicastMessage(
      const MessageVector& msgs, ReturnVector& rets, ResponseCallback* cb )
   {
      if ( rets.size() < N )
      {
         throw QSESBQSReturnsBad( __FILE__ , __LINE__ , rets.size() , N );
      }

      // Make sure there's a message to send.
      unsigned char* key = 0;
      MessageVector::const_iterator item = msgs.begin();
      MessageVector::const_iterator end = msgs.end();
      for ( ; (item != end) && (0 == key) ; ++item )
      {
         if ( 0 < item->length() )
         {
            key = extractKey( item->buffer() );
         }
      }
      if ( 0 == key )
      {
         return false;
      }
      bool ownKey = true;

      if ( 0 != cb )
      {
         // Register this message with the ResponseTracker
         ResponseInfo* pRI = new ResponseInfo( cb );
         m_responseTracker->insert( key, pRI );
         ownKey = false;
      }

      struct timeval tv;
      gettimeofday( &tv, 0 );

      unsigned int nerror = 0;
      ResponseInfo* pRI = (*m_responseTracker)( key );
      for ( int i = 0 ; i < N ; ++i )
      {
         if ( rets[i].returnCode() == RemoteServerReturn::kFailure )
         {
            if ( 0 != pRI )
            {
               // This relies on the internal map, because ResponseInfo doesn't
               // in principle know how many servers to keep track of.
               typedef ResponseInfo::ServerMap ServerMap;
               const ServerMap& sMap = pRI->map();
               ServerMap::const_iterator smItr = sMap.find( i );
               if ( smItr != sMap.end() )
               {
                  // This server has already been contacted successfully.
                  rets[i].setReturnCode( RemoteServerReturn::kSuccess );
                  continue;
               }
            }
            rets[i].updateTime( tv );
         }
         else
         {
            // This server has already been contacted successfully.
            continue;
         }
         if ( (0 != m_servers[i]) && (0 < msgs[i].length()) )
         {
            try
            {
               m_servers[i]->sendTo( msgs[i], rets[i] );
               if ( rets[i].errorCode() != RemoteServerReturn::kNone )
               {
                  ++nerror;
               }
               if ( 0 != pRI )
               {
                  // We only count success when a response is received.
                  rets[i].setReturnCode(RemoteServerReturn::kFailure);
               }
            }
            catch ( ... )
            {
               ++nerror;
               delete m_servers[i];
               m_servers[i] = 0;
            }
         }
         // If we have a message but the server's not there, it's an error.
         else if ( ( 0 < msgs[i].length() ) && ( 0 == m_servers[i] ) )
         {
            // We don't have a connection to this server, but do we
            // actually need one at this point?
            if ( ( rets[i].returnCode() != RemoteServerReturn::kSuccess ) &&
                 ( rets[i].errorCode() != RemoteServerReturn::kNone ) )
            {
               ++nerror;
            }
         }
            
      }
      // Old semantics
      //return ( nerror <= faultsTolerated() );
      if ( ownKey ) delete [] key;
      return ( 0 == nerror );
   }

   template<int N, int T>
   void
   StaticByzantineQuorumSystem<N,T>::poll(
      MessageDispatcher& requestDispatcher,
      QuorumDispatcher& responseDispatcher )
   {
      // bitmaps for select(2)
      fd_set read_set;
      fd_set write_set;
      fd_set error_set;
      FD_ZERO(&read_set);
      FD_ZERO(&write_set);
      FD_ZERO(&error_set);

      // loop over sockets to remote servers setting read, write, and
      // error bitmaps for select(2)
      int width = 0;
      for ( int i = 0 ; i < N ; ++i )
      {
         if ( 0 != m_servers[i] )
         {
            try
            {
               int skt = m_servers[i]->set_fd(&read_set, SocketBase::kRead);
               m_servers[i]->set_fd(&write_set, SocketBase::kWrite);
               m_servers[i]->set_fd(&error_set, SocketBase::kError);
               width = ( skt > width )? skt : width;
            }
            catch ( ... )
            {
               delete m_servers[i];
               m_servers[i] = 0;
               m_buffers[i].first = 0;
               if ( 0 != m_buffers[i].second )
               {
                  delete m_buffers[i].second;
                  m_buffers[i].second = 0;
               }
            }
         }
      }

      // we want select(2) to be non-blocking, so we set the timeout to 0
      struct timeval tv;
      tv.tv_sec = 0;
      tv.tv_usec = 0;
      if ( 0 > select(width+1,&read_set,&write_set,&error_set,&tv) )
      {
         // wrong exception to throw!
         throw QSESocketBaseReadFailed( __FILE__ , __LINE__ , 0 , errno );
      }

      // now check for data to be read
      for ( int i = 0 ; i < N ; ++i )
      {
         if ( 0 != m_servers[i] )
         {
            try
            {
               // is there data to read?
               if ( m_servers[i]->isset_fd(&read_set, SocketBase::kRead) )
               {
                  // we may have read a partial message previously
                  if ( 0 == m_buffers[i].second )
                  {
                     m_buffers[i].first = 0;
                     m_buffers[i].second = new Message;
                  }
                  // grab everything we can without blocking, up to the
                  // full length of a message -- this will append to an
                  // existing buffer
                  RemoteServerReturn rsReturn;
                  m_buffers[i].first =
                     m_servers[i]->receiveFrom( *(m_buffers[i].second),
                                                rsReturn,
                                                m_buffers[i].first );
                  // have we read the entire message?
                  if ( ( 0 == m_buffers[i].first ) &&
                       ( m_buffers[i].second->length() > 0 ) )
                  {
                     Message* msg = m_buffers[i].second;
                     // we no longer need the buffer
                     m_buffers[i].second = 0;
                     if ( m_responseTracker->check( msg->buffer(),
                                                    msg->length(),
                                                    i ) )
                     {
                        unsigned char* msgID = extractKey( msg->buffer() );
                        // We're done with msg -- all its data has been copied.
                        delete msg;
                        ResponseInfo* respInfo = (*m_responseTracker)(msgID);
                        // If we've collected enough responses, then dispatch
                        // the response information to the functional object
                        // passed to this method.
                        if ( ( respInfo->failures()  >  faultsTolerated() ) ||
                             ( respInfo->successes() >= quorumSize()      ) )
                        {
                           m_responseTracker->remove(msgID);
                           responseDispatcher( msgID, respInfo );
                           return;
                        }
                        // Now clean up the message ID, since no one else will.
                        delete [] msgID;
                     }
                     else
                     {
                        requestDispatcher( msg );
                     }
                  }
               }
               // used to manage internal state of the socket
               m_servers[i]->isset_fd(&write_set, SocketBase::kWrite);
               if ( m_servers[i]->isset_fd(&error_set, SocketBase::kError) )
               {
               }
            }
            catch ( ... )
            {
               delete m_servers[i];
               m_servers[i] = 0;
               m_buffers[i].first = 0;
               if ( 0 != m_buffers[i].second )
               {
                  delete m_buffers[i].second;
                  m_buffers[i].second = 0;
               }
            }
         }
      }
   }

   template<int N, int T>
   void
   StaticByzantineQuorumSystem<N,T>::setServer( unsigned int i ,
                                                RemoteServer* server )
   {
      if ( i < N )
      {
         m_servers[i] = server;
      }
   }

   template<int N, int T>
   const RemoteServer*
   StaticByzantineQuorumSystem<N,T>::server(unsigned int i) const
   {
      if ( i < N )
      {
         return m_servers[i];
      }
      return 0;
   }


   template< int T >
   class SimpleStaticByzantineQuorumSystem :
         public StaticByzantineQuorumSystem< 3*T + 1 , T >
   {
      public :
         SimpleStaticByzantineQuorumSystem( ResponseTracker* aRT ) :
            StaticByzantineQuorumSystem<3*T+1,T>( aRT ) {}
         virtual ~SimpleStaticByzantineQuorumSystem() {}
   };

}

#endif /* __CODEX_QUORUM_STATICBYZANTINEQUORUMSYSTEM_H__ */

---