1 /*

     2  *    Copyright 2004-2006 Intel Corporation

     3  * 

     4  *    Licensed under the Apache License, Version 2.0 (the "License");

     5  *    you may not use this file except in compliance with the License.

     6  *    You may obtain a copy of the License at

     7  * 

     8  *        http://www.apache.org/licenses/LICENSE-2.0

     9  * 

    10  *    Unless required by applicable law or agreed to in writing, software

    11  *    distributed under the License is distributed on an "AS IS" BASIS,

    12  *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

    13  *    See the License for the specific language governing permissions and

    14  *    limitations under the License.

    15  */

    16 

    17 #ifdef HAVE_CONFIG_H

    18 #  include <dtn-config.h>

    19 #endif

    20 

    21 #include <oasys/io/IO.h>

    22 #include <oasys/tclcmd/TclCommand.h>

    23 #include <oasys/util/Time.h>

    24 

    25 #include "Bundle.h"

    26 #include "BundleActions.h"

    27 #include "BundleEvent.h"

    28 #include "BundleDaemon.h"

    29 #include "BundleStatusReport.h"

    30 #include "BundleTimestamp.h"

    31 #include "CustodySignal.h"

    32 #include "ExpirationTimer.h"

    33 #include "FragmentManager.h"

    34 #include "contacts/Link.h"

    35 #include "contacts/Contact.h"

    36 #include "contacts/ContactManager.h"

    37 #include "conv_layers/ConvergenceLayer.h"

    38 #include "reg/AdminRegistration.h"

    39 #include "reg/APIRegistration.h"

    40 #include "reg/PingRegistration.h"

    41 #include "reg/Registration.h"

    42 #include "reg/RegistrationTable.h"

    43 #include "routing/BundleRouter.h"

    44 #include "routing/RouteTable.h"

    45 #include "session/Session.h"

    46 #include "storage/BundleStore.h"

    47 #include "storage/RegistrationStore.h"

    48 #include "bundling/S10Logger.h"

    49 

    50 #ifdef BSP_ENABLED

    51 #  include "security/Ciphersuite.h"

    52 #  include "security/SPD.h"

    53 #  include "security/KeyDB.h"

    54 #endif

    55 

    56 namespace dtn {

    57 

    58 template <>

    59 BundleDaemon* oasys::Singleton<BundleDaemon, false>::instance_ = NULL;

    60 

    61 BundleDaemon::Params::Params()

    62     :  early_deletion_(true),

    63        suppress_duplicates_(true),

    64        accept_custody_(true),

    65        reactive_frag_enabled_(true),

    66        retry_reliable_unacked_(true),

    67        test_permuted_delivery_(false),

    68        injected_bundles_in_memory_(false) {}

    69 

    70 BundleDaemon::Params BundleDaemon::params_;

    71 

    72 bool BundleDaemon::shutting_down_ = false;

    73 

    74 //----------------------------------------------------------------------

    75 BundleDaemon::BundleDaemon()

    76     : BundleEventHandler("BundleDaemon", "/dtn/bundle/daemon"),

    77       Thread("BundleDaemon", CREATE_JOINABLE)

    78 {

    79     // default local eid

    80     local_eid_.assign(EndpointID::NULL_EID());

    81 

    82     actions_ = NULL;

    83     eventq_ = NULL;

    84     

    85     memset(&stats_, 0, sizeof(stats_));

    86 

    87     all_bundles_     = new BundleList("all_bundles");

    88     pending_bundles_ = new BundleList("pending_bundles");

    89     custody_bundles_ = new BundleList("custody_bundles");

    90 

    91     contactmgr_ = new ContactManager();

    92     fragmentmgr_ = new FragmentManager();

    93     reg_table_ = new RegistrationTable();

    94 

    95     router_ = 0;

    96 

    97     app_shutdown_proc_ = NULL;

    98     app_shutdown_data_ = NULL;

    99 

   100     rtr_shutdown_proc_ = 0;

   101     rtr_shutdown_data_ = 0;

   102 }

   103 

   104 //----------------------------------------------------------------------

   105 BundleDaemon::~BundleDaemon()

   106 {

   107     delete pending_bundles_;

   108     delete custody_bundles_;

   109     

   110     delete contactmgr_;

   111     delete fragmentmgr_;

   112     delete reg_table_;

   113     delete router_;

   114 

   115     delete actions_;

   116     delete eventq_;

   117 }

   118 

   119 //----------------------------------------------------------------------

   120 void

   121 BundleDaemon::do_init()

   122 {

   123     actions_ = new BundleActions();

   124     eventq_ = new oasys::MsgQueue<BundleEvent*>(logpath_);

   125     eventq_->notify_when_empty();

   126     BundleProtocol::init_default_processors();

   127 #ifdef BSP_ENABLED

   128     Ciphersuite::init_default_ciphersuites();

   129     SPD::init();

   130     KeyDB::init();

   131 #endif

   132 }

   133 

   134 //----------------------------------------------------------------------

   135 void

   136 BundleDaemon::post(BundleEvent* event)

   137 {

   138     instance_->post_event(event);

   139 }

   140 

   141 //----------------------------------------------------------------------

   142 void

   143 BundleDaemon::post_at_head(BundleEvent* event)

   144 {

   145     instance_->post_event(event, false);

   146 }

   147 

   148 //----------------------------------------------------------------------

   149 bool

   150 BundleDaemon::post_and_wait(BundleEvent* event,

   151                             oasys::Notifier* notifier,

   152                             int timeout, bool at_back)

   153 {

   154     /*

   155      * Make sure that we're either already started up or are about to

   156      * start. Otherwise the wait call below would block indefinitely.

   157      */

   158     ASSERT(! oasys::Thread::start_barrier_enabled());

   159     

   160     ASSERT(event->processed_notifier_ == NULL);

   161     event->processed_notifier_ = notifier;

   162     if (at_back) {

   163         post(event);

   164     } else {

   165         post_at_head(event);

   166     }

   167     return notifier->wait(NULL, timeout);

   168 }

   169 

   170 //----------------------------------------------------------------------

   171 void

   172 BundleDaemon::post_event(BundleEvent* event, bool at_back)

   173 {

   174     log_debug("posting event (%p) with type %s (at %s)",

   175               event, event->type_str(), at_back ? "back" : "head");

   176     event->posted_time_.get_time();

   177     eventq_->push(event, at_back);

   178 }

   179 

   180 //----------------------------------------------------------------------

   181 void

   182 BundleDaemon::get_routing_state(oasys::StringBuffer* buf)

   183 {

   184     router_->get_routing_state(buf);

   185     contactmgr_->dump(buf);

   186 }

   187 

   188 //----------------------------------------------------------------------

   189 void

   190 BundleDaemon::get_bundle_stats(oasys::StringBuffer* buf)

   191 {

   192     buf->appendf("%zu pending -- "

   193                  "%zu custody -- "

   194                  "%u received -- "

   195                  "%u delivered -- "

   196                  "%u generated -- "

   197                  "%u transmitted -- "

   198                  "%u expired -- "

   199                  "%u duplicate -- "

   200                  "%u deleted -- "

   201                  "%u injected",

   202                  pending_bundles()->size(),

   203                  custody_bundles()->size(),

   204                  stats_.received_bundles_,

   205                  stats_.delivered_bundles_,

   206                  stats_.generated_bundles_,

   207                  stats_.transmitted_bundles_,

   208                  stats_.expired_bundles_,

   209                  stats_.duplicate_bundles_,

   210                  stats_.deleted_bundles_,

   211                  stats_.injected_bundles_);

   212 }

   213 

   214 //----------------------------------------------------------------------

   215 void

   216 BundleDaemon::get_daemon_stats(oasys::StringBuffer* buf)

   217 {

   218     buf->appendf("%zu pending_events -- "

   219                  "%u processed_events -- "

   220                  "%zu pending_timers",

   221                  event_queue_size(),

   222                  stats_.events_processed_,

   223                  oasys::TimerSystem::instance()->num_pending_timers());

   224 }

   225 

   226 

   227 //----------------------------------------------------------------------

   228 void

   229 BundleDaemon::reset_stats()

   230 {

   231     memset(&stats_, 0, sizeof(stats_));

   232 

   233     oasys::ScopeLock l(contactmgr_->lock(), "BundleDaemon::reset_stats");

   234     

   235     const LinkSet* links = contactmgr_->links();

   236     LinkSet::const_iterator iter;

   237     for (iter = links->begin(); iter != links->end(); ++iter) {

   238         (*iter)->reset_stats();

   239     }

   240 }

   241 

   242 //----------------------------------------------------------------------

   243 void

   244 BundleDaemon::generate_status_report(Bundle* orig_bundle,

   245                                      BundleStatusReport::flag_t flag,

   246                                      status_report_reason_t reason)

   247 {

   248     log_debug("generating return receipt status report, "

   249               "flag = 0x%x, reason = 0x%x", flag, reason);

   250     

   251     Bundle* report = new Bundle();

   252     BundleStatusReport::create_status_report(report, orig_bundle,

   253                                              local_eid_, flag, reason);

   254     

   255     BundleReceivedEvent e(report, EVENTSRC_ADMIN);

   256     handle_event(&e);

   257 	s10_bundle(S10_TXADMIN,report,NULL,0,0,orig_bundle,"status report");

   258 }

   259 

   260 //----------------------------------------------------------------------

   261 void

   262 BundleDaemon::generate_custody_signal(Bundle* bundle, bool succeeded,

   263                                       custody_signal_reason_t reason)

   264 {

   265     if (bundle->local_custody()) {

   266         log_err("send_custody_signal(*%p): already have local custody",

   267                 bundle);

   268         return;

   269     }

   270 

   271     if (bundle->custodian().equals(EndpointID::NULL_EID())) {

   272         log_err("send_custody_signal(*%p): current custodian is NULL_EID",

   273                 bundle);

   274         return;

   275     }

   276     

   277     Bundle* signal = new Bundle();

   278     CustodySignal::create_custody_signal(signal, bundle, local_eid_,

   279                                          succeeded, reason);

   280     

   281     BundleReceivedEvent e(signal, EVENTSRC_ADMIN);

   282     handle_event(&e);

   283 	s10_bundle(S10_TXADMIN,signal,NULL,0,0,bundle,"custody signal");

   284 

   285 }

   286 

   287 //----------------------------------------------------------------------

   288 void

   289 BundleDaemon::cancel_custody_timers(Bundle* bundle)

   290 {

   291     oasys::ScopeLock l(bundle->lock(), "BundleDaemon::cancel_custody_timers");

   292     

   293     CustodyTimerVec::iterator iter;

   294     for (iter =  bundle->custody_timers()->begin();

   295          iter != bundle->custody_timers()->end();

   296          ++iter)

   297     {

   298         bool ok = (*iter)->cancel();

   299         if (!ok) {

   300             log_crit("unexpected error cancelling custody timer for bundle *%p",

   301                      bundle);

   302         }

   303         

   304         // the timer will be deleted when it bubbles to the top of the

   305         // timer queue

   306     }

   307     

   308     bundle->custody_timers()->clear();

   309 }

   310 

   311 //----------------------------------------------------------------------

   312 void

   313 BundleDaemon::accept_custody(Bundle* bundle)

   314 {

   315     log_info("accept_custody *%p", bundle);

   316     

   317     if (bundle->local_custody()) {

   318         log_err("accept_custody(*%p): already have local custody",

   319                 bundle);

   320         return;

   321     }

   322 

   323     if (bundle->custodian().equals(local_eid_)) {

   324         log_err("send_custody_signal(*%p): "

   325                 "current custodian is already local_eid",

   326                 bundle);

   327         return;

   328     }

   329     

   330     // send a custody acceptance signal to the current custodian (if

   331     // it is someone, and not the null eid)

   332     if (! bundle->custodian().equals(EndpointID::NULL_EID())) {

   333         generate_custody_signal(bundle, true, BundleProtocol::CUSTODY_NO_ADDTL_INFO);

   334     }

   335 	// next line is  for S10

   336 	EndpointID prev_custodian=bundle->custodian();

   337 

   338     // now we mark the bundle to indicate that we have custody and add

   339     // it to the custody bundles list

   340     bundle->mutable_custodian()->assign(local_eid_);

   341     bundle->set_local_custody(true);

   342     actions_->store_update(bundle);

   343     

   344     custody_bundles_->push_back(bundle);

   345 

   346     // finally, if the bundle requested custody acknowledgements,

   347     // deliver them now

   348     if (bundle->custody_rcpt()) {

   349         generate_status_report(bundle, 

   350                                BundleStatusReport::STATUS_CUSTODY_ACCEPTED);

   351     }

   352 	s10_bundle(S10_TAKECUST,bundle,prev_custodian.c_str(),0,0,NULL,NULL);

   353 }

   354 

   355 //----------------------------------------------------------------------

   356 void

   357 BundleDaemon::release_custody(Bundle* bundle)

   358 {

   359     log_info("release_custody *%p", bundle);

   360     

   361     if (!bundle->local_custody()) {

   362         log_err("release_custody(*%p): don't have local custody",

   363                 bundle);

   364         return;

   365     }

   366 

   367     cancel_custody_timers(bundle);

   368 

   369     bundle->mutable_custodian()->assign(EndpointID::NULL_EID());

   370     bundle->set_local_custody(false);

   371     actions_->store_update(bundle);

   372 

   373     custody_bundles_->erase(bundle);

   374 }

   375 

   376 //----------------------------------------------------------------------

   377 void

   378 BundleDaemon::deliver_to_registration(Bundle* bundle,

   379                                       Registration* registration)

   380 {

   381     ASSERT(!bundle->is_fragment());

   382 

   383     ForwardingInfo::state_t state = bundle->fwdlog()->get_latest_entry(registration);

   384     if (state != ForwardingInfo::NONE)

   385     {

   386         ASSERT(state == ForwardingInfo::DELIVERED);

   387         log_debug("not delivering bundle *%p to registration %d (%s) "

   388                   "since already delivered",

   389                   bundle, registration->regid(),

   390                   registration->endpoint().c_str());

   391         return;

   392     }

   393 

   394     

   395     // if this is a session registration and doesn't have either the

   396     // SUBSCRIBE or CUSTODY bits (i.e. it's publish-only), don't

   397     // deliver the bundle

   398     if (registration->session_flags() == Session::PUBLISH)

   399     {

   400         log_debug("not delivering bundle *%p to registration %d (%s) "

   401                   "since it's a publish-only session registration",

   402                   bundle, registration->regid(),

   403                   registration->endpoint().c_str());

   404         return;

   405     }

   406 

   407     log_debug("delivering bundle *%p to registration %d (%s)",

   408               bundle, registration->regid(),

   409               registration->endpoint().c_str());

   410 

   411     if (registration->deliver_if_not_duplicate(bundle)) {

   412         // XXX/demmer this action could be taken from a registration

   413         // flag, i.e. does it want to take a copy or the actual

   414         // delivery of the bundle

   415         bundle->fwdlog()->add_entry(registration,

   416                                     ForwardingInfo::FORWARD_ACTION,

   417                                     ForwardingInfo::DELIVERED);

   418     } else {

   419         log_notice("suppressing duplicate delivery of bundle *%p "

   420                    "to registration %d (%s)",

   421                    bundle, registration->regid(),

   422                    registration->endpoint().c_str());

   423     }

   424 }

   425 

   426 //----------------------------------------------------------------------

   427 bool

   428 BundleDaemon::check_local_delivery(Bundle* bundle, bool deliver)

   429 {

   430     log_debug("checking for matching registrations for bundle *%p", bundle);

   431 

   432     RegistrationList matches;

   433     RegistrationList::iterator iter;

   434 

   435     reg_table_->get_matching(bundle->dest(), &matches);

   436 

   437     if (deliver) {

   438         ASSERT(!bundle->is_fragment());

   439         for (iter = matches.begin(); iter != matches.end(); ++iter) {

   440             Registration* registration = *iter;

   441             deliver_to_registration(bundle, registration);

   442         }

   443     }

   444 

   445     return (matches.size() > 0) || bundle->dest().subsume(local_eid_);

   446 }

   447 

   448 //----------------------------------------------------------------------

   449 void

   450 BundleDaemon::check_and_deliver_to_registrations(Bundle* bundle, const EndpointID& reg_eid)

   451 {

   452     int num;

   453     log_debug("checking for matching entries in table for %s", reg_eid.c_str());

   454 

   455     RegistrationList matches;

   456     RegistrationList::iterator iter;

   457 

   458     num = reg_table_->get_matching(reg_eid, &matches);

   459 

   460     for (iter = matches.begin(); iter != matches.end(); ++iter)

   461     {

   462         Registration* registration = *iter;

   463         deliver_to_registration(bundle, registration);

   464     }

   465 }

   466 

   467 //----------------------------------------------------------------------

   468 void

   469 BundleDaemon::handle_bundle_delete(BundleDeleteRequest* request)

   470 {

   471     if (request->bundle_.object()) {

   472         log_info("BUNDLE_DELETE: bundle *%p (reason %s)",

   473                  request->bundle_.object(),

   474                  BundleStatusReport::reason_to_str(request->reason_));

   475         delete_bundle(request->bundle_, request->reason_);

   476     }

   477 }

   478 

   479 //----------------------------------------------------------------------

   480 void

   481 BundleDaemon::handle_bundle_accept(BundleAcceptRequest* request)

   482 {

   483     *request->result_ =

   484         router_->accept_bundle(request->bundle_.object(), request->reason_);

   485 

   486     log_info("BUNDLE_ACCEPT_REQUEST: bundle *%p %s (reason %s)",

   487              request->bundle_.object(),

   488              *request->result_ ? "accepted" : "not accepted",

   489              BundleStatusReport::reason_to_str(*request->reason_));

   490 }

   491     

   492 //----------------------------------------------------------------------

   493 void

   494 BundleDaemon::handle_bundle_received(BundleReceivedEvent* event)

   495 {

   496     const BundleRef& bundleref = event->bundleref_;

   497     Bundle* bundle = bundleref.object();

   498 

   499     // update statistics and store an appropriate event descriptor

   500     const char* source_str = "";

   501     switch (event->source_) {

   502     case EVENTSRC_PEER:

   503         stats_.received_bundles_++;

   504 		if (event->link_.object()) {

   505 			s10_bundle(S10_RX,bundle,event->link_.object()->nexthop(),0,0,NULL,"link");

   506 		} else {

   507 			s10_bundle(S10_RX,bundle,event->prevhop_.c_str(),0,0,NULL,"nolink");

   508 		}

   509         break;

   510         

   511     case EVENTSRC_APP:

   512         stats_.received_bundles_++;

   513         source_str = " (from app)";

   514 		if (event->registration_ != NULL) {

   515 			s10_bundle(S10_FROMAPP,bundle,event->registration_->endpoint().c_str(),0,0,NULL,NULL);

   516 		} else {

   517 			s10_bundle(S10_FROMAPP,bundle,"dunno",0,0,NULL,NULL);

   518 		}

   519         break;

   520         

   521     case EVENTSRC_STORE:

   522         source_str = " (from data store)";

   523 		s10_bundle(S10_FROMDB,bundle,NULL,0,0,NULL,NULL);

   524         break;

   525         

   526     case EVENTSRC_ADMIN:

   527         stats_.generated_bundles_++;

   528         source_str = " (generated)";

   529         break;

   530         

   531     case EVENTSRC_FRAGMENTATION:

   532         stats_.generated_bundles_++;

   533         source_str = " (from fragmentation)";

   534 		s10_bundle(S10_OHCRAP,bundle,NULL,0,0,NULL,"__FILE__:__LINE__");

   535         break;

   536 

   537     case EVENTSRC_ROUTER:

   538         stats_.generated_bundles_++;

   539         source_str = " (from router)";

   540 		s10_bundle(S10_OHCRAP,bundle,NULL,0,0,NULL,"__FILE__:__LINE__");

   541         break;

   542 

   543     default:

   544 		s10_bundle(S10_OHCRAP,bundle,NULL,0,0,NULL,"__FILE__:__LINE__");

   545         NOTREACHED;

   546     }

   547 

   548     // if debug logging is enabled, dump out a verbose printing of the

   549     // bundle, including all options, otherwise, a more terse log

   550     if (log_enabled(oasys::LOG_DEBUG)) {

   551         oasys::StaticStringBuffer<1024> buf;

   552         buf.appendf("BUNDLE_RECEIVED%s: prevhop %s (%u bytes recvd)\n",

   553                     source_str, event->prevhop_.c_str(), event->bytes_received_);

   554         bundle->format_verbose(&buf);

   555         log_multiline(oasys::LOG_DEBUG, buf.c_str());

   556     } else {

   557         log_info("BUNDLE_RECEIVED%s *%p prevhop %s (%u bytes recvd)",

   558                  source_str, bundle, event->prevhop_.c_str(), event->bytes_received_);

   559     }

   560     

   561     // log the reception in the bundle's forwarding log

   562     if (event->source_ == EVENTSRC_PEER && event->link_ != NULL)

   563     {

   564         bundle->fwdlog()->add_entry(event->link_,

   565                                     ForwardingInfo::FORWARD_ACTION,

   566                                     ForwardingInfo::RECEIVED);

   567     }

   568     else if (event->source_ == EVENTSRC_APP)

   569     {

   570         if (event->registration_ != NULL) {

   571             bundle->fwdlog()->add_entry(event->registration_,

   572                                         ForwardingInfo::FORWARD_ACTION,

   573                                         ForwardingInfo::RECEIVED);

   574         }

   575     }

   576 

   577     // log a warning if the bundle doesn't have any expiration time or

   578     // has a creation time that's in the future. in either case, we

   579     // proceed as normal

   580     if (bundle->expiration() == 0) {

   581         log_warn("bundle id %d arrived with zero expiration time",

   582                  bundle->bundleid());

   583     }

   584 

   585     u_int32_t now = BundleTimestamp::get_current_time();

   586     if ((bundle->creation_ts().seconds_ > now) &&

   587         (bundle->creation_ts().seconds_ - now > 30000))

   588     {

   589         log_warn("bundle id %d arrived with creation time in the future "

   590                  "(%llu > %u)",

   591                  bundle->bundleid(), bundle->creation_ts().seconds_, now);

   592     }

   593 

   594     /*

   595      * If a previous hop block wasn't included, but we know the remote

   596      * endpoint id of the link where the bundle arrived, assign the

   597      * prevhop_ field in the bundle so it's available for routing.

   598      */

   599     if (event->source_ == EVENTSRC_PEER)

   600     {

   601         if (bundle->prevhop()       == EndpointID::NULL_EID() ||

   602             bundle->prevhop().str() == "")

   603         {

   604             bundle->mutable_prevhop()->assign(event->prevhop_);

   605         }

   606 

   607         if (bundle->prevhop() != event->prevhop_)

   608         {

   609             log_warn("previous hop mismatch: prevhop header contains '%s' but "

   610                      "convergence layer indicates prevhop is '%s'",

   611                      bundle->prevhop().c_str(),

   612                      event->prevhop_.c_str());

   613         }

   614     }

   615     

   616     /*

   617      * Check if the bundle isn't complete. If so, do reactive

   618      * fragmentation.

   619      */

   620     if (event->source_ == EVENTSRC_PEER) {

   621         ASSERT(event->bytes_received_ != 0);

   622         fragmentmgr_->try_to_convert_to_fragment(bundle);

   623     }

   624 

   625     /*

   626      * validate a bundle, including all bundle blocks, received from a peer

   627      */

   628     if (event->source_ == EVENTSRC_PEER) { 

   629 

   630         /*

   631          * Check all BlockProcessors to validate the bundle.

   632          */

   633         status_report_reason_t

   634             reception_reason = BundleProtocol::REASON_NO_ADDTL_INFO,

   635             deletion_reason = BundleProtocol::REASON_NO_ADDTL_INFO;

   636 

   637         bool valid = BundleProtocol::validate(bundle,

   638                                               &reception_reason,

   639                                               &deletion_reason);

   640         

   641         /*

   642          * Send the reception receipt if requested within the primary

   643          * block or some other error occurs that requires a reception

   644          * status report but may or may not require deleting the whole

   645          * bundle.

   646          */

   647         if (bundle->receive_rcpt() ||

   648             reception_reason != BundleProtocol::REASON_NO_ADDTL_INFO)

   649         {

   650             generate_status_report(bundle, BundleStatusReport::STATUS_RECEIVED,

   651                                    reception_reason);

   652         }

   653 

   654         /*

   655          * If the bundle is valid, probe the router to see if it wants

   656          * to accept the bundle.

   657          */

   658         bool accept_bundle = false;

   659         if (valid) {

   660             int reason = BundleProtocol::REASON_NO_ADDTL_INFO;

   661             accept_bundle = router_->accept_bundle(bundle, &reason);

   662             deletion_reason = static_cast<BundleProtocol::status_report_reason_t>(reason);

   663         }

   664         

   665         /*

   666          * Delete a bundle if a validation error was encountered or

   667          * the router doesn't want to accept the bundle, in both cases

   668          * not giving the reception event to the router.

   669          */

   670         if (!accept_bundle) {

   671             delete_bundle(bundleref, deletion_reason);

   672             event->daemon_only_ = true;

   673             return;

   674         }

   675     }

   676     

   677     /*

   678      * Check if the bundle is a duplicate, i.e. shares a source id,

   679      * timestamp, and fragmentation information with some other bundle

   680      * in the system.

   681      */

   682     Bundle* duplicate = find_duplicate(bundle);

   683     if (duplicate != NULL) {

   684         log_notice("got duplicate bundle: %s -> %s creation %llu.%llu",

   685                    bundle->source().c_str(),

   686                    bundle->dest().c_str(),

   687                    bundle->creation_ts().seconds_,

   688                    bundle->creation_ts().seqno_);

   689 		s10_bundle(S10_DUP,bundle,NULL,0,0,NULL,"__FILE__:__LINE__");

   690 

   691         stats_.duplicate_bundles_++;

   692         

   693         if (bundle->custody_requested() && duplicate->local_custody())

   694         {

   695             generate_custody_signal(bundle, false,

   696                                     BundleProtocol::CUSTODY_REDUNDANT_RECEPTION);

   697         }

   698 

   699         if (params_.suppress_duplicates_) {

   700             // since we don't want the bundle to be processed by the rest

   701             // of the system, we mark the event as daemon_only (meaning it

   702             // won't be forwarded to routers) and return, which should

   703             // eventually remove all references on the bundle and then it

   704             // will be deleted

   705             event->daemon_only_ = true;

   706             return;

   707         }

   708 

   709         // The BP says that the "dispatch pending" retention constraint

   710         // must be removed from this bundle if there is a duplicate we

   711         // currently have custody of. This would cause the bundle to have

   712         // no retention constraints and it now "may" be discarded. Assuming

   713         // this means it is supposed to be discarded, we have to suppress

   714         // a duplicate in this situation regardless of the parameter

   715         // setting. We would then be relying on the custody transfer timer

   716         // to cause a new forwarding attempt in the case of routing loops

   717         // instead of the receipt of a duplicate, so in theory we can indeed

   718         // suppress this bundle. It may not be strictly required to do so,

   719         // in which case we can remove the following block.

   720         if (bundle->custody_requested() && duplicate->local_custody()) {

   721             event->daemon_only_ = true;

   722             return;

   723         }

   724 

   725     }

   726 

   727     /*

   728      * Add the bundle to the master pending queue and the data store

   729      * (unless the bundle was just reread from the data store on startup)

   730      *

   731      * Note that if add_to_pending returns false, the bundle has

   732      * already expired so we immediately return instead of trying to

   733      * deliver and/or forward the bundle. Otherwise there's a chance

   734      * that expired bundles will persist in the network.

   735      */

   736     bool ok_to_route =

   737         add_to_pending(bundle, (event->source_ != EVENTSRC_STORE));

   738 

   739     if (!ok_to_route) {

   740         event->daemon_only_ = true;

   741         return;

   742     }

   743     

   744     /*

   745      * If the bundle is a custody bundle and we're configured to take

   746      * custody, then do so. In case the event was delivered due to a

   747      * reload from the data store, then if we have local custody, make

   748      * sure it's added to the custody bundles list.

   749      */

   750     if (bundle->custody_requested() && params_.accept_custody_

   751         && (duplicate == NULL || !duplicate->local_custody()))

   752     {

   753         if (event->source_ != EVENTSRC_STORE) {

   754             accept_custody(bundle);

   755         

   756         } else if (bundle->local_custody()) {

   757             custody_bundles_->push_back(bundle);

   758         }

   759     }

   760 

   761     /*

   762      * If this bundle is a duplicate and it has not been suppressed, we

   763      * can assume the bundle it duplicates has already been delivered or

   764      * added to the fragment manager if required, so do not do so again.

   765      * We can bounce out now.

   766      * XXX/jmmikkel If the extension blocks differ and we care to

   767      * do something with them, we can't bounce out quite yet.

   768      */

   769     if (duplicate != NULL) {

   770         return;

   771     }

   772 

   773     /*

   774      * Check if this is a complete (non-fragment) bundle that

   775      * obsoletes any fragments that we know about.

   776      */

   777     if (! bundle->is_fragment()) {

   778         fragmentmgr_->delete_obsoleted_fragments(bundle);

   779     }

   780 

   781     /*

   782      * Deliver the bundle to any local registrations that it matches,

   783      * unless it's generated by the router or is a bundle fragment.

   784      * Delivery of bundle fragments is deferred until after re-assembly.

   785      */

   786     bool is_local =

   787         check_local_delivery(bundle,

   788                              (event->source_       != EVENTSRC_ROUTER) &&

   789                              (bundle->is_fragment() == false));

   790     

   791     /*

   792      * Re-assemble bundle fragments that are destined to the local node.

   793      */

   794     if (bundle->is_fragment() && is_local) {

   795         log_debug("deferring delivery of bundle *%p "

   796                   "since bundle is a fragment", bundle);

   797         fragmentmgr_->process_for_reassembly(bundle);

   798     }

   799 

   800     /*

   801      * Finally, bounce out so the router(s) can do something further

   802      * with the bundle in response to the event.

   803      */

   804 }

   805 

   806 //----------------------------------------------------------------------

   807 void

   808 BundleDaemon::handle_bundle_transmitted(BundleTransmittedEvent* event)

   809 {

   810     Bundle* bundle = event->bundleref_.object();

   811 

   812     LinkRef link = event->link_;

   813     ASSERT(link != NULL);

   814     

   815     log_debug("trying to find xmit blocks for bundle id:%d on link %s",

   816               bundle->bundleid(),link->name());

   817     BlockInfoVec* blocks = bundle->xmit_blocks()->find_blocks(link);

   818     

   819     // Because a CL is running in another thread or process (External CLs),

   820     // we cannot prevent all redundant transmit/cancel/transmit_failed messages.

   821     // If an event about a bundle bound for particular link is posted after another,

   822     // which it might contradict, the BundleDaemon need not reprocess the event.

   823     // The router (DP) might, however, be interested in the new status of the send.

   824     if(blocks == NULL)

   825     {

   826         log_info("received a redundant/conflicting bundle_transmit event about "

   827                  "bundle id:%d -> %s (%s)",

   828                  bundle->bundleid(),

   829                  link->name(),

   830                  link->nexthop());

   831         return;

   832     }

   833     

   834     /*

   835      * Update statistics and remove the bundle from the link inflight

   836      * queue. Note that the link's queued length statistics must

   837      * always be decremented by the full formatted size of the bundle,

   838      * yet the transmitted length is only the amount reported by the

   839      * event.

   840      */

   841     size_t total_len = BundleProtocol::total_length(blocks);

   842     

   843     stats_.transmitted_bundles_++;

   844     

   845     link->stats()->bundles_transmitted_++;

   846     link->stats()->bytes_transmitted_ += event->bytes_sent_;

   847 

   848     // remove the bundle from the link's in flight queue

   849     if (link->del_from_inflight(event->bundleref_, total_len)) {

   850         log_debug("removed bundle id:%d from link %s inflight queue",

   851                  bundle->bundleid(),

   852                  link->name());

   853     } else {

   854         log_warn("bundle id:%d not on link %s inflight queue",

   855                  bundle->bundleid(),

   856                  link->name());

   857     }

   858     

   859     // verify that the bundle is not on the link's to-be-sent queue

   860     if (link->del_from_queue(event->bundleref_, total_len)) {

   861         log_warn("bundle id:%d unexpectedly on link %s queue in transmitted event",

   862                  bundle->bundleid(),

   863                  link->name());

   864     }

   865     

   866     log_info("BUNDLE_TRANSMITTED id:%d (%u bytes_sent/%u reliable) -> %s (%s)",

   867              bundle->bundleid(),

   868              event->bytes_sent_,

   869              event->reliably_sent_,

   870              link->name(),

   871              link->nexthop());

   872 	s10_bundle(S10_TX,bundle,link->nexthop(),0,0,NULL,NULL);

   873 

   874 

   875     /*

   876      * If we're configured to wait for reliable transmission, then

   877      * check the special case where we transmitted some or all a

   878      * bundle but nothing was acked. In this case, we create a

   879      * transmission failed event in the forwarding log and don't do

   880      * any of the rest of the processing below.

   881      *

   882      * Note also the special care taken to handle a zero-length

   883      * bundle. XXX/demmer this should all go away when the lengths

   884      * include both the header length and the payload length (in which

   885      * case it's never zero).

   886      *

   887      * XXX/demmer a better thing to do (maybe) would be to record the

   888      * lengths in the forwarding log as part of the transmitted entry.

   889      */

   890     if (params_.retry_reliable_unacked_ &&

   891         link->is_reliable() &&

   892         (event->bytes_sent_ != event->reliably_sent_) &&

   893         (event->reliably_sent_ == 0))

   894     {

   895         bundle->fwdlog()->update(link, ForwardingInfo::TRANSMIT_FAILED);

   896         log_debug("trying to delete xmit blocks for bundle id:%d on link %s",

   897                   bundle->bundleid(),link->name());

   898         BundleProtocol::delete_blocks(bundle, link);

   899 

   900         log_warn("XXX/demmer fixme transmitted special case");

   901         

   902         return;

   903     }

   904 

   905     /*

   906      * Grab the latest forwarding log state so we can find the custody

   907      * timer information (if any).

   908      */

   909     ForwardingInfo fwdinfo;

   910     bool ok = bundle->fwdlog()->get_latest_entry(link, &fwdinfo);

   911     if(!ok)

   912     {

   913         oasys::StringBuffer buf;

   914         bundle->fwdlog()->dump(&buf);

   915         log_debug("%s",buf.c_str());

   916     }

   917     ASSERTF(ok, "no forwarding log entry for transmission");

   918     // ASSERT(fwdinfo.state() == ForwardingInfo::QUEUED);

   919     if (fwdinfo.state() != ForwardingInfo::QUEUED) {

   920         log_err("*%p fwdinfo state %s != expected QUEUED",

   921                 bundle, ForwardingInfo::state_to_str(fwdinfo.state()));

   922     }

   923     

   924     /*

   925      * Update the forwarding log indicating that the bundle is no

   926      * longer in flight.

   927      */

   928     log_debug("updating forwarding log entry on *%p for *%p to TRANSMITTED",

   929               bundle, link.object());

   930     bundle->fwdlog()->update(link, ForwardingInfo::TRANSMITTED);

   931                             

   932     /*

   933      * Check for reactive fragmentation. If the bundle was only

   934      * partially sent, then a new bundle received event for the tail

   935      * part of the bundle will be processed immediately after this

   936      * event.

   937      */

   938     if (link->reliable_) {

   939         fragmentmgr_->try_to_reactively_fragment(bundle,

   940                                                  blocks,

   941                                                  event->reliably_sent_);

   942     } else {

   943         fragmentmgr_->try_to_reactively_fragment(bundle,

   944                                                  blocks,

   945                                                  event->bytes_sent_);

   946     }

   947 

   948     /*

   949      * Remove the formatted block info from the bundle since we don't

   950      * need it any more.

   951      */

   952     log_debug("trying to delete xmit blocks for bundle id:%d on link %s",

   953               bundle->bundleid(),link->name());

   954     BundleProtocol::delete_blocks(bundle, link);

   955     blocks = NULL;

   956 

   957     /*

   958      * Generate the forwarding status report if requested

   959      */

   960     if (bundle->forward_rcpt()) {

   961         generate_status_report(bundle, BundleStatusReport::STATUS_FORWARDED);

   962     }

   963     

   964     /*

   965      * Schedule a custody timer if we have custody.

   966      */

   967     if (bundle->local_custody()) {

   968         bundle->custody_timers()->push_back(

   969             new CustodyTimer(fwdinfo.timestamp(),

   970                              fwdinfo.custody_spec(),

   971                              bundle, link));

   972         

   973         // XXX/TODO: generate failed custodial signal for "forwarded

   974         // over unidirectional link" if the bundle has the retention

   975         // constraint "custody accepted" and all of the nodes in the

   976         // minimum reception group of the endpoint selected for

   977         // forwarding are known to be unable to send bundles back to

   978         // this node

   979     }

   980 }

   981 

   982 //----------------------------------------------------------------------

   983 void

   984 BundleDaemon::handle_bundle_delivered(BundleDeliveredEvent* event)

   985 {

   986     // update statistics

   987     stats_.delivered_bundles_++;

   988     

   989     /*

   990      * The bundle was delivered to a registration.

   991      */

   992     Bundle* bundle = event->bundleref_.object();

   993 

   994     log_info("BUNDLE_DELIVERED id:%d (%zu bytes) -> regid %d (%s)",

   995              bundle->bundleid(), bundle->payload().length(),

   996              event->registration_->regid(),

   997              event->registration_->endpoint().c_str());

   998 	s10_bundle(S10_DELIVERED,bundle,event->registration_->endpoint().c_str(),0,0,NULL,NULL);

   999 

  1000     /*

  1001      * Generate the delivery status report if requested.

  1002      */

  1003     if (bundle->delivery_rcpt())

  1004     {

  1005         generate_status_report(bundle, BundleStatusReport::STATUS_DELIVERED);

  1006     }

  1007 

  1008     /*

  1009      * If this is a custodial bundle and it was delivered, we either

  1010      * release custody (if we have it), or send a custody signal to

  1011      * the current custodian indicating that the bundle was

  1012      * successfully delivered, unless there is no current custodian

  1013      * (the eid is still dtn:none).

  1014      */

  1015     if (bundle->custody_requested())

  1016     {

  1017         if (bundle->local_custody()) {

  1018             release_custody(bundle);

  1019 

  1020         } else if (bundle->custodian().equals(EndpointID::NULL_EID())) {

  1021             log_info("custodial bundle *%p delivered before custody accepted",

  1022                      bundle);

  1023 

  1024         } else {

  1025             generate_custody_signal(bundle, true,

  1026                                     BundleProtocol::CUSTODY_NO_ADDTL_INFO);

  1027         }

  1028     }

  1029 }

  1030 

  1031 //----------------------------------------------------------------------

  1032 void

  1033 BundleDaemon::handle_bundle_expired(BundleExpiredEvent* event)

  1034 {

  1035     // update statistics

  1036     stats_.expired_bundles_++;

  1037     

  1038     const BundleRef& bundle = event->bundleref_;

  1039 

  1040     log_info("BUNDLE_EXPIRED *%p", bundle.object());

  1041 

  1042     // note that there may or may not still be a pending expiration

  1043     // timer, since this event may be coming from the console, so we

  1044     // just fall through to delete_bundle which will cancel the timer

  1045 

  1046     delete_bundle(bundle, BundleProtocol::REASON_LIFETIME_EXPIRED);

  1047     

  1048     // fall through to notify the routers

  1049 }

  1050 

  1051 //----------------------------------------------------------------------

  1052 void

  1053 BundleDaemon::handle_bundle_send(BundleSendRequest* event)

  1054 {

  1055     LinkRef link = contactmgr_->find_link(event->link_.c_str());

  1056     if (link == NULL){

  1057         log_err("Cannot send bundle on unknown link %s", event->link_.c_str()); 

  1058         return;

  1059     }

  1060 

  1061     BundleRef br = event->bundle_;

  1062     if (! br.object()){

  1063         log_err("NULL bundle object in BundleSendRequest");

  1064         return;

  1065     }

  1066 

  1067     ForwardingInfo::action_t fwd_action =

  1068         (ForwardingInfo::action_t)event->action_;

  1069 

  1070     actions_->queue_bundle(br.object(), link,

  1071         fwd_action, CustodyTimerSpec::defaults_);

  1072 }

  1073 

  1074 //----------------------------------------------------------------------

  1075 void

  1076 BundleDaemon::handle_bundle_cancel(BundleCancelRequest* event)

  1077 {

  1078     BundleRef br = event->bundle_;

  1079 

  1080     if(!br.object()) {

  1081         log_err("NULL bundle object in BundleCancelRequest");

  1082         return;

  1083     }

  1084 

  1085     // If the request has a link name, we are just canceling the send on

  1086     // that link.

  1087     if (!event->link_.empty()) {

  1088         LinkRef link = contactmgr_->find_link(event->link_.c_str());

  1089         if (link == NULL) {

  1090             log_err("BUNDLE_CANCEL no link with name %s", event->link_.c_str());

  1091             return;

  1092         }

  1093 

  1094         log_info("BUNDLE_CANCEL bundle %d on link %s", br->bundleid(),

  1095                 event->link_.c_str());

  1096         

  1097         actions_->cancel_bundle(br.object(), link);

  1098     }

  1099     

  1100     // If the request does not have a link name, the bundle itself has been

  1101     // canceled (probably by an application).

  1102     else {

  1103         delete_bundle(br);

  1104     }

  1105 }

  1106 

  1107 //----------------------------------------------------------------------

  1108 void

  1109 BundleDaemon::handle_bundle_cancelled(BundleSendCancelledEvent* event)

  1110 {

  1111     Bundle* bundle = event->bundleref_.object();

  1112     LinkRef link = event->link_;

  1113     

  1114     log_info("BUNDLE_CANCELLED id:%d -> %s (%s)",

  1115             bundle->bundleid(),

  1116             link->name(),

  1117             link->nexthop());

  1118     

  1119     log_debug("trying to find xmit blocks for bundle id:%d on link %s",

  1120               bundle->bundleid(), link->name());

  1121     BlockInfoVec* blocks = bundle->xmit_blocks()->find_blocks(link);

  1122     

  1123     // Because a CL is running in another thread or process (External CLs),

  1124     // we cannot prevent all redundant transmit/cancel/transmit_failed 

  1125     // messages. If an event about a bundle bound for particular link is 

  1126     // posted after  another, which it might contradict, the BundleDaemon 

  1127     // need not reprocess the event. The router (DP) might, however, be 

  1128     // interested in the new status of the send.

  1129     if (blocks == NULL)

  1130     {

  1131         log_info("received a redundant/conflicting bundle_cancelled event "

  1132                  "about bundle id:%d -> %s (%s)",

  1133                  bundle->bundleid(),

  1134                  link->name(),

  1135                  link->nexthop());