Node.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 <iostream>
#include <netinet/in.h>

#include "Node.h"
#include "Link.h"
#include "RoutingTable.h"
#include "VolatileBundleStore.h"
#include "PersistentBundleStore.h"
#include "Exception.h"

using namespace DTN;

Node::Node() :
   m_seqNum( 0U ),
   m_consumer( 0 ),
   m_volatileCap( 4096 ),
   m_volatileStore( 0 ),
   m_persistentCap( 4096 ),
   m_persistentStore( 0 ),
   m_vstorePol( 0 ),
   m_forwPol( 0 ),
   m_custPol( 0 )
{
}

Node::~Node()
{
   if ( 0 != m_consumer )
   {
      delete m_consumer;
   }
   if ( 0 != m_volatileStore )
   {
      delete m_volatileStore;
   }
   if ( 0 != m_persistentStore )
   {
      delete m_persistentStore;
   }
   if ( 0 != m_vstorePol )
   {
      delete m_vstorePol;
   }
   if ( 0 != m_forwPol )
   {
      delete m_forwPol;
   }
   if ( 0 != m_custPol )
   {
      delete m_custPol;
   }
   LinkList::iterator llItr = m_links.begin();
   LinkList::iterator llEnd = m_links.end();
   for ( ; llItr != llEnd ; ++llItr )
   {
      Link* l = *llItr;
      if ( 0 != l ) delete l;
   }
}

void
Node::forward( Bundle* b )
{
   if ( 0 == b ) return;
   b->send() = m_addr;
   if ( 0 == m_forwPol )
   {
      // We can't forward bundles at all, so we drop them.
      std::cerr << "No forwarding policy exists for this node.\n"
                << "All bundles will be dropped." << std::endl;
      drop(b);
      return;
   }

   // Check for bundle expiration.
   if ( b->expired() )
   {
      drop(b,ExpiryDrop::inst);
      return;
   }

   // Check for custody transfer.
   if ( 0 != m_custPol )
   {
      CustodyPolicy::CustodyReturn cr = m_custPol->takeCustody(*b);
      if ( CustodyPolicy::kCustodyTaken == cr )
      {
         if ( 0 != m_consumer )
         {
            m_consumer->custody(*b);
            m_consumer->persistentStore(*b);
         }
         acknowledge(*b);
         b->custodian() = addr();
      }
      else if ( CustodyPolicy::kDropBundle == cr )
      {
         drop(b,CustodyPolicyDrop::inst);
         return;
      }
   }

   // Try to forward.
   if ( ! m_forwPol->forward(b) )
   {
      // Do we have local storage (including a simple FIFO queue)?
      if ( 0 != m_vstorePol )
      {
         // We couldn't forward immediately, so try to store.
         BundlePointer p = m_vstorePol->store(b);
         if ( p.isNull() )
         {
            drop(b,FullQueueDrop::inst);
         }
         else
         {
            // We stored the bundle; let the forwarding policy know so
            // that it can update its cache.
            m_forwPol->cache(p);
         }
      }
      else
      {
         // We can't store, so we have to drop.
         drop(b,FullQueueDrop::inst);
      }
   }
}

void
Node::broadcast( Bundle* b )
{
   if ( 0 == b ) return;

   if ( ! b->expired() )
   {
      LinkList::const_iterator itr = m_links.begin();
      LinkList::const_iterator end = m_links.end();
      for ( ; itr != end ; ++itr )
      {
         if ( 0 != *itr )
         {
            Bundle* pB = b->clone();
            pB->destination() = (*itr)->remoteAddr();
            pB->recv() =  (*itr)->remoteAddr();
            pB->custodian() = m_addr;
            forward( pB );
         }
      }
   }

   delete b;
}

bool
Node::addPersistent( const Bundle& b )
{
   if ( 0 == m_persistentStore ) return false;
   unsigned int size = b.size();
   unsigned int avail = persistentCap() - usedPersistentCap();
   if ( size > avail ) return false;
   BundlePointer ptr = m_persistentStore->addBundle( b.clone() );
   if ( ! ptr.isNull() )
   {
      if ( 0 != m_consumer ) m_consumer->persistentStore( b );
      return true;
   }
   return false;
}

bool
Node::validate( const BundlePointer& p )
{
   if ( p.isNull() ) return false;
   if ( 0 == m_volatileStore ) return false;
   if ( ! m_volatileStore->validatePointer(p) ) return false;
   return true;
}

bool
Node::validatePersistent( const BundlePointer& p )
{
   if ( p.isNull() ) return false;
   if ( 0 == m_persistentStore ) return false;
   if ( ! m_persistentStore->validatePointer(p) ) return false;
   return true;
}

