---

InitActivity.cc

/*
 * 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: InitActivity.cc,v 1.3 2004/05/19 15:56:24 mmarsh Exp $
//
// $Log: InitActivity.cc,v $
// Revision 1.3  2004/05/19 15:56:24  mmarsh
// Added copyright and license statements.
//
// Revision 1.2  2003/11/04 22:04:05  mmarsh
// General code cleanup.
//
//

#include "CODEX_Server/ServerResponseEvent.h"
#include "InitActivity.h"
#include "InitEvent.h"
#include "Message.h"
#include "EstablishCallback.h"
#include "CODEX_Server/UnicastRequestEvent.h"
#include "StateInfo.h"
#include "CODEX_Server/ServerState.h"

#include "CODEX_Server/ServerExceptions.h"

#include "timing.h"

using namespace CODEX_APSS;

InitActivity::InitActivity(
   CODEX_Events::DeadPileType& deadPile,
   CODEX_Events::QType& eventQueue,
   unsigned int version,
   unsigned int coordinator,
   unsigned int self,
   unsigned int num,
   const unsigned char* key,
   const SubshareList& subshares,
   CODEX_Server::ServerResponseHandler* responseHandler,
   CODEX_Server::UnicastRequestHandler* requestHandler ) :
   CODEX_Events::Activity( deadPile, eventQueue ),
   m_version( version ),
   m_coordinator( coordinator ),
   m_self( self ),
   m_num( num ),
   m_subshares( subshares ),
   m_responseHandler( responseHandler ),
   m_requestHandler( requestHandler ),
   m_failures( 0 )
{
   memcpy( m_key, key, CODEX_Server::ServerState::nMID );
   m_numShares = 0;
   SubshareList::const_iterator itr = subshares.begin();
   SubshareList::const_iterator end = subshares.end();
   for ( ; itr != end ; ++itr )
   {
      if ( itr->second.initialized() ) ++m_numShares;
   }
}

InitActivity::~InitActivity()
{
   EvidenceMap::iterator itr = m_evidence.begin();
   EvidenceMap::iterator end = m_evidence.end();
   for ( ; itr != end ; ++itr )
   {
      if ( 0 != itr->second )
      {
         delete itr->second;
      }
   }
}

bool
InitActivity::handler( CODEX_Events::CloseEvent& event )
{
   sendEvent( 0, event.source() );
   enableTerminate();
   return true;
}

bool
InitActivity::handler( InitEvent& event )
{
#ifdef TIMING
   ActiveTimer.start();
#endif
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   StateInfo* stateInfo = StateInfo::instance();

   // Loop over splittings of shares
   SubshareList::const_iterator listItr = m_subshares.begin();
   SubshareList::const_iterator listEnd = m_subshares.end();
   for ( ; listItr != listEnd ; ++listItr )
   {
      CODEX_Quorum::MessageVector mv;
      mv.resize( ShareType::NumShares );

      if ( ! listItr->second.initialized() )
      {
         continue;
      }

      const unsigned char* seqNum = serverState->newSequenceNumber();

      // Loop over recipients
      for ( unsigned int server = 0 ;
            server < CODEX_Server::ServerState::nServers ;
            ++server )
      {
         // Distribute appropriate subshares for this server.
         ShareType shares(listItr->first,server);

         // Compose the message.
         EstablishMsg em( m_version, listItr->second, m_self, server, shares );

         // Sign the message.
         BIGNUM * digest = 0;
         CODEX_Ciphers::RSASignature* signature = 0;
         try
         {
            const CODEX_Ciphers::HashFunction& hashFunc =
               serverState->hashFunc();
            digest = em.digest( hashFunc );
            const CODEX_Ciphers::RSAPrivateKey& key =
               serverState->privateKey();
            signature = key.sign( digest );
            BN_free(digest);
            digest = 0;
            if ( 0 == signature )
            {
               throw CODEX_Exceptions::BignumNullException( __FILE__ ,
                                                            __LINE__ );
            }
         }
         catch ( ... )
         {
            if ( 0 != digest ) BN_free( digest );
            if ( 0 != signature ) delete signature;
#ifdef TIMING
            ActiveTimer.stop();
#endif
            throw;
         }
         SignedEstablishMsg signedEM( em, *signature );
         delete signature;

         // Compose the request.
         unsigned char selfChar =
            m_self | CODEX_Server::ServerState::OutgoingMask;
         mv[server].fill( selfChar );
         mv[server].fill( seqNum, CODEX_Server::ServerState::nMID );
         mv[server].fill( stateInfo->domain() );
         mv[server].fill( kEstablishMsg | SignatureMask );
         int length = signedEM.marshal(0);
         unsigned char* buffer = new unsigned char[length];
         unsigned char* pBuffer = buffer;
         signedEM.marshal(&pBuffer);
         mv[server].fill( buffer, length );
         delete [] buffer;
      }

      // Create the callback for this request.
      EstablishCallback* cb =
         new EstablishCallback( this, m_version, listItr->second );

      // Create and send the unicast event.
      sendEvent( new CODEX_Server::UnicastRequestEvent( this,
                                                        m_requestHandler,
                                                        mv,
                                                        seqNum,
                                                        cb ),
                 0 );
   }

   sendEvent( 0, event.source() );
#ifdef TIMING
   InitActIdleTimer[m_num].start();
   ActiveTimer.stop();
#endif
   return true;
}

void
InitActivity::addFailure()
{
#ifdef TIMING
   InitActIdleTimer[m_num].stop();
#endif

   ++m_failures;

#ifdef TIMING
   InitActIdleTimer[m_num].start();
#endif
}

void
InitActivity::addEvidence( const SublabelType& label,
                           EvidenceVector* evidence )
{
#ifdef TIMING
   InitActIdleTimer[m_num].stop();
#endif
   if ( 0 == evidence )
   {
#ifdef TIMING
      InitActIdleTimer[m_num].start();
#endif
      return;
   }

   // Accumulate evidence
   EvidenceMap::iterator itr = m_evidence.find( label );
   if ( m_evidence.end() != itr )
   {
      // The label is already present.
      delete evidence;
#ifdef TIMING
      InitActIdleTimer[m_num].start();
#endif
      return;
   }
   m_evidence[ label ] = evidence;

   // Check for finished condition
   StateInfo* stateInfo = StateInfo::instance();
   CODEX_Server::ServerState* serverState =
      CODEX_Server::ServerState::instance();
   CODEX_Quorum::QuorumSystem* qs = serverState->quorumSystem();
   if ( 0 == qs )
   {
      throw CODEX_Server::NoQuorumSystemException( __FILE__ , __LINE__ );
   }
//   if ( m_evidence.size() > qs->faultsTolerated() )
   if ( m_evidence.size() + m_failures == m_numShares )
   {
      EstablishedArray ea;
      EvidenceMap::const_iterator mItr = m_evidence.begin();
      EvidenceMap::const_iterator mEnd = m_evidence.end();
      for ( ; mItr != mEnd ; ++mItr )
      {
         EvidenceVector::const_iterator vItr = mItr->second->begin();
         EvidenceVector::const_iterator vEnd = mItr->second->end();
         for ( ; vItr != vEnd ; ++vItr )
         {
            ea.append( new SignedEstablishedMsg( *vItr ) );
         }
      }

      ContributeMsg response( m_version, m_coordinator, m_self, ea );

      BIGNUM * digest = 0;
      CODEX_Ciphers::RSASignature* signature = 0;
      try
      {
         const CODEX_Ciphers::HashFunction& hashFunc = serverState->hashFunc();
         digest = response.digest( hashFunc );
         const CODEX_Ciphers::RSAPrivateKey& key = serverState->privateKey();
         signature = key.sign( digest );
         BN_free(digest);
         digest = 0;
         if ( 0 == signature )
         {
            throw CODEX_Exceptions::BignumNullException( __FILE__ , __LINE__ );
         }
      }
      catch ( ... )
      {
         if ( 0 != digest ) BN_free( digest );
         if ( 0 != signature ) delete signature;
#ifdef TIMING
         InitActIdleTimer[m_num].start();
#endif
         throw;
      }

      SignedContributeMsg signedResponse( response, *signature );
      delete signature;

      stateInfo->addInitResponse( signedResponse );

      CODEX_Quorum::Message m;

      // Fill message headers.
      m.fill( m_coordinator );
      m.fill( m_key, CODEX_Server::ServerState::nMID );
      m.fill( stateInfo->domain() );
      m.fill( kContributeMsg | SignatureMask );
      int length = signedResponse.marshal(0);
      unsigned char* buffer = new unsigned char[length];
      unsigned char* pBuffer = buffer;
      signedResponse.marshal(&pBuffer);
      m.fill( buffer, length );
      delete [] buffer;

      CODEX_Server::ServerResponseEvent* outEvent =
         new CODEX_Server::ServerResponseEvent( this,
                                                m_responseHandler,
                                                m,
                                                m_coordinator );
      sendEvent( outEvent, 0 );
      sendEvent( new CODEX_Events::CloseEvent( this, this ), 0 );

#ifdef TIMING
      InitActTimer[m_num].stop();
#endif

   }
#ifdef TIMING
   else // Only start again if we're waiting for more responses.
   {
      InitActIdleTimer[m_num].start();
   }
#endif
}

---