ChangeLogTag:Wed Dec 30 17:37:19 1998 Carlos O'Ryan <coryan@cs.wustl.edu>

7 files changed, 120 insertions, 18 deletions

diff --git a/TAO/ChangeLog-98c b/TAO/ChangeLog-98c
index 30d989ff4e9..de083f35da3 100644
--- a/TAO/ChangeLog-98c
+++ b/TAO/ChangeLog-98c
@@ -1,3 +1,25 @@
+Wed Dec 30 17:37:19 1998  Carlos O'Ryan  <coryan@cs.wustl.edu>
+
+	* orbsvcs/orbsvcs/Event/EC_Gateway_UDP.h: 
+	* orbsvcs/orbsvcs/Event/EC_Gateway_UDP.i: 
+	* orbsvcs/orbsvcs/Event/EC_Gateway_UDP.cpp: 
+	* orbsvcs/tests/EC_Mcast/EC_Mcast.h: 
+	* orbsvcs/tests/EC_Mcast/EC_Mcast.cpp: 
+	  Detecting if a fragment comes from the current process is harder
+	  than I thought: just using the local address of the seding
+	  socket is not enough, because in some platform this address has
+	  the ipaddr set to 0 (the source address is chosen by the kernel
+	  after routing considerations).
+	  We use the ACE facilities to find the list of local ipaddresses
+	  to work around this problem.
+
+	* orbsvcs/tests/EC_Mcast/EC_Mcast.cpp: 
+	  Adjusted the timeout value for the UDP fragments, it was too
+	  low.
+
+	* orbsvcs/tests/EC_Mcast/README: 
+	  Updated the documentation.
+
 Wed Dec 30 16:43:35 1998  Vishal Kachroo  <vishal@merengue.cs.wustl.edu>
 
 	The following files have been added to the newly created
@@ -59,7 +81,7 @@ Wed Dec 30 16:26:51 1998  Vishal Kachroo  <vishal@merengue.cs.wustl.edu>
 
 
 Wed Dec 30 16:19:06 1998  Pradeep Gore  <pradeep@flamenco.cs.wustl.edu>
-	added the following files for the CosEC test to orbsvcs
+
 	* orbsvcs/tests/CosEC_Basic:
 	* orbsvcs/tests/CosEC_Basic/README:
 	* orbsvcs/tests/CosEC_Basic/CosECConsumer.cpp:
@@ -69,6 +91,7 @@ Wed Dec 30 16:19:06 1998  Pradeep Gore  <pradeep@flamenco.cs.wustl.edu>
 	* orbsvcs/tests/CosEC_Basic/CosEC_Basic.cpp:
 	* orbsvcs/tests/CosEC_Basic/CosEC_Basic.h:
 	* orbsvcs/tests/CosEC_Basic/Makefile:
