Sun Apr 2 20:19:36 2000 Vishal Kachroo <vishal@cs.wustl.edu>

9 files changed, 191 insertions, 55 deletions

diff --git a/ace/QoS_Manager.cpp b/ace/QoS_Manager.cpp
new file mode 100644
index 00000000000..699cbecba1a
--- /dev/null
+++ b/ace/QoS_Manager.cpp
@@ -0,0 +1,47 @@
+// QoS_Manager.cpp
+// $Id$
+
+#define ACE_BUILD_DLL
+#include "ace/QoS_Manager.h"
+
+ACE_RCSID(ace, QoS_Manager, "$Id $")
+
+ACE_ALLOC_HOOK_DEFINE(ACE_QOS_MANAGER)
+
+ACE_QoS_Manager::ACE_QoS_Manager (void)
+{}
+
+ACE_QoS_Manager::~ACE_QoS_Manager (void)
+{}
+
+// Adds the given session to the list of session objects joined by
+// this socket.
+
+int 
+ACE_QoS_Manager::join_qos_session (ACE_QoS_Session *qos_session)
+{
+  if (this->qos_session_set ().insert (qos_session) != 0)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       "Error in adding a new session to the "
+                       "socket session set\n"),
+                      -1);
+  return 0;
+}
+
+// Returns the QoS session set for this socket.
+
+ACE_Unbounded_Set <ACE_QoS_Session *>
+ACE_QoS_Manager::qos_session_set (void)
+{
+  return this->qos_session_set_;
+}
+
+#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
+template class ACE_Unbounded_Set<ACE_QoS_Session *>;
+template class ACE_Node<ACE_QoS_Session *>;
+template class ACE_Unbounded_Set_Iterator<ACE_QoS_Session *>;
+#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
+#pragma instantiate ACE_Unbounded_Set<ACE_QoS_Session *>
+#pragma instantiate ACE_Node<ACE_QoS_Session *>
+#pragma instantiate ACE_Unbounded_Set_Iterator<ACE_QoS_Session *>
+#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/ace/QoS_Manager.h b/ace/QoS_Manager.h
new file mode 100644
index 00000000000..aed2772207e
--- /dev/null
+++ b/ace/QoS_Manager.h
@@ -0,0 +1,69 @@
+/* -*- C++ -*- */
+// $Id$
+
+//============================================================================
+//
+// = LIBRARY
+//    ace
+//
+// = FILENAME
+//    QoS_Manager.h
+//
+// = AUTHOR
+//    Vishal Kachroo
+//
+//============================================================================
+
+#ifndef ACE_QOS_MANAGER_H
+#define ACE_QOS_MANAGER_H
+#include "ace/pre.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "ace/Addr.h"
+#include "ace/IPC_SAP.h"
+#include "ace/Containers_T.h"
+
+class ACE_QoS_Session;
+
+class ACE_Export ACE_QoS_Manager
+{
+  // = TITLE
+  //     This class manages the QoS sessions associated with ACE_SOCK.
+  //
+  // = DESCRIPTION 
+  //     This class provides functions to manage the QoS
+  //     associated with a socket.  The idea is to keep the management of
+  //     QoS for a socket separate from the socket itself. Currently, the
+  //     manager is used to manage the QoS session set. It will handle more
+  //     responsibilities in the future.
+
+public:
+  ACE_QoS_Manager (void);
+  ~ACE_QoS_Manager (void);
+  // Default ctor/dtor.
+  
+  int join_qos_session (ACE_QoS_Session *qos_session);
+  // Join the given QoS session. A socket can join multiple QoS
+  // sessions.  This call adds the given QoS session to the list of
+  // QoS sessions that the socket has already joined.
+
+  typedef ACE_Unbounded_Set <ACE_QoS_Session *> ACE_QOS_SESSION_SET;
+
+  ACE_QOS_SESSION_SET qos_session_set (void);
+  // Get the QoS session set.
+
+private:
+
+  ACE_QOS_SESSION_SET qos_session_set_;
+  // Set of QoS sessions that this socket has joined.
+};
+
+#include "ace/post.h"
+#endif /* ACE_QOS_MANAGER_H */
+
+
+
+
diff --git a/ace/QoS_Session.h b/ace/QoS_Session.h
index 31d17b7b771..0e145c432c4 100644
--- a/ace/QoS_Session.h
+++ b/ace/QoS_Session.h
@@ -25,7 +25,7 @@
 #endif /* ACE_LACKS_PRAGMA_ONCE */
 
 class ACE_SOCK;
