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

/*************************************************************
   appmacedon: A simple MACEDON application
   Adolfo Rodriguez
   Usage:
   appmacedon [DEFAULT PARAMETER FILE] [OPTIONS]
     Options: 
       -protocol INT 
          (Which protocol to use from macedon.protocols)
       -bootstrap IP_ADDR 
          (IP of bootstrap node, typically the "root")
       -port INT 
          (Network port to use)
       -multicast_port INT 
          (Network multicast port, not currently implemented)
       -degree INT 
          (Maximum node fan-out)
       -run_time DOUBLE 
          (Amount of time to run)
       -hold_time DOUBLE 
          (Time after starting to disable overlay transformations)
       -streaming_time DOUBLE 
          (Time after starting of when to begin streaming)
       -streaming_rate INT 
          (Rate to send in Kbps)
       -data_packet_size INT 
          (Size in bytes of data packets)
       -parent_filename FILENAME 
          (File specifying nodes' parents)
       -topology INT 
          (Topology ID, 
           only for inclusion in trace entries, not used)
       -topology_size INT 
          (Size of topo (number of total nodes), 
           only for inclusion in trace entries, not used)
       -topology_seed INT 
          (Random seed, 
           only for inclusion in trace entries, not used)
       -num_nodes INT 
          (Number of nodes in overlay, 
           only for inclusion in trace entries, not used)
**************************************************************/

#include <sys/time.h>
#include <sys/resource.h>
#include <unistd.h>
#include <sys/types.h>
#include <netinet/in.h> 
#include <sys/socket.h>
#include <arpa/inet.h>
#include <math.h>
#include "macedon_api.h"
#include "ext.h"

#define GROUPID 0xcaca1234   // some arbitrary group id

// This app has two flavors:  a multicast/collect app and a P2P driver.  
// To switch (un)comment the following line:
//#define APP_P2P

// Defining the below causes the app to stream at the specified streaming_rate.
// Otherwise, it will not send any data.
#define APP_STREAM

// Globals used only by this test program
macedon_Agent *globalmacedon;
double packet_spacing;  // spacing between pkt sends
int gotten=0;   // number of pkts received
double avg_lat=0.0;  // the average latency of received pkts
char *message;   // the message we are sending
bool receiver_started; //marks whether the receiver has been started.  Used to drop early packets.

pthread_mutex_t ready_lock;
pthread_cond_t wait_for_ready;

void start_sender(double, double, int);
void start_receiver(double, double);

double now_time()
{
  double ret;
  timeval clock_;
  struct timezone tz_;
  gettimeofday(&clock_, &tz_);
  ret = (clock_.tv_sec+(double)clock_.tv_usec/1000000.0);
  return ret;
}

int my_forward(char *msg, int size, int type, int nextkey, macedon_key nextkeyhash)
{  // called by MACEDON to determine if "next hop" should be taken
  return 0;
}

void my_deliver(char *msg, int size, int type)
{  // called by MACEDON when data is delivered
 if (!receiver_started) {
  return;
 }
 if(size != parameters.getint("data_packet_size")) {
  printf("ERROR: incorrect data size %d (should be %d)\n",size,parameters.getint("data_packet_size"));
 }
 if (size > sizeof(double)) {
   double delay = now_time() - *((double *)msg);
   if (delay > 1000.0) {
     printf("Exception: weird time sent detected in app: %f %f\n", now_time(), *((double *)msg));
   }
   else {
     if (avg_lat > 0.0)
       avg_lat = (avg_lat * gotten + delay)/(gotten + 1);
     else 
       avg_lat = delay;
   }
 }
 gotten++;
 return;
}

int my_upcall(int operation, void *arg) {
  printf("my_upcall called op %d\n", operation);
  fflush(NULL);
}

