test_sink.cc

topo_gen基于NS2的场景自动生成软件源代码

//
// test_sink.cc     : Test Sink Main File
// author           : Fabio Silva
//
// Copyright (C) 2000-2003 by the University of Southern California
// $Id: test_sink.cc,v 1.1 2005/04/22 06:08:36 fstann Exp $
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License,
// version 2, as published by the Free Software Foundation.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along
// with this program; if not, write to the Free Software Foundation, Inc.,
// 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
//
//

#include "test_sink.hh"

TestSinkApp *test_sink_app;

void signal_handler(int p)
{
  test_sink_app->stop();
  exit(0);
}

void TestSinkReceive::recv(NRAttrVec *data, NR::handle my_handle)
{
  app_->recv(data, my_handle);
}

void TestSinkApp::stop()
{
  EventsList::iterator event_iterator;
  EventEntry *event_entry;
  FILE *outfile = NULL;
  char outfile_name[100];

  if (my_id_ != -1)
    sprintf(outfile_name, "/tmp/sink-%d.out", my_id_);
  else
    sprintf(outfile_name, "/tmp/sink.out");

  outfile = fopen(outfile_name, "a");

  if (outfile){
    fprintf(outfile, "Sink %d received %d messages !\n",
	    my_id_, num_msg_recv_);

    for (event_iterator = events_list_.begin();
	 event_iterator != events_list_.end();
	 event_iterator++){
      event_entry = *event_iterator;
      fprintf(outfile, "Event: Source %d, delay %.5f\n",
	      event_entry->source_id_, event_entry->latency_);
    }
    
    fclose(outfile);
  }

  DiffPrint(DEBUG_ALWAYS,
	    "Stopping... number of messages received %d !\n",
	    num_msg_recv_);
}

void TestSinkApp::recv(NRAttrVec *data, NR::handle my_handle)
{
  NRSimpleAttribute<void *> *timeAttr = NULL;
  NRSimpleAttribute<int> *nodeAttr = NULL;
  EventTime *probe_event;
  EventEntry *event_entry;
  long delay_seconds;
  long delay_useconds;
  float total_delay;
  struct timeval tmv;
  int source_id;

  GetTime(&tmv);

  if (tmv.tv_sec > start_time_.tv_sec + EXPERIMENT_WARM_UP_TIME){
    // Count this message
    num_msg_recv_++;
  }

  timeAttr = TimeAttr.find(data);
  nodeAttr = NodeAttr.find(data);

  if (!timeAttr || !nodeAttr){
    DiffPrint(DEBUG_ALWAYS, "Received a BAD packet !\n");
    PrintAttrs(data);
    return;
  }

  // Calculate latency
  probe_event = (EventTime *) timeAttr->getVal();
  delay_seconds = tmv.tv_sec;
  delay_useconds = tmv.tv_usec;

  if ((delay_seconds < probe_event->seconds_) ||
      ((delay_seconds == probe_event->seconds_) &&
       (delay_useconds < probe_event->useconds_))){
    // Time's not synchronized
    delay_seconds = -1;
    delay_useconds = 0;
    DiffPrint(DEBUG_ALWAYS, "Error calculating delay !\n");
  }
  else{
    delay_seconds = delay_seconds - probe_event->seconds_;
    if (delay_useconds < probe_event->useconds_){
      delay_seconds--;
      delay_useconds = delay_useconds + 1000000;
    }
    delay_useconds = delay_useconds - probe_event->useconds_;
    total_delay = (float) (1.0 * delay_seconds) +
      ((float) delay_useconds / 1000000.0);

    source_id = nodeAttr->getVal();

    DiffPrint(DEBUG_ALWAYS,
	      "Received data from source %d, total latency = %f!\n",
	      source_id, total_delay);

    if (tmv.tv_sec > start_time_.tv_sec + EXPERIMENT_WARM_UP_TIME){
      event_entry = new EventEntry(source_id, total_delay);
      events_list_.push_back(event_entry);
    }
  }
}

handle TestSinkApp::setupSubscription()
{
  NRAttrVec attrs;

  attrs.push_back(NRClassAttr.make(NRAttribute::IS,
				   NRAttribute::INTEREST_CLASS));
  // Use push or pull semantics
  if (using_push_){
    attrs.push_back(NRScopeAttr.make(NRAttribute::IS,
				     NRAttribute::NODE_LOCAL_SCOPE));
    if (using_gear_){
      attrs.push_back(LatitudeAttr.make(NRAttribute::IS, lat_pt_));
      attrs.push_back(LongitudeAttr.make(NRAttribute::IS, long_pt_));
    }
  }
  else{
    attrs.push_back(NRScopeAttr.make(NRAttribute::IS,
				     NRAttribute::GLOBAL_SCOPE));
    attrs.push_back(NRProtocolAttr.make(NRAttribute::IS,
					NRAttribute::ONE_PHASE_PULL_PROTOCOL));
    if (using_gear_){
      attrs.push_back(LatitudeAttr.make(NRAttribute::GE, lat_min_));
      attrs.push_back(LatitudeAttr.make(NRAttribute::LE, lat_max_));
      attrs.push_back(LongitudeAttr.make(NRAttribute::GE, long_min_));
      attrs.push_back(LongitudeAttr.make(NRAttribute::LE, long_max_));
    }
  }

  attrs.push_back(TargetAttr.make(NRAttribute::EQ, "Elephant"));

  handle h = dr_->subscribe(&attrs, mr_);

  ClearAttrs(&attrs);

  return h;
}

