agent.cc

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

//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.

#include "agent.h"
#include "ip.h"
#include "nsport.h"
#include "config.h"
#include "macedon.h"
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <stdlib.h>

#ifdef GCC3
#include <ext/hash_map>
using namespace __gnu_cxx;
#else
#include <hash_map>
#endif


#ifndef _SOLARIS
extern "C" {int gethostname(char *name, size_t len);}
#endif
int defttl = 100;
char my_hostname[80];
short global_port = - 1;
//short global_multicast_port = - 1;

pthread_cond_t Agent::sending_cv;

pthread_t Agent::childtid;
#ifdef RWLOCK
pthread_rwlock_t Agent::agentrwlock;
#else
pthread_mutex_t Agent::agentlock = PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP; 
//  pthread_mutex_t Agent::agentlock = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
int Agent::lock_initialized = 0;
#endif

int Agent::lock_count;
double Agent::lock_taken;
int Agent::locked_for_writing;
pthread_t Agent::lock_owner;


Agent::Agent(int arg)
{
  int yes = 1;
  lock_count= 0;
  struct hostent *phe;
  struct in_addr addr;
  lock_owner = (pthread_t)0;

//    if (!lock_initialized)
//      {
//        pthread_mutexattr_t    attr;

//        pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
//        pthread_mutex_init(&agentlock, &attr);
//        printf("agentlock initialized.\n");
//        lock_initialized = 1;
//      }


//  if ( (sockfd = socket(PF_INET, SOCK_DGRAM,0)) < 0)
//  {
//    printf("Could not obtain unicast socket.\n");
//    exit(41);
//  }

//  if ( (multisockfd = socket(PF_INET, SOCK_DGRAM,0)) < 0)
//  {
//    printf("Could not obtain multicast socket.\n");
//    exit(42);
//  }

//#ifndef NSPORT_COMPRESS_ADDRESS
//  setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR,
//      (int *)&yes, sizeof(yes));
//#endif
//  setsockopt(multisockfd, SOL_SOCKET, SO_REUSEADDR,
//      (int *)&yes, sizeof(yes));

//  bzero ((char *)&readsocket, sizeof(readsocket));
//  readsocket.sin_family = AF_INET;
//  if (parameters.getint("port") > 0)
//  {
//    realport = htons((short unsigned int)parameters.getint("port"));
//  } 
//  else {
//    realport = htons((short unsigned int)NSPORT_PORT);	
//  }


//  readsocket.sin_port = realport;
//  readsocket.sin_addr.s_addr = htonl(INADDR_ANY);

//  setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_LOOP, &yes, sizeof(yes));
  //        printf("trying to bind to read port %d global read for %d\n", realport,global_port);

//  while (bind(sockfd, (struct sockaddr *) & readsocket,
//        sizeof (readsocket)) == -1)
//  {
//    printf("Could not bind to read port %d global read for %d\n", ntohs(realport),global_port);
//#ifdef NSPORT_COMPRESS_ADDRESS
//    printf("Trying next port %d.\n", realport+1);
//    realport = realport++;
//    readsocket.sin_port = realport;
//#else
//    exit(43);
//#endif
//  }

//  bzero ((char *)&multisocket, sizeof(multisocket));
//  multisocket.sin_family = AF_INET;
//  if (global_multicast_port==-1)
//  {
//    realport = htons((short unsigned int)NSPORT_MULTI_PORT);
//  }
//  else
//  {
//    realport = htons((short unsigned int)global_multicast_port);	
//  }


//  multisocket.sin_port = realport;
//  multisocket.sin_addr.s_addr = htonl(INADDR_ANY);
//  setsockopt(multisockfd, IPPROTO_IP, IP_MULTICAST_LOOP, &yes, sizeof(yes));

//  if (bind(multisockfd, (struct sockaddr *) & multisocket,
//        sizeof (multisocket)) == -1)
//  {
//    printf("Could not bind to multicast port %d\n", multisocket.sin_port);
//    perror("Could not bind to multicast port ");
//    //        exit(44);
//  }

  if(!parameters.isset("local_address")) {  
    if (gethostname(my_hostname, sizeof(my_hostname)) == -1)
    {
      printf("Bad host name.\n");
      exit(45);
    }
    if (!parameters.getint("quiet")) {
      printf("My host name %s.\n", my_hostname);
    }
    if ((phe = gethostbyname(my_hostname)) == 0)
    {
      printf("Bad host lookup.\n");
      exit(324);
    }

    memcpy(&addr, phe->h_addr_list[0], sizeof(struct in_addr));

    realaddr = addr.s_addr;
#ifdef NSPORT_COMPRESS_ADDRESS
    here_.addr_ = convert_address(realaddr, realport);
#else
    here_.addr_ = realaddr;
#endif

    if (here_.addr_==0x0100007f) { // loopback address
      if(phe->h_addr_list[1]) {
        memcpy(&addr, phe->h_addr_list[1], sizeof(struct in_addr));

        realaddr = addr.s_addr;
#ifdef NSPORT_COMPRESS_ADDRESS
        here_.addr_ = convert_address(realaddr, realport);
#else
        here_.addr_ = realaddr;
#endif
        printf("We received the loopback address and decided to take the next address %x.\n", here_.addr_);
      } else {
        printf("We could not determine our ipaddr (found only the loopback address).  Shamefully refusing to use the loopback address.\n");
        exit(231);
      }
    }

  } else {
    inet_pton(AF_INET, parameters.getstr("local_address"), &addr);
    realaddr = addr.s_addr;
#ifdef NSPORT_COMPRESS_ADDRESS
    here_.addr_ = convert_address(realaddr, realport);
#else
    here_.addr_ = realaddr;
#endif
  }
  
  local_addr = here_.addr_;
  if (!parameters.getint("quiet")) {
    printf("Set here address to %x\n", here_.addr_);
  }

  return;
}

  int 