-#include "ace/Containers_T.h"
+class ACE_QoS_Manager;
 
 typedef int ACE_Protocol_ID;
 // IPPROTO_UDP or IPPROTO_TCP.
@@ -65,6 +65,7 @@ public:
   // Returns the QoS in the current session.
    
   virtual int qos (ACE_SOCK *socket,
+                   ACE_QoS_Manager *qos_manager,
                    const ACE_QoS &ace_qos) = 0;
   // Set QoS for the current session. The socket parameter is used to confirm if
   // this QoS session was subscribed to by the socket.
@@ -81,7 +82,11 @@ public:
   // Invoking this method is an indication of a QoS event occurring, that may have 
   // resulted in a change of QoS for the underlying session. This method updates 
   // the QoS object associated with this session.
- 
+
+  virtual ACE_End_Point_Type flags (void) const = 0;
+  virtual void flags (const ACE_End_Point_Type flags) = 0;
+  // Get/Set methods for the flags_.
+
   virtual int session_id (void) const = 0;
   // Get the session id.
   
diff --git a/ace/QoS_Session_Impl.cpp b/ace/QoS_Session_Impl.cpp
index 5c98e57703f..95084e69716 100644
--- a/ace/QoS_Session_Impl.cpp
+++ b/ace/QoS_Session_Impl.cpp
@@ -4,6 +4,7 @@
 #define ACE_BUILD_DLL
 
 #include "ace/SOCK.h"
+#include "ace/QoS_Manager.h"
 #include "ace/QoS_Session_Impl.h"
 
 #if !defined (__ACE_INLINE__)
@@ -156,21 +157,27 @@ ACE_RAPI_Session::open (ACE_INET_Addr dest_addr,
   this->dest_addr_ = dest_addr;
   this->protocol_id_ = protocol_id;
   
+  rapi_eventinfo_t RSVP_arg;      /*RSVP callback argument*/
+
   // Open a RAPI session. Note "this" is being passed as an argument to 
   // the callback function. The callback function uses this argument to 
   // update the QoS of this session based on the RSVP event it receives.
-  if ((this->session_id_ = rapi_session((sockaddr *) dest_addr.get_addr (), 
-                                       protocol_id,
-                                       0, 
-                                       rsvp_callback, 
-                                       (void *) this, 
-                                       &rsvp_error)) == NULL_SID)
+  
+  if ((this->session_id_ = rapi_session((struct sockaddr *) dest_addr.get_addr (), 
+                                        protocol_id,
+                                        0, 
+                                        rsvp_callback, 
+                                        //(void *) this, 
+                                        (void *) &RSVP_arg,
+                                        &rsvp_error)) == NULL_SID)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "rapi_session () call fails. Error\n"),
                       -1);
   else
     ACE_DEBUG ((LM_DEBUG,
-                "rapi_session () call succeeds\n"));
+                "rapi_session () call succeeds\n"
+                "Session ID = %d\n",
+                this->session_id_));
   
   return 0;
 }
@@ -184,15 +191,21 @@ ACE_RAPI_Session::close (void)
                        "Can't release RSVP session:\n\t%s\n",
                        rapi_errlist[rsvp_error]),
                       -1);
