macedon_swp.cc

这是一个著名的应用层组播中间件的源码

/*   { */
/*     // ack the packet */
/*     ack_pkt = macedon_pkt_alloc(); */
/*     ack_hdrip = gethdrip( ack_pkt ); */
/* #ifndef ALLINONEHDR */
/*     ack_pkt->hdrlen_ = sizeof(hdr_cmn) + sizeof(hdr_ip) + sizeof(hdr_basic); */
/* #else */
/*     ack_pkt->hdrlen_ = sizeof(hdr_cmn) + sizeof(hdr_ip) + sizeof(macedon_fields); */
/* #endif */
/*     // ack_hdrlift = gethdrlift( ack_pkt ); */
/*     ack_mf = getmacedonfields ( ack_pkt ); */
/*     ack_hdrcmn = gethdrcmn( ack_pkt ); */
/*     ack_hdrip->fid_ = MH_TYPE_ACK; */
/*     ack_mf->mh_type_ = MH_TYPE_ACK; */
/*     ack_hdrip->dst_.addr_ = recv_hdrip->src_.addr_; */
/*     ack_hdrip->dst_.port_ = 1; */
/*     // ack_hdrip->src_.port_ = 0; */
/*     ack_mf->mh_time_added = search->time_added; */
/*     ack_mf->mh_ack_seqno_ =  recv_mf->mh_send_seqno_; */
/*     ack_hdrcmn->size() = 0; */
/* #ifdef MACEDON_TRACE_RELIAB */
/*     debug_macro("Reliab: acking message %d to %x(%x) from port %x first time \n", ack_mf->mh_ack_seqno_, ack_hdrip->dst_.addr_, ack_hdrip->dst_.port_, ack_hdrip->src_.port_); */
/* #endif */
/*     macedon_send( ack_pkt ); */
/* #ifdef MACEDON_TRACE_RELIAB */
/*     debug_macro("Reliab: first ack packet sent to %x\n", ack_hdrip->dst_.addr_); */
/* #endif */

/*     // update the next expected pkt sequence number */
/*     search->ne_recv_seqno++; */

/*     return 0; */
/*   } */
/* } */

/*   void */
/* MACEDON_Agent::ack_in(Packet *pkt) */
/* { */
/*   macedon_neighbor_entry *search; */
/*   macedon_packet_qent *to_free; */
/*   Packet *pkt_to_free; */
/*   hdr_ip* recv_hdrip     = gethdrip( pkt ); */
/*   // hdr_lift* recv_hdrlift = gethdrlift( pkt ); */
/*   struct macedon_fields *recv_mf = getmacedonfields( pkt ); */
/*   //  hdr_lift* tofree_hdrlift; */
/*   struct macedon_fields *tofree_mf; */
/*   int i, count_to_free; */

/*   // process acks even if not in the tree, Adolfo */
/*   //  if (lift_fsm_ != LIFT_FSM_IN_TREE)    */
/*   //  return; */

/* #ifdef MACEDON_TRACE_RELIAB_DETAIL */
/*   debug_macro("Reliab: searching for neighbor \n"); */
/* #endif */

/*   // first get the neighbor entry */
/*   search = neighbors_; */
/*   while (search && search->ne_addr!= recv_hdrip->src_.addr_) */
/*     search = search->ne_next_entry; */

/*   // it better have worked */
/*   if (!search) */
/*   { */
/*     debug_macro("Exception: I could not find my neighbor! I was looking for %x, yet all I found was: ", recv_hdrip->src_.addr_); */
/*     search = neighbors_; */
/*     while (search && search->ne_addr!= recv_hdrip->src_.addr_) */
/*     { */
/*       debug_macro("%x ", search->ne_addr); */
/*       search = search->ne_next_entry; */
/*     } */
/*     return; */
/*   } */

