6 files changed, 254 insertions, 30 deletions

diff --git a/CIAO/DAnCE/ExecutionManager/Execution_Manager.cpp b/CIAO/DAnCE/ExecutionManager/Execution_Manager.cpp
index 71d2280ef31..7e35841fa41 100644
--- a/CIAO/DAnCE/ExecutionManager/Execution_Manager.cpp
+++ b/CIAO/DAnCE/ExecutionManager/Execution_Manager.cpp
@@ -5,7 +5,10 @@
 
 // Include Name Service header
 #include "orbsvcs/CosNamingC.h"
-#include "tao/Utils/Implicit_Deactivator.h"
+#include "tao/RTCORBA/RTCORBA.h"
+#include "tao/RTCORBA/RT_Policy_i.h"
+#include "tao/RTPortableServer/RTPortableServer.h"
+#include "tao/Strategies/advanced_resource.h"
 
 #include "ace/SString.h"
 #include "ace/Read_Buffer.h"
@@ -22,11 +25,12 @@ namespace CIAO
     const char *pid_file_name_ = 0;
     static bool register_with_ns_ = false;
     static bool write_to_ior_ = false;
+    static bool rt_corba_enabled = false;
 
     bool
     parse_args (int argc, char *argv[])
     {
-      ACE_Get_Opt get_opts (argc, argv, "o:i:np:");
+      ACE_Get_Opt get_opts (argc, argv, "o:i:nrp:");
       int c;
       while ((c = get_opts ()) != -1)
         switch (c)
@@ -41,6 +45,9 @@ namespace CIAO
           case 'n':
             register_with_ns_ = true;
             break;
+          case 'r':
+            rt_corba_enabled = true;
+            break;
           case 'p':
             pid_file_name_ = get_opts.opt_arg ();
             break;
@@ -156,39 +163,99 @@ namespace CIAO
           if (!parse_args (argc, argv))
             return -1;
 
+          // RTORB.
+          CORBA::Object_var object =
+            orb->resolve_initial_references ("RTORB");
+          RTCORBA::RTORB_var rt_orb = RTCORBA::RTORB::_narrow (object.in ());
+
+          if (CORBA::is_nil (rt_orb.in ()))
+            ACE_ERROR_RETURN ((LM_ERROR,
+                               "(%P|%t) CIAO_ExecutionManager: "
+                               "Nil RT_ORB panic error, returning \n"),
+                               -1);
+
           // Get reference to Root POA.
           CORBA::Object_var obj
             = orb->resolve_initial_references ("RootPOA"
                                                ACE_ENV_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
-          PortableServer::POA_var poa =
+          PortableServer::POA_var root_poa =
             PortableServer::POA::_narrow (obj.in ()
                                           ACE_ENV_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
 
-          if (CORBA::is_nil (poa.in ()))
+          if (CORBA::is_nil (root_poa.in ()))
             ACE_ERROR_RETURN ((LM_ERROR,
                                "(%P|%t) CIAO_ExecutionManager: "
-                               "Nil POA panic error, returning \n"),
+                               "Nil Root POA panic error, returning \n"),
+                               -1);
+
+          // POAManager.
+          PortableServer::POAManager_var poa_manager =
+            root_poa->the_POAManager ();
+
+          if (poa_manager.in () == 0)
+            ACE_ERROR_RETURN ((LM_ERROR,
+                               "(%P|%t) CIAO_ExecutionManager: "
+                               "Nil POA Manager panic error, returning \n"),
+                               -1);
+
+          PortableServer::POA_var child_poa;
+          if (rt_corba_enabled)
+            {
+              // Create child POA with RTCORBA::ClientProtocolPolicy set.
+              CORBA::PolicyList poa_policy_list;
+              poa_policy_list.length (1);
+              poa_policy_list[0] =
+                rt_orb->create_priority_model_policy (RTCORBA::CLIENT_PROPAGATED,
+                                                      0
+                                                      ACE_ENV_ARG_PARAMETER);
+              ACE_TRY_CHECK;
+
+              child_poa =
+                root_poa->create_POA ("Child_POA",
+                                      poa_manager.in (),
+                                      poa_policy_list);
+
+
+            }
+          else
+            {
+              child_poa =
+                root_poa->create_POA ("Child_POA",
+                                      poa_manager.in (),
+                                      0);
+            }
+
+          if (CORBA::is_nil (child_poa.in ()))
+            ACE_ERROR_RETURN ((LM_ERROR,
+                              "(%P|%t) CIAO_ExecutionManager: "
+                              "Nil Child POA panic error, returning \n"),
                               -1);
 
-          // Create and install the CIAO Daemon servant
+          // Create and install the CIAO Daemon servant on child POA
           Execution_Manager_Impl *daemon_servant = 0;
           ACE_NEW_RETURN (daemon_servant,
                           Execution_Manager_Impl(orb.in (),
-                                                 poa.in (),
+                                                 child_poa.in (),
                                                  init_file_name),
                           -1);
 
-          // Implicit activation
+          // Explicit activation through the child POA
           PortableServer::ServantBase_var safe_daemon (daemon_servant);
 
-          CIAO::ExecutionManagerDaemon_var daemon =
-            daemon_servant->_this ();
+          PortableServer::ObjectId_var id =
+            child_poa->activate_object (daemon_servant);
+
+          CORBA::Object_var daemon_obj =
+            child_poa->id_to_reference (id.in ());
 
-          TAO::Utils::Implicit_Deactivator de (daemon_servant);
+          CIAO::ExecutionManagerDaemon_var daemon = 
+            CIAO::ExecutionManagerDaemon::_narrow (daemon_obj.in ());
+
+          // Register to naming service
           bool retval = false;
 
           if (register_with_ns_)
@@ -213,17 +280,7 @@ namespace CIAO
             return -1;
 
           // Activate POA manager
-          PortableServer::POAManager_var mgr =
-            poa->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER);
-          ACE_TRY_CHECK;
-
-          if (mgr.in () == 0)
-            ACE_ERROR_RETURN ((LM_ERROR,
-                               "(%P|%t) CIAO_ExecutionManager: "
-                               "Nil POA Manager error, returning \n"),
-                              -1);
-
-          mgr->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
+          poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
           // End Deployment part
@@ -236,13 +293,9 @@ namespace CIAO
           orb->run (ACE_ENV_SINGLE_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
-          // Forget the pointer. The POA will take care of it during
-          // destroy.
-          (void) de.release ();
-
-          poa->destroy (1,
-                        1
-                        ACE_ENV_ARG_PARAMETER);
+          root_poa->destroy (1,
+                             1
+                             ACE_ENV_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
           orb->destroy (ACE_ENV_SINGLE_ARG_PARAMETER);
diff --git a/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.cpp b/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.cpp
index 85f7ebd005d..deb633f37c4 100644
--- a/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.cpp
+++ b/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.cpp
@@ -1,6 +1,7 @@
 // $Id$
 
 #include "Execution_Manager_Impl.h"
+#include "tao/RTCORBA/RTCORBA.h"
 #include "ciao/CIAO_common.h"
 #include "DomainApplicationManager/DomainApplicationManager_Impl.h"
 
@@ -41,6 +42,11 @@ namespace CIAO
 
       if (CIAO::debug_level () > 9)
         ACE_DEBUG ((LM_DEBUG,
+                    "CIAO (%P|%t) Execution Manager Running on CORBA Priority <%d> \n",
+                    this->get_current_thread_priority ()));
+
+      if (CIAO::debug_level () > 9)
+        ACE_DEBUG ((LM_DEBUG,
                     "CIAO (%P|%t) Domain Application Manager "
                     "invoked CIAO_Execution_Manager: preparePlan \n"));
 
@@ -510,6 +516,39 @@ namespace CIAO
       return node_app._retn ();
     }
 
+    CORBA::Short
+    Execution_Manager_Impl::
+    get_current_thread_priority ()
+      ACE_THROW_SPEC ((
+        ::CORBA::SystemException))
+    {
+      // Use RTCurrent to find out the CORBA priority of the current
+      // thread.
+
+      CORBA::Object_var obj =
+        this->orb_->resolve_initial_references ("RTCurrent" ACE_ENV_ARG_PARAMETER);
+      ACE_CHECK;
+
+      RTCORBA::Current_var current =
+        RTCORBA::Current::_narrow (obj.in () ACE_ENV_ARG_PARAMETER);
+      ACE_CHECK;
+
+      if (CORBA::is_nil (obj.in ()))
+        {
+          ACE_ERROR ((LM_ERROR,
+                      "DAnCE (%P|%t) ExecutionManager_Impl.cpp -"
+                      "CIAO::Execution_Manager_Impl::get_current_thread_priority -"
+                      "Unable to get current thread handld.\n"));
+          ACE_THROW (CORBA::INTERNAL ());
+        }
+
+      CORBA::Short servant_thread_priority =
+        current->the_priority (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_CHECK;
+
+      return servant_thread_priority;
+    }
+
     void
     Execution_Manager_Impl::
     add_shared_component (const Component_Binding_Info & comp)
diff --git a/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.h b/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.h
index cbbdd3094a1..58812c6116b 100644
--- a/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.h
+++ b/CIAO/DAnCE/ExecutionManager/Execution_Manager_Impl.h
@@ -155,6 +155,10 @@ namespace CIAO
         ::CORBA::SystemException,
         ::Deployment::InvalidConnection));
 
+      CORBA::Short get_current_thread_priority (void)
+      ACE_THROW_SPEC ((
+        ::CORBA::SystemException));
+
     protected:
       /// Destructor.
       virtual ~Execution_Manager_Impl (void);
diff --git a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher.cpp b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher.cpp
index 1d023c774bc..108dc855530 100644
--- a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher.cpp
+++ b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher.cpp
@@ -186,6 +186,7 @@ namespace CIAO
           if (parse_args (argc, argv) == false)
             return -1;
 
+//          Plan_Launcher_i launcher (orb.in ());
           Plan_Launcher_i launcher;
 
           if (!launcher.init (em_use_naming ? 0 : em_ior_file,
diff --git a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.cpp b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.cpp
index 73d1890288b..66e1078c031 100644
--- a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.cpp
+++ b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.cpp
@@ -3,6 +3,10 @@
 #include "orbsvcs/CosNamingC.h"
 #include "Config_Handlers/XML_File_Intf.h"
 #include "Config_Handlers/DnC_Dump.h"
+#include "tao/RTCORBA/RTCORBA.h"
+#include "tao/RTCORBA/Priority_Mapping_Manager.h"
+#include "tao/ORB_Core.h"
+#include "ace/Sched_Params.h"
 
 namespace CIAO
 {
@@ -38,7 +42,14 @@ namespace CIAO
     }
 
     Plan_Launcher_i::Plan_Launcher_i ()
-      : em_ (), pg_ ()
+      : orb_ (0), em_ (), pg_ ()
+    {
+    }
+
+    Plan_Launcher_i::Plan_Launcher_i (CORBA::ORB_ptr orb)
+      : orb_ (CORBA::ORB::_duplicate  (orb))
+      , em_ ()
+      , pg_ ()
     {
     }
 
@@ -50,6 +61,8 @@ namespace CIAO
                            const char *rm_name
                            ACE_ENV_ARG_DECL)
     {
+      this->orb_ = CORBA::ORB::_duplicate  (orb);
+
       CORBA::Object_var obj;
 
       // EM
@@ -80,6 +93,15 @@ namespace CIAO
                       " Manager ref \n"));
         }
 
