macedon.h

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

//Copyright (c) 2004, Charles Killian, Adolfo Rodriguez, Dejan Kostic, Sooraj Bhat, and Amin Vahdat
//All rights reserved.
//
//Redistribution and use in source and binary forms, with or without
//modification, are permitted provided that the following conditions are met:
//
//   * Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//   * Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//   * Neither the names of Duke University nor The University of
//     California, San Diego, nor the names of its contributors
//     may be used to endorse or promote products derived from
//     this software without specific prior written permission.
//
//THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
//IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
//DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE
//FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
//DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
//SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
//OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
//USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

/*

Adolfo Rodriguez


*/

#ifndef MACEDONH
#define MACEDONH

#include "macedon-defines.h"
#include "agent.h"
#include "macedon_transport.h"
#include "macedon_tcp_transport.h"
#include "macedon_udp_transport.h"
#include "massert.h"

#define ORACLE_COST 0
#define HEARTBEAT_SEND 4.0  
#define HEARTBEAT_TIMEOUT 10.0

#define MACEDON_SANITY_CHECKS
#define MACEDON_MAX_TRACE_SIZE                   50000
extern char trace_buf_[MACEDON_MAX_TRACE_SIZE+5];

extern pthread_mutex_t debug_lock;

#define MACEDON_MAX_PACKET_SIZE 1472

// Utility timer pops this often
#define MACEDON_UTIL_TIMER_VALUE             0.1

#define MH_TYPE_DATA                       10
#define MH_TYPE_ACK                        20

// MACEDON engine packets are always LESS THAN 100
#define MH_TYPE_HIGHER                    100

class MACEDON_Timer;
class MACEDON_Agent;

class neighbor {
public:
  neighbor() : ipaddr(0), delay(0), time_last_heard(Scheduler::instance().clock()) {}
  int ipaddr;
  double delay;
  double time_last_heard;
} ; 
  
// lock stuff
struct macedon_lock_info
{
  int lock_;
  int source_of_lock_;
  double time_of_lock_;
};

class macedon_packet_qent
{
 public:
  macedon_packet_qent();

  Packet *qe_pkt;
  class macedon_packet_qent *qe_next;
  double qe_time_sent;
  int qe_retransmitted;
};


class macedon_neighbor_entry
{
 public:
  macedon_neighbor_entry(int);

  int ne_addr;
  int ne_send_seqno;     // next sequence to send
  int ne_ack_seqno;      // acked sequence number
  int ne_recv_seqno;     // next sequence number expected
  // double ne_last_sent;   // timestamp of last data sent
  struct macedon_neighbor_entry *ne_next_entry;   // next neighbor entry pointer
  struct macedon_packet_qent *ne_pktq;  // queue of packets to be sent to this node
  struct macedon_packet_qent *ne_sent;
  int ne_num_pkts;       // num pkts waiting in the queue
  int ne_num_pkts_not_sent;
  int time_added;
  int remote_time_added;
  double ne_time_to_kill;
  double ne_rtt_deviation;
  double ne_rtt_average;
  double ne_retransmit_timeout;
};


struct macedon_fields
{
  // common header
  int mh_type_;  

  // allows for layering of Agents
  int mh_dest_addr_;
  int mh_src_addr_;

  // reliable data delivery support
  /*   int mh_send_seqno_; */
  /*   int mh_ack_seqno_; */
  /*   int mh_time_added; */

#ifdef VARIABLE_HEADERS
  int mh_expanded_size_;  //the total size of this pkt, expanded
#endif
};

struct oracle_ent {
  int from;
  int to;
  double cost;
} ;


struct hdr_basic
{
  struct macedon_fields nice_mf_;
  static int offset_;
  inline static int& offset() { return offset_; }
  inline static hdr_basic* access(const Packet* p)
  {
    return (hdr_basic*) p->access(offset_);
  }
};

typedef int ( MACEDON_Agent::*agent_timer_handler)(int);

