diffusion.cc

R. Lo Cigno, P. Larcheri 802.11e closed-loop scheduling Description: ns-2 package for 802.11e clo

	if (filter_entry->priority_ <= new_priority){
	  forwardMessage(msg, filter_entry);
	  break;
	}
      }

      if (filter_list_itr == filter_list->end())
	filter_is_last = true;

    }
    else{
      filter_is_last = true;
    }

    if (filter_is_last){
      // Forward message to the network or the destination application
      sendMessage(msg);
    }

    filter_list->clear();

    delete filter_list;

    break;

  default:

    DiffPrint(DEBUG_ALWAYS, "Error: Unknown control message received !\n");

    break;
  }
}

void DiffusionCoreAgent::logControlMessage(Message *msg, int command,
					   int param1, int param2)
{
  // Logs the incoming message
}

#ifdef NS_DIFFUSION
DiffusionCoreAgent::DiffusionCoreAgent(DiffRoutingAgent *diffrtg, int nodeid)
{
#else
DiffusionCoreAgent::DiffusionCoreAgent(int argc, char **argv)
{
  int opt;
  int debug_level;
#endif // NS_DIFFUSION
  DeviceList *in_devices, *out_devices, *local_out_devices;
  DiffusionIO *device;
  TimerCallback *callback;
  char *scadds_env;
  long stop_time;
  struct timeval tv;
#ifdef IO_LOG
  IOLog *pseudo_io_device;
#endif // IO_LOG

  opterr = 0;
  config_file_ = NULL;
  stop_time = 0;

  scadds_env = getenv("scadds_addr");
  diffusion_port_ = DEFAULT_DIFFUSION_PORT;

#ifndef NS_DIFFUSION
  // Parse command line options
  while (1){
    opt = getopt(argc, argv, "f:hd:vt:p:");

    switch(opt){

    case 'p': 

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

      break;

    case 't':

      stop_time = atol(optarg);
      if (stop_time <= 0){
	DiffPrint(DEBUG_ALWAYS, "Diffusion Error: stop time must be > 0\n");
	exit(-1);
      }
      else{
	DiffPrint(DEBUG_ALWAYS, "%s will stop after %ld seconds\n",
		  PROGRAM, stop_time);
      }

      break;

    case 'h':

      usage();

      break;

    case 'v':

      DiffPrint(DEBUG_ALWAYS, "\n%s %s\n", PROGRAM, RELEASE);
      exit(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();
      }

      global_debug_level = debug_level;

      break;

    case 'f':

      if (!strncasecmp(optarg, "-", 1)){
	DiffPrint(DEBUG_ALWAYS, "Error: Parameter is missing !\n");
	usage();
      }

      config_file_ = strdup(optarg);

      break;

    case '?':

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

      break;

    case ':':

      DiffPrint(DEBUG_ALWAYS, "Parameter missing !\n");
      usage();

      break;
    }

    if (opt == -1)
      break;
  }

  if (!config_file_)
    config_file_ = strdup(DEFAULT_CONFIG_FILE);

  // Get diffusion ID
  if (scadds_env != NULL){
    my_id_ = atoi(scadds_env);
  }
  else{
    DiffPrint(DEBUG_ALWAYS,
	      "Diffusion : scadds_addr not set. Using random id.\n");

    // Generate random ID
    do{
      GetTime(&tv);
      SetSeed(&tv);
      my_id_ = GetRand();
    }
    while(my_id_ == LOCALHOST_ADDR || my_id_ == BROADCAST_ADDR);
  }
#else
  my_id_ = nodeid;
#endif // !NS_DIFFUSION

  // Initialize variables
  lon_ = 0.0;
  lat_ = 0.0;

#ifdef STATS
  stats_ = new DiffusionStats(my_id_);
#  ifndef WIRED
#     ifdef USE_RPC
  rpcstats_ = new RPCStats(my_id_);
#     endif // USE_RPC
#  endif // !WIRED
#endif // STATS

  GetTime(&tv);
  SetSeed(&tv);
  pkt_count_ = GetRand();
  random_id_ = GetRand();

  Tcl_InitHashTable(&htable_, 2);

  // Initialize EventQueue
  timers_manager_ = new TimerManager;

  // Create regular timers
  callback = new NeighborsTimeoutTimer(this);
  timers_manager_->addTimer(NEIGHBORS_DELAY, callback);

  callback = new FilterTimeoutTimer(this);
  timers_manager_->addTimer(FILTER_DELAY, callback);

  if (stop_time > 0){
    callback = new DiffusionStopTimer(this);
    timers_manager_->addTimer((stop_time * 1000), callback);
  }

  GetTime(&tv);

  // Print Initialization message
  DiffPrint(DEBUG_ALWAYS, "Diffusion : starting at time %ld:%ld\n",
	    tv.tv_sec, tv.tv_usec);
  DiffPrint(DEBUG_ALWAYS, "Diffusion : Node id = %d\n", my_id_);

  // Initialize diffusion io devices
#ifdef IO_LOG
  pseudo_io_device = new IOLog(my_id_);
  in_devices_.push_back(pseudo_io_device);
  out_devices_.push_back(pseudo_io_device);

  in_devices = &(pseudo_io_device->in_devices_);
  out_devices = &(pseudo_io_device->out_devices_);
  local_out_devices = &(local_out_devices_);
#else
  in_devices = &(in_devices_);
  out_devices = &(out_devices_);
  local_out_devices = &(local_out_devices_);
#endif // IO_LOG

#ifdef NS_DIFFUSION
  device = new LocalApp(diffrtg);
  local_out_devices->push_back(device);

  device = new LinkLayerAbs(diffrtg);
  out_devices->push_back(device);
#endif // NS_DIFFUSION

#ifdef UDP
  device = new UDPLocal(&diffusion_port_);
  in_devices->push_back(device);
  local_out_devices->push_back(device);

#ifdef WIRED
  device = new UDPWired(config_file_);
  out_devices->push_back(device);
#endif // WIRED
#endif // UDP

#ifdef USE_RPC
  device = new RPCIO();
  in_devices->push_back(device);
  out_devices->push_back(device);
#endif // USE_RPC

#ifdef USE_MOTE_NIC
  device = new MOTEIO();
  in_devices->push_back(device);
  out_devices->push_back(device);
#endif // USE_MOTE_NIC

#ifdef USE_WINSNG2
  device = new WINSNG2();
  in_devices->push_back(device);
  out_devices->push_back(device);
#endif // USE_WINSNG2
}

HashEntry * DiffusionCoreAgent::getHash(unsigned int pkt_num,
					 unsigned int rdm_id)
{
  unsigned int key[2];

  key[0] = pkt_num;
  key[1] = rdm_id;

  Tcl_HashEntry *entryPtr = Tcl_FindHashEntry(&htable_, (char *)key);

  if (entryPtr == NULL)
    return NULL;

  return (HashEntry *)Tcl_GetHashValue(entryPtr);
}

void DiffusionCoreAgent::putHash(unsigned int pkt_num,
				 unsigned int rdm_id)
{
  Tcl_HashEntry *tcl_hash_entry;
  HashEntry *hash_entry;
  HashList::iterator hash_itr;
  unsigned int key[2];
  int new_hash_key;

  if (hash_list_.size() == HASH_TABLE_MAX_SIZE){
    // Hash table reached maximum size

    for (int i = 0; ((i < HASH_TABLE_REMOVE_AT_ONCE)
		     && (hash_list_.size() > 0)); i++){
      hash_itr = hash_list_.begin();
      tcl_hash_entry = *hash_itr;
      hash_entry = (HashEntry *) Tcl_GetHashValue(tcl_hash_entry);
      delete hash_entry;
      hash_list_.erase(hash_itr);
      Tcl_DeleteHashEntry(tcl_hash_entry);
    }
  }

  key[0] = pkt_num;
  key[1] = rdm_id;

  tcl_hash_entry = Tcl_CreateHashEntry(&htable_, (char *)key, &new_hash_key);

  if (new_hash_key == 0){
    DiffPrint(DEBUG_IMPORTANT, "Key already exists in hash !\n");
    return;
  }

  hash_entry = new HashEntry;

  Tcl_SetHashValue(tcl_hash_entry, hash_entry);
  hash_list_.push_back(tcl_hash_entry);
}

#ifndef NS_DIFFUSION
void DiffusionCoreAgent::recvPacket(DiffPacket pkt)
{
  struct hdr_diff *dfh = HDR_DIFF(pkt);
  Message *rcv_message = NULL;
  int8_t version, msg_type;
  u_int16_t data_len, num_attr, source_port;
  int32_t rdm_id, pkt_num, next_hop, last_hop;   

  // Read header
  version = DIFF_VER(dfh);
  msg_type = MSG_TYPE(dfh);
  source_port = ntohs(SRC_PORT(dfh));
  pkt_num = ntohl(PKT_NUM(dfh));
  rdm_id = ntohl(RDM_ID(dfh));
  num_attr = ntohs(NUM_ATTR(dfh));
  next_hop = ntohl(NEXT_HOP(dfh));
  last_hop = ntohl(LAST_HOP(dfh));
  data_len = ntohs(DATA_LEN(dfh));

  // Packet is good, create a message
  rcv_message = new Message(version, msg_type, source_port, data_len,
			    num_attr, pkt_num, rdm_id, next_hop, last_hop);

  // Read all attributes into the Message structure
  rcv_message->msg_attr_vec_ = UnpackAttrs(pkt, num_attr);

  // Process the incoming message
  recvMessage(rcv_message);

  // Don't forget to message when we're done
  delete rcv_message;

  delete [] pkt;
}
#endif // !NS_DIFFUSION

void DiffusionCoreAgent::recvMessage(Message *msg)
{
  Tcl_HashEntry *tcl_hash_entry;
  unsigned int key[2];

  // Check version
  if (msg->version_ != DIFFUSION_VERSION)
    return;

  // Check for ID conflict
  if (msg->last_hop_ == my_id_){
    DiffPrint(DEBUG_ALWAYS, "Error: A diffusion ID conflict has been detected !\n");
    exit(-1);
  }

  // Address filtering
  if ((msg->next_hop_ != BROADCAST_ADDR) &&
      (msg->next_hop_ != LOCALHOST_ADDR) &&
      (msg->next_hop_ != my_id_))
    return;

  // Control Messages are unique and don't go to the hash
  if (msg->msg_type_ != CONTROL){
    // Hash table keeps info about packets
  
    key[0] = msg->pkt_num_;
    key[1] = msg->rdm_id_;
    tcl_hash_entry = Tcl_FindHashEntry(&htable_, (char *) key);

    if (tcl_hash_entry != NULL){
      DiffPrint(DEBUG_DETAILS, "Received old message !\n");
      msg->new_message_ = 0;
    }
    else{
      // Add message to the hash table
      putHash(key[0], key[1]);
      msg->new_message_ = 1;
    }
  }

#ifdef STATS
  stats_->logIncomingMessage(msg);
#endif // STATS

  // Check if it's a control of a regular message
  if (msg->msg_type_ == CONTROL)
    processControlMessage(msg);
  else
    processMessage(msg);
}

#ifndef NS_DIFFUSION
int main(int argc, char **argv)
{
  agent = new DiffusionCoreAgent(argc, argv);

  signal(SIGINT, signal_handler);

  agent->run();

  return 0;
}
#endif // !NS_DIFFUSION