EpidemicStore.cc

// Copyright 2008 Michael Marsh, University of Maryland.
//
// This file is part of pydtn.
//
// pydtn is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// pydtn is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with pydtn.  If not, see <http://www.gnu.org/licenses/>.
//
// The views and conclusions contained in the software and documentation
// are those of the authors and should not be interpreted as representing
// official policies, either expressed or implied, of the University
// of Maryland.
//
// pydtn extends and embeds the Python interpreter, which is
// Copyright 2001-2006 Python Software Foundation, All Rights Reserved,
// and is released under the PSF License Agreement.
//
// RANLUX random number generation uses the Boost library,
// Copyright 1994-2006 by various authors (details in individual files),
// which is released under the Boost Software License, Version 1.0.

#include "simlpy/interpreter_defs.h"
#include "simlpy/interpreter_hooks.h"

#include "EpidemicStore.h"
#include "dtn/Bundle.h"
#include "simlpy/Clock.h"

#include <iostream>

using namespace Epidemic;

//===================
// EpidemicBundle
//===================
EpidemicBundle::EpidemicBundle( DTN::Bundle* b ) :
   m_bundle( b ),
   m_timeAdded( Clock::time() ),
   m_numTries( 0 )
{
   if ( 0 != b )
   {
      m_originator = b->source();
      m_seqNum = b->seqNum();
      m_digest = EpidemicStore::digest(*b);
   }
}

EpidemicBundle::EpidemicBundle( const DTN::ByteString& originator,
                                uint32_t seqNum ) :
   m_bundle( 0 ),
   m_originator( originator ),
   m_seqNum( seqNum ),
   m_numTries( 0 )
{
   m_digest = EpidemicStore::digest(originator,seqNum);
}

EpidemicBundle::EpidemicBundle( const DTN::ByteString& digest ) :
   m_bundle( 0 ),
   m_numTries( 0 ),
   m_digest( digest )
{
}

EpidemicBundle::~EpidemicBundle()
{
}

DTN::Bundle*
EpidemicBundle::bundle() const
{
   return m_bundle;
}

bool
EpidemicBundle::operator<( const EpidemicBundle& b ) const
{
   return m_digest < b.m_digest ;
}


//==================
// EpidemicStore
//==================

EpidemicStore::EpidemicStore( DTN::Node* owner, uint32_t h ) :
   DTN::PersistentBundleStore(owner),
   m_max_hops(h),
   m_bytesUsed(0)
{
}

EpidemicStore::~EpidemicStore()
{
}

DTN::BundlePointer
EpidemicStore::p_addBundle( DTN::Bundle* b )
{
   if ( 0 == b ) return DTN::BundlePointer();

   // Reject bundles we've already seen.
   if ( m_history.end() != m_history.find( digest(*b) ) )
      return DTN::BundlePointer();

   // Insert the new bundle into the store.
   std::pair< iterator , bool > r = m_list.insert( b );
   if ( r.second )
   {
      m_bytesUsed += b->size();

      // We'll never remove digests from this list.
      m_history.insert( r.first->digest() );

      if ( ( 0 == m_max_hops ) || ( b->hopCount() < m_max_hops ) )
      {
         m_digestMap.insert( DigestMap::value_type( r.first->digest(),
                                                    r.first ) );
      }
      else
      {
         m_destMap[b->destination()].push_back( r.first );
      }
      return DTN::BundlePointer( new EpidemicStoreItr(*this,r.first) );
   }
   return DTN::BundlePointer();
}