void
Node::remove( Bundle* b, bool dealloc, const DropCause& c )
{
   if ( 0 == m_volatileStore ) return;
   m_volatileStore->deleteBundle( b );
   if ( dealloc ) drop(b,c);
}

void
Node::removePersistent( Bundle* b, bool dropping )
{
   if ( 0 == m_persistentStore ) return;
   if ( 0 != m_custPol )
   {
      if ( ! m_custPol->policyRemove(*b) ) return;
   }
   m_persistentStore->deleteBundle( b );
   recordPersistentRemove(*b);
   if ( dropping )
   {
      drop(b, CustodyPolicyDrop::inst);
   }
   else
   {
      delete b;
   }
}

void
Node::send( Bundle* b, Link& l )
{
   if ( 0 == b ) return;
   if ( ! ( b->type() & Bundle::kBcast ) )
   {
      b->recv() = l.remoteAddr();
   }
   if ( 0 != m_consumer ) m_consumer->send( *b );
   l.send(b);
}

void
Node::recv( Bundle* b )
{
   if ( 0 == b ) return;

   if ( ! ( ours(b->recv()) || ( b->type() & Bundle::kBcast ) ) )
   {
      // This wasn't meant for us.  Drop silently.
      delete b;
      return;
   }

   b->incrHop();
   if ( 0 != m_consumer ) m_consumer->recv(*b);

   if ( ours(b->destination()) || ( b->type() & Bundle::kBcast ) )
   {
      consume(b);
   }
   else
   {
      forward(b);
   }
}

void
Node::forwardOn( Link& l )
{
   if ( 0 != m_forwPol )
   {
      while ( 1 )
      {
         Bundle* b = m_forwPol->forwardOn(l);
         if ( 0 == b ) return;
         if ( b->expired() )
         {
            drop(b, ExpiryDrop::inst);
         }
         else
         {
            forward(b);
            if ( ! l.available() )
            {
               return;
            }
         }
      }
   }
}

void
Node::retryStored()
{
   if ( 0 != m_custPol )
   {
      m_custPol->retry();
   }
}

void
Node::drop( Bundle* b, const DropCause& c )
{
   if ( 0 != m_consumer )
   {
      m_consumer->drop(*b,c);
   }
   delete b;
}

void
Node::recordPersistentRemove( const Bundle& b, bool cleanup )
{
   if ( 0 != m_consumer )
   {
      m_consumer->persistentRemove(b,cleanup);
   }
}

void
Node::setAddr( const ByteString& addr )
{
   m_addr = addr;
}

bool
Node::ours( const ByteString& destination )
{
   return ( destination == m_addr );
}

void
Node::setConsumer( Consumer* c )
{
   if ( 0 == c ) return;
   if ( 0 != m_consumer )
   {
      delete m_consumer;
   }
   m_consumer = c;
}

size_t
Node::usedVolatileCap( bool cleanup ) const
{
   if ( 0 == m_volatileStore ) return 0;
   return m_volatileStore->bytesUsed(cleanup);
}

size_t
Node::volatileBundles( bool cleanup ) const
{
   if ( 0 == m_volatileStore ) return 0;
   return m_volatileStore->bundles(cleanup);
}

size_t
Node::usedPersistentCap( bool cleanup ) const
{
   if ( 0 == m_persistentStore ) return 0;
   return m_persistentStore->bytesUsed(cleanup);
}

size_t
Node::persistentBundles( bool cleanup ) const
{
   if ( 0 == m_persistentStore ) return 0;
   return m_persistentStore->bundles(cleanup);
}

BundlePointer
Node::cachedVolatile()
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( BundlePointer() );
}

const BundlePointer
Node::cachedVolatile() const
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( BundlePointer() );
}

BundlePointer
Node::cachedVolatile( const BundlePointer& p )
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( p );
}

const BundlePointer
Node::cachedVolatile( const BundlePointer& p ) const
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( p );
}

BundlePointer
Node::cachedVolatile( const ByteString& sender,
                      uint32_t seqNum )
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( sender, seqNum );
}

const BundlePointer
Node::cachedVolatile( const ByteString& sender,
                      uint32_t seqNum ) const
{
   if ( 0 == m_volatileStore )
   {
      return BundlePointer();
   }
   return m_volatileStore->getPointer( sender, seqNum );
}

BundlePointer
Node::cachedPersistent()
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( BundlePointer() );
}

const BundlePointer
Node::cachedPersistent() const
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( BundlePointer() );
}

BundlePointer
Node::cachedPersistent( const BundlePointer& p )
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( p );
}