Agent::command(int argc, const char*const* argv)
{
  return 0;

}


//  void 
//Agent::send(Packet *pkt, int size)
//{
//  unsigned char *mydata, *mydata2;
//  int mysize;
//  hdr_cmn* recv_hdrcmn = (hdr_cmn *)pkt->bits_; 
//  hdr_ip* recv_hdrip     = (hdr_ip *)(pkt->bits_ + hdr_ip::offset_);
//  struct macedon_fields *recv_mf; 
//
//  bzero ((char *)&sendsocket, sizeof(sendsocket));
//  sendsocket.sin_family = AF_INET;
//
//  //  printf("Agent send called to send packet to destination %x, hdr size %d\n", recv_hdrip->dst_.addr_, pkt->hdrlen_);
//
//#ifdef NSPORT_COMPRESS_ADDRESS
//  sendsocket.sin_port = extract_port(recv_hdrip->dst_.addr_);
//  int realaddrfake = htons((short unsigned int)NSPORT_COMPRESS_ADDRESS);
//
//  sendsocket.sin_addr.s_addr = 
//    convert_address(recv_hdrip->dst_.addr_, realaddrfake);
//#else
//  sendsocket.sin_addr.s_addr = recv_hdrip->dst_.addr_;
//  if(global_port == -1)
//    sendsocket.sin_port = htons((short unsigned int)NSPORT_PORT);
//  else
//    sendsocket.sin_port = htons((short unsigned int)global_port);
//#endif
//
//  char*target =  inet_ntoa(sendsocket.sin_addr);
//  char*target_hostname=NULL;
//  struct hostent *hp;
//  char **p;
//
////    hp = gethostbyaddr((char *)&recv_hdrip->dst_.addr_, sizeof (recv_hdrip->dst_.addr_), AF_INET);
////    if (hp == NULL) {
////      (void) printf("host information for %s not found\n",target);
////      target_hostname= "UNKNOWN";
////      //      exit (33);
////    }
////    else
////    {
////      p = hp->h_addr_list;
////      struct in_addr in;
//
////      (void) memcpy(&in.s_addr, *p, sizeof (in.s_addr));
////      target_hostname= hp->h_name;
////    }
//  //    if (recv_hdrip->ttl_)
//  //      {
//  //        setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL,
//  //  		 (char *)&(recv_hdrip->ttl_), sizeof(recv_hdrip->ttl_));
//  //        setsockopt(sockfd, IPPROTO_IP, IP_TTL,
//  //  		 (char *)&(recv_hdrip->ttl_), sizeof(recv_hdrip->ttl_));
//  //      }
//  //    else
//  //      {
//  //        setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL,
//  //                   (char *)&(defttl), sizeof(defttl));
//  //        setsockopt(sockfd, IPPROTO_IP, IP_TTL,
//  //                   (char *)&(defttl), sizeof(defttl));
//  //      }
//
//
//  mysize = pkt->hdrlen_ + recv_hdrcmn->size_;
//  recv_hdrcmn->hdrsize_ = pkt->hdrlen_;
//  recv_mf = getmacedonfields( pkt ); 
//#ifdef MACEDON_TRACE_SEND_RECEIVE
//  printf("Sending out pkt of type %d to %x at time %f size %d data_size %d, data: ", recv_mf->mh_type_, sendsocket.sin_addr, Scheduler::instance().clock(), mysize, recv_hdrcmn->size_);
//  if (recv_hdrcmn->size_ > 0)
//  {
//    //      struct macedon_fields *temp_mf = (struct macedon_fields *)pkt->data_; 
//    //      printf("%d\n", temp_mf->mh_type_);
//    //       int tmplen = recv_hdrcmn->size_;
//    //       unsigned int *tmpdata  = (unsigned int *)(pkt->data_);
//    //       while (tmplen) 
//    // 	{
//    // 	  printf("%.8x ", *tmpdata);
//    // 	  tmpdata = tmpdata++;
//    // 	  tmplen-=sizeof(unsigned int);
//    // 	}
//    printf("\n");
//  }
//  else
//    printf("\n");
//#endif
//
//#ifdef TRANSPORT_MASSIVE_TRACE
//  printf("swp_transport -- sending pkt size %d %d %d to %.8x\n", pkt->hdrlen_+pkt->size_, pkt->hdrlen_, pkt->size_, sendsocket.sin_addr);
//  printf("swp_transport %.8x: header hex ", sendsocket.sin_addr);
//  dump_hex(pkt->bits_, pkt->hdrlen_);
//  printf("swp_transport %.8x: data hex ", sendsocket.sin_addr);
//  dump_hex(pkt->data_, pkt->size_);
//#endif
//
//  mydata = (unsigned char *)malloc(mysize);
//  memcpy (mydata, pkt->bits_, pkt->hdrlen_);
//  if (pkt->data_)
//  {
//    mydata2 = mydata+pkt->hdrlen_;
//    memcpy (mydata2, pkt->data_, recv_hdrcmn->size_);
//    //      printf("Sent %d data + %d\n",  recv_hdrcmn->size_, pkt->hdrlen_);
//  }
//
//  if (sendto(sockfd, mydata, mysize, 0, 
//        (const sockaddr *)&sendsocket, 
//        sizeof(sendsocket)) < 0)
//  {
//    perror ("Error in sending packet.\n");
//    exit(46);
//  }
//
//  Packet::free(pkt);
//  ::free(mydata);
//
//}
//

Packet*
Agent::allocpkt() const
{
  Packet* p = Packet::alloc();
  return (p);
}
const char* get_hostname( int from )
{
//   static hash_map<int,string> dns_cache;
//   if (dns_cache[from] != "") {
//     return dns_cache[from].c_str();
//   }

  if (from == -1)
  {
    return my_hostname;
  }
  struct sockaddr_in in;
  in.sin_addr.s_addr= from;
  return inet_ntoa(in.sin_addr); 

 struct hostent *hp;
 char **p;

 hp = gethostbyaddr((char *)&from, sizeof (from), AF_INET);
 if (hp == NULL) {
   //      (void) printf("get_hostname host information for %x not found\n",from);
   (void) printf("get_hostname host information for %x(%s) not found, returning IP\n",from, inet_ntoa(in.sin_addr));
   return inet_ntoa(in.sin_addr); 
   //exit (3);
 }
 p = hp->h_addr_list;

//   dns_cache[from] = hp->h_name;

 return hp->h_name;
}