void
EpidemicStore::p_deleteBundle( DTN::Bundle* b )
{
   if ( 0 == b ) return;

   // If it's in the transferable bundle structure, remove it.
   m_digestMap.erase( digest(*b) );

   // Checking for the bundle in the last-hop delivery structure is
   // more complicated.  We begin by seeing if there are any bundles
   // for the particular destination.
   DestMap::iterator ditr = m_destMap.find( b->destination() );
   if ( m_destMap.end() != ditr )
   {
      // Having found an entry for this destination, we have to loop
      // over bundles looking for a match.  The set of iterators is
      // the second element of the pair iterator from the map.
      DestList::iterator itr = ditr->second.begin();
      DestList::iterator end = ditr->second.end();
      for ( ; itr != end ; ++itr )
      {
         // The addresses of the bundles should match, so we can do an
         // easy comparison.
         if ( (*itr)->bundle() == b )
         {
            // We've found the bundle's iterator, so first we erase
            // it.
            ditr->second.erase(itr);

            // Now we check to see if that was the last bundle for the
            // destination.  If so, we'll remove the map entry
            // completely.  This prevents cruft from accumulating.
            if ( 0 == ditr->second.size() )
            {
               m_destMap.erase(ditr);
            }

            // The bundle will only appear once, so we can exit the
            // loop.  What's more, we've invalidated the iterator
            // we're using in the loop, and might have invalidated the
            // map iterator as well.  All said, we've done all we need
            // to do here.
            break;
         }
      }
   }

   // Finally, get rid of the bundle from the persistent store and
   // decrement the usage.
   if ( 0 != m_list.erase( b ) )
   {
      m_bytesUsed -= b->size();
   }
}

void
EpidemicStore::consume( DTN::Bundle* b )
{
   if ( 0 == b ) return;

   // Don't store ACKs.
   if ( DTN::Bundle::kACK & b->type() ) return;

   // Don't store non-custodial bundles.
   if ( ! (DTN::Bundle::kCustodial & b->type()) ) return;

   m_history.insert( digest(*b) );
}

DTN::BundlePointer
EpidemicStore::p_getPointer( const DTN::BundlePointer& p )
{
   if ( p.isNull() )
   {
      if ( m_list.empty() )
      {
         return DTN::BundlePointer();
      }
      return DTN::BundlePointer(
         new EpidemicStoreItr(*this,m_list.begin()) );
   }
   iterator itr = m_list.lower_bound( p.repr()->bundle() );
   if ( m_list.end() != itr )
   {
      return DTN::BundlePointer( new EpidemicStoreItr(*this,itr) );
   }
   return DTN::BundlePointer();
}

DTN::BundlePointer
EpidemicStore::p_getPointer( const DTN::ByteString& sender, uint32_t seqNum )
{
   iterator itr = m_list.find( EpidemicBundle(sender,seqNum) );
   if ( m_list.end() == itr )
   {
      return DTN::BundlePointer();
   }
   return DTN::BundlePointer( new EpidemicStoreItr(*this,itr) );
}

DTN::BundlePointer
EpidemicStore::getPointer( const DTN::ByteString& digest )
{
   clean();
   iterator itr = m_list.find( EpidemicBundle(digest) );
   if ( m_list.end() == itr )
   {
      return DTN::BundlePointer();
   }
   return DTN::BundlePointer( new EpidemicStoreItr(*this,itr) );
}

bool
EpidemicStore::p_validatePointer( const DTN::BundlePointer& p )
{
   if ( p.isNull() ) return false;
   iterator itr = m_list.find( p.repr()->bundle() );
   if ( m_list.end() == itr ) return false;
   return true;
}

void
EpidemicStore::p_shrinkStore( size_t s )
{
   // This loop is not efficient.  If we wanted to be efficient, we'd
   // store a separate list of bundles sorted by time added.  However,
   // we don't expect this to be called often, so a little
   // inefficiency here does no real harm.  This is, after all, fairly
   // major node reconfiguration.
   while ( m_bytesUsed > s )
   {
      if ( 0 == m_list.size() ) return;
      BundleList::reverse_iterator itr = m_list.rbegin();
      BundleList::reverse_iterator end = m_list.rend();
      BundleList::reverse_iterator last = itr;
      for ( ; itr != end ; ++itr )
      {
         if ( itr->timeAdded() > last->timeAdded() )
         {
            last = itr;
         }
      }
      m_owner->recordPersistentRemove( *(last->bundle()) );
      p_deleteBundle(last->bundle());
      delete last->bundle();
   }
}

EpidemicStore::iterator
EpidemicStore::begin()
{
   return m_list.begin();
}

