WrapNode.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 <iostream>

#include "Globals.h"
#include "WrapNode.h"
#include "WrapLink.h"
#include "parsers.h"
#include "MockBundle.h"
#include "PydtnBundle.h"
#include "ForwardingEntity.h"
#include "CustodyEntity.h"
#include "BundleEvent.h"
#include "ResendEvent.h"
#include "FlatStore.h"
#include "PersistentStore.h"
#include "DumpConsumer.h"
#include "ConsumerAlias.h"
#include "GlobalStatistics.h"
#include "Tracer.h"
#include "TraceConsumer.h"

#include "simlpy/Clock.h"
#include "simlpy/Registry.h"
#include "simlpy/interpreter_hooks.h"
#include "simlpy/SimulationException.h"

#include "dtn/DropTail.h"

struct timeval
DTN::dtn_time()
{
   return Clock::time().tv;
}

WrapNode::WrapNode() :
   m_bundleLifetime( Globals::bundleLifetime() ),
   m_resendPeriod( Globals::resendPeriod() ),
   m_resending( false )
{
   m_node.setVolatileCap( Globals::volatileCap() );
   m_node.setPersistentCap( Globals::persistentCap() );
   m_consumers = new DTN::ConsumerChain(&m_node);
   m_node.setConsumer( m_consumers );
   m_consumers->addConsumer( new DumpConsumer(&m_node) );
   GlobalStatistics* gs = GlobalStatistics::instance();
   if ( 0 != gs )
   {
      gs->addNode(&m_node);
      m_consumers->addConsumer( new ConsumerAlias(&m_node,gs) );
   }
   if ( 0 != GlobalTracer::instance() )
   {
      m_consumers->addConsumer( new TraceConsumer(&m_node) );
   }
   m_node.setVolatileStorePolicy( new DTN::DropTail(&m_node) );
   m_node.setForwardingPolicy( 0 ); // gets set in configure()
   m_node.setVolatileStore( new FlatStore(&m_node) );
   m_node.setPersistentStore( new PersistentStore(this) );
}

WrapNode::~WrapNode()
{
   // Note: We do not own the memory for m_consumers, so we don't delete it.
   CollectorList::iterator itr = m_collectors.begin();
   CollectorList::iterator end = m_collectors.end();
   for ( ; itr != end ; ++itr )
   {
      if ( 0 != *itr )
      {
         delete *itr;
      }
   }
}

Entity*
WrapNode::create() const
{
   return new WrapNode;
}

void
WrapNode::configure( const ArgList& args )
{
   unsigned int nargs = args.size();
   if ( nargs < 2 )
   {
      std::cerr << "config() takes at least 2 arguments ("
                << nargs << " given)"
                << std::endl;
      throw SimulationException( __FILE__ , __LINE__ );
   }
   std::string command = parse_string(args[0]);
   if ( "addr" == command )
   {
      std::string addrString = parse_string( args[1] );
      DTN::ByteString a;
      for ( unsigned int i = 0 ; i < addrString.length() ; ++i )
      {
         a += addrString[i];
      }
      m_node.setAddr( a );
   }
   else if ( "capacity" == command )
   {
      double dcap = parse_double( args[1] );
      // convert from KB to B
      size_t cap = (size_t)(1024 * dcap);
      m_node.setVolatileCap(cap);
   }
   else if ( "forwarding" == command )
   {
      ForwardingEntity* fe = 0;
      try
      {
         ItemWrapper item = resolve_symbol( args[1] );
         if ( 0 != item.type.find( "forwarding:" ) )
         {
            throw SimulationException( __FILE__ , __LINE__ );
         }
         fe = (ForwardingEntity*)(item.item);

         DTN::ForwardingPolicy* fwd = fe->generatePolicy( &m_node );
         if ( 0 == fwd )
         {
            throw SimulationException( __FILE__ , __LINE__ );
         }
         m_node.setForwardingPolicy( fwd );
      }
      catch ( ... )
      {
         std::cerr << "The second argument of config('forwarding',pol) "
                   << "must be an entity\ncapable of generating a "
                   << "forwarding policy"
                   << std::endl;
      }
   }
   else if ( "stable_store" == command )
   {
      double dcap = parse_double( args[1] );
      // convert from KB to B
      size_t cap = (size_t)(1024 * dcap);
      m_node.setPersistentCap(cap);
   }
   else if ( "custody" == command )
   {
      Entity* e = 0;
      std::string ptype = parse_string( args[1] );
      try
      {
         e = Registry::lookup( "custody:" + ptype );
      }
      catch ( ... )
      {
         std::cerr << "The custody policy type " << ptype
                   << " is not recognized"
                   << std::endl;
         throw;
      }
      CustodyEntity* ce = (CustodyEntity*)( e->create() );
      if ( 0 == ce )
      {
         throw SimulationException( __FILE__ , __LINE__ );
      }
      m_node.setCustodyPolicy( ce->generatePolicy( &m_node ) );
      delete ce;
   }
   else if ( "bundle_lifetime" == command )
   {
      m_bundleLifetime = parse_time(args[1]);
   }
   else if ( "resend" == command )
   {
      m_resendPeriod = parse_time(args[1]);
   }
   else
   {
      std::cerr << "Unrecognized configuration option: " << command
                << std::endl;
      throw SimulationException( __FILE__ , __LINE__ );
   }

   // We've been re-configured, so it's time to create trace information.
   GlobalTracer* gt = GlobalTracer::instance();
   if ( 0 != gt )
   {
      gt->node(*this);
   }
}