const BundlePointer
Node::cachedPersistent( const BundlePointer& p ) const
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( p );
}

BundlePointer
Node::cachedPersistent( const ByteString& sender,
                        uint32_t seqNum )
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( sender, seqNum );
}

const BundlePointer
Node::cachedPersistent( const ByteString& sender,
                        uint32_t seqNum ) const
{
   if ( 0 == m_persistentStore )
   {
      return BundlePointer();
   }
   return m_persistentStore->getPointer( sender, seqNum );
}

void
Node::setVolatileStore( VolatileBundleStore* s )
{
   if ( 0 == s ) return;
   if ( 0 != m_volatileStore )
   {
      delete m_volatileStore;
   }
   m_volatileStore = s;
   if ( 0 != m_vstorePol ) m_vstorePol->setStore( m_volatileStore );
   if ( 0 != m_forwPol ) m_forwPol->setStore( m_volatileStore );
}

void
Node::setPersistentStore( PersistentBundleStore* s )
{
   if ( 0 == s ) return;
   if ( 0 != m_persistentStore )
   {
      delete m_persistentStore;
   }
   m_persistentStore = s;
   if ( 0 != m_custPol ) m_custPol->setStore( m_persistentStore );
}

void
Node::setVolatileCap( size_t cap )
{
   if ( 0 != m_volatileStore )
   {
      if ( cap < m_volatileCap ) m_volatileStore->shrinkStore(cap);
   }
   m_volatileCap = cap;
}

void
Node::setPersistentCap( size_t cap )
{
   if ( 0 != m_persistentStore )
   {
      if ( cap < m_persistentCap ) m_persistentStore->shrinkStore(cap);
   }
   m_persistentCap = cap;
}

void
Node::setVolatileStorePolicy( VolatileStorePolicy* p )
{
   if ( 0 == p ) return;
   if ( 0 != m_vstorePol ) delete m_vstorePol;
   m_vstorePol = p;
   m_vstorePol->setStore( m_volatileStore );
}

void
Node::setForwardingPolicy( ForwardingPolicy* p )
{
   if ( 0 == p ) return;
   if ( 0 != m_forwPol ) delete m_forwPol;
   m_forwPol = p;
   m_forwPol->setStore( m_volatileStore );
}

void
Node::setCustodyPolicy( CustodyPolicy* p )
{
   if ( 0 == p ) return;
   if ( 0 != m_custPol ) delete m_custPol;
   m_custPol = p;
   m_custPol->setStore( m_persistentStore );
}

void
Node::addLink( Link* l )
{
   if ( 0 != l )
   {
      m_links.insert( l );
   }
}

void
Node::signalExhausted( const Bundle& b )
{
   if ( 0 != m_consumer )
   {
      m_consumer->exhausted(b);
   }
}

void
Node::consume( Bundle* b )
{
   if ( 0 == b ) return;
   Bundle::BundleType t = b->type();

   if ( 0 != m_persistentStore )
   {
      m_persistentStore->consume(b);
   }
   if ( 0 != m_consumer )
   {
      (*m_consumer)(*b);
   }

   if ( Bundle::kACK & t )
   {
      // Clean out the persistent store.
      if ( 0 != m_persistentStore )
      {
         BundlePointer p = m_persistentStore->getPointer( b->source(),
                                                          b->seqNum() );
         if ( ! p.isNull() )
         {
            Bundle* pB = p.repr()->bundle();
            if ( (0 != m_custPol) && m_custPol->policyRemove( *pB ) )
            {
               m_persistentStore->deleteBundle( pB );
               recordPersistentRemove(*pB);
               delete pB;
            }
         }
      }
   }
   else
   {
      // Generate an acknowledgment.
      acknowledge( *b );
   }

   delete b;
}

void
Node::acknowledge( const Bundle& b )
{
   // Make sure this is an acknowledgeable bundle.
   Bundle::BundleType t = b.type();
   if ( Bundle::kNoACK & t )
   {
      return;
   }

   struct timeval tv = dtn_time();

   // Use the same time-to-live as the original Bundle, since we really
   // have no better guess.
   struct timeval e;
   e.tv_sec = tv.tv_sec + b.expiry().tv_sec - b.created().tv_sec;
   e.tv_usec = tv.tv_usec + b.expiry().tv_usec - b.created().tv_usec;
   e.tv_sec += ( e.tv_usec / 1000000 );
   e.tv_usec = ( e.tv_usec % 1000000 );

   Bundle* ack = new Bundle( ntohl(b.seqNum()),
                             b.source(),
                             b.custodian(),
                             tv,
                             e,
                             b.type() | Bundle::kACK );
   ack->custodian() = addr();
   forward( ack );
}

---