EpidemicApp.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 "EpidemicApp.h"

#include "simlpy/interpreter_hooks.h"
#include "simlpy/SimulationException.h"
#include "dtn/Node.h"
#include "pydtn/parsers.h"
#include <iostream>

using namespace BasicEpidemic;

// Bookkeeping: initialize static data
DTN::ByteString EpidemicApp::kIdentifier((unsigned char*)"basicepidemic");
unsigned char EpidemicApp::kSummaryVector = 0;
unsigned char EpidemicApp::kDataRequest = 1;

// Construct the protocol instance.  See python_init.cc for actual
// protocol instantiation.
EpidemicApp::EpidemicApp( Mobility::MobileNode* n,
                          DTN::Link* l,
                          const struct timeval& t ) :
   MobileForwarding::Protocol(n,l,t)
{
}

// Clean up
EpidemicApp::~EpidemicApp()
{
   LinkMap::iterator itr = m_linkMap.begin();
   LinkMap::iterator end = m_linkMap.end();
   for ( ; itr != end ; ++itr )
   {
      if ( 0 != itr->second )
      {
         delete itr->second;
      }
   }
}


//======================================================================
// Define the MobileForwarding::Protocol methods.
//
// We're going to do this in the order that they're listed in
// Protocol.h, so that it's easier to follow.
//======================================================================

// Return the unique identifier, which we've stored as static data.
// This could have been inlined in the header (and thus been more
// efficient), but for clarity we put it here.
const DTN::ByteString&
EpidemicApp::appID() const
{
   return kIdentifier;
}

// We've seen a new neighbor.  This will require exchanging state.
void
EpidemicApp::newNeighbor( const DTN::ByteString& addr )
{
   if ( 0 == m_owner ) return;

   // Make sure there's a mock link for this address.
   if ( m_linkMap.end() == m_linkMap.find(addr) )
   {
      m_linkMap[addr] = new Mobility::MockLink(link(),addr);
   }

   // Construct the summary vector.
   DTN::ByteString data;
   data.push_back( kSummaryVector );
   DTN::BundlePointer itr = m_owner->cachedPersistent();
   for ( ; ! itr.isNull() ; itr = itr.next() )
   {
      data.append( digest(itr) );
   }
   if ( 1 == data.length() )
   {
      // We're not storing anything, so there's no point in sending the
      // summary vector.
      return;
   }

   if ( verbosity() > 1 )
   {
      std::cout << "sending summary vector from "
                << stringify(m_owner->addr())
                << " to "
                << stringify(addr)
                << " ["
                << stringify(data)
                << "]"
                << std::endl;
   }

   // Create a bundle to carry the message.
   struct timeval now = DTN::dtn_time();

   struct timeval exp = DTN::dtn_time();
   exp.tv_sec += lifetime().tv_sec;
   exp.tv_usec += lifetime().tv_usec;
   while ( exp.tv_usec > 1000000 )
   {
      exp.tv_usec -= 1000000;
      exp.tv_sec += 1;
   }

   DTN::Bundle* b = new DTN::Bundle( m_owner->nextSeq(),
                                     m_owner->addr(),
                                     addr,
                                     data,
                                     now,
                                     exp,
                                     DTN::Bundle::kData,
                                     appID() );
   b->recv() = addr; // We have to set the receiver now.

   // Send the message via the node.
   m_owner->forward(b);
}