/*   if(search->remote_time_added == 0) { */
/*     debug_macro("Reliab: setting remote_time_added to %d just added peer %.8x\n",recv_mf->mh_time_added, search->ne_addr); */
/*     search->remote_time_added = recv_mf->mh_time_added; */
/*   } */
/*   else if(search->remote_time_added < recv_mf->mh_time_added) { */
/*     debug_macro("INFO: Detected new time added from neighbor, resetting reliab connection\n"); */
/*     debug_macro("Reliab: resetting remote_time_added to %d from %d existing peer %.8x\n",recv_mf->mh_time_added, search->remote_time_added, search->ne_addr); */
/*     if(recv_mf->mh_ack_seqno_ == 0) { */
/*       search->ne_send_seqno = 1; */
/*       search->ne_ack_seqno = 0; */
/*       search->ne_recv_seqno = 1; */
/*       macedon_packet_qent *to_retrans = search->ne_sent; */
/*       while (to_retrans) */
/*       { */
/*         macedon_packet_qent *to_free = to_retrans; */
/*         Packet *pkt_to_free; */
/*         pkt_to_free =  to_free->qe_pkt; */
/*         macedon_pkt_free (pkt_to_free); */
/*         to_retrans = to_retrans->qe_next; */
/*         delete to_free; */
/*       } */
/*       search->ne_sent = NULL; */
/*     } else { */
/*       debug_macro("Reliab: Things seem kosher -- going on with life."); */
/*     } */
/*     search->remote_time_added = recv_mf->mh_time_added; */
/*   } */
/*   else { */
/* //     debug_macro("Reliab: matched remote_time_added %d existing peer %.8x\n",recv_mf->mh_time_added, search->ne_addr); */
/*   } */

/*   //  #ifdef MACEDON_TRACE_RELIAB */
/*   //    debug_macro("Reliab: ack: found my neighbor %x, ack %d, expecting to get %d\n", search->ne_addr, recv_hdrlift->lh_mf_.mh_ack_seqno_, search->ne_ack_seqno+1 ); */
/*   //  #endif */

/* #ifdef MACEDON_SANITY_CHECKS */
/*   if (search->ne_num_pkts > MACEDON_PACKET_LEAK) */
/*   { */
/*     debug_macro("Exception: got ack %d with neighbor %x with big queue %d, expecting to get %d\n", recv_mf->mh_ack_seqno_, search->ne_addr, search->ne_num_pkts, search->ne_ack_seqno+1 ); */
/*     exit(54); */
/*   } */
/* #endif */

/*   // only bump the ack count if we have not seen this ack before */
/*   if (search->ne_ack_seqno+1 <= recv_mf->mh_ack_seqno_ && */
/*       search->ne_ack_seqno+MACEDON_MAX_RELIAB_WINDOW >= recv_mf->mh_ack_seqno_) */
/*   { */
/*     count_to_free = recv_mf->mh_ack_seqno_ - search->ne_ack_seqno; */
/* #ifdef MACEDON_TRACE_RELIAB */
/*     debug_macro("Reliab: message %d from %x acked, will free %d\n", recv_mf->mh_ack_seqno_, recv_hdrip->src_.addr_, count_to_free); */
/* #endif */

/*     for (i=0; i<count_to_free; i++) */
/*     { */
/*       // free the packet which has been acked */

/*       // first get the last packet in the queue, which is the oldest  */
/*       // and guaranteed to be the packet being acked */
/*       to_free = search->ne_sent; */

/*       if (!search->ne_sent) */
/*       { */
/*         printf("Exception: I received an ack for %x, but I have nothing in my send queue! count = %x, currently acked is %x\n",recv_mf->mh_ack_seqno_, count_to_free, search->ne_ack_seqno ); */
/*         exit(454); */

/*       } */

/*       search->ne_sent = search->ne_sent->qe_next; */
/*       pkt_to_free =  to_free->qe_pkt; */
/*       search->ne_num_pkts--; */

/*       //  tofree_hdrlift = gethdrlift( pkt_to_free ); */
/*       tofree_mf = getmacedonfields( pkt_to_free ); */

/*       if (tofree_mf->mh_send_seqno_ > recv_mf->mh_ack_seqno_) */
/*       { */
/*         printf("Exception: Trying to free a packet that is not acked!\n"); */
/*         exit(454); */
/*       } */

/*       if(1==count_to_free && !to_free->qe_retransmitted) */
/*       { */
/*         double curtime = Scheduler::instance().clock(); */
/*         double roundtrip = curtime - to_free->qe_time_sent; */
/*         if (roundtrip>0) */
/*         { */
/*           if (search->ne_rtt_average != MACEDON_RETRANS_TIMEOUT ) */
/*           { */
/*             search->ne_rtt_average = RTT_ALPHA*search->ne_rtt_average+(1-RTT_ALPHA)* roundtrip; */
/*           } */
/*           else */
/*           { */
/*             search->ne_rtt_average = roundtrip; */
/*           } */

/*           double difference =  search->ne_rtt_average - roundtrip; */

/*           if (difference < 0) */
/*           { */
/*             difference = - difference; */
/*           } */
/*           if (search->ne_rtt_deviation>0) */
/*           { */
/*             search->ne_rtt_deviation=RTT_ALPHA*search->ne_rtt_deviation+(1-RTT_ALPHA)*difference; */
/*           } */
/*           else */
/*           { */
/*             search->ne_rtt_deviation=(1-RTT_ALPHA)*difference; */
/*           } */

