peer_agent.h

升级版的p2p文件共享linux环境ns2仿真软件

/*-------------------------------------------------------------------------*/
/* Packet-level peer-to-peer and Gnutella simulator                        */
/* Author: Qi He <http://www.cc.gatech.edu/~qhe> 01Aug2003                 */
/* $Revision:$ $Name:$ $Date:$                                             */
/*-------------------------------------------------------------------------*/
#ifndef __PEER_AGENT_H_
#define __PEER_AGENT_H_

#include <list>
#include <map>

// Some useful ifdefs
#ifdef GTNETS
typedef IPAddr_t NodeAddr_t;
typedef PortId_t Port_t;
typedef PacketData Packet;
typedef TCP Socket;
#else
#include <stdarg.h>
#include "agent.h"
#include "node.h"
#include "packet.h"
#include "scheduler.h"
#include "tclcl.h"
#include "timer-handler.h"
#include "nssocket.h"

typedef unsigned short Word_t;
typedef unsigned char Byte_t;
#endif

/* gnutella protocol related macros */
#define LISTEN_PORT 6346 //default Gnutella listening port
#define PING_INTERVAL  120//PING INTERVAL (seconds)
#define CONN_INTERVAL 60 //Connection Watchdog Interval (seconds)
#define INIT_TTL 7  //default TTL
//modify by zdh
#define INIT_QUERY_TTL 12  //default TTL
#define KNOWNCACHE_LIMIT 60 //known servents cache size limit
#define CONN_TIMEOUT 60 //connection request timeout
#define MAX_DEGREE 3 //maximum number of connections
#define BK_THRESH 60
#define DESC_TIMEOUT 300 //5 mins per 
#define MAX_DESC 1000 //max descriptor held by a peer
#define SMP_INTERVAL 1000 //report available member every 200 seconds
#define RATE_LIMIT 10 //incoming rate limit

/*boostrap related*/
#define MAX_BOOTREPLY 60 //the maximum number of peers in a bootstrap reply 
#define PUBLISH_PROB 100 //probabiltiy a node publishes its presence
#define BOOTSERVER 1 //nodeID of the default bootserver

#define QUERY_TIMEOUT 120 //outstanding query timeout in the UMASS model
#define HIT_PROB 0.2 //a random hit probability for query
#define QRP_HITPROB 0.2 //probability forwarding a query to a leaf

//add by zdh 04-03
#define segment_play_time 0.1
#define start_up_time 0.1

#define GNUTELLA_BOOTSTRAP "LIME BOOTSTRAP"
#define GNUTELLA_BOOTSTRAP_RES "LIME BOOTSTRAP RES"
#define GNUTELLA_CONN "GNUTELLA CONNECT/0.4\n\n"
#define GNUTELLA_OK "GNUTELLA OK\n\n"
#define GNUTELLA_REJ "GNUTELLA REJ\n\n"
#define ULTRA_CONN "GNUTELLA CONNECT/0.6\nX-Ultrapeer:True"
#define LEAF_CONN "GNUTELLA CONNECT/0.6\nX-Ultrapeer:False"
#define ULTRA_OK "GNUTELLA/0.6 200 OK\nX-Ultrapeer:True"
#define BOOTCACHE_UPDATE "BOOTSTRAP UPDATE"

/*lengths of the corresponding null-terminated string */
/*notice that the length includes the null terminator */
#define BOOTSTRAP_LEN 15
#define BOOTSTRAP_RESLEN 19 
#define CONNREJ_LEN 15
#define CONN_LEN 23
#define CONNOK_LEN 14
#define ULTRACONN_LEN 38
#define LEAFCONN_LEN 39
#define ULTRAOK_LEN 37
#define CACHEUPDATE_LEN 17

/* some fixed lengths of message */
#define DESC_HDRLEN 23 //descriptor header
#define PONG_LEN 14 //PONG
#define QUERYHIT_FIXLEN 16 //fixed part of a query

/* message ID returned by ParseMsg()*/
#define MSG_UNKNOWN -1
#define MSG_ILLFORMAT 0
#define MSG_CONNREQ 1
#define MSG_CONNOK 2
#define MSG_BOOTSTRAP 3
#define MSG_BOOTSTRAP_RES 4
#define MSG_CONNREJ 5

/* Ultrapeer-Leaf arch related */
#define MSG_ULTRACONNREQ 6
#define MSG_LEAFCONNREQ 7
/* prioritized messages */
#define MSG_QUERYHIT  8
#define MSG_QUERY 9
#define MSG_PONG 10
#define MSG_PING 11
#define MSG_BOOTCACHE_UPDATE 12
#define MSG_PUSH 13

#define REQ_PING 0
#define REQ_QUERY 1

/* Gnutella Msg type */
#define PING 0x00
#define PONG 0x01
#define QUERY 0x80
#define QUERYHIT 0x81
#define PUSH 0x40