void
WrapNode::emit( const ArgList& args )
{
   unsigned int nargs = args.size();
   if ( nargs < 3 )
   {
      std::cerr << "emit() takes at least 3 arguments ("
                << nargs << " given)"
                << std::endl;
      throw SimulationException( __FILE__ , __LINE__ );
   }

   Time t = parse_time(args[0]);

   std::string command = parse_string(args[1]);
   if ( "data" == command )
   {
      if ( nargs < 4 )
      {
         std::cerr << "emit(t,\"data\") takes at least 4 arguments ("
                   << nargs << " given)"
                   << std::endl;
         throw SimulationException( __FILE__ , __LINE__ );
      }
      // emit's args: time, type, destination node, data, [lifetime]
      WrapNode* dest = parse_node( args[2] );
      if ( 0 == dest )
      {
         std::cerr << "The third argument of emit() must be a node"
                   << std::endl;
         throw SimulationException( __FILE__ , __LINE__ );
      }
      std::string dataString = parse_string(args[3]);
      DTN::ByteString dat;
      for ( unsigned int i = 0 ; i < dataString.length() ; ++i )
      {
         dat += dataString[i];
      }
      DTN::Bundle::BundleType type =
         DTN::Bundle::kData | DTN::Bundle::kCustodial;
      Time e = t;
      if ( 4 == nargs )
      {
         e += m_bundleLifetime;
      }
      else
      {
         e += parse_time(args[4]);
      }
      DTN::Bundle* b =
         new DTN::Bundle( m_node.nextSeq(),
                          addr(),
                          dest->addr(),
                          dat,
                          t.tv,
                          e.tv,
                          type );
      if ( 0 != b )
      {
         Clock::schedule( new BundleEvent(this,this,t,t,b,true) );
      }
   }
   else if ( "datafake" == command )
   {
      if ( nargs < 4 )
      {
         std::cerr << "emit(t,\"datafake\") takes at least 4 arguments ("
                   << nargs << " given)"
                   << std::endl;
         throw SimulationException( __FILE__ , __LINE__ );
      }
      // emit's args: time, type, destination node, data size, [lifetime]
      WrapNode* dest = parse_node( args[2] );
      if ( 0 == dest )
      {
         std::cerr << "The third argument of emit() must be a node"
                   << std::endl;
         throw SimulationException( __FILE__ , __LINE__ );
      }

      unsigned long int dat = parse_long(args[3]);

      DTN::Bundle::BundleType type =
         DTN::Bundle::kData | DTN::Bundle::kCustodial;
      Time e = t;
      if ( 4 == nargs )
      {
         e += m_bundleLifetime;
      }
      else
      {
         e += parse_time(args[4]);
      }
      DTN::Bundle* b =
         new MockBundle( m_node.nextSeq(),
                         addr(),
                         dest->addr(),
                         dat,
                         t.tv,
                         e.tv,
                         type );
      if ( 0 != b )
      {
         Clock::schedule( new BundleEvent(this,this,t,t,b,true) );
      }
   }
   else if ( "bcast" == command )
   {
      if ( nargs < 3 )
      {
         std::cerr << "emit(t,\"bcast\") takes at least 3 arguments ("
                   << nargs << " given)"
                   << std::endl;
         throw SimulationException( __FILE__ , __LINE__ );
      } 
      // emit's args: time, type, data, [TTL]
      std::string dataString = parse_string(args[2]);
      DTN::ByteString dat;
      for ( unsigned int i = 0 ; i < dataString.length() ; ++i )
      {
         dat += dataString[i];
      }
      DTN::Bundle::BundleType type = DTN::Bundle::kData | DTN::Bundle::kBcast;
      Time e = t;
      if ( 3 == nargs )
      {
         e += m_bundleLifetime;
      }
      else
      {
         e += parse_time(args[3]);
      }
      DTN::Bundle* b =
         new DTN::Bundle( m_node.nextSeq(),
                          addr(),
                          addr(),
                          dat,
                          t.tv,
                          e.tv,
                          type );
      if ( 0 != b )
      {
         Clock::schedule( new BundleEvent(this,this,t,t,b,false) );
      }
   }
   else if ( "bundle" == command )
   {
      if ( ! PydtnBundle_Check(args[2]) )
      {
         PyErr_SetString(PyExc_TypeError,
                         "pydtn.Bundle expected as third argument");
         throw SimulationException( __FILE__ , __LINE__ );
      }
      PydtnBundle* pb = (PydtnBundle*)(args[2]);
      if ( 0 != pb->b )
      {
         Clock::schedule( new BundleEvent(this,this,t,t,pb->b->clone(),true) );
      }
   }
   else
   {
      std::cerr << "Unrecognized event: " << command << std::endl;
      throw SimulationException( __FILE__ , __LINE__ );
   }
}