+	  Added the following files for the CosEC test to orbsvcs
 
 1998-12-30  Seth Widoff  <swidoff@cs.cmu.edu>
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.cpp
index 584d4e6316c..2e4893f6b55 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.cpp
@@ -155,7 +155,7 @@ TAO_ECG_UDP_Sender::push (const RtecEventComm::EventSet &events,
 
   if (events.length () == 0)
     {
-      ACE_DEBUG ((LM_DEBUG, "no events\n"));
+      // ACE_DEBUG ((LM_DEBUG, "no events\n"));
       return;
     }
 
@@ -451,6 +451,48 @@ TAO_ECG_UDP_Sender::compute_fragment_count (const ACE_Message_Block* begin,
 
 // ****************************************************************
 
+TAO_ECG_UDP_Out_Endpoint::~TAO_ECG_UDP_Out_Endpoint (void)
+{
+  delete[] this->ifs_;
+  this->ifs_ = 0;
+}
+
+CORBA::Boolean
+TAO_ECG_UDP_Out_Endpoint::is_loopback (const ACE_INET_Addr& from)
+{
+  if (this->port_number_ == 0)
+    {
+      // Cache the port number...
+      ACE_INET_Addr local_addr;
+      if (this->dgram ().get_local_addr (local_addr) == -1)
+        return 0;
+      this->port_number_ = local_addr.get_port_number ();
+    }
+
+  // Most of the time the port number is enough to determine if the
+  // message is remote, only when the local port number and the remote
+  // port number match we have to look at the local ip addresses.
+  if (from.get_port_number () != this->port_number_)
+    return 0;
+
+  if (this->ifs_ == 0)
+    {
+      ACE::get_ip_interfaces (this->if_count_, this->ifs_);
+    }
+
+  for (ACE_INET_Addr* i = this->ifs_;
+       i != this->ifs_ + this->if_count_;
+       ++i)
+    {
+      if ((*i).get_ip_address () == from.get_ip_address ())
+        return 1;
+    }
+  return 0;
+}
+
+
+// ****************************************************************
+
 #if 0
 TAO_ECG_UDP_Request_Entry::TAO_ECG_UDP_Request_Entry (void)
   : request_size_ (0),
@@ -617,7 +659,7 @@ void
 TAO_ECG_UDP_Receiver::init (RtecEventChannelAdmin::EventChannel_ptr lcl_ec,
                             RtecScheduler::Scheduler_ptr lcl_sched,
                             const char* lcl_name,
-                            const ACE_INET_Addr& ignore_from,
+                            TAO_ECG_UDP_Out_Endpoint* ignore_from,
 			    RtecUDPAdmin::AddrServer_ptr addr_server,
                             ACE_Reactor *reactor,
                             const ACE_Time_Value &expire_interval,
@@ -755,7 +797,8 @@ TAO_ECG_UDP_Receiver::handle_input (ACE_SOCK_Dgram& dgram)
 
   // This is to avoid receiving the events we send; notice that we
   // must read the message to drop it...
-  if (from == this->ignore_from_)
+  if (this->ignore_from_ != 0
+      && this->ignore_from_->is_loopback (from))
     {
       n = dgram.recv (header, sizeof(header), from);
       // ACE_DEBUG ((LM_DEBUG,
@@ -795,7 +838,7 @@ TAO_ECG_UDP_Receiver::handle_input (ACE_SOCK_Dgram& dgram)
     }
 
   //  ACE_DEBUG ((LM_DEBUG,
-  //              "ECG_UDP_Receiver (%P|%t): msg = %d, from = (%u:%d)"
+  //              "ECG_UDP_Receiver (%P|%t): msg = %d, from = (%u:%d) "
   //              "fragment = %d/%d\n",
   //              request_id,
   //              from.get_ip_address (), from.get_port_number (),
@@ -921,8 +964,10 @@ TAO_ECG_UDP_Receiver::handle_input (ACE_SOCK_Dgram& dgram)
       TAO_CHECK_ENV;
 
       //      ACE_DEBUG ((LM_DEBUG,
-      //                  "TAO_ECG_UDP_Received (%P|%t): push %d\n",
-      //                  request_id));
+      //                  "TAO_ECG_UDP_Received (%P|%t): push %d "
+      //                  "from = (%u:%d)\n",
+      //                  request_id,
+      //                  from.get_ip_address (), from.get_port_number ()));
     }
   TAO_CATCHANY
     {
@@ -963,6 +1008,10 @@ TAO_ECG_UDP_Receiver::handle_timeout (const ACE_Time_Value& /* tv */,
           Request_Map_Entry& entry = *j;
           ++j;
           {
+            // ACE_DEBUG ((LM_DEBUG,
+            //            "TAO_ECG_UDP_Receiver::handle_timeout (%P|%t) "
+            //            "msg = %d\n",
+            //            entry.ext_id_.request_id));
             delete entry.int_id_;
             this->request_map_.unbind (&entry);
           }
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.h b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.h
index aed0977befd..70c3df83ba7 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.h
@@ -234,6 +234,9 @@ public:
   TAO_ECG_UDP_Out_Endpoint (void);
   // Constructor
 
+  ~TAO_ECG_UDP_Out_Endpoint (void);
+  // Constructor
+
   ACE_SOCK_Dgram& dgram (void);
   // Obtain the datagram associated with this endpoint. Clients of
   // this class must open, and register (if necessary) this datagram.
@@ -241,12 +244,25 @@ public:
   CORBA::ULong next_request_id (void);
   // Obtain the next request id.
 
+  CORBA::Boolean is_loopback (const ACE_INET_Addr& from);
+  // The endpoint can detect if a data-gram was sent by itself, this
+  // is useful to ignore or remove messages sent by the same process.
+
 private:
   ACE_Atomic_Op<ACE_SYNCH_MUTEX,CORBA::ULong> request_id_generator_;
   // The request id....
 
   ACE_SOCK_Dgram dgram_;
   // The datagram....
+
+  u_short port_number_;
+  // cache the port-number so we can quickly determine if an event is
+  // coming from another endpoint.
+
+  size_t if_count_;
+  ACE_INET_Addr* ifs_;
+  // Keep the list of local interfaces, needed for the is_loopback
+  // method.
 };
 
 // ****************************************************************
@@ -413,7 +429,7 @@ public:
   void init (RtecEventChannelAdmin::EventChannel_ptr lcl_ec,
              RtecScheduler::Scheduler_ptr lcl_sched,
              const char* lcl_name,
-             const ACE_INET_Addr& ignore_from,
+             TAO_ECG_UDP_Out_Endpoint* ignore_from,
              RtecUDPAdmin::AddrServer_ptr addr_server,
              ACE_Reactor *reactor,
              const ACE_Time_Value &expire_interval,
@@ -423,6 +439,8 @@ public:
   // connect to; it also requires to build an RT_Info for the local
   // scheduler.
   // The <reactor> is used to receive timeout events..
+  // The <ignore_from> endpoint is used to remove events generated by
+  // the same process.
   // @@ TODO part of the RT_Info is hardcoded, we need to make it
   // parametric.
 
@@ -465,7 +483,7 @@ private:
   RtecEventChannelAdmin::ProxyPushConsumer_var consumer_proxy_;
   // We talk to the EC (as a consumer) using this proxy.
 
-  ACE_INET_Addr ignore_from_;
+  TAO_ECG_UDP_Out_Endpoint* ignore_from_;
   // Ignore any events coming from this IP addres.
 
   RtecUDPAdmin::AddrServer_var addr_server_;
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.i b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.i
index 16a85affc05..5d494997767 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.i
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_UDP.i
@@ -4,6 +4,9 @@
 
 ACE_INLINE
 TAO_ECG_UDP_Out_Endpoint::TAO_ECG_UDP_Out_Endpoint (void)
+  : port_number_ (0),
+    if_count_ (0),
+    ifs_ (0)
 {
 }
 
diff --git a/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.cpp b/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.cpp
index e82d1fbb033..6776ef78093 100644
--- a/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.cpp
+++ b/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.cpp
@@ -221,9 +221,6 @@ ECM_Driver::run (int argc, char* argv[])
       if (this->orb_->run () == -1)
         ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "orb->run"), -1);
 
-      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: shutdown the EC\n"));
-      ec_impl.shutdown ();
-
       this->dump_results ();
 
       this->close_receivers (TAO_TRY_ENV);
@@ -234,11 +231,15 @@ ECM_Driver::run (int argc, char* argv[])
       this->close_federations (TAO_TRY_ENV);
       TAO_CHECK_ENV;
 
+      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: shutdown the EC\n"));
+      ec_impl.shutdown ();
+
       ACE_DEBUG ((LM_DEBUG, "EC_Mcast: shutdown grace period\n"));
 
       ACE_Time_Value tv (5, 0);
       if (this->orb_->run (&tv) == -1)
         ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "orb->run"), -1);