#define TYPE_LEGACY 0 
#define TYPE_LEAF 1
#define TYPE_ULTRA 2
#define TYPE_ANY 3

#define SOCK_LEGACY 0
#define SOCK_LEAF 1
#define SOCK_ULTRA 2
#define SOCK_BOOT 3
#define SOCK_BOOT_WAIT 4

typedef enum {NOT_CONNECTED, TRANSPORT_CONNECTED, GNUTELLA_CONNECTED} ConnState_t; //pending conn req state

typedef enum {PS_OFFLINE, PS_ACTIVE, PS_IDLE, PS_OUTSYS} PeerState_t; //peer state in the UMASS model

typedef int (*Upcall_Proto) (va_list); //upcalls from PeerAgent to PeerApp
typedef char DescriptorID_t [16];//Gnutella Descriptor ID
typedef char ServentID_t [16]; //Gnutella Servent ID

/* Structures to represent Gnutella msgs */
typedef struct {
    DescriptorID_t id_;
    Byte_t payload_desc_;
    Byte_t TTL_;
    Byte_t hops_;
    int length_;
} desc_hdr; //Descriptor header of a Gnutella Message

typedef struct desc_cache_{
	Word_t segment_id_;
	Byte_t hops_;
	double buffer_time_;
//	struct desc_cache_ *next;
}desc_cache;

//add by zdh 04-04
typedef struct desc_segment_{
	double Tdelay_;
	Byte_t searched_;
	Byte_t received_;
	double tdelay_;
}desc_segment_;

typedef struct {
    Word_t port_; 
    NodeAddr_t addr_;
    int file_shared_;
    int byte_shared_;
} tPong;  //Pong
  
typedef struct {
    Word_t minSpeed_;
    char *criteria_;
} tQuery; //Query

typedef struct {
    Byte_t num_hits_;
    Word_t port_;
    NodeAddr_t addr_;
    int speed_;
    Word_t segmentID_;
} tQueryHit; //QueryHit

typedef struct {
  ServentID_t srvid_;
  int idx_;
  NodeAddr_t addr_;
  Word_t port_;
} tPush; //Push	 

class PeerApp;
class PeerAgent;
class GnutellaApp;
class GnutellaAgent;
class GnutellaMsg;
class PendingConnEntry;
class PendingReqEntry;
class PeerEntry;
class DescEntry;
class SockEntry;
class ConnTimer;
class PingTimer;
class WatchDog;
class ServentRec;
class PeerSys;
class ActivityController;
class SmpBootServer;
class GServentRec;
class GC;
class BootServerRec;

typedef std::map<NodeAddr_t, PendingConnEntry> PendingConns_t;
typedef std::list<DescEntry> DescMap_t;
typedef std::map<NodeAddr_t, SockEntry> SockMap_t;
typedef std::list<ServentRec> ServentMap_t;
typedef std::map<NodeAddr_t, PeerEntry> PeerMap_t;
typedef std::list<PendingReqEntry> ReqList_t;
typedef std::map<NodeAddr_t, GServentRec> GServentMap_t;
typedef std::list<BootServerRec> BServerList_t;

/**********************************************************************/
/****                Part I: GnutellaAgent related                     ****/

/* Pending Connection Requests, maintained by GnutellaAgent */
class PendingConnEntry {
 public:
  PendingConnEntry(NodeAddr_t peer, double timeout, double tstamp, ConnState_t state) {
    peer_ = peer;
    timeout_ = timeout;
    start_ = tstamp;
    state_ = state;
  };

  void setState(ConnState_t connState);
  NodeAddr_t peer_;
  double timeout_;
  double start_; 
  ConnState_t state_;
};

/* Descriptor Cache */
class DescEntry {
 public:
  DescEntry(char *id, Socket *incoming, double tstamp) {
    memcpy((char *)&id_, id, 16);
    sock_ = incoming;
    tstamp_ = tstamp;
  }

  DescriptorID_t id_;
  Socket *sock_;
  double tstamp_;
};

/* Garbage Collector for DescEntry */
class GC: public TimerHandler {
 public:
  GC(GnutellaAgent *agent) {
    agent_ = agent;
  }

  GnutellaAgent *agent_;
  void expire(Event *);
};

/* Sockets maintained by a GnutellaAgent */
class SockEntry {
 public:
  SockEntry(Socket *sock, NodeAddr_t peer, Byte_t leaf);

  Socket *sock_;
  NodeAddr_t peer_;
  Byte_t type_;
};

/* bootstrap server record */
class BootServerRec {
 public:
  BootServerRec(NodeAddr_t addr, Word_t port) {
    addr_ = addr;
    port_ = port;
  };

  BootServerRec(NodeAddr_t addr) {
    addr_ = addr;
    port_ = LISTEN_PORT;
  };

  NodeAddr_t addr_;
  Word_t port_;
};

/* Servent cache */
class ServentRec {
 public:
  ServentRec(Byte_t, NodeAddr_t, int, int);