+  else
+    ACE_DEBUG ((LM_DEBUG,
+                "rapi session with id %d released successfully.\n",
+               this->session_id_));
   return 0;
 }
 
 int 
 ACE_RAPI_Session::qos (ACE_SOCK *socket,
+                       ACE_QoS_Manager *qos_manager,
                        const ACE_QoS &ace_qos)
 {
   ACE_UNUSED_ARG (socket);
-
+  ACE_UNUSED_ARG (qos_manager);
+  
   // If sender : call sending_qos ()
   // If receiver : call receiving_qos ()
   // If both : call sending_qos () and receiving_qos ()
@@ -212,7 +225,7 @@ ACE_RAPI_Session::sending_qos (const ACE_QoS &ace_qos)
 {
   
   ACE_Flow_Spec sending_flowspec = ace_qos.sending_flowspec ();
-  rapi_tspec_t *t_spec = init_tspec_simplified (sending_flowspec);
+  rapi_tspec_t *t_spec = this->init_tspec_simplified (sending_flowspec);
   
   if (t_spec == 0)
     ACE_ERROR_RETURN ((LM_ERROR,
@@ -249,8 +262,11 @@ ACE_RAPI_Session::sending_qos (const ACE_QoS &ace_qos)
               sending_flowspec.ttl ()));
   
   // @@Hardcoded port. This should be changed later.
-  ACE_INET_Addr sender_addr (8001); 
-  
+  ACE_INET_Addr sender_addr (9090); 
+
+  ACE_DEBUG ((LM_DEBUG,
+              "Making the rapi_sender () call\n"));
+
   // Set the Sender TSpec for this QoS session.
   if(rapi_sender(this->session_id_, 
                  0, 
@@ -309,7 +325,13 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
               flow_spec->specbody_qosx.xspec_M));
 
   // @@Hardcoded port. This should be changed later.
-  ACE_INET_Addr receiver_addr (8002); 
+  //  ACE_INET_Addr receiver_addr (8002); 
+
+  //  ACE_INET_Addr receiver_addr; 
+
+  sockaddr_in Receiver_host;
+
+  Receiver_host.sin_addr.s_addr = INADDR_ANY;
 
   // Set the Receiver FlowSpec for this QoS session.
   // @@The filter style is hardcoded to WildCard. This can be changed later.
@@ -318,7 +340,8 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
                    // Setting the RAPI_REQ_CONFIRM flag requests confirmation 
                    // of the resevation, by means of a confirmation upcall of
                    // type RAPI_RESV_CONFIRM.
-                   (sockaddr *)receiver_addr.get_addr (), 
+                   //                   (sockaddr *)receiver_addr.get_addr (), 
+                   (sockaddr *)&Receiver_host, 
                    RAPI_RSTYLE_WILDCARD,
                    // This applies the flowspec to all the senders. Given this, 
                    // @@I am passing the filter_spec to be null, hoping this will work.
@@ -333,6 +356,9 @@ ACE_RAPI_Session::receiving_qos (const ACE_QoS &ace_qos)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "rapi_reserve () error:\n\tRESV Generation can't be started\n"),
                       -1);
+  else
+    ACE_DEBUG ((LM_DEBUG,
+                "rapi_reserve () call succeeds \n"));
 
   return 0;
 }
@@ -373,7 +399,7 @@ ACE_RAPI_Session::init_tspec_simplified (const ACE_Flow_Spec &flow_spec)
   ctxp->xtspec_m  = flow_spec.minimum_policed_size (); // Minimum policed unit.
   
   // @@Hardcoded for the time being.
-  ctxp->xtspec_M  = 65535;                             // Maximum SDU size. 
+  ctxp->xtspec_M  = 1024;                             // Maximum SDU size. 
   
   t_spec->len = sizeof(rapi_hdr_t) + sizeof(qos_tspecx_t);
   t_spec->form = RAPI_TSPECTYPE_Simplified;
@@ -409,6 +435,8 @@ ACE_RAPI_Session::init_flowspec_simplified(const ACE_Flow_Spec &flow_spec)
       // Note there is no break !!
 
     case QOS_CNTR_LOAD:
+      ACE_DEBUG ((LM_DEBUG,
+                  "QOS_CONTROLLED_LOAD\n"));
       csxp->spec_type = flow_spec.service_type ();        // qos_service_type
       csxp->xspec_r = flow_spec.token_rate ();            // Token Bucket Average Rate (B/s)
       csxp->xspec_b = flow_spec.token_bucket_size ();     // Token Bucket Rate (B)
@@ -466,18 +494,24 @@ ACE_GQoS_Session::close (void)
 // Set the QoS for this GQoS session.  
 int 
 ACE_GQoS_Session::qos (ACE_SOCK *socket,
+                       ACE_QoS_Manager *qos_manager,
                        const ACE_QoS &ace_qos)
 {
   
-  // Confirm if the current session is one of the QoS sessions subscribed
-  // to by the given socket.
+  // Confirm if the current session is one of the QoS sessions
+  // subscribed to by the given socket.
+
+  //if (socket->qos_session_set ().find (this) == -1)
+
+  // @@Vishal : Need to relate the below to the socket (as above)
+  // instead of the QoS Manager.
 
-  if (socket->qos_session_set ().find (this) == -1)
+  if (qos_manager->qos_session_set ().find (this) == -1)
     ACE_ERROR_RETURN ((LM_ERROR,
                        "This QoS session was not subscribed to"
                        " by the socket\n"),
                       -1);
-
+  
   // Set the QOS according to the supplied ACE_QoS. The I/O control
   // code used under the hood is SIO_SET_QOS.
 
diff --git a/ace/QoS_Session_Impl.h b/ace/QoS_Session_Impl.h
index d1429ba4d47..698ae478f82 100644
--- a/ace/QoS_Session_Impl.h
+++ b/ace/QoS_Session_Impl.h
@@ -56,6 +56,7 @@ public:
   // Returns the QoS for this RAPI session.
   
   virtual int qos (ACE_SOCK *socket,
+                   ACE_QoS_Manager *qos_manager,
                    const ACE_QoS &ace_qos);
   // Set QoS for this RAPI session. The socket parameter is used to confirm if
   // this QoS session was subscribed to by the socket.
@@ -72,6 +73,10 @@ public:
   // It is a mechanism of updating the QoS for this session asynchronously, as
   // RSVP events occur.
 
+  virtual ACE_End_Point_Type flags (void) const;
+  virtual void flags (const ACE_End_Point_Type flags);
+  // Get/Set methods for the flags_.
+
   virtual int session_id (void) const;
   // Get the RAPI session id.
   
@@ -144,6 +149,7 @@ public:
   // Returns the QoS for this GQoS session.
 
   virtual int qos (ACE_SOCK *socket,
+                   ACE_QoS_Manager *qos_manager,
                    const ACE_QoS &ace_qos);
   // Set QoS for this GQoS session. The socket parameter is used to confirm if
   // this QoS session was subscribed to by the socket.
@@ -159,6 +165,10 @@ public:
   // Calls the ioctl (ACE_SIO_GET_QOS). It is a mechanism of updating the 
   // QoS for this session asynchronously, as RSVP events occur.
 
+  virtual ACE_End_Point_Type flags (void) const;
+  virtual void flags (const ACE_End_Point_Type flags);
+  // Get/Set methods for the flags_.
+
   virtual ACE_INET_Addr dest_addr (void) const;
   // Get the destination address for this GQoS session.
 
@@ -192,3 +202,5 @@ private:
 
 #include "ace/post.h"
 #endif /* ACE_QOS_SESSION_IMPL_H */
+
+
diff --git a/ace/SOCK.cpp b/ace/SOCK.cpp
index 8edfb747169..59edba0abb6 100644
--- a/ace/SOCK.cpp
+++ b/ace/SOCK.cpp
@@ -106,28 +106,6 @@ ACE_SOCK::open (int type,
   return 0;
 }
 
-// Adds the given session to the list of session objects joined by
-// this socket.
-
-int 
-ACE_SOCK::join_qos_session (ACE_QoS_Session *qos_session)
-{
-  if (this->qos_session_set ().insert (qos_session) != 0)
-    ACE_ERROR_RETURN ((LM_ERROR,
-                       "Error in adding a new session to the "
-                       "socket session set\n"),
-                      -1);
-  return 0;
-}
-
-// Returns the QoS session set for this socket.
-
-ACE_Unbounded_Set <ACE_QoS_Session *>
-ACE_SOCK::qos_session_set (void)
-{
-  return this->qos_session_set_;
-}
-
 // General purpose constructor for performing server ACE_SOCK
 // creation.
 
diff --git a/ace/SOCK.h b/ace/SOCK.h
index 84e4d7af2bf..4c02306a523 100644
--- a/ace/SOCK.h
+++ b/ace/SOCK.h
@@ -88,16 +88,6 @@ public:
             u_long flags,
             int reuse_addr);
   // Wrapper around the QoS-enabled <WSASocket> function.