class MACEDON_Agent : public Agent
, 	   public macedon_transport_receiver
{
public:
  MACEDON_Agent(packet_t arg);

  hdr_ip* gethdrip( Packet* );
  hdr_cmn* gethdrcmn( Packet* );
  virtual int macedon_downcall_ext(int operation, void *arg)=0;
  virtual void macedon_register_handlers(macedon_forward_handler, macedon_deliver_handler, macedon_notify_handler=0, macedon_upcall_handler=0)=0;
  virtual int macedon_create_group(macedon_key groupID)=0; 
  virtual void macedon_join(macedon_key groupID)=0; 
  virtual void macedon_leave(macedon_key groupID)=0; 
  virtual int macedon_route(macedon_key dest, char *msg, int size, int transport)=0;  
  virtual int macedon_routeIP(macedon_key dest, char *msg, int size, int transport)=0;   
  virtual int macedon_multicast(macedon_key groupID, char *msg, int size, int transport)=0; 
  virtual int macedon_anycast(macedon_key groupID, char *msg, int size, int transport)=0; 
  virtual int macedon_collect(macedon_key groupID, char *msg, int size, int transport)=0;
  virtual struct macedon_fields *getmacedonfields(Packet *)=0;  
  virtual void recv(Packet *p, Handler *h)=0;
  virtual int command(int argc, const char*const *argv)=0;
  virtual void sendmsg(int nbytes, AppData *data, const char *flags=0)=0;
  virtual void check_neighbors(double curtime)=0;
  virtual void exit_agent()=0;

  // general stuff
  MACEDON_Agent *base_agent;
  int source_;    
  int num_nodes_;
  double time_booted;
  int topo_;
  int topo_nodes_;
  int seed_;
  int degree_;
  int macedon_sendret;
  int keep_quiet;

  // lock functions
  void lock_on( int );
  void lock_off();
  int  test_lock();
  int  test_lock_source();
#ifdef RWLOCK
  void Lock_Read();
#endif
  void Lock_Write();
  void Lock();
  void Unlock();

  // oracle stuff
  double oracle_get_overlay_cost(int from, int to);
  void read_oracle_file();

  // neighbor stuff
  macedon_neighbor_entry *get_neighbor(int);
  void queue_pkt(macedon_neighbor_entry *, macedon_packet_qent *);

  // /SWP/reliable data functions
  //int reliab_send(Packet *);
  //void send_next_in_queue(class macedon_neighbor_entry *);
  Packet * macedon_pkt_copy(Packet *);
  //int macedon_send(Packet *);
  //int reliab_recv(Packet *);
  //void ack_in(Packet *);
  //void retrans_check();
  Packet *macedon_pkt_alloc();
  void macedon_pkt_free(Packet *);

  double smooth_latency (double newv, double old, double alpha);

  // trace functions
  virtual void trace_print();

  int my_address()
  {
    return (here_.addr_);
  }

  // oracle stuff
  struct oracle_ent *oracle;
  int oracle_nodes;

  // thread stuff
  struct macedon_thread_qent *threads;
  
  // join stuff
  struct macedon_packet_qent *queued_join_;

  // reliable delivery
  class macedon_neighbor_entry *neighbors_;
  int packet_count_;  
  
#ifdef MACEDON_SANITY_CHECKS
  int mem_pkts_recvd_;
  int mem_pkts_sent_;
  int mem_pkts_alloced_;
  int mem_pkts_freed_;
#endif
 
  struct macedon_lock_info lockstr_;

   static const double RTT_ALPHA  = 0.9;
};

class MACEDON_Timer : public TimerHandler {
public: 
  MACEDON_Timer( MACEDON_Agent *a, agent_timer_handler handler, int periodic = 1, int target=0)
    : TimerHandler()  
  { 
    agent_ = a; 
    handler_= handler;
    periodic_= periodic; 
    running = 0; 
    canceled = 0;
    target_=target;    
  }
  
  void expire(Event *e) {
    running = 0;
    if (canceled )
      return ;
    
    (agent_->*handler_) (target_);
    
    if (periodic_ && !canceled) {
      if (interval_ < 0.0) {
	//printf("Exception, reschedding 4 for negative value %f\n",
	//   interval_ );
	exit (18);
      }
      resched(interval_);
      running = 1;
    }
  }
  
  void start( double interval)
  {  
    interval_= interval;
    running = 1;
    canceled = 0;
    resched(interval_);
  }
  
  void stop( )
  {
    canceled = 1;
    
    if(running)
      force_cancel();
    
#ifndef NSPORT
    (void)Scheduler::instance().cancel(&event_);
#endif
  }
  
  int target_;
  int running;
  int canceled; 
  
protected:
  MACEDON_Agent *agent_;
  agent_timer_handler handler_;
  int periodic_;
  double interval_;
};

int random_integer( int maximum);
int macedon_hash (int);
double wall_clock ();
void dump_hex(void *, int);
int isinrange(int, int, int);
int isinrangeright(int, int, int);
int better_or_equal (double a, double b);
int round_double (double value);

#endif