+
     }
   TAO_CATCH (CORBA::SystemException, sys_ex)
     {
@@ -312,6 +313,12 @@ ECM_Driver::open_senders (RtecEventChannelAdmin::EventChannel_ptr ec,
       _env.exception (new CORBA::COMM_FAILURE (CORBA::COMPLETED_NO));
       return;
     }
+  ACE_INET_Addr ignore_from;
+  this->endpoint_.dgram ().get_local_addr (ignore_from);
+  ACE_DEBUG ((LM_DEBUG, "ECM_Driver::open_senders - "
+              "local endpoint = (%u:%d)\n",
+              ignore_from.get_ip_address (),
+              ignore_from.get_port_number ()));
   for (int i = 0; i < this->all_federations_count_; ++i)
     {
       this->all_federations_[i]->open (&this->endpoint_,
@@ -342,6 +349,7 @@ ECM_Driver::open_receivers (RtecEventChannelAdmin::EventChannel_ptr ec,
     {
       this->local_federations_[i]->open_receiver (ec,
                                                   scheduler,
+                                                  &this->endpoint_,
                                                   _env);
       TAO_CHECK_ENV_RETURN_VOID (_env);
     }
@@ -1016,6 +1024,7 @@ ECM_Local_Federation::ECM_Local_Federation (ECM_Federation *federation,
      last_publication_change_ (0),
      last_subscription_change_ (0),
      mcast_eh_ (&receiver_),
+     seed_ (0),
      subscription_change_period_ (10000),
      publication_change_period_ (10000)
 {
@@ -1167,6 +1176,7 @@ ECM_Local_Federation::consumer_push (ACE_hrtime_t,
 void
 ECM_Local_Federation::open_receiver (RtecEventChannelAdmin::EventChannel_ptr ec,
                                      RtecScheduler::Scheduler_ptr scheduler,
+                                     TAO_ECG_UDP_Out_Endpoint* ignore_from,
                                      CORBA::Environment &_env)
 {
   const int bufsize = 512;
@@ -1180,14 +1190,12 @@ ECM_Local_Federation::open_receiver (RtecEventChannelAdmin::EventChannel_ptr ec,
 
   ACE_Reactor* reactor = TAO_ORB_Core_instance ()->reactor ();
 
-  ACE_INET_Addr local_addr;
-  this->federation_->sender_local_addr (local_addr);
   // @@ This should be parameters...
-  ACE_Time_Value expire_interval (0, 1000); // 1 milli second
-  const int max_timeouts = 3; // expire the message after 3 msecs
+  ACE_Time_Value expire_interval (1, 0);
+  const int max_timeouts = 5;
   this->receiver_.init (ec, scheduler,
                         buf,
-                        local_addr,
+                        ignore_from,
                         addr_server.in (),
                         reactor, expire_interval, max_timeouts,
                         _env);
diff --git a/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.h b/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.h
index aff711b14e6..b05aa7aedb2 100644
--- a/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.h
+++ b/TAO/orbsvcs/tests/EC_Mcast/EC_Mcast.h
@@ -300,6 +300,7 @@ public:
 
   void open_receiver (RtecEventChannelAdmin::EventChannel_ptr ec,
                       RtecScheduler::Scheduler_ptr scheduler,
+                      TAO_ECG_UDP_Out_Endpoint* ignore_from,
                       CORBA::Environment &_env);
   // Connect the UDP receiver to the EC.
 
diff --git a/TAO/orbsvcs/tests/EC_Mcast/README b/TAO/orbsvcs/tests/EC_Mcast/README
index 2a28adf83fe..9f195081d11 100644
--- a/TAO/orbsvcs/tests/EC_Mcast/README
+++ b/TAO/orbsvcs/tests/EC_Mcast/README
@@ -13,7 +13,7 @@ on each process.
 	To make the test interesting you should run at least two
 copies of the test, on different windows (to keep the output appart):
 
-$ ./EC_Mcast -p ECM1.pid -c sample.cfg -n 200 -t 500000 -f Set01
+$ ./EC_Mcast -p ECM1.pid -c sample.cfg -n 200 -t 500000 -f Set02
 $ ./EC_Mcast -p ECM2.pid -c sample.cfg -n 200 -t 500000 -f Set02
 
 	the test will report the number of events received, if you run