void TestSinkApp::run()
{
  subHandle_ = setupSubscription();

  // Do nothing
  while (1){
    sleep(1000);
  }
}

void TestSinkApp::usage(char *s){
  DiffPrint(DEBUG_ALWAYS, "Usage: %s [-d debug] [-p port] [-s] [-g] [-r] [-h]\n\n", s);
  DiffPrint(DEBUG_ALWAYS, "\t-d - Sets debug level (0-10)\n");
  DiffPrint(DEBUG_ALWAYS, "\t-p - Uses port 'port' to talk to diffusion\n");
  DiffPrint(DEBUG_ALWAYS,
	    "\t-s - Uses push semantics (default: one-phase-pull)\n");
  DiffPrint(DEBUG_ALWAYS, "\t-g - Uses Gear\n");
  DiffPrint(DEBUG_ALWAYS, "\t-h - Prints this information\n");
  DiffPrint(DEBUG_ALWAYS, "\n");
  exit(0);
}

void TestSinkApp::parseCommandLine(int argc, char **argv)
{
  u_int16_t diff_port = DEFAULT_DIFFUSION_PORT;
  int debug_level;
  int opt;

  config_file_ = NULL;
  using_push_ = false;
  using_gear_ = false;
  opterr = 0;
  lat_min_ = -1;
  lat_max_ = -1;
  long_min_ = -1;
  long_max_ = -1;
  lat_pt_ = -1;
  long_pt_ = -1;

  while (1){
    opt = getopt(argc, argv, "sghd:p:i:j:k:l:x:y:");
    switch (opt){

    case 'p':

      diff_port = (u_int16_t) atoi(optarg);
      if ((diff_port < 1024) || (diff_port >= 65535)){
	DiffPrint(DEBUG_ALWAYS, "Error: Diffusion port must be between 1024 and 65535 !\n");
	exit(-1);
      }

      break;

    case 'i':

      long_min_ = (float) atof(optarg);

      break;

    case 'j':

      long_max_ = (float) atof(optarg);

      break;

    case 'k':

      lat_min_ = (float) atof(optarg);

      break;

    case 'l':

      lat_max_ = (float) atof(optarg);

      break;

    case 'x':

      long_pt_ = (float) atof(optarg);

      break;

    case 'y':

      lat_pt_ = (float) atof(optarg);

      break;

    case 'h':

      usage(argv[0]);

      break;

    case 'd':

      debug_level = atoi(optarg);

      if (debug_level < 1 || debug_level > 10){
	DiffPrint(DEBUG_ALWAYS, "Error: Debug level outside range or missing !\n");
	usage(argv[0]);
      }

      global_debug_level = debug_level;

      break;

    case 's':

      using_push_ = true;

      break;

    case 'g':

      using_gear_ = true;

      break;

    case '?':

      DiffPrint(DEBUG_ALWAYS,
		"Error: %c isn't a valid option or its parameter is missing !\n", optopt);
      usage(argv[0]);

      break;

    case ':':

      DiffPrint(DEBUG_ALWAYS, "Parameter missing !\n");
      usage(argv[0]);

      break;

    }

    if (opt == -1)
      break;
  }

  diffusion_port_ = diff_port;

  // Check for gear coordinates input
  if (using_gear_){
    if (using_push_){
      // For push, we must have a region
      if (lat_pt_ == -1 || long_pt_ == -1){
	DiffPrint(DEBUG_ALWAYS, "Using push and gear but not point set !\n");
	exit(-1);
      }
    }
    else{
      if (lat_min_ == -1 || lat_max_ == -1 ||
	  long_min_ == -1 || long_max_ == 1){
	DiffPrint(DEBUG_ALWAYS, "Using pull and gear but region not set !\n");
	exit(-1);
      }
    }
  }
}

TestSinkApp::TestSinkApp(int argc, char **argv)
{
  int sleep_time;
  char *sim_id;

  num_msg_recv_ = 0;

  mr_ = new TestSinkReceive(this);

  parseCommandLine(argc, argv);
  dr_ = NR::createNR(diffusion_port_);

  // Initialize Experiment
  sim_id = getenv("SIM_ID");
  if (sim_id)
    my_id_ = atoi(sim_id);
  else
    my_id_ = -1;

  GetTime(&start_time_);

  SetSeed(&start_time_);
  sleep_time = STARTUP_DELAY_TIME +
    (int) ((STARTUP_DELAY_JITTER * (GetRand() * 1.0 / RAND_MAX)
	    - (STARTUP_DELAY_JITTER / 2)));
  //  DiffPrint(DEBUG_ALWAYS, "Sink: Sleeping for %d seconds !\n",
  //	    sleep_time);
  sleep(sleep_time);
  DiffPrint(DEBUG_ALWAYS, "Sink Initialized !\n");
}

#ifndef USE_SINGLE_ADDRESS_SPACE
int main(int argc, char **argv)
{
  TestSinkApp *app;

  app = new TestSinkApp(argc, argv);

  test_sink_app = app;
  signal(SIGINT, signal_handler);

  app->run();

  return 0;
}
#endif // !USE_SINGLE_ADDRESS_SPACE