EpidemicStore::iterator
EpidemicStore::lower_bound( const key_type& x )
{
   return m_list.lower_bound( x );
}

EpidemicStore::iterator
EpidemicStore::upper_bound( const key_type& x )
{
   return m_list.upper_bound( x );
}

EpidemicStore::iterator
EpidemicStore::end()
{
   return m_list.end();
}

EpidemicStore::const_iterator
EpidemicStore::begin() const
{
   return m_list.begin();
}

EpidemicStore::const_iterator
EpidemicStore::lower_bound( const key_type& x ) const
{
   return m_list.lower_bound( x );
}

EpidemicStore::const_iterator
EpidemicStore::upper_bound( const key_type& x ) const
{
   return m_list.upper_bound( x );
}

EpidemicStore::const_iterator
EpidemicStore::end() const
{
   return m_list.end();
}

DTN::ByteString*
EpidemicStore::summaryVector() const
{
   DTN::ByteString* retval = new DTN::ByteString;
   if ( 0 == retval ) return 0;
   const_iterator itr = m_list.begin();
   const_iterator end = m_list.end();
   for ( ; itr != end ; ++itr )
   {
      retval->append( itr->digest() );
   }
   return retval;
}

DTN::ByteString
EpidemicStore::digest( const DTN::Bundle& b )
{
   return digest( b.source(), b.seqNum() );
}

DTN::ByteString
EpidemicStore::digest( const DTN::ByteString& originator, uint32_t seqNum )
{
   DTN::ByteString retval;
   unsigned char n = 4 + originator.length();
   retval.push_back(n);
   union
   {
         uint32_t i;
         unsigned char c[4];
   } seq;
   seq.i = seqNum;
   retval.append(seq.c,4);
   retval.append(originator);
   return retval;
}

EpidemicStore::DigestList*
EpidemicStore::parseSummaryVector( const DTN::ByteString& sv )
{
   DigestList* retval = 0;
   if ( 0 == sv.length() ) return retval;
   retval = new DigestList;
   for ( unsigned int i = 0 ; i < sv.length() ; )
   {
      unsigned char c = sv[i];
      retval->push_back( DTN::ByteString(sv,i,c+1) );
      i += c+1;
   }
   return retval;
}

void
EpidemicStore::forwardLastHop( const DTN::ByteString& addr, DTN::Node& node )
{
   DestMap::iterator ditr = m_destMap.find( addr );
   if ( m_destMap.end() == ditr ) return;

   DestList::iterator itr = ditr->second.begin();
   DestList::iterator end = ditr->second.end();
   for ( ; itr != end ; ++itr )
   {
      // Copy the bundle.  The iterator stores an iterator, hence the
      // double-dereference.
      DTN::Bundle* pB = (*itr)->bundle()->clone();
      // Set the next-hop receiver to the requesting node.
      pB->recv() = addr;
      // Enqueue the bundle.
      node.forward(pB);
   }
}

bool
EpidemicStore::seen( const DTN::ByteString& digest )
{
   return ( m_history.end() != m_history.find( digest ) );
}

//=====================
// EpidemicStoreItr
//=====================

EpidemicStoreItr::EpidemicStoreItr(
   EpidemicStore& store,
   EpidemicStore::iterator itr ) :
   DTN::BundlePointerRepr(),
   m_store( store ),
   m_itr( itr )
{
}

EpidemicStoreItr::~EpidemicStoreItr()
{
}

bool
EpidemicStoreItr::operator==( const DTN::BundlePointerRepr& b )
{
   return false;
}

DTN::BundlePointerRepr*
EpidemicStoreItr::next()
{
   EpidemicStore::iterator itr = m_store.upper_bound( *m_itr );
   if ( itr != m_store.end() )
   {
      return new EpidemicStoreItr( m_store, itr );
   }
   return 0;
}

DTN::BundlePointerRepr*
EpidemicStoreItr::clone()
{
   return new EpidemicStoreItr( m_store, m_itr );
}

DTN::Bundle*
EpidemicStoreItr::bundle() const
{
   return m_itr->bundle();
}

---