InterpreterItem
WrapNode::get( const ArgList& args )
{
   unsigned int nargs = args.size();
   if ( 0 == nargs )
   {
      std::cerr << "No attribute specified" << std::endl;
      throw SimulationException( __FILE__ , __LINE__ );
   }

   std::string attr = parse_string(args[0]);

   if ( "addr" == attr )
   {
      return construct_string( addrString() );
   }
   if ( "bundle_lifetime" == attr )
   {
      double t =
         m_bundleLifetime.tv.tv_sec +
         1.0 * m_bundleLifetime.tv.tv_usec / Time::MILLION;
      return construct_double( t );
   }
   if ( "resend_period" == attr )
   {
      double t =
         m_resendPeriod.tv.tv_sec +
         1.0 * m_resendPeriod.tv.tv_usec / Time::MILLION;
      return construct_double( t );
   }
   if ( ( "volatile_capacity" == attr ) ||
        ( "queue_capacity" == attr ) )
   {
      return construct_long( m_node.volatileCap() );
   }
   if ( ( "persistent_capacity" == attr ) ||
        ( "stable_capacity" == attr ) )
   {
      return construct_long( m_node.persistentCap() );
   }
   if ( ( "volatile_used" == attr ) ||
        ( "queue_used" == attr ) )
   {
      ArgList alist;
      alist.push_back( construct_long( m_node.usedVolatileCap() ) );
      alist.push_back( construct_long( m_node.volatileBundles() ) );
      return construct_list( alist );
   }
   if ( ( "persistent_used" == attr ) ||
        ( "stable_used" == attr ) )
   {
      ArgList alist;
      alist.push_back( construct_long( m_node.usedPersistentCap() ) );
      alist.push_back( construct_long( m_node.persistentBundles() ) );
      return construct_list( alist );
   }

   if ( "forwarding" == attr )
   {
      std::cerr << "not yet implemented" << std::endl;
      Entity::ArgList subList( args.begin()+1, args.end() );
   }
   if ( "storage" == attr )
   {
      std::cerr << "not yet implemented" << std::endl;
   }
   if ( "custody" == attr )
   {
      std::cerr << "not yet implemented" << std::endl;
   }

   // Pass the request up.
   return Entity::get(args);
}

void
WrapNode::addLink( DTN::Link* l )
{
   if ( 0 == l ) return;
   m_node.addLink(l);
}

void
WrapNode::addConsumer( DTN::Consumer* c )
{
   if ( 0 == c ) return;
   if ( 0 == m_consumers )
   {
      m_node.setConsumer(c);
      return;
   }
   m_consumers->addConsumer(c);
}

void
WrapNode::addApplication( DTN::Application* a )
{
   if ( 0 == a ) return;
   a->setOwner(&m_node);
   addConsumer(a);
}

void
WrapNode::addCollector( NodeCollector* c )
{
   if ( 0 == c ) return;
   m_collectors.push_back(c);
}

void
WrapNode::setVolatileStorePolicy( DTN::VolatileStorePolicy* p )
{
   if ( 0 == p ) return;
   m_node.setVolatileStorePolicy(p);
}