/*           search->ne_retransmit_timeout =  search->ne_rtt_average + 4*search->ne_rtt_deviation; */
/* #ifdef MACEDON_TRACE_RELIAB_DETAIL */
/*           debug_macro("Reliab: ack from %x with rtt %f new retransmit_timeout %f (=%f+4*%f)\n", recv_hdrip->src_.addr_, roundtrip, search->ne_retransmit_timeout, search->ne_rtt_average, search->ne_rtt_deviation); */
/* #endif  */
/*         } */
/*       } */
/* #ifdef MACEDON_TRACE_RELIAB_DETAIL */
/*       debug_macro("Reliab: freeing packet %d (%x) from %x queue, %d left\n", tofree_mf->mh_send_seqno_, to_free, recv_hdrip->src_.addr_,  */
/*           search->ne_num_pkts); */
/* #endif  */
/*       macedon_pkt_free (pkt_to_free); */
/*       delete to_free; */
/*     } */
/*     search->ne_ack_seqno = recv_mf->mh_ack_seqno_; */

/*     // see if there is anything in the queue waiting to go */
/*     while ( search->ne_send_seqno <= search->ne_ack_seqno+MACEDON_MAX_RELIAB_WINDOW  */
/*         && search->ne_pktq) */
/*       send_next_in_queue(search);     */
/*   } */
/*   else if (search->ne_ack_seqno+1 > recv_mf->mh_ack_seqno_ ) */
/*   { */
/*     // force restransmission since a pkt probably got lost.  the other guy is acking a pkt again.  but only do it if there is something un-acked in queue. */
/*     if (search->ne_sent) */
/*     { */
/* #ifdef MACEDON_TRACE_RELIAB */
/*       debug_macro("Reliab: retransimission forced since expected %d and got %d\n", search->ne_ack_seqno+1, recv_mf->mh_ack_seqno_ ); */
/* #endif */
/*       //	search->ne_last_sent = 0; */
/*       search->ne_sent->qe_time_sent = 0; */
/*       retrans_check(); */
/*     } */
/*   } */
/*   else */
/*   { */
/*     //  #ifdef MACEDON_TRACE_RELIAB */
/*     //        debug_macro("Reliab: bad ack %d, expecting to get %d\n", recv_hdrlift->lh_mf_.mh_ack_seqno_, search->ne_ack_seqno+1 ); */
/*     //  #endif */
/*   } */
/* } */

/*   void */
/* MACEDON_Agent::retrans_check() */
/* { */
/*   Packet *send_pkt; */
/*   macedon_packet_qent *to_retrans; */
/*   macedon_neighbor_entry *search; */
/*   hdr_ip* dest_hdrip; */
/*   //  hdr_lift* dest_hdrlift;       */
/*   struct macedon_fields *dest_mf; */
/*   double curtime = Scheduler::instance().clock(); */

/*   // loop through neighbor entries and check whether any need retransmission */
/*   search = neighbors_; */
/*   while (search) */
/*   { */
/* #ifdef MACEDON_TRACE_RELIAB */
/*     debug_macro("Reliab: checking to see if retrans is needed for %x.  the search ptr is %x, first pkt in sent q is %x \n", search->ne_addr, search, search->ne_sent); */
/* #endif  */
/*     to_retrans = search->ne_sent; */

/*     while (to_retrans) */
/*     { */

/*       //      if (curtime - search->ne_last_sent > MACEDON_RETRANS_TIMEOUT */
/*       //  && search->ne_sent) */
/*       if (curtime-to_retrans->qe_time_sent >  MACEDON_RETRANS_TIMEOUT) */
/*       { */

/* #ifdef MACEDON_TRACE_RELIAB_DETAIL */
/*         debug_macro("Reliab: retransmitting packet ent %x, packet %x\n", to_retrans, to_retrans->qe_pkt); */
/* #endif  */

/*         // copy the pkt */
/*         send_pkt = macedon_pkt_copy(to_retrans->qe_pkt); */
/*         if (send_pkt == 0) */
/*         { */
/*           debug_macro("Reliab: copy failed, not good, but we can continue \n"); */
/*           return; */
/*         } */