  Word_t port_; 
  NodeAddr_t addr_;
  int file_shared_;
  int byte_shared_;
};

/* Outstanding connection requests timer, per GnutellaAgent */
class ConnTimer: public TimerHandler {
 public:
  ConnTimer(GnutellaAgent *);

  void expire(Event *);
  GnutellaAgent *agent_;
};

#ifdef GTNETS
class PeerAgent: public Application {
#else
class PeerAgent: public SocketApp {
#endif
 public:
  PeerAgent(PeerApp *);
  PeerAgent(NodeAddr_t addr) {firewalled_ = FALSE;};

  PeerApp *app_;   
  SockMap_t lsocks_; //list of active sockets
  int firewalled_; //whether I am firewalled

  //upcalls from Socket
  virtual int upcall_recv(Socket *, PacketData *, Handler *) {};
  virtual void upcall_connected(Socket *) {};
  virtual void upcall_passconn(Socket *) {};
  virtual void upcall_closing(Socket *) {};
  virtual void upcall_send(Socket *) {};
};

// the base class that maintains statistics (calculated out of samples)
class BasicStat {
public:
  BasicStat(NodeAddr_t addr) {cnt_=0; tstamp_=0; avg_=0; std_=0; addr_ = addr; online_time_=0; last_online_=-1; cumtime_ = 0; prev_ = 0;};

  void increment() {
    if(tstamp_==0) 
      tstamp_ = NOW;
    else {
      double dtime = NOW - tstamp_;
      avg_ = (avg_ * (double)cnt_ +  dtime)/ (double)(cnt_+1);
      std_ = (pow((dtime - avg_), 2) + std_ * cnt_)/(double)(cnt_+1);    
      tstamp_ = NOW;
    }
    cnt_++;    
  };

  void increment(int value) {
    if(tstamp_==0) {
      tstamp_ = NOW;
      prev_ = value;
    } else {
      double dtime = NOW - tstamp_;
      avg_ = (avg_*cumtime_ + (double)prev_*dtime)/ (double)(cumtime_+dtime);
      cumtime_ += dtime;
      tstamp_ = NOW;
      prev_ = value;
    }
    cnt_++;
  }

  void increment2(int value) {
    avg_ = (avg_*(double)cnt_ + value)/(double)(cnt_+1);
    std_ = (pow((value-avg_), 2) + std_*(double)cnt_)/(double)(cnt_+1);
    cnt_++;
  }

  int total() {
    return cnt_;
  }

  void print(int i) {
    if(cnt_>1) {	
      double tmp = online_time_;
      if(last_online_!=-1) 
      	tmp += NOW - last_online_;

      //statistics 
      switch(i) {
      case 0:
  	printf("Unicast drop (%d)\t%d\t%f\t%f\n", addr_, cnt_, avg_, tmp);
	break;
	
      case 1:
  	printf("Broadcast drop (%d)\t%d\t%f\t%f\n", addr_, cnt_, avg_, tmp);
	break;
	
      case 2:
	printf("Degree (%d)\t%d\t%f\t%f\n", addr_, cnt_, avg_, tmp);
	break;

      case 3:
	printf("Recv rate (%d)\t%d\t%f\t%f\n", addr_, cnt_, avg_, tmp);
	break;
	
      default:
	break;
      }
      fflush(NULL);
    }
  }

  void online() {
    last_online_ = NOW;
  }

  void offline() {
    if (last_online_!=-1) {
  	online_time_ += NOW - last_online_;
	last_online_ = -1;
    }
    tstamp_ = 0;
  }

 protected:
  int cnt_; //number of samples
  double tstamp_; //timestamp of last sample
  double avg_;  //average of the samples
  double std_;  //standard deviation of the samples
  double online_time_; //time since online, updated when going offline
  double last_online_; //last time instance switching to online state
  double cumtime_; //time since online, updated on every sample
  int prev_; //previous sample value
  NodeAddr_t addr_;  //address of the peer

};

/* GnutellaAgent: implement a Peer-to-Peer protocol          */
/*  maintains network connections to neighbours          */
class GnutellaAgent: public PeerAgent {
 public:
  GnutellaAgent(GnutellaApp *);
  GnutellaAgent(NodeAddr_t);

  ConnTimer *conn_timer_; //outstanding conn request timer
  GnutellaApp *gapp_; //Gnutella App associated with the Agent
  GnutellaMsg *msgHandle_; //message parser
  PendingConns_t conn_pending_; //outstanding conn requests
  DescMap_t desc_cache_; //descriptor cache
  GC *gc_; //Garbage Collector
  ReqList_t pending_req_; //outstanding request list
  ServentID_t mySrvId_; //servent ID

  //flow control options
  int rate_limit_;   //rate limit per socket
  int use_prio_;     //whether to use priority message queuing for sockets

  //basic statistics
  BasicStat *bkblock_;
  BasicStat *fwblock_;
  BasicStat *rcvRate_;