开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Windows编程 > 通讯编程 > 查看源码
filter_core.cc
资源名称：ns-allinone-2.33.tar.gz　[点击查看]
上传用户：rrhhcc
上传日期：2015-12-11
资源大小：54129k
文件大小：36k
源码类别：

通讯编程

开发平台：

Visual C++

filter_core.cc：源码内容
  1. // 
  2. // filter_core.cc  : Main Diffusion program
  3. // authors         : Chalermek Intanagonwiwat and Fabio Silva
  4. //
  5. // Copyright (C) 2000-2003 by the University of Southern California
  6. // $Id: filter_core.cc,v 1.4 2005/09/13 04:53:47 tomh Exp $
  7. //
  8. // This program is free software; you can redistribute it and/or
  9. // modify it under the terms of the GNU General Public License,
  10. // version 2, as published by the Free Software Foundation.
  11. //
  12. // This program is distributed in the hope that it will be useful,
  13. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15. // GNU General Public License for more details.
  16. //
  17. // You should have received a copy of the GNU General Public License along
  18. // with this program; if not, write to the Free Software Foundation, Inc.,
  19. // 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
  20. //
  21. // Linking this file statically or dynamically with other modules is making
  22. // a combined work based on this file.  Thus, the terms and conditions of
  23. // the GNU General Public License cover the whole combination.
  24. //
  25. // In addition, as a special exception, the copyright holders of this file
  26. // give you permission to combine this file with free software programs or
  27. // libraries that are released under the GNU LGPL and with code included in
  28. // the standard release of ns-2 under the Apache 2.0 license or under
  29. // otherwise-compatible licenses with advertising requirements (or modified
  30. // versions of such code, with unchanged license).  You may copy and
  31. // distribute such a system following the terms of the GNU GPL for this
  32. // file and the licenses of the other code concerned, provided that you
  33. // include the source code of that other code when and as the GNU GPL
  34. // requires distribution of source code.
  35. //
  36. // Note that people who make modified versions of this file are not
  37. // obligated to grant this special exception for their modified versions;
  38. // it is their choice whether to do so.  The GNU General Public License
  39. // gives permission to release a modified version without this exception;
  40. // this exception also makes it possible to release a modified version
  41. // which carries forward this exception.
  43. #include "filter_core.hh"
  45. #ifndef NS_DIFFUSION
  46. DiffusionCoreAgent *agent;
  47. #endif // !NS_DIFFUSION
  49. class HashEntry {
  50. public:
  51.   bool dummy;
  53.   HashEntry() { 
  54.     dummy  = false;
  55.   }
  56. };
  58. class NeighborEntry {
  59. public:
  60.   int32_t id;
  61.   struct timeval tmv;
  63.   NeighborEntry(int _id) : id(_id)
  64.   {
  65.     GetTime(&tmv);
  66.   }
  67. };
  69. int NeighborsTimeoutTimer::expire()
  70. {
  71.   agent_->neighborsTimeout();
  73.   return 0;
  74. }
  76. int FilterTimeoutTimer::expire()
  77. {
  78.   agent_->filterTimeout();
  80.   return 0;
  81. }
  83. int DiffusionStopTimer::expire()
  84. {
  85.   agent_->timeToStop();
  86. #ifndef NS_DIFFUSION
  87.   exit(0);
  88. #endif // !NS_DIFFUSION
  90.   // Never gets here !
  91.   return 0;
  92. }
  94. void DiffusionCoreAgent::timeToStop()
  95. {
  96. #ifdef STATS
  97.   char out_filename[100];
  98.   FILE *outfile = NULL;
  100.   if (stats_){
  101.     sprintf(out_filename, "/tmp/diffusion-%d.out", my_id_);
  102.     outfile = fopen(out_filename, "w");
  104.     if (outfile == NULL){
  105.       DiffPrint(DEBUG_ALWAYS,
  106. "Diffusion Error: Cannot create %sn", out_filename);
  107.       return;
  108.     }
  110.     stats_->printStats(stdout);
  111.     if (outfile){
  112.       stats_->printStats(outfile);
  113.       fclose(outfile);
  114.     }
  115.   }
  116. #endif // STATS
  117. }
  119. #ifndef NS_DIFFUSION
  121. void signal_handler(int p)
  122. {
  123.   agent->timeToStop();
  124.   exit(0);
  125. }
  127. void DiffusionCoreAgent::usage(char *s)
  128. {
  130.   DiffPrint(DEBUG_ALWAYS, "Usage: %s [-d debug] [-f filename] [-t stoptime] [-v] [-h] [-p port]", s);
  131. #ifdef IO_LOG
  132.   DiffPrint(DEBUG_ALWAYS, " [-l]");
  133. #endif // IO_LOG
  134. #ifdef STATS
  135.   DiffPrint(DEBUG_ALWAYS, " [-s] [-i warm_up_time]");
  136. #endif // STATS
  137.   DiffPrint(DEBUG_ALWAYS, "nn");
  138.   DiffPrint(DEBUG_ALWAYS, "t-d - Sets debug level (0-10)n");
  139.   DiffPrint(DEBUG_ALWAYS, "t-t - Stops after stoptime secondsn");
  140.   DiffPrint(DEBUG_ALWAYS, "t-f - Uses filename as the config filen");
  141.   DiffPrint(DEBUG_ALWAYS, "t-v - Prints diffusion versionn");
  142.   DiffPrint(DEBUG_ALWAYS, "t-h - Prints this informationn");
  143.   DiffPrint(DEBUG_ALWAYS, "t-p - Sets diffusion port to portn");
  144. #ifdef IO_LOG
  145.   DiffPrint(DEBUG_ALWAYS, "t-l - Turns on i/o loggingn");
  146. #endif // IO_LOG
  147. #ifdef STATS
  148.   DiffPrint(DEBUG_ALWAYS, "t-s - Disables statisticsn");
  149.   DiffPrint(DEBUG_ALWAYS, "t-i - Ignores traffic from the first warm_up_time seconds for statsn");
  150. #endif // STATS
  152.   DiffPrint(DEBUG_ALWAYS, "n");
  154.   exit(0);
  155. }
  157. void DiffusionCoreAgent::run()
  158. {
  159.   DeviceList::iterator device_itr;
  160.   DiffPacket in_pkt;
  161.   fd_set fds;
  162.   bool flag;
  163.   int status, max_sock, fd;
  164.   struct timeval tv;
  166.   // Main Select Loop
  167.   while (1){
  169.     // Wait for incoming packets
  170.     FD_ZERO(&fds);
  171.     max_sock = 0;
  173.     // Figure out how much time to wait
  174.     timers_manager_->nextTimerTime(&tv);
  175.     if (tv.tv_sec == 0 && tv.tv_usec == 0){
  176.       // Timer has expired !
  177.       timers_manager_->executeAllExpiredTimers();
  178.       continue;
  179.     }
  181.     for (device_itr = in_devices_.begin();
  182.  device_itr != in_devices_.end(); ++device_itr){
  183.       (*device_itr)->addInFDS(&fds, &max_sock);
  184.     }
  186.     status = select(max_sock+1, &fds, NULL, NULL, &tv);
  188.     if (status == 0){
  189.       // We process all expired timers
  190.       timers_manager_->executeAllExpiredTimers();
  191.     }
  193.     // Check for new packets
  194.     if (status > 0){
  195.       do{
  196. flag = false;
  197. for (device_itr = in_devices_.begin();
  198.      device_itr != in_devices_.end(); ++device_itr){
  199.   fd = (*device_itr)->checkInFDS(&fds);
  200.   if (fd != -1){
  201.     // Message waiting
  202.     in_pkt = (*device_itr)->recvPacket(fd);
  204.     if (in_pkt)
  205.       recvPacket(in_pkt);
  207.     // Clear this fd
  208.     FD_CLR(fd, &fds);
  209.     status--;
  210.     flag = true;
  211.   }
  212. }
  213.       } while ((status > 0) && (flag == true));
  214.     }
  216.     // This should not happen
  217.     if (status < 0){
  218.       DiffPrint(DEBUG_IMPORTANT, "Select returned %dn", status);
  219.     }
  220.   }
  221. }
  222. #endif // !NS_DIFFUSION
  224. void DiffusionCoreAgent::neighborsTimeout()
  225. {
  226.   struct timeval tmv;
  227.   NeighborEntry *neighbor_entry;
  228.   NeighborList::iterator neighbor_itr;
  230.   DiffPrint(DEBUG_MORE_DETAILS, "Neighbors Timeout !n");
  232.   GetTime(&tmv);
  234.   neighbor_itr = neighbor_list_.begin();
  236.   while(neighbor_itr != neighbor_list_.end()){
  237.     neighbor_entry = *neighbor_itr;
  238.     if (tmv.tv_sec > neighbor_entry->tmv.tv_sec + NEIGHBORS_TIMEOUT){
  239.       // This neighbor expired
  240.       neighbor_itr = neighbor_list_.erase(neighbor_itr);
  241.       delete neighbor_entry;
  242.     }
  243.     else{
  244.       neighbor_itr++;
  245.     }
  246.   }
  247. }
  249. void DiffusionCoreAgent::filterTimeout()
  250. {
  251.   struct timeval tmv;
  252.   FilterEntry *filter_entry;
  253.   FilterList::iterator filter_itr;
  255.   DiffPrint(DEBUG_MORE_DETAILS, "Filter Timeout !n");
  257.   GetTime(&tmv);
  259.   filter_itr = filter_list_.begin();
  261.   while(filter_itr != filter_list_.end()){
  262.     filter_entry = *filter_itr;
  263.     if (tmv.tv_sec > filter_entry->tmv_.tv_sec + FILTER_TIMEOUT){
  265.       // This filter expired
  266.       DiffPrint(DEBUG_NO_DETAILS, "Filter %d, %d, %d timed out !n",
  267. filter_entry->agent_, filter_entry->handle_,
  268. filter_entry->priority_);
  269.       filter_itr = filter_list_.erase(filter_itr);
  270.       delete filter_entry;
  271.     }
  272.     else{
  273.       filter_itr++;
  274.     }
  275.   }
  276. }
  278. void DiffusionCoreAgent::sendMessage(Message *msg)
  279. {
  280.   Tcl_HashEntry *tcl_hash_entry;
  281.   unsigned int key[2];
  282.   Message *send_message;
  283.   
  284.   send_message = new Message(DIFFUSION_VERSION, msg->msg_type_, diffusion_port_,
  285.      0, 0, msg->pkt_num_, msg->rdm_id_,
  286.      msg->next_hop_, 0);
  288.   send_message->msg_attr_vec_ = CopyAttrs(msg->msg_attr_vec_);
  289.   send_message->num_attr_ = send_message->msg_attr_vec_->size();
  290.   send_message->data_len_ = CalculateSize(send_message->msg_attr_vec_);
  292.   // Adjust message size for logging and check hash
  293.   key[0] = msg->pkt_num_;
  294.   key[1] = msg->rdm_id_;
  295.   tcl_hash_entry = Tcl_FindHashEntry(&htable_, (char *) key);
  296.   if (tcl_hash_entry)
  297.     msg->new_message_ = 0;
  298.   else
  299.     msg->new_message_ = 1;
  301.   send_message->new_message_ = msg->new_message_;
  303.   // Check if message goes to an agent or the network
  304.   if (msg->next_port_){
  305.     // Message goes to an agent
  306.     send_message->last_hop_ = LOCALHOST_ADDR;
  308.     // If it's a local message, it has to go to a local agent
  309.     if (send_message->next_hop_ != LOCALHOST_ADDR){
  310.       DiffPrint(DEBUG_ALWAYS, "Error: Message destination is a local agent but next_hop != LOCALHOST_ADDR !n");
  311.       delete send_message;
  312.       return;
  313.     }
  315.     // Send the message to the agent specified
  316.     sendMessageToLibrary(send_message, msg->next_port_);
  317.   }
  318.   else{
  319.     // Message goes to the network
  320.     send_message->last_hop_ = my_id_;
  322. #ifdef STATS
  323.     if (stats_)
  324.       stats_->logOutgoingMessage(send_message);
  325. #endif // STATS
  327.     // Add message to the hash table      
  328.     if (tcl_hash_entry == NULL)
  329.       putHash(key[0], key[1]);
  330.     else
  331.       DiffPrint(DEBUG_DETAILS, "Node%d: Message being sent is an old message !n", my_id_);
  333.     // Send Message
  334.     sendMessageToNetwork(send_message);
  335.   }
  337.   delete send_message;
  338. }
  340. void DiffusionCoreAgent::forwardMessage(Message *msg, FilterEntry *filter_entry)
  341. {
  342.   RedirectMessage *original_hdr;
  343.   NRAttribute *original_header_attr;
  344.   Message *send_message;
  346.   // Create an attribute with the original header
  347.   original_hdr = new RedirectMessage(msg->new_message_, msg->msg_type_,
  348.      msg->source_port_, msg->data_len_,
  349.      msg->num_attr_, msg->rdm_id_,
  350.      msg->pkt_num_, msg->next_hop_,
  351.      msg->last_hop_, filter_entry->handle_,
  352.      msg->next_port_);
  354.   original_header_attr = OriginalHdrAttr.make(NRAttribute::IS,
  355.       (void *)original_hdr,
  356.       sizeof(RedirectMessage));
  358.   send_message = new Message(DIFFUSION_VERSION, REDIRECT, diffusion_port_, 0,
  359.      0, pkt_count_, random_id_, LOCALHOST_ADDR, my_id_);
  361.   // Increment pkt_counter
  362.   pkt_count_++;
  364.   // Duplicate the message's attributes
  365.   send_message->msg_attr_vec_ = CopyAttrs(msg->msg_attr_vec_);
  366.   
  367.   // Add the extra attribute
  368.   send_message->msg_attr_vec_->push_back(original_header_attr);
  369.   send_message->num_attr_ = send_message->msg_attr_vec_->size();
  370.   send_message->data_len_ = CalculateSize(send_message->msg_attr_vec_);
  372.   sendMessageToLibrary(send_message, filter_entry->agent_);
  374.   delete send_message;
  375.   delete original_hdr;
  376. }
  378. #ifndef NS_DIFFUSION
  379. void DiffusionCoreAgent::sendMessageToLibrary(Message *msg, u_int16_t agent_id)
  380. {
  381.   DiffPacket out_pkt = NULL;
  382.   struct hdr_diff *dfh;
  383.   int len;
  384.   char *pos;
  386.   out_pkt = AllocateBuffer(msg->msg_attr_vec_);
  387.   dfh = HDR_DIFF(out_pkt);
  389.   pos = (char *) out_pkt;
  390.   pos = pos + sizeof(struct hdr_diff);
  392.   len = PackAttrs(msg->msg_attr_vec_, pos);
  394.   LAST_HOP(dfh) = htonl(msg->last_hop_);
  395.   NEXT_HOP(dfh) = htonl(msg->next_hop_);
  396.   DIFF_VER(dfh) = msg->version_;
  397.   MSG_TYPE(dfh) = msg->msg_type_;
  398.   DATA_LEN(dfh) = htons(len);
  399.   PKT_NUM(dfh) = htonl(msg->pkt_num_);
  400.   RDM_ID(dfh) = htonl(msg->rdm_id_);
  401.   NUM_ATTR(dfh) = htons(msg->num_attr_);
  402.   SRC_PORT(dfh) = htons(msg->source_port_);
  404.   sendPacketToLibrary(out_pkt, sizeof(struct hdr_diff) + len, agent_id);
  406.   delete [] out_pkt;
  407. }
  408. #else
  409. void DiffusionCoreAgent::sendMessageToLibrary(Message *msg, u_int16_t agent_id)
  410. {
  411.   Message *send_message;
  412.   DeviceList::iterator device_itr;
  413.   int len;
  415.   send_message = CopyMessage(msg);
  416.   len = CalculateSize(send_message->msg_attr_vec_);
  417.   len = len + sizeof(struct hdr_diff);
  419.   for (device_itr = local_out_devices_.begin();
  420.        device_itr != local_out_devices_.end(); ++device_itr){
  421.     (*device_itr)->sendPacket((DiffPacket) send_message, len, agent_id);
  422.   }
  423. }
  424. #endif // !NS_DIFFUSION
  426. #ifndef NS_DIFFUSION
  427. void DiffusionCoreAgent::sendMessageToNetwork(Message *msg)
  428. {
  429.   DiffPacket out_pkt = NULL;
  430.   struct hdr_diff *dfh;
  431.   int len;
  432.   char *pos;
  434.   out_pkt = AllocateBuffer(msg->msg_attr_vec_);
  435.   dfh = HDR_DIFF(out_pkt);
  437.   pos = (char *) out_pkt;
  438.   pos = pos + sizeof(struct hdr_diff);
  440.   len = PackAttrs(msg->msg_attr_vec_, pos);
  442.   LAST_HOP(dfh) = htonl(msg->last_hop_);
  443.   NEXT_HOP(dfh) = htonl(msg->next_hop_);
  444.   DIFF_VER(dfh) = msg->version_;
  445.   MSG_TYPE(dfh) = msg->msg_type_;
  446.   DATA_LEN(dfh) = htons(len);
  447.   PKT_NUM(dfh) = htonl(msg->pkt_num_);
  448.   RDM_ID(dfh) = htonl(msg->rdm_id_);
  449.   NUM_ATTR(dfh) = htons(msg->num_attr_);
  450.   SRC_PORT(dfh) = htons(msg->source_port_);
  452.   sendPacketToNetwork(out_pkt, sizeof(struct hdr_diff) + len, msg->next_hop_);
  454.   delete [] out_pkt;
  455. }
  456. #else
  457. void DiffusionCoreAgent::sendMessageToNetwork(Message *msg)
  458. {
  459.   Message *send_message;
  460.   int len;
  461.   int32_t dst;
  462.   DeviceList::iterator device_itr;
  464.   send_message = CopyMessage(msg);
  465.   len = CalculateSize(send_message->msg_attr_vec_);
  466.   len = len + sizeof(struct hdr_diff);
  467.   dst = send_message->next_hop_;
  469.   for (device_itr = out_devices_.begin();
  470.        device_itr != out_devices_.end(); ++device_itr){
  471.     (*device_itr)->sendPacket((DiffPacket) send_message, len, dst);
  472.   }
  473. }
  474. #endif // !NS_DIFFUSION
  476. void DiffusionCoreAgent::sendPacketToLibrary(DiffPacket pkt, int len,
  477.      u_int16_t dst)
  478. {
  479.   DeviceList::iterator device_itr;
  481.   for (device_itr = local_out_devices_.begin();
  482.        device_itr != local_out_devices_.end(); ++device_itr){
  483.     (*device_itr)->sendPacket(pkt, len, dst);
  484.   }
  485. }
  487. void DiffusionCoreAgent::sendPacketToNetwork(DiffPacket pkt, int len, int dst)
  488. {
  489.   DeviceList::iterator device_itr;
  491.   for (device_itr = out_devices_.begin();
  492.        device_itr != out_devices_.end(); ++device_itr){
  493.     (*device_itr)->sendPacket(pkt, len, dst);
  494.   }
  495. }
  497. void DiffusionCoreAgent::updateNeighbors(int id)
  498. {
  499.   NeighborList::iterator neighbor_itr;
  500.   NeighborEntry *neighbor_entry;
  502.   if (id == LOCALHOST_ADDR || id == my_id_)
  503.     return;
  505.   for (neighbor_itr = neighbor_list_.begin();
  506.        neighbor_itr != neighbor_list_.end(); ++neighbor_itr){
  507.     if ((*neighbor_itr)->id == id)
  508.       break;
  509.   }
  511.   if (neighbor_itr == neighbor_list_.end()){
  512.     // This is a new neighbor
  513.     neighbor_entry = new NeighborEntry(id);
  514.     neighbor_list_.push_front(neighbor_entry);
  515.   }
  516.   else{
  517.     // Just update the neighbor timeout
  518.     GetTime(&((*neighbor_itr)->tmv));
  519.   }
  520. }
  522. FilterEntry * DiffusionCoreAgent::findFilter(int16_t handle, u_int16_t agent)
  523. {
  524.   FilterList::iterator filter_itr;
  525.   FilterEntry *filter_entry;
  527.   for (filter_itr = filter_list_.begin();
  528.        filter_itr != filter_list_.end(); ++filter_itr){
  529.     filter_entry = *filter_itr;
  530.     if (handle != filter_entry->handle_ || agent != filter_entry->agent_)
  531.       continue;
  533.     // Found
  534.     return filter_entry;
  535.   }
  536.   return NULL;
  537. }
  539. FilterEntry * DiffusionCoreAgent::deleteFilter(int16_t handle, u_int16_t agent)
  540. {
  541.   FilterList::iterator filter_itr = filter_list_.begin();
  542.   FilterEntry *filter_entry = NULL;
  544.   while (filter_itr != filter_list_.end()){
  545.     filter_entry = *filter_itr;
  546.     if (handle == filter_entry->handle_ && agent == filter_entry->agent_){
  547.       filter_list_.erase(filter_itr);
  548.       break;
  549.     }
  550.     filter_entry = NULL;
  551.     filter_itr++;
  552.   }
  553.   return filter_entry;
  554. }
  556. bool DiffusionCoreAgent::addFilter(NRAttrVec *attrs, u_int16_t agent,
  557.    int16_t handle, u_int16_t priority)
  558. {
  559.   FilterList::iterator filter_itr;
  560.   FilterEntry *filter_entry;
  562.   filter_itr = filter_list_.begin();
  563.   while (filter_itr != filter_list_.end()){
  564.     filter_entry = *filter_itr;
  565.     if (filter_entry->priority_ == priority)
  566.       return false;
  567.     filter_itr++;
  568.   }
  570.   filter_entry = new FilterEntry(handle, priority, agent);
  572.   // Copy the Attribute Vector
  573.   filter_entry->filter_attrs_ = CopyAttrs(attrs);
  575.   // Add this filter to the filter list
  576.   filter_list_.push_back(filter_entry);
  578.   return true;
  579. }
  581. FilterList::iterator DiffusionCoreAgent::findMatchingFilter(NRAttrVec *attrs,
  582.     FilterList::iterator filter_itr)
  583. {
  584.   FilterEntry *filter_entry;
  586.   for (;filter_itr != filter_list_.end(); ++filter_itr){
  587.     filter_entry = *filter_itr;
  589.     if (OneWayMatch(filter_entry->filter_attrs_, attrs)){
  590.       // That's a match !
  591.       break;
  592.     }
  593.   }
  594.   return filter_itr;
  595. }
  597. bool DiffusionCoreAgent::restoreOriginalHeader(Message *msg)
  598. {
  599.   NRAttrVec::iterator attr_itr = msg->msg_attr_vec_->begin();
  600.   NRSimpleAttribute<void *> *original_header_attr = NULL;
  601.   RedirectMessage *original_hdr;
  603.   // Find original Header
  604.   original_header_attr = OriginalHdrAttr.find_from(msg->msg_attr_vec_,
  605.    attr_itr, &attr_itr);
  606.   if (!original_header_attr){
  607.     DiffPrint(DEBUG_ALWAYS, "Error: DiffusionCoreAgent::ProcessControlMessage couldn't find the OriginalHdrAttr !n");
  608.     return false;
  609.   }
  611.   // Restore original Header
  612.   original_hdr = (RedirectMessage *) original_header_attr->getVal();
  614.   msg->msg_type_ = original_hdr->msg_type_;
  615.   msg->source_port_ = original_hdr->source_port_;
  616.   msg->pkt_num_ = original_hdr->pkt_num_;
  617.   msg->rdm_id_ = original_hdr->rdm_id_;
  618.   msg->next_hop_ = original_hdr->next_hop_;
  619.   msg->last_hop_ = original_hdr->last_hop_;
  620.   msg->new_message_ = original_hdr->new_message_;
  621.   msg->num_attr_ = original_hdr->num_attr_;
  622.   msg->data_len_ = original_hdr->data_len_;
  623.   msg->next_port_ = original_hdr->next_port_;
  625.   // Delete attribute from original set
  626.   msg->msg_attr_vec_->erase(attr_itr);
  627.   delete original_header_attr;
  629.   return true;
  630. }
  632. FilterList * DiffusionCoreAgent::getFilterList(NRAttrVec *attrs)
  633. {
  634.   FilterList *matching_filter_list = new FilterList;
  635.   FilterList::iterator known_filters_itr, filter_list_itr;
  636.   FilterEntry *matching_filter_entry, *filter_entry;
  638.   // We need to come up with a list of filters to call
  639.   // F1 will be called before F2 if F1->priority > F2->priority
  641.   known_filters_itr = findMatchingFilter(attrs, filter_list_.begin());
  643.   while (known_filters_itr != filter_list_.end()){
  644.     // We have a match !
  645.     matching_filter_entry = *known_filters_itr;
  647.     for (filter_list_itr = matching_filter_list->begin();
  648.  filter_list_itr != matching_filter_list->end(); ++filter_list_itr){
  649.       filter_entry = *filter_list_itr;
  651.       // Figure out where to insert 
  652.       if (matching_filter_entry->priority_ > filter_entry->priority_)
  653. break;
  654.     }
  656.     // Insert matching filter in the list
  657.     matching_filter_list->insert(filter_list_itr, matching_filter_entry);
  659.     // Continue the search
  660.     known_filters_itr++;
  661.     known_filters_itr = findMatchingFilter(attrs, known_filters_itr);
  662.   }
  663.   return matching_filter_list;
  664. }
  666. u_int16_t DiffusionCoreAgent::getNextFilterPriority(int16_t handle,
  667.     u_int16_t priority,
  668.     u_int16_t agent)
  669. {
  670.   FilterList::iterator filter_itr;
  671.   FilterEntry *filter_entry;
  673.   if ((priority < FILTER_MIN_PRIORITY) ||
  674.       (priority > FILTER_KEEP_PRIORITY))
  675.     return FILTER_INVALID_PRIORITY;
  677.   if (priority < FILTER_KEEP_PRIORITY)
  678.     return (priority - 1);
  680.   filter_itr = filter_list_.begin();
  682.   while (filter_itr != filter_list_.end()){
  683.     filter_entry = *filter_itr;
  685.     if ((filter_entry->handle_ == handle) && (filter_entry->agent_ == agent)){
  686.       // Found this filter
  687.       return (filter_entry->priority_ - 1);
  688.     }
  690.     filter_itr++;
  691.   }
  693.   return FILTER_INVALID_PRIORITY;
  694. }
  696. void DiffusionCoreAgent::processMessage(Message *msg)
  697. {
  698.   FilterList *filter_list;
  699.   FilterList::iterator filter_list_itr;
  700.   FilterEntry *filter_entry;
  702.   filter_list = getFilterList(msg->msg_attr_vec_);
  704.   // Ok, we have a list of Filters to call. Send this message
  705.   // to the first filter on this list
  706.   if (filter_list->size() > 0){
  707.     filter_list_itr = filter_list->begin();
  708.     filter_entry = *filter_list_itr;
  710.     forwardMessage(msg, filter_entry);
  711.     filter_list->clear();
  712.   }
  713.   delete filter_list;
  714. }
  716. void DiffusionCoreAgent::processControlMessage(Message *msg)
  717. {
  718.   NRSimpleAttribute<void *> *ctrl_msg_attr = NULL;
  719.   NRAttrVec::iterator attr_itr;
  720.   ControlMessage *control_blob = NULL;
  721.   FilterList *filter_list;
  722.   FilterList::iterator filter_list_itr;
  723.   FilterEntry *filter_entry;
  724.   int command, param1, param2;
  725.   u_int16_t priority, source_port, new_priority;
  726.   int16_t handle;
  727.   bool filter_is_last = false;
  729.   // Control messages should not come from other nodes
  730.   if (msg->last_hop_ != LOCALHOST_ADDR){
  731.     DiffPrint(DEBUG_ALWAYS,
  732.       "Error: Received control message from another node !n");
  733.     return;
  734.   }
  736.   // Find the control attribute
  737.   attr_itr = msg->msg_attr_vec_->begin();
  738.   ctrl_msg_attr = ControlMsgAttr.find_from(msg->msg_attr_vec_,
  739.    attr_itr, &attr_itr);
  741.   if (!ctrl_msg_attr){
  742.     // Control message is invalid
  743.     DiffPrint(DEBUG_ALWAYS, "Error: Control message received is invalid !n");
  744.     return;
  745.   }
  747.   // Extract the control message info
  748.   control_blob = (ControlMessage *) ctrl_msg_attr->getVal();
  749.   command = control_blob->command_;
  750.   param1 = control_blob->param1_;
  751.   param2 = control_blob->param2_;
  753.   // Filter API definitions
  754.   //
  755.   // command = ADD_UPDATE_FILTER
  756.   // param1  = priority
  757.   // param2  = handle
  758.   // attrs   = other attrs specify the filter
  759.   // 
  760.   // Remarks: If this filter is already present for this module,
  761.   //          we don't create a new one. A filter is identified
  762.   //          by the handle and the originating agent. The filter
  763.   //          gets refreshed if it already exists. If attrs and
  764.   //          handle are the same, we update the priority.
  765.   //
  766.   //
  767.   // command = REMOVE_FILTER
  768.   // param1  = handle
  769.   //
  770.   // Remarks: Remove the filter identified by (agent, handle)
  771.   //          If it's not found, a warning message is generated.
  772.   //
  773.   //
  774.   // Remarks: Send message from a local App to another App or
  775.   //          a neighbor. If agent_id is zero, the packet goes
  776.   //          out to the network. Otherwise, it goes to the
  777.   //          agent_id located on this node.
  778.   //
  779.   //
  780.   // command = SEND_MESSAGE
  781.   // param1  = handle
  782.   // param2  = priority
  783.   //
  784.   // Remarks: Send this message to the next filter or to a local
  785.   //          application. We have to assemble the list again
  786.   //          and figure out the current agent's position on the
  787.   //          list. Then, we send to the next guy. If there is
  788.   //          no other filter in the list, we try to send it to
  789.   //          the network, if next_hop contains a node address.
  791.   logControlMessage(msg, command, param1, param2);
  793.   // First we remove the control attribute from the message
  794.   msg->msg_attr_vec_->erase(attr_itr);
  795.   delete ctrl_msg_attr;
  797.   switch(command){
  798.   case ADD_UPDATE_FILTER:
  800.     priority = param1;
  801.     handle = param2;
  803.     filter_entry = findFilter(handle, msg->source_port_);
  805.     if (filter_entry){
  806.       // Filter already present, must be an update message
  807.       if (PerfectMatch(filter_entry->filter_attrs_, msg->msg_attr_vec_)){
  808. // Attrs also match, let's update the filter's timeout
  809. GetTime(&(filter_entry->tmv_));
  811. // Check if the priority has changed...
  812. if (priority == filter_entry->priority_){
  813.   // Nothing to do !
  814.   DiffPrint(DEBUG_SOME_DETAILS, "Filter %d, %d, %d refreshed.n",
  815.     filter_entry->agent_, filter_entry->handle_,
  816.     filter_entry->priority_);
  817. }
  818. else{
  819.   // Update the priority
  820.   DiffPrint(DEBUG_NO_DETAILS,
  821.     "Updated priority of filter %d, %d, %d to %dn",
  822.     msg->source_port_, handle, filter_entry->priority_, priority);
  823.   filter_entry->priority_ = priority;
  824. }
  826. break;
  827.       }
  828.       else{
  829. // Filter attributes have changed ! This is not allowed !
  830. DiffPrint(DEBUG_ALWAYS,
  831.   "Filter attributes cannot change during an update !n");
  832. break;
  833.       }
  834.     }
  835.     else{
  836.       // This is a new filter
  837.       if (!addFilter(msg->msg_attr_vec_, msg->source_port_, handle, priority)){
  838. DiffPrint(DEBUG_ALWAYS, "Failed to add filter %d, %d, %dn",
  839.   msg->source_port_, handle, priority);
  840.       }
  841.       else{
  842. DiffPrint(DEBUG_NO_DETAILS, "Adding filter %d, %d, %dn",
  843.   msg->source_port_, handle, priority);
  844.       }
  845.     }
  847.     break;
  849.   case REMOVE_FILTER:
  851.     handle = param1;
  852.     filter_entry = deleteFilter(handle, msg->source_port_);
  853.     if (filter_entry){
  854.       // Filter deleted
  855.       DiffPrint(DEBUG_NO_DETAILS, "Filter %d, %d, %d deleted.n",
  856. filter_entry->agent_, filter_entry->handle_,
  857. filter_entry->priority_);
  859.       delete filter_entry;
  860.     }
  861.     else{
  862.       DiffPrint(DEBUG_ALWAYS, "Couldn't find filter to delete !n");
  863.     }
  865.     break;
  867.   case SEND_MESSAGE:
  869.     handle = param1;
  870.     priority = param2;
  871.     source_port = msg->source_port_;
  873.     if (!restoreOriginalHeader(msg))
  874.       break;
  876.     new_priority = getNextFilterPriority(handle, priority, source_port);
  878.     if (new_priority == FILTER_INVALID_PRIORITY)
  879.       break;
  881.     // Now process the incoming message
  882.     filter_list = getFilterList(msg->msg_attr_vec_);
  884.     // Find the filter after the 'current' filter on the list
  885.     if (filter_list->size() > 0){
  886.       for (filter_list_itr = filter_list->begin();
  887.    filter_list_itr != filter_list->end(); ++filter_list_itr){
  888. filter_entry = *filter_list_itr;
  889. if (filter_entry->priority_ <= new_priority){
  890.   forwardMessage(msg, filter_entry);
  891.   break;
  892. }
  893.       }
  895.       if (filter_list_itr == filter_list->end())
  896. filter_is_last = true;
  898.     }
  899.     else{
  900.       filter_is_last = true;
  901.     }
  903.     if (filter_is_last){
  904.       // Forward message to the network or the destination application
  905.       sendMessage(msg);
  906.     }
  908.     filter_list->clear();
  910.     delete filter_list;
  912.     break;
  914.   case ADD_TO_BLACKLIST:
  916.     DiffPrint(DEBUG_IMPORTANT, "Diffusion: Adding node %d to blacklist !n",
  917.       param1);
  918.     black_list_.push_front(param1);
  920.     break;
  922.   case CLEAR_BLACKLIST:
  924.     DiffPrint(DEBUG_IMPORTANT, "Diffusion: Clearing blacklist !n");
  926.     black_list_.clear();
  928.     break;
  930.   default:
  932.     DiffPrint(DEBUG_ALWAYS, "Error: Unknown control message received !n");
  934.     break;
  935.   }
  936. }
  938. void DiffusionCoreAgent::logControlMessage(Message *msg, int command,
  939.    int param1, int param2)
  940. {
  941.   // Logs the incoming message
  942. }
  944. #ifdef NS_DIFFUSION
  945. DiffusionCoreAgent::DiffusionCoreAgent(DiffRoutingAgent *diffrtg, int nodeid)
  946. {
  947. #else
  948. DiffusionCoreAgent::DiffusionCoreAgent(int argc, char **argv)
  949. {
  950.   int opt;
  951.   int debug_level;
  952. #endif // NS_DIFFUSION
  953.   DeviceList *in_devices, *out_devices, *local_out_devices;
  954.   DiffusionIO *device;
  955.   TimerCallback *callback;
  956.   char *node_id_env;
  957. #ifdef USE_EMSIM
  958.   char *sim_id = getenv("SIM_ID");
  959.   char *sim_group = getenv("SIM_GROUP");
  960.   int32_t group_id;
  961. #endif // USE_EMSIM
  962.   long stop_time;
  963.   struct timeval tv;
  964. #ifdef IO_LOG
  965.   IOLog *pseudo_io_device;
  966.   bool use_io_log = false;
  967. #endif // IO_LOG
  968. #ifdef STATS
  969.   bool use_io_stats = true;
  970.   int stats_warm_up_time = 0;
  971. #endif // STATS
  972.   //bool node_id_configured = false;
  974.   opterr = 0;
  975.   config_file_ = NULL;
  976.   stop_time = 0;
  978.   node_id_env = getenv("node_addr");
  979.   diffusion_port_ = DEFAULT_DIFFUSION_PORT;
  981. #ifndef NS_DIFFUSION
  982.   // Parse command line options
  983.   while (1){
  984.     opt = getopt(argc, argv, COMMAND_LINE_ARGS);
  986.     switch(opt){
  988.     case 'p':
  990.       diffusion_port_ = (u_int16_t) atoi(optarg);
  991.       if ((diffusion_port_ < 1024) || (diffusion_port_ >= 65535)){
  992. DiffPrint(DEBUG_ALWAYS,
  993.   "Diffusion Error: Port must be between 1024 and 65535 !n");
  994. exit(-1);
  995.       }
  997.       break;
  999.     case 't':
  1001.       stop_time = atol(optarg);
  1002.       if (stop_time <= 0){
  1003. DiffPrint(DEBUG_ALWAYS, "Diffusion Error: stop time must be > 0n");
  1004. exit(-1);
  1005.       }
  1006.       else{
  1007. DiffPrint(DEBUG_ALWAYS, "%s will stop after %ld secondsn",
  1008.   PROGRAM, stop_time);
  1009.       }
  1011.       break;
  1013. #ifdef IO_LOG
  1014.     case 'l':
  1016.       use_io_log = true;
  1018.       break;
  1020. #endif // IO_LOG
  1022. #ifdef STATS
  1023.     case 's':
  1025.       use_io_stats = false;
  1027.       break;
  1029.     case 'i':
  1031.       stats_warm_up_time = atoi(optarg);
  1032.       if (stats_warm_up_time < 0){
  1033. DiffPrint(DEBUG_ALWAYS, "Diffusion Error: warm_up_time must be > 0n");
  1034. exit(-1);
  1035.       }
  1037.       break;
  1038. #endif // STATS
  1040.     case 'h':
  1042.       usage(argv[0]);
  1044.       break;
  1046.     case 'v':
  1048.       DiffPrint(DEBUG_ALWAYS, "n%s %sn", PROGRAM, RELEASE);
  1049.       exit(0);
  1051.       break;
  1053.     case 'd':
  1055.       debug_level = atoi(optarg);
  1057.       if (debug_level < 1 || debug_level > 10){
  1058. DiffPrint(DEBUG_ALWAYS,
  1059.   "Error: Debug level outside range or missing !n");
  1060. usage(argv[0]);
  1061.       }
  1063.       global_debug_level = debug_level;
  1065.       break;
  1067.     case 'f':
  1069.       if (!strncasecmp(optarg, "-", 1)){
  1070. DiffPrint(DEBUG_ALWAYS, "Error: Parameter is missing !n");
  1071. usage(argv[0]);
  1072.       }
  1074.       config_file_ = strdup(optarg);
  1076.       break;
  1078.     case '?':
  1080.       DiffPrint(DEBUG_ALWAYS,
  1081. "Error: %c isn't a valid option or its parameter is missing !n",
  1082. optopt);
  1083.       usage(argv[0]);
  1085.       break;
  1087.     case ':':
  1089.       DiffPrint(DEBUG_ALWAYS, "Parameter missing !n");
  1090.       usage(argv[0]);
  1092.       break;
  1093.     }
  1095.     if (opt == -1)
  1096.       break;
  1097.   }
  1099.   if (!config_file_)
  1100.     config_file_ = strdup(DEFAULT_CONFIG_FILE);
  1102.   // Get diffusion ID
  1103.   if (!node_id_configured){
  1104.     // Try to get id from environment variable
  1105.     if (node_id_env != NULL){
  1106.       my_id_ = atoi(node_id_env);
  1107.       node_id_configured = true;
  1108.     }
  1109.   }
  1111. #ifdef USE_EMSIM
  1112.   if (!node_id_configured){
  1113.     // Try to read groups and node id from emsim environment variables
  1114.     if (sim_id && sim_group){
  1115.       my_id_ = atoi(sim_id);
  1116.       group_id = atoi(sim_group);
  1117.       diffusion_port_ = diffusion_port_ + my_id_ + (100 * group_id);
  1118.       node_id_configured = true;
  1119.     }
  1120.   }
  1121. #endif // USE_EMSIM
  1123.   // Use random node id if user has not specified it
  1124.   if (!node_id_configured){
  1125.     DiffPrint(DEBUG_ALWAYS, "Diffusion : node_addr not set. Using random id.n");
  1127.     // Generate random ID
  1128.     do{
  1129.       GetTime(&tv);
  1130.       SetSeed(&tv);
  1131.       my_id_ = GetRand();
  1132.     }
  1133.     while(my_id_ == LOCALHOST_ADDR || my_id_ == BROADCAST_ADDR);
  1134.   }
  1136. #else
  1137.   my_id_ = nodeid;
  1138. #endif // !NS_DIFFUSION
  1140.   // Initialize variables
  1141.   lon_ = 0.0;
  1142.   lat_ = 0.0;
  1144. #ifdef STATS
  1145.   if (use_io_stats)
  1146.     stats_ = new DiffusionStats(my_id_, stats_warm_up_time);
  1147.   else
  1148.     stats_ = NULL;
  1149. #endif // STATS
  1151.   GetTime(&tv);
  1152.   SetSeed(&tv);
  1153.   pkt_count_ = GetRand();
  1154.   random_id_ = GetRand();
  1156.   Tcl_InitHashTable(&htable_, 2);
  1158.   // Initialize EventQueue
  1159.   timers_manager_ = new TimerManager;
  1161.   // Create regular timers
  1162.   callback = new NeighborsTimeoutTimer(this);
  1163.   timers_manager_->addTimer(NEIGHBORS_DELAY, callback);
  1165.   callback = new FilterTimeoutTimer(this);
  1166.   timers_manager_->addTimer(FILTER_DELAY, callback);
  1168.   if (stop_time > 0){
  1169.     callback = new DiffusionStopTimer(this);
  1170.     timers_manager_->addTimer((stop_time * 1000), callback);
  1171.   }
  1173.   GetTime(&tv);
  1175.   // Print Initialization message
  1176.   DiffPrint(DEBUG_ALWAYS, "Diffusion : starting at time %ld:%ldn",
  1177.     tv.tv_sec, tv.tv_usec);
  1178.   DiffPrint(DEBUG_ALWAYS, "Diffusion : Node id = %dn", my_id_);
  1180.   // Initialize diffusion io devices
  1181. #ifdef IO_LOG
  1182.   if (use_io_log){
  1183.     pseudo_io_device = new IOLog(my_id_);
  1184.     in_devices_.push_back(pseudo_io_device);
  1185.     out_devices_.push_back(pseudo_io_device);
  1187.     in_devices = &(pseudo_io_device->in_devices_);
  1188.     out_devices = &(pseudo_io_device->out_devices_);
  1189.     local_out_devices = &(local_out_devices_);
  1190.   }
  1191.   else{
  1192.     in_devices = &(in_devices_);
  1193.     out_devices = &(out_devices_);
  1194.     local_out_devices = &(local_out_devices_);
  1195.   }
  1196. #else
  1197.   in_devices = &(in_devices_);
  1198.   out_devices = &(out_devices_);
  1199.   local_out_devices = &(local_out_devices_);
  1200. #endif // IO_LOG
  1202. #ifdef NS_DIFFUSION
  1203.   device = new LocalApp(diffrtg);
  1204.   local_out_devices->push_back(device);
  1206.   device = new LinkLayerAbs(diffrtg);
  1207.   out_devices->push_back(device);
  1208. #endif // NS_DIFFUSION
  1210. #ifdef UDP
  1211.   device = new UDPLocal(&diffusion_port_);
  1212.   in_devices->push_back(device);
  1213.   local_out_devices->push_back(device);
  1215. #ifdef WIRED
  1216.   device = new UDPWired(config_file_);
  1217.   out_devices->push_back(device);
  1218. #endif // WIRED
  1219. #endif // UDP
  1221. #ifdef USE_RPC
  1222.   device = new RPCIO();
  1223.   in_devices->push_back(device);
  1224.   out_devices->push_back(device);
  1225. #endif // USE_RPC
  1227. #ifdef USE_MOTE_NIC
  1228.   device = new MOTEIO();
  1229.   in_devices->push_back(device);
  1230.   out_devices->push_back(device);
  1231. #endif // USE_MOTE_NIC
  1233. #ifdef USE_SMAC
  1234.   device = new SMAC();
  1235.   in_devices->push_back(device);
  1236.   out_devices->push_back(device);
  1237. #endif // USE_SMAC
  1239. #ifdef USE_EMSTAR
  1240. #ifdef USE_EMSIM
  1241.   device = new Emstar(my_id_, group_id, true);
  1242. #else
  1243.   device = new Emstar();
  1244. #endif // USE_EMSIM
  1245.   in_devices->push_back(device);
  1246.   out_devices->push_back(device);
  1247. #endif // USE_EMSTAR
  1249. #ifdef USE_WINSNG2
  1250.   device = new WINSNG2();
  1251.   in_devices->push_back(device);
  1252.   out_devices->push_back(device);
  1253. #endif // USE_WINSNG2
  1254. }
  1256. HashEntry * DiffusionCoreAgent::getHash(unsigned int pkt_num,
  1257.  unsigned int rdm_id)
  1258. {
  1259.   unsigned int key[2];
  1261.   key[0] = pkt_num;
  1262.   key[1] = rdm_id;
  1264.   Tcl_HashEntry *entryPtr = Tcl_FindHashEntry(&htable_, (char *)key);
  1266.   if (entryPtr == NULL)
  1267.     return NULL;
  1269.   return (HashEntry *)Tcl_GetHashValue(entryPtr);
  1270. }
  1272. void DiffusionCoreAgent::putHash(unsigned int pkt_num,
  1273.  unsigned int rdm_id)
  1274. {
  1275.   Tcl_HashEntry *tcl_hash_entry;
  1276.   HashEntry *hash_entry;
  1277.   HashList::iterator hash_itr;
  1278.   unsigned int key[2];
  1279.   int new_hash_key;
  1281.   if (hash_list_.size() == HASH_TABLE_MAX_SIZE){
  1282.     // Hash table reached maximum size
  1284.     for (int i = 0; ((i < HASH_TABLE_REMOVE_AT_ONCE)
  1285.      && (hash_list_.size() > 0)); i++){
  1286.       hash_itr = hash_list_.begin();
  1287.       tcl_hash_entry = *hash_itr;
  1288.       hash_entry = (HashEntry *) Tcl_GetHashValue(tcl_hash_entry);
  1289.       delete hash_entry;
  1290.       hash_list_.erase(hash_itr);
  1291.       Tcl_DeleteHashEntry(tcl_hash_entry);
  1292.     }
  1293.   }
  1295.   key[0] = pkt_num;
  1296.   key[1] = rdm_id;
  1298.   tcl_hash_entry = Tcl_CreateHashEntry(&htable_, (char *)key, &new_hash_key);
  1300.   if (new_hash_key == 0){
  1301.     DiffPrint(DEBUG_IMPORTANT, "Key already exists in hash !n");
  1302.     return;
  1303.   }
  1305.   hash_entry = new HashEntry;
  1307.   Tcl_SetHashValue(tcl_hash_entry, hash_entry);
  1308.   hash_list_.push_back(tcl_hash_entry);
  1309. }
  1311. #ifndef NS_DIFFUSION
  1312. void DiffusionCoreAgent::recvPacket(DiffPacket pkt)
  1313. {
  1314.   struct hdr_diff *dfh = HDR_DIFF(pkt);
  1315.   Message *rcv_message = NULL;
  1316.   int8_t version, msg_type;
  1317.   u_int16_t data_len, num_attr, source_port;
  1318.   int32_t rdm_id, pkt_num, next_hop, last_hop;   
  1320.   // Read header
  1321.   version = DIFF_VER(dfh);
  1322.   msg_type = MSG_TYPE(dfh);
  1323.   source_port = ntohs(SRC_PORT(dfh));
  1324.   pkt_num = ntohl(PKT_NUM(dfh));
  1325.   rdm_id = ntohl(RDM_ID(dfh));
  1326.   num_attr = ntohs(NUM_ATTR(dfh));
  1327.   next_hop = ntohl(NEXT_HOP(dfh));
  1328.   last_hop = ntohl(LAST_HOP(dfh));
  1329.   data_len = ntohs(DATA_LEN(dfh));
  1331.   // Packet is good, create a message
  1332.   rcv_message = new Message(version, msg_type, source_port, data_len,
  1333.     num_attr, pkt_num, rdm_id, next_hop, last_hop);
  1335.   // Read all attributes into the Message structure
  1336.   rcv_message->msg_attr_vec_ = UnpackAttrs(pkt, num_attr);
  1338.   // Process the incoming message
  1339.   recvMessage(rcv_message);
  1341.   // Don't forget to message when we're done
  1342.   delete rcv_message;
  1344.   delete [] pkt;
  1345. }
  1346. #endif // !NS_DIFFUSION
  1348. void DiffusionCoreAgent::recvMessage(Message *msg)
  1349. {
  1350.   BlackList::iterator black_list_itr;
  1351.   Tcl_HashEntry *tcl_hash_entry;
  1352.   unsigned int key[2];
  1354.   // Check version
  1355.   if (msg->version_ != DIFFUSION_VERSION)
  1356.     return;
  1358.   // Check for ID conflict
  1359.   if (msg->last_hop_ == my_id_){
  1360.     DiffPrint(DEBUG_ALWAYS, "Error: A diffusion ID conflict has been detected !n");
  1361.     exit(-1);
  1362.   }
  1364.   // Address filtering
  1365.   if ((msg->next_hop_ != BROADCAST_ADDR) &&
  1366.       (msg->next_hop_ != LOCALHOST_ADDR) &&
  1367.       (msg->next_hop_ != my_id_))
  1368.     return;
  1370.   // Blacklist filtering
  1371.   black_list_itr = black_list_.begin();
  1372.   while (black_list_itr != black_list_.end()){
  1373.     if (*black_list_itr == msg->last_hop_){
  1374.       DiffPrint(DEBUG_DETAILS, "Ignoring message from blacklisted node %d !n",
  1375. msg->last_hop_);
  1376.       return;
  1377.     }
  1378.     black_list_itr++;
  1379.   }
  1381.   // Control Messages are unique and don't go to the hash
  1382.   if (msg->msg_type_ != CONTROL){
  1383.     // Hash table keeps info about packets
  1384.   
  1385.     key[0] = msg->pkt_num_;
  1386.     key[1] = msg->rdm_id_;
  1387.     tcl_hash_entry = Tcl_FindHashEntry(&htable_, (char *) key);
  1389.     if (tcl_hash_entry != NULL){
  1390.       DiffPrint(DEBUG_DETAILS, "Node%d: Received old message !n", my_id_);
  1391.       msg->new_message_ = 0;
  1392.     }
  1393.     else{
  1394.       // Add message to the hash table
  1395.       putHash(key[0], key[1]);
  1396.       msg->new_message_ = 1;
  1397.     }
  1398.   }
  1400. #ifdef STATS
  1401.   if (stats_)
  1402.     stats_->logIncomingMessage(msg);
  1403. #endif // STATS
  1405.   // Check if it's a control of a regular message
  1406.   if (msg->msg_type_ == CONTROL)
  1407.     processControlMessage(msg);
  1408.   else
  1409.     processMessage(msg);
  1410. }
  1412. #ifndef USE_SINGLE_ADDRESS_SPACE
  1413. int main(int argc, char **argv)
  1414. {
  1415.   agent = new DiffusionCoreAgent(argc, argv);
  1417.   signal(SIGINT, signal_handler);
  1419.   agent->run();
  1421.   return 0;
  1422. }
  1423. #endif // !USE_SINGLE_ADDRESS_SPACE