// We received a protocol message from another node.
//
// There are two types of protocol messages that we might receive:
// 1) summary vectors
// 2) data requests
void
EpidemicApp::process( const DTN::Bundle& b )
{
   if ( 0 == m_owner ) return;

   // All we have is a message type, which means there's no meaningful
   // content.
   if ( b.payload().length() < 2 ) return;

   // If we've received a summary vector, look through it for
   // bundles we haven't seen and send a data request.
   if ( kSummaryVector == b.payload()[0] )
   {
      if ( verbosity() > 1 )
      {
         std::cout << "processing a summary vector at "
                   << stringify( m_owner->addr())
                   << " from "
                   << stringify( b.source() )
                   << std::endl;
      }

      // Break the summary vector into distinct message digests.
      DigestList* dl = parseSummaryVector( b.payload().substr(1) );
      if ( 0 == dl ) return;

      // Compose the request.  If we don't have the message with a
      // specified digest, add the digest to our request list.
      DTN::ByteString data;
      data.push_back( kDataRequest );
      DigestList::iterator itr = dl->begin();
      DigestList::iterator end = dl->end();
      bool haveReq = false;
      for ( ; itr != end ; ++itr )
      {
         // Look in the data store for this message.
         DTN::BundlePointer bp = storedBundle(*itr);
         if ( bp.isNull() )
         {
            // We don't have it, so add it to the request.
            haveReq = true;
            data.append( *itr );
         }
      }
      delete dl;

      // If there was nothing of interest in the summary vector, we
      // have nothing more to do.
      if ( ! haveReq ) return;

      if ( verbosity() > 1 )
      {
         std::cout << "sending data request from "
                   << stringify(m_owner->addr())
                   << " to "
                   << stringify(b.source())
                   << " ["
                   << stringify(data)
                   << "]"
                   << std::endl;
      }

      // Construct the bundle with the protocol's data request.
      struct timeval now = DTN::dtn_time();

      struct timeval exp = DTN::dtn_time();
      exp.tv_sec += lifetime().tv_sec;
      exp.tv_usec += lifetime().tv_usec;
      while ( exp.tv_usec > 1000000 )
      {
         exp.tv_usec -= 1000000;
         exp.tv_sec += 1;
      }

      DTN::Bundle* pB = new DTN::Bundle( m_owner->nextSeq(),
                                         m_owner->addr(),
                                         b.source(),
                                         data,
                                         now,
                                         exp,
                                         DTN::Bundle::kData,
                                         appID() );
      // We have to set the receiver now.
      pB->recv() = b.source();

      // Send the message via the node.
      m_owner->forward(pB);
      return;
   }

   // If we've received a data request, enqueue the specified bundles.
   if ( kDataRequest == b.payload()[0] )
   {
      // Break the summary vector into distinct message digests.
      DigestList* dl = parseSummaryVector( b.payload().substr(1) );
      if ( 0 == dl ) return;

      // Go through the specified bundles, scheduling them to be sent
      // to the node that sent us this request.
      DigestList::iterator itr = dl->begin();
      DigestList::iterator end = dl->end();
      for ( ; itr != end ; ++itr )
      {
         // Look in the data store for the message.
         DTN::BundlePointer bp = storedBundle(*itr);
         if ( ! bp.isNull() )
         {
            // Copy the bundle.
            DTN::Bundle* pB = bp.repr()->bundle()->clone();
            // Set the next-hop receiver to the requesting node.
            pB->recv() = b.source();
            // Enqueue the bundle.
            m_owner->forward(pB);
         }
      }
      delete dl;

      return;
   }

   // The message type was invalid.  We'll ignore this.
}

// Retrieve the aliased link for a particular destination.
DTN::Link*
EpidemicApp::getLink( const DTN::ByteString& addr )
{
   LinkMap::iterator itr = m_linkMap.find(addr);
   if ( m_linkMap.end() == itr ) return 0;
   return itr->second;
}

// Make sure the no-ACK flag is set.
void
EpidemicApp::tweak( DTN::Bundle* b )
{
   if ( 0 == b )
   {
      throw SimulationException( __FILE__ , __LINE__ );
   }
   b->type() |= DTN::Bundle::kNoACK;
}


//==================
// Utility methods
//==================

// Compute the digest for a message.
DTN::ByteString
EpidemicApp::digest( const DTN::BundlePointer& p )
{
   DTN::ByteString retval;
   if ( p.isNull() ) return retval;
   const DTN::Bundle* b = p.repr()->bundle();
   unsigned char n = 4 + b->source().length();
   retval.push_back(n);
   union
   {
         uint32_t i;
         unsigned char c[4];
   } seq;
   seq.i = b->seqNum();
   retval.append(seq.c,4);
   retval.append(b->source());
   return retval;
}

// Split a summary vector into individual message digests.
EpidemicApp::DigestList*
EpidemicApp::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;
}

// Find the stored bundle corresponding to a digest.
DTN::BundlePointer
EpidemicApp::storedBundle( const DTN::ByteString& d )
{
   union
   {
         uint32_t i;
         unsigned char c[4];
   } seqNum;
   seqNum.c[0] = d[1];
   seqNum.c[1] = d[2];
   seqNum.c[2] = d[3];
   seqNum.c[3] = d[4];
   
   DTN::ByteString sender = d.substr(5);

   return m_owner->cachedPersistent( sender, seqNum.i );
}

---