int main (int argc, char **argv)
{

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

  // First load running parameters 
  parameters.loadparams(argc, argv);
  parameters.print(stdout);
  
  int protocol = parameters.getint("protocol");
  double runtime = parameters.getdouble("run_time");
  unsigned long source = inet_addr(parameters.getstr("bootstrap"));
  printf("Source is %.8x\n",source);

  // Allocate a message and fill it with something
  message = (char *)malloc(parameters.getint("data_packet_size"));
  for (int i=0; i<parameters.getint("data_packet_size")-1; i++) {
    message[i] = '.';
  }
  message[parameters.getint("data_packet_size")-1] = '\0';

  // Determine the ammount of time between each packet
  packet_spacing = (double)(parameters.getint("data_packet_size")) *8/(1000.0*(double) parameters.getint("streaming_rate"));

#ifdef APP_STREAM
  printf("appmacedon: Sending packets every %f seconds.\n", packet_spacing);
#endif  

  // Initialize the random seed
  double time_r = now_time();
  srand48((int)(1000000 * (time_r - (int)time_r)));
  time_r += parameters.getdouble("streaming_time");
  
  globalmacedon = macedon_init(source, protocol);   
  globalmacedon->macedon_register_handlers((macedon_forward_handler)&my_forward, (macedon_deliver_handler)&my_deliver, NULL, (macedon_upcall_handler)&my_upcall);

  if (globalmacedon->local_addr == source) {  // I am the "root"
    printf("Automatically starting as sender.\n");
    start_sender(runtime, parameters.getdouble("streaming_time"), protocol);
  }
  else {
#ifndef APP_P2P
    printf("Automatically starting as receiver.\n");
    start_receiver(runtime, parameters.getdouble("streaming_time"));
#else
    printf("Automatically starting as sender.\n");
    start_sender(runtime, parameters.getdouble("streaming_time"), protocol);
#endif
  }
  
  // Print receiver statistics.
  int window;
  if (gotten)
    window = gotten-1;
  else
    window = 0;
  double end = now_time();
  if(parameters.getdouble("streaming_time") < runtime) {
    printf("REPLAY got %d pkts: avg bw %.2f Kbps time %lf avg lat %.8f\n", gotten, ((double)(window)* parameters.getint("data_packet_size")*8/(1000*(end-time_r))), end-time_r, avg_lat);
  }
  printf("Exiting at time %lf\n", now_time());
  fflush(NULL);
  
  // All done.
  globalmacedon->macedon_exit();
  _exit(0);
  return 0;

}


int 
stream_data(double runtime, double stream_time)
{
  int pcount = 0;
  double start = now_time(); 
  double now = start;
  double prev_sent = now- packet_spacing;
  char qchar = 'A';
  int qcount=0;
  while (now - start < runtime - stream_time) {
    *(double *)message = now_time(); 
    int mindex = pcount/26 % (parameters.getint("data_packet_size")-sizeof(double));
    if (mindex == 0) {
      qchar = (char) ((int) qchar + qcount%26 );
      qcount ++;
    }
    char pchar = (char)( (int)'A' + pcount%26 );
    message [mindex+sizeof(double)] = pchar;
#ifdef APP_STREAM
#ifdef APP_P2P
    int mydest = (unsigned int)(drand48() * (pow(2.0,32.0)-1.0));
    //      printf("appmacedon: %lf %x P2Ping %.8x size %d\n", now, (int)globalmacedon->local_addr,mydest, parameters.getint("data_packet_size"));
    globalmacedon->macedon_route(mydest, message, parameters.getint("data_packet_size"), PRIORITY_MED);
#else 
    //    printf("%lf appmacedon: %lf %lf %x root disseminating %d, streamrate %d, actual %f, pcount %d, timespent %f\n", now, now-prev_sent, packet_spacing, (int)globalmacedon->local_addr, parameters.getint("data_packet_size"), parameters.getint("streaming_rate"), ((double)(pcount-1)*parameters.getint("data_packet_size") *8/(1000*(now-start))), pcount, now-start);
    globalmacedon->macedon_multicast(GROUPID, message, parameters.getint("data_packet_size"), PRIORITY_MED);
#endif
    pcount++;
#endif
    prev_sent = now;
    message[mindex] = qchar;
    now = now_time(); 
    double time_sending = now - start;
    // figure out how many pkts we should have sent
    int should_have_sent = (int)(time_sending / packet_spacing);
    if (pcount >= should_have_sent)
      // only sleep if we have sent what we should have
      globalmacedon->macedon_wait(packet_spacing);
    now = now_time();
  }
  
  return pcount;
}

void
start_sender(double runtime, double stream_time, int protocol)
{
  int i;
  int pcount = 0;
  int old_bullet=0;

  if(stream_time < runtime) {
    globalmacedon->macedon_wait(stream_time);  
    receiver_started=true;
    double start = now_time();

#ifndef APP_P2P
    printf("appmacedon: creating group with id %x, time %f\n", GROUPID, now_time());
    globalmacedon->macedon_create_group(GROUPID); 
#endif
    pcount = stream_data(runtime, stream_time);
    double end = now_time();
    if (pcount==0)
      pcount=1;
    printf("REPLAY sent %d pkts: avg bw %.2f Kbps time %lf\n",pcount, ((double)(pcount-1)*(parameters.getint("data_packet_size")) *8/(1000*(end-start))), end-start);
  } else {
    globalmacedon->macedon_wait(runtime);  
    
  }
  
  
  return;
}


void
start_receiver(double runtime, double stream_time)
{
  if(stream_time < runtime) {
    globalmacedon->macedon_wait(stream_time);

    receiver_started=true;
    printf("appmacedon: joining group with id %x, time %f\n", GROUPID, now_time());
    fflush(stdout);
    globalmacedon->macedon_join(GROUPID);
    //NOTE: Could be bad
    globalmacedon->macedon_wait(runtime-stream_time);   // wait til the end of the run
  } else {
    globalmacedon->macedon_wait(runtime);

  }

  return;
}