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

#ifndef AGENTH
#define AGENTH
#include <pthread.h>
#include <sys/types.h>
#include <netinet/in.h> 
#include <sys/socket.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <math.h>
#include "scheduler.h"
#include "config.h"
#include "params.h"
#include "packet.h"
#include "macedon_api.h"

const char* get_hostname( int from = -1);

// To enable read/write locks
/*  #define RWLOCK */
#define AGENT_LOCK_DEBUG 0
#define AGENT_LOCK_TIME 0

typedef int packet_t;

class Agent : public macedon_Agent
{
public:
  // general stuff
  int pt_type;
  static pthread_t childtid;
  static pthread_cond_t sending_cv;
#ifdef RWLOCK
  static pthread_rwlock_t agentrwlock;
#else
  static pthread_mutex_t agentlock;
  static int lock_initialized;
#endif
  static int locked_for_writing;
  static int lock_count;
  static  double lock_taken;
  static pthread_t lock_owner;

  // my address stuff
  ns_addr_t here_;
  int realport;
  int realaddr;

  // unicast socket
//  int sockfd;
//  struct sockaddr_in sendsocket;
//  struct sockaddr_in readsocket;

  // multicast socket
  //int multisockfd;
  //struct sockaddr_in multisocket;

  // other agent required stuff
  virtual int& size() { return size_; }
  int size_;                      // fixed packet size
  Agent(int);

/*  #ifdef RWLOCK */
  inline void Lock_Read();
/*  #endif */

  inline void Lock_Write();  // Just calls the base lock
  inline void Lock();
  inline void Unlock();

  // required agent functions
  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 void macedon_init()=0;
  virtual void trace_print()=0;
  virtual struct macedon_fields *getmacedonfields(Packet *)=0;  
  Packet *allocpkt() const;
//  void send(Packet *, int);
  int command(int argc, const char*const* argv);
};

// ---------------------------------------------- 
// lock functions, read and base(write)
// ---------------------------------------------- 

inline void  Agent::Lock_Read ()
{
#ifdef RWLOCK
#if AGENT_LOCK_DEBUG > 1
  printf("%d Agent_lock in %d %d: %d %d %x %d %x %d \n", pthread_self(),
      lock_count,
      lock_owner,
      agentlock.__m_reserved,
      agentlock.__m_count,
      agentlock.__m_owner,
      agentlock.__m_kind,
      agentlock.__m_lock.__status,
      agentlock.__m_lock.__spinlock
      );
#endif
  if (lock_count==0 ||
      lock_owner != pthread_self()) {
#if AGENT_LOCK_DEBUG > 2
    printf("%d Agent_lock going\n" , pthread_self());
    fflush(stdout);
    if (lock_owner != pthread_self() && lock_owner
        && (int)pthread_self() > 20000) {
      printf("%d diff thread locking Agent_unlock owner is %d \n", pthread_self(),lock_owner);
      fflush(stdout);
    }
#endif
    int code =pthread_rwlock_rdlock( &(agentrwlock) );
    if (code) {
      printf("lock unsuccessful %s",strerror(errno));
      fflush(stdout);
      exit(15);
    }

#if AGENT_LOCK_DEBUG > 2
    printf("%d Agent_lock gone\n" , pthread_self());
#endif
    lock_owner = pthread_self();
  }
  lock_count++;
#if AGENT_LOCK_DEBUG > 0
  printf("%d Agent_lock out %d\n" , pthread_self(),lock_count);
#endif
}
#else
  Lock();
}

#endif

inline void  Agent::Lock_Write ()
{
  Lock();
}