void
WrapNode::setForwardingPolicy( DTN::ForwardingPolicy* p )
{
   if ( 0 == p ) return;
   m_node.setForwardingPolicy(p);
}

void
WrapNode::setVolatileStore( DTN::VolatileBundleStore* s )
{
   if ( 0 == s ) return;
   m_node.setVolatileStore(s);
}

void
WrapNode::setPersistentStore( DTN::PersistentBundleStore* s )
{
   if ( 0 == s ) return;
   m_node.setPersistentStore(s);
}

bool
WrapNode::handler( BundleEvent& event )
{
   if ( event.storeLocally() )
   {
      if ( ! m_node.addPersistent(event.data()) )
      {
         m_node.drop( event.pData() );
         return false;
      }
   }
   if ( ( event.source() == this ) &&
        ( m_node.addr() == event.data().send() ) )
   {
      if ( event.data().type() & DTN::Bundle::kBcast )
      {
         m_node.broadcast(event.pData());
      }
      else
      {
         m_node.forward(event.pData());
      }
   }
   else
   {
      if ( ( m_node.addr() == event.data().recv() ) ||
           ( event.data().type() & DTN::Bundle::kBcast ) )
      {
         m_node.recv(event.pData());
      }
      else
      {
         // silently ignore this bundle
         return true;
      }
   }
   if ( 0 != m_node.usedPersistentCap() )
   {
      if ( ! m_resending )
      {
         m_resending = true;
         if ( m_resendPeriod > Time(0,0) )
         {
            Clock::schedule(
               new ResendEvent( this,
                                this,
                                Clock::time() + m_resendPeriod ) );
         }
      }
   }
   return true;
}

bool
WrapNode::handler( ResendEvent& event )
{
//    std::cout << "(t=" << Clock::time() << ") resending stored bundles"
//              << std::endl;
   m_node.retryStored();
   if ( 0 == m_node.usedPersistentCap() )
   {
      m_resending = false;
   }
   else
   {
      if ( m_resendPeriod > Time(0,0) )
      {
         Time t = Clock::time() + m_resendPeriod;
         Clock::schedule( new ResendEvent(this,this,t) );
      }
   }
   return true;
}

void
WrapNode::finalize() const
{
   collect();
}

void
WrapNode::collect() const
{
   CollectorList::const_iterator itr = m_collectors.begin();
   CollectorList::const_iterator end = m_collectors.end();
   for ( ; end != itr ; ++itr )
   {
      if ( 0 != (*itr) ) (**itr)(*this);
   }
}

void
WrapNode::setAddr( const DTN::ByteString& a )
{
   m_node.setAddr(a);
}

const DTN::ByteString&
WrapNode::addr() const
{
   return m_node.addr();
}

const std::string&
WrapNode::addrString() const
{
   if ( 0 == m_addrString.length() )
   {
      const DTN::ByteString& b = m_node.addr();
      bool printable = true;
      for ( unsigned int i = 0 ; i < b.length() ; ++i )
      {
         if ( !isprint(b[i]) ) printable = false;
         if ( isspace(b[i]) ) printable = false;
      }
      if ( ! printable )
      {
         m_addrString += "0x";
      }
      for ( unsigned int i = 0 ; i < b.length() ; ++i )
      {
         if ( printable )
         {
            m_addrString += b[i];
         }
         else
         {
            char tmp[3];
            ::sprintf(tmp,"%02X",b[i]);
            m_addrString += tmp[0];
            m_addrString += tmp[1];
         }
      }
   }
   return m_addrString;
}

void
WrapNode::forwardOn( DTN::Link& l )
{
   m_node.forwardOn(l);
}

void
WrapNode::drop( DTN::Bundle* b, const DTN::DropCause& c )
{
   if ( 0 == b ) return;
   m_node.drop(b,c);
}

size_t
WrapNode::volatileBundles() const
{
   return m_node.volatileBundles();
}

DTN::BundlePointer
WrapNode::cachedVolatile()
{
   return m_node.cachedVolatile();
}

const DTN::BundlePointer
WrapNode::cachedVolatile() const
{
   return m_node.cachedVolatile();
}

size_t
WrapNode::persistentBundles() const
{
   return m_node.persistentBundles();
}

DTN::BundlePointer
WrapNode::cachedPersistent()
{
   return m_node.cachedPersistent();
}

const DTN::BundlePointer
WrapNode::cachedPersistent() const
{
   return m_node.cachedPersistent();
}

---