Added Thread Pools

3 files changed, 193 insertions, 27 deletions

diff --git a/TAO/performance-tests/RTCorba/Thread_Pool/client.cpp b/TAO/performance-tests/RTCorba/Thread_Pool/client.cpp
index 38595f56f19..5e0817ef9f8 100644
--- a/TAO/performance-tests/RTCorba/Thread_Pool/client.cpp
+++ b/TAO/performance-tests/RTCorba/Thread_Pool/client.cpp
@@ -24,11 +24,12 @@ static int done = 0;
 static CORBA::ULong time_for_test = 10;
 static CORBA::ULong work = 10;
 static CORBA::ULong max_throughput_timeout = 5;
+static int set_priority = 1;
 
 int
 parse_args (int argc, char *argv[])
 {
-  ACE_Get_Opt get_opts (argc, argv, "hxk:r:c:w:t:");
+  ACE_Get_Opt get_opts (argc, argv, "hxk:r:c:w:t:p:");
   int c;
 
   while ((c = get_opts ()) != -1)
@@ -62,6 +63,10 @@ parse_args (int argc, char *argv[])
         time_for_test = ACE_OS::atoi (get_opts.optarg);
         break;
 
+      case 'p':
+        set_priority = ACE_OS::atoi (get_opts.optarg);
+        break;
+
       case '?':
       default:
         ACE_ERROR_RETURN ((LM_ERROR,
@@ -69,10 +74,11 @@ parse_args (int argc, char *argv[])
                            "-h <show history> "
                            "-x [shutdown server] "
                            "-k <ior> "
-                           "-i <iterations for slowest paced worker> "
                            "-r <rates file> "
                            "-c <number of continuous workers> "
-                           "-w <work in milli seconds> "
+                           "-w <work> "
+                           "-t <time for test> "
+                           "-p <set priorities> "
                            "\n",
                            argv [0]),
                           -1);
@@ -234,18 +240,21 @@ Paced_Worker::svc (void)
 
   ACE_TRY_NEW_ENV
     {
-      this->current_->the_priority (this->priority_,
-                                    ACE_TRY_ENV);
-      ACE_TRY_CHECK;
+      if (set_priority)
+        {
+          this->current_->the_priority (this->priority_,
+                                        ACE_TRY_ENV);
+          ACE_TRY_CHECK;
 
-      CORBA::Boolean result =
-        priority_mapping_.to_native (this->priority_,
-                                     native_priority);
-      if (!result)
-        ACE_ERROR_RETURN ((LM_ERROR,
-                           "Error in converting CORBA priority %d to native priority\n",
-                           this->priority_),
-                          -1);
+          CORBA::Boolean result =
+            priority_mapping_.to_native (this->priority_,
+                                         native_priority);
+          if (!result)
+            ACE_ERROR_RETURN ((LM_ERROR,
+                               "Error in converting CORBA priority %d to native priority\n",
+                               this->priority_),
+                              -1);
+        }
 
       this->start_of_test_ =
         ACE_OS::gettimeofday ();
@@ -303,8 +312,8 @@ Paced_Worker::svc (void)
       ACE_DEBUG ((LM_DEBUG,
                   "Priority = %d/%d; Rate = %d/sec; Iterations = %d; "
                   "deadlines made = %d; deadlines missed = %d; Success = %d%%\n",
-                  this->priority_,
-                  native_priority,
+                  set_priority ? this->priority_ : -1,
+                  set_priority ? native_priority : -1,
                   this->rate_,
                   this->history_.max_samples (),
                   this->history_.sample_count (),
diff --git a/TAO/performance-tests/RTCorba/Thread_Pool/rates b/TAO/performance-tests/RTCorba/Thread_Pool/rates
index 105d7d9ad3a..7d0dd3894a5 100644
--- a/TAO/performance-tests/RTCorba/Thread_Pool/rates
+++ b/TAO/performance-tests/RTCorba/Thread_Pool/rates
@@ -1 +1 @@
-100
\ No newline at end of file
+25 50 75
\ No newline at end of file
diff --git a/TAO/performance-tests/RTCorba/Thread_Pool/server.cpp b/TAO/performance-tests/RTCorba/Thread_Pool/server.cpp
index edb6eaac2fb..62f63780cad 100644
--- a/TAO/performance-tests/RTCorba/Thread_Pool/server.cpp
+++ b/TAO/performance-tests/RTCorba/Thread_Pool/server.cpp
@@ -43,17 +43,17 @@ test_i::method (CORBA::ULong work,
                 CORBA::Environment &)
   ACE_THROW_SPEC ((CORBA::SystemException))
 {
+  if (TAO_debug_level > 0)
+    ACE_DEBUG ((LM_DEBUG,
+                "test_i::method: %hd units of work\n",
+                work));
+
   const u_long prime_number = 9619;
 
   for (; work != 0; work--)
     ACE::is_prime (prime_number,
                    2,
                    prime_number / 2);
-
-  if (TAO_debug_level > 0)
-    ACE_DEBUG ((LM_DEBUG,
-                "test_i::method: %hd units of work\n",
-                work));
 }
 
 PortableServer::POA_ptr
@@ -71,12 +71,15 @@ test_i::shutdown (CORBA::Environment& ACE_TRY_ENV)
   ACE_CHECK;
 }
 
-const char *ior_output_file = "ior";
+static const char *ior_output_file = "ior";
+static CORBA::ULong static_threads = 1;
+static CORBA::ULong dynamic_threads = 0;
+static CORBA::ULong number_of_lanes = 0;
 
 int
 parse_args (int argc, char *argv[])
 {
-  ACE_Get_Opt get_opts (argc, argv, "o:s:d:t:");
+  ACE_Get_Opt get_opts (argc, argv, "o:s:l:");
   int c;
 
   while ((c = get_opts ()) != -1)
@@ -86,11 +89,21 @@ parse_args (int argc, char *argv[])
         ior_output_file = get_opts.optarg;
         break;
 
+      case 's':
+        static_threads = ACE_OS::atoi (get_opts.optarg);
+        break;
+
+      case 'l':
+        number_of_lanes = ACE_OS::atoi (get_opts.optarg);
+        break;
+
       case '?':
       default:
         ACE_ERROR_RETURN ((LM_ERROR,
                            "usage:  %s "
                            "-o <iorfile> "
+                           "-s <static threads> "
+                           "-l <lanes> "
                            "\n",
                            argv [0]),
                           -1);
@@ -155,15 +168,159 @@ main (int argc, char *argv[])
         root_poa->the_POAManager (ACE_TRY_ENV);
       ACE_TRY_CHECK;
 
+      object =
+        orb->resolve_initial_references ("RTORB",
+                                         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      RTCORBA::RTORB_var rt_orb =
+        RTCORBA::RTORB::_narrow (object.in (),
+                                 ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      object =
+        orb->resolve_initial_references ("RTCurrent",
+                                         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      RTCORBA::Current_var current =
+        RTCORBA::Current::_narrow (object.in (),
+                                   ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      RTCORBA::Priority default_thread_priority =
+        current->the_priority (ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
       int result =
         parse_args (argc, argv);
       if (result != 0)
         return result;
 
-      test_i servant (orb.in (),
-                      root_poa.in ());
+      CORBA::ULong stacksize = 0;
+      CORBA::Boolean allow_request_buffering = 0;
+      CORBA::ULong max_buffered_requests = 0;
+      CORBA::ULong max_request_buffer_size = 0;
+
+      RTCORBA::ThreadpoolId threadpool_id;
+      CORBA::PolicyList policies;
+
+      if (number_of_lanes != 0)
+        {
+          CORBA::Boolean allow_borrowing = 0;
+
+          RTCORBA::ThreadpoolLanes lanes;
+          lanes.length (number_of_lanes);
+
+          RTCORBA::PriorityBands bands;
+          bands.length (number_of_lanes);
+
+          CORBA::Short priority_range =
+            RTCORBA::maxPriority - RTCORBA::minPriority;
+
+          ACE_DEBUG ((LM_DEBUG,
+                      "\nUsing %d lanes\n",
+                      number_of_lanes));
+
+          for (CORBA::ULong i = 0;
+               i < number_of_lanes;
+               ++i)
+            {
+              CORBA::Short high_priority =
+                CORBA::Short (
+                  ACE_OS::floor ((priority_range /
+                                  double (number_of_lanes)) *
+                                 (i + 1)));
+
+              CORBA::Short low_priority =
+                CORBA::Short (
+                  ACE_OS::ceil ((priority_range /
+                                 double (number_of_lanes)) *
+                                i));
+
+              lanes[i].lane_priority = high_priority;
+              lanes[i].static_threads = static_threads;
+              lanes[i].dynamic_threads = dynamic_threads;
+
+              bands[i].high = high_priority;
+              bands[i].low = low_priority;
+
+              ACE_DEBUG ((LM_DEBUG,
+                          "%d: [%d %d] ",
+                          i + 1,
+                          low_priority,
+                          high_priority));
+            }
+
+          ACE_DEBUG ((LM_DEBUG,
+                      "\n\n"));
+
+          threadpool_id =
+            rt_orb->create_threadpool_with_lanes (stacksize,
+                                                  lanes,
+                                                  allow_borrowing,
+                                                  allow_request_buffering,
+                                                  max_buffered_requests,
+                                                  max_request_buffer_size,
+                                                  ACE_TRY_ENV);
+          ACE_TRY_CHECK;
+
+          policies.length (policies.length () + 1);
+          policies[policies.length () - 1] =
+            rt_orb->create_priority_banded_connection_policy (bands,
+                                                              ACE_TRY_ENV);
+          ACE_TRY_CHECK;
+
+          policies.length (policies.length () + 1);
+          policies[policies.length () - 1] =
+            rt_orb->create_priority_model_policy (RTCORBA::CLIENT_PROPAGATED,
+                                                  0,
+                                                  ACE_TRY_ENV);
+          ACE_TRY_CHECK;
+        }
+      else
+        {
+          threadpool_id =
+            rt_orb->create_threadpool (stacksize,
+                                       static_threads,
+                                       dynamic_threads,
+                                       default_thread_priority,
+                                       allow_request_buffering,
+                                       max_buffered_requests,
+                                       max_request_buffer_size,
+                                       ACE_TRY_ENV);
+          ACE_TRY_CHECK;
+        }
+
+      policies.length (policies.length () + 1);
+      policies[policies.length () - 1] =
+        root_poa->create_implicit_activation_policy
+        (PortableServer::IMPLICIT_ACTIVATION,
+         ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      policies.length (policies.length () + 1);
+      policies[policies.length () - 1] =
+        rt_orb->create_threadpool_policy (threadpool_id,
+                                          ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      PortableServer::POA_var poa =
+        root_poa->create_POA ("RT POA",
+                              poa_manager.in (),
+                              policies,
+                              ACE_TRY_ENV);
+      ACE_TRY_CHECK;
+
+      test_i *servant =
+        new test_i (orb.in (),
+                    poa.in ());
+
+      PortableServer::ServantBase_var safe_servant (servant);
+      ACE_UNUSED_ARG (safe_servant);
+
       test_var test =
-        servant._this (ACE_TRY_ENV);
+        servant->_this (ACE_TRY_ENV);
       ACE_TRY_CHECK;
 
       result =