inline void  Agent::Lock ()
{
#if AGENT_LOCK_DEBUG > 1
  printf("%d Agent_lock in %d %d: %d %d %x %d %x %d \n", pthread_self(),
      lock_count,
      lock_owner,
      agentlock.__m_reserved,
      agentlock.__m_count,
      agentlock.__m_owner,
      agentlock.__m_kind,
      agentlock.__m_lock.__status,
      agentlock.__m_lock.__spinlock
      );
  //        fflush(stdout);
#endif
  if (lock_count==0 || // no one has the lock
#ifdef RWLOCK
      (lock_count>0 && lock_owner==pthread_self() && 
       locked_for_writing==0) || // I have the lock for reading only
#endif
      lock_owner != pthread_self() // I am not the owner
     ) {  // need to actually perform lock
#if AGENT_LOCK_DEBUG > 2
    fflush(stdout);
    if (lock_owner != pthread_self() && lock_owner
        && (int)pthread_self() > 20000) {
      printf("%d diff thread locking Agent_unlock owner is %d \n", pthread_self(),lock_owner);
      fflush(stdout);
      //  	      abort();
    }
#endif
#ifdef RWLOCK
    int local_count=0;
    if (lock_count>0 && lock_owner==pthread_self() && 
        locked_for_writing==0) { // I have the lock for reading only
      // The idea here is that a read lock is *converted* into a
      //  write lock. The lock_count is kept locally in case another thread
      //  crept in and stole the lock.  
      local_count = lock_count;
      int code =pthread_rwlock_unlock( &(agentrwlock) );
      if (code) {
        printf("read unlock unsuccessful %s",strerror(errno));
        fflush(stdout);
        exit(15);
      }
    }
    int code =pthread_rwlock_wrlock( &(agentrwlock) );
    if (local_count) // restore lock_count for the odd case
      lock_count = local_count;
    locked_for_writing=1;
#if AGENT_LOCK_TIME
    lock_taken =  Scheduler::instance().clock();
#endif
#else
    int code =pthread_mutex_lock(&(agentlock));
    locked_for_writing=1;
#endif
    if (code) {
      printf("lock unsuccessful %s",strerror(errno));
      fflush(stdout);
      exit(15);
    }

#if AGENT_LOCK_DEBUG > 2
    printf("%d Agent_lock gone\n" , pthread_self());
#endif
    lock_owner = pthread_self();
  }
  lock_count++;
#if AGENT_LOCK_DEBUG > 0
  printf("%d Agent_lock out %d\n" , pthread_self(),lock_count);
#endif
}

// ---------------------------------------------- 
// unlock
// ---------------------------------------------- 

inline void  Agent::Unlock ()
{
#if AGENT_LOCK_DEBUG > 1
  printf("%d Agent_unlock in %d %d: %d %d %x %d %x %d \n", pthread_self(),
      lock_count,
      lock_owner,
      agentlock.__m_reserved,
      agentlock.__m_count,
      agentlock.__m_owner,
      agentlock.__m_kind,
      agentlock.__m_lock.__status,
      agentlock.__m_lock.__spinlock
      );
  //    fflush(stdout);
#endif
  //        agentlock.__m_lock.__status =1;
  if (lock_owner != pthread_self())
  {
    printf("%d wrong thread unlocking Agent_unlock owner is %d \n", pthread_self(),lock_owner);
    fflush(stdout);
    abort();
  }
  if (lock_count <= 0)
  {
    printf("%d lock_count %d <= 0\n", pthread_self(), lock_count);
    fflush(stdout);
    abort();
  }

  lock_count--;
  if (lock_count==0)
  {
#if AGENT_LOCK_TIME
     if (locked_for_writing )
      {
	double lock_t = Scheduler::instance().clock() - lock_taken;
	  if (lock_t > 0.25)
	    {
	      printf ("%f lock taken for eternity: %f\n", Scheduler::instance().clock(), lock_t);
	      fflush(stdout);
//  	      abort();
      	    }
      }
#endif
    lock_owner = (pthread_t) 0;
    locked_for_writing = 0;
#if AGENT_LOCK_DEBUG > 2
    printf("%d Agent_unlock going\n" , pthread_self());
    fflush(stdout);
#endif

#ifdef RWLOCK      
    int code =pthread_rwlock_unlock( &(agentrwlock) );
#else
    int code =pthread_mutex_unlock(&(agentlock));
#endif
    if (code) {
      printf("unlock unsuccessful %s",strerror(errno));
      fflush(stdout);
      exit(15);
    }

#if AGENT_LOCK_DEBUG > 2
    printf("%d Agent_unlock gone\n" , pthread_self());
#endif
  }
#if AGENT_LOCK_DEBUG > 0
  printf("%d Agent_unlock out %d %d\n", pthread_self(), lock_count,lock_owner );
#endif
}

#endif