+      // Check whether the client_propagate priority model has been set
+      // on the ExecutionManager
+      if (this->is_client_propagated_model ())
+        {
+          // Set the priority of the current thread, so it can be propagated
+          // to the ExecutionManager
+          this->set_current_priority (2);
+        }
+
       if (use_repoman)
         {
           return pg_.init (orb, rm_use_naming, rm_name);
@@ -430,5 +452,97 @@ namespace CIAO
 
       return CORBA::string_dup (plan.UUID.in ());
     }
+
+    bool
+    Plan_Launcher_i::set_current_priority (CORBA::Short desired_priority)
+    {
+      CORBA::Object_var object =
+        this->orb_->resolve_initial_references ("RTCurrent" ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+      RTCORBA::Current_var current =
+        RTCORBA::Current::_narrow (object.in () ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      object = this->orb_->resolve_initial_references ("PriorityMappingManager"
+                                                       ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+      RTCORBA::PriorityMappingManager_var mapping_manager =
+        RTCORBA::PriorityMappingManager::_narrow (object.in ()
+                                                  ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      RTCORBA::PriorityMapping *pm =
+        mapping_manager->mapping ();
+
+      int sched_policy =
+        this->orb_->orb_core ()->orb_params ()->ace_sched_policy ();
+
+      int max_priority =
+        ACE_Sched_Params::priority_max (sched_policy);
+      int min_priority =
+        ACE_Sched_Params::priority_min (sched_policy);
+
+      CORBA::Short native_priority =
+        (max_priority + min_priority) / 2;
+
+      // CORBA::Short desired_priority = 0;
+
+      //if (pm->to_CORBA (native_priority, desired_priority) == 0)
+      //  ACE_ERROR_RETURN ((LM_ERROR,
+      //                     "Cannot convert native priority %d to corba priority\n",
+      //                     native_priority),
+      //                     false);
+
+
+      current->the_priority (desired_priority ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      CORBA::Short priority =
+        current->the_priority (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      if (desired_priority != priority)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                            "ERROR: Unable to set thread "
+                            "priority to %d\n", desired_priority),
+                          false);
+
+      return true;
+    }
+
+    bool
+    Plan_Launcher_i::is_client_propagated_model ()
+    {
+      CORBA::Policy_var policy =
+        this->em_->_get_policy (RTCORBA::PRIORITY_MODEL_POLICY_TYPE
+                                ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      RTCORBA::PriorityModelPolicy_var priority_policy =
+        RTCORBA::PriorityModelPolicy::_narrow (policy.in () ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      if (CORBA::is_nil (priority_policy.in ()))
+      {
+        ACE_DEBUG ((LM_DEBUG,
+          ACE_TEXT ("CIAO::Plan_Launcher_i: " )
+          ACE_TEXT ("In ExecutionManager, the Priority Model Policy not exposed!\n")));
+        return false;
+      }
+
+      RTCORBA::PriorityModel priority_model =
+        priority_policy->priority_model (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      if (priority_model != RTCORBA::CLIENT_PROPAGATED)
+      {
+        ACE_DEBUG ((LM_DEBUG,
+          ACE_TEXT ("CIAO::Plan_Launcher_i: " )
+          ACE_TEXT ("The Priority Model of ExecutionManager is not CLIENT_PROPAGATED !\n")));
+        return false;
+      }
+       
+      return true;
+    }
   }
 }
diff --git a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.h b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.h
index 23ea6abc722..d8896607c90 100644
--- a/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.h
+++ b/CIAO/DAnCE/Plan_Launcher/Plan_Launcher_Impl.h
@@ -15,6 +15,7 @@
 #include "DAnCE/Interfaces/ExecutionManagerDaemonC.h"
 #include "DAnCE/Plan_Generator/Plan_Generator_Impl.h"
 
+
 namespace CIAO
 {
   namespace Plan_Launcher
@@ -38,6 +39,7 @@ namespace CIAO
 
       Plan_Launcher_i ();
 
+      Plan_Launcher_i (CORBA::ORB_ptr orb);
 
       bool init (const char *em_ior,
                  CORBA::ORB_ptr orb,
@@ -94,7 +96,18 @@ namespace CIAO
       void destroy_dam_by_plan (const char * plan_uuid
                                 ACE_ENV_ARG_DECL_WITH_DEFAULTS);
 
+      // Change the priority of this thread
+      bool set_current_priority (CORBA::Short priority);
+
+    protected:
+      // Check that the object is configured with CLIENT_PROPAGATED
+      // PriorityModelPolicy.
+      bool is_client_propagated_model (void);
+
     private:
+      /// Cached ORB pointer
+      CORBA::ORB_var orb_;
+
       ::CIAO::ExecutionManagerDaemon_var em_;
 
       /// Local map for DAMs, to save expensive UUID lookups.