/*         // setup pkt */
/*         dest_hdrip = gethdrip(send_pkt); */
/*         dest_hdrip->dst_.port_ = 1; */
/*         //	  dest_hdrlift = gethdrlift(send_pkt); */
/*         dest_mf = getmacedonfields( send_pkt ); */
/* #ifdef MACEDON_SANITY_CHECKS */
/*         //	      if (search->ne_num_pkts > MACEDON_PACKET_LEAK) */
/*         //		{ */
/*         //		  debug_macro("Exception: Big queue (%d) to %x. Retransmitting packet %d\n", search->ne_num_pkts, search->ne_addr, dest_mf->mh_send_seqno_); */
/*         //    } */
/* #endif */


/* #ifdef MACEDON_TRACE_RELIAB */
/*         debug_macro("Reliab: resending message %d of type %d to %x(%x) from %x(%x)\n", dest_mf->mh_send_seqno_, dest_mf->mh_type_, dest_hdrip->dst_.addr_, dest_hdrip->dst_.port_, dest_hdrip->src_.addr_, dest_hdrip->src_.port_); */
/* #endif */
/*         dest_mf->mh_time_added = search->time_added; */
/*         // search->ne_last_sent = Scheduler::instance().clock(); */
/*         to_retrans->qe_time_sent =  Scheduler::instance().clock(); */
/*         to_retrans->qe_retransmitted = 1; */
/*         // send the copy */
/*         //	      debug_macro("Reliab: retrans message %d of type %d to %x(%x) from %x(%x)\n", dest_mf->mh_send_seqno_, dest_mf->mh_type_, dest_hdrip->dst_.addr_, dest_hdrip->dst_.port_, dest_hdrip->src_.addr_, dest_hdrip->src_.port_); */
/*         macedon_send( send_pkt ); */
/*       } */
/*       to_retrans = to_retrans->qe_next; */
/*     } */

/* #ifdef MACEDON_TRACE_RELIAB */
/*     debug_macro("Reliab: going to next one:  %x \n", search->ne_next_entry); */
/* #endif   */
/*     search = search->ne_next_entry; */
/*   } */
/* #ifdef MACEDON_TRACE_RELIAB_DETAIL */
/*   debug_macro("Reliab: done with retrans check!\n"); */
/* #endif  */
/* } */


Packet *MACEDON_Agent::macedon_pkt_alloc()
{
  Packet *p;
  struct macedon_packet_qent *qentry;
  int diff;
  int max_queue=0;
  int max_queue_not_sent=0;
  int max_to_whom;
  int join_packets_queued =0;
  int reliab_packets_queued =0;
  int reliab_packets_not_sent=0;
  macedon_neighbor_entry *search;

  p = allocpkt();

#ifdef MACEDON_TRACE_RELIAB_DETAIL
  debug_macro( "Reliab: allocated packet %x\n", p);
#endif


#ifdef MACEDON_SANITY_CHECKS
  mem_pkts_alloced_++;
  diff = mem_pkts_alloced_ + mem_pkts_recvd_ - mem_pkts_freed_ - mem_pkts_sent_;

  if ( diff > MACEDON_PACKET_LEAK)
  {
    qentry = queued_join_;
    while (qentry)
    {
      join_packets_queued++;
      qentry = qentry->qe_next;
    }

    search = neighbors_;
    while (search)
    {
      if (search->ne_num_pkts > max_queue)
      {
        max_to_whom = search->ne_addr;
        max_queue = search->ne_num_pkts;
        max_queue_not_sent = search->ne_num_pkts_not_sent;
      }
      reliab_packets_queued += search->ne_num_pkts;
      reliab_packets_not_sent += search->ne_num_pkts_not_sent;
      search = search->ne_next_entry;
    }

    debug_macro("Exception: Detected a packet leak of %d: %d %d %d %d. Had %d joins queued and %d pkts in reliab queues with %d not sent (max %d with %d not sent to %x)\n", diff, mem_pkts_alloced_, mem_pkts_recvd_, mem_pkts_freed_, mem_pkts_sent_, join_packets_queued, reliab_packets_queued, reliab_packets_not_sent,  max_queue, max_queue_not_sent, max_to_whom);
  }
#endif
  return p;
}

void MACEDON_Agent::macedon_pkt_free(Packet *pkt)
{
#ifdef MACEDON_SANITY_CHECKS
  mem_pkts_freed_++;
#endif
  Packet::free(pkt);  

#ifdef MACEDON_TRACE_RELIAB_DETAIL
  debug_macro( "Reliab: freed packet %x\n", pkt);
#endif

}



Packet *MACEDON_Agent::macedon_pkt_copy(Packet *pkt)
{

  Packet *p;

  p = pkt->copy();
#ifdef MACEDON_SANITY_CHECKS
  mem_pkts_alloced_++;
#endif
  return p;
}