-
-  int join_qos_session (ACE_QoS_Session *qos_session);
-  // Join the given QoS session. A socket can join multiple QoS
-  // sessions.  This call adds the given QoS session to the list of
-  // QoS sessions that the socket has already joined.
-
-  typedef ACE_Unbounded_Set <ACE_QoS_Session *> ACE_QOS_SESSION_SET;
-
-  ACE_QOS_SESSION_SET qos_session_set (void);
-  // Get the QoS session set.
   
 protected:
   ACE_SOCK (int type,
@@ -121,8 +111,6 @@ protected:
   // Default constructor is private to prevent instances of this class
   // from being defined.
 
-  ACE_QOS_SESSION_SET qos_session_set_;
-  // Set of QoS sessions that this socket has joined.
 };
 
 #if !defined (ACE_LACKS_INLINE_FUNCTIONS)
diff --git a/ace/SOCK_Dgram_Mcast.cpp b/ace/SOCK_Dgram_Mcast.cpp
index 56b815087e0..febe7b1b41c 100644
--- a/ace/SOCK_Dgram_Mcast.cpp
+++ b/ace/SOCK_Dgram_Mcast.cpp
@@ -394,6 +394,7 @@ ACE_SOCK_Dgram_Mcast::subscribe (const ACE_INET_Addr &mcast_addr,
                                  ACE_Protocol_Info *protocolinfo,
                                  ACE_SOCK_GROUP g,
                                  u_long flags,
+                                 ACE_QoS_Manager *qos_manager,
                                  ACE_QoS_Session *qos_session)
 {
   ACE_TRACE ("ACE_SOCK_Dgram_Mcast::subscribe");
@@ -430,7 +431,7 @@ ACE_SOCK_Dgram_Mcast::subscribe (const ACE_INET_Addr &mcast_addr,
       {
 
         // Subscribe to the QoS session.
-        if (this->join_qos_session (qos_session) == -1)
+        if (qos_manager->join_qos_session (qos_session) == -1)
           ACE_ERROR_RETURN ((LM_ERROR,
                              "Unable to join QoS Session\n"),
                             -1);
diff --git a/ace/SOCK_Dgram_Mcast.h b/ace/SOCK_Dgram_Mcast.h
index b727eaa024f..e8ad5a1cb71 100644
--- a/ace/SOCK_Dgram_Mcast.h
+++ b/ace/SOCK_Dgram_Mcast.h
@@ -21,6 +21,7 @@
 #include "ace/pre.h"
 
 #include "ace/SOCK_Dgram.h"
+#include "ace/QoS_Manager.h"
 
 #if !defined (ACE_LACKS_PRAGMA_ONCE)
 # pragma once
@@ -76,6 +77,7 @@ public:
                  ACE_Protocol_Info *protocolinfo = 0,
                  ACE_SOCK_GROUP g = 0,
                  u_long flags = 0,
+                 ACE_QoS_Manager *qos_manager = 0,
                  ACE_QoS_Session *qos_session = 0);
   // This is a QoS-enabled method for joining a multicast group, which
   // passes <qos_params> via <ACE_OS::join_leaf>.  The network