*** empty log message ***

3 files changed, 380 insertions, 0 deletions

diff --git a/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsp b/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsp
new file mode 100644
index 00000000000..f2104b1bbea
--- /dev/null
+++ b/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsp
@@ -0,0 +1,102 @@
+# Microsoft Developer Studio Project File - Name="Queue_Based_Workers" - Package Owner=<4>

+# Microsoft Developer Studio Generated Build File, Format Version 6.00

+# ** DO NOT EDIT **

+

+# TARGTYPE "Win32 (x86) Console Application" 0x0103

+

+CFG=Queue_Based_Workers - Win32 Debug

+!MESSAGE This is not a valid makefile. To build this project using NMAKE,

+!MESSAGE use the Export Makefile command and run

+!MESSAGE 

+!MESSAGE NMAKE /f "Queue_Based_Workers.mak".

+!MESSAGE 

+!MESSAGE You can specify a configuration when running NMAKE

+!MESSAGE by defining the macro CFG on the command line. For example:

+!MESSAGE 

+!MESSAGE NMAKE /f "Queue_Based_Workers.mak" CFG="Queue_Based_Workers - Win32 Debug"

+!MESSAGE 

+!MESSAGE Possible choices for configuration are:

+!MESSAGE 

+!MESSAGE "Queue_Based_Workers - Win32 Release" (based on "Win32 (x86) Console Application")

+!MESSAGE "Queue_Based_Workers - Win32 Debug" (based on "Win32 (x86) Console Application")

+!MESSAGE 

+

+# Begin Project

+# PROP AllowPerConfigDependencies 0

+# PROP Scc_ProjName ""

+# PROP Scc_LocalPath ""

+CPP=cl.exe

+RSC=rc.exe

+

+!IF  "$(CFG)" == "Queue_Based_Workers - Win32 Release"

+

+# PROP BASE Use_MFC 0

+# PROP BASE Use_Debug_Libraries 0

+# PROP BASE Output_Dir "Release"

+# PROP BASE Intermediate_Dir "Release"

+# PROP BASE Target_Dir ""

+# PROP Use_MFC 0

+# PROP Use_Debug_Libraries 0

+# PROP Output_Dir "Release"

+# PROP Intermediate_Dir "Release"

+# PROP Ignore_Export_Lib 0

+# PROP Target_Dir ""

+# ADD BASE CPP /nologo /W3 /GX /O2 /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c

+# ADD CPP /nologo /MD /W3 /GX /O2 /I "..\..\.." /D "WIN32" /D "NDEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD /c

+# ADD BASE RSC /l 0x409 /d "NDEBUG"

+# ADD RSC /l 0x409 /d "NDEBUG"

+BSC32=bscmake.exe

+# ADD BASE BSC32 /nologo

+# ADD BSC32 /nologo

+LINK32=link.exe

+# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /machine:I386

+# ADD LINK32 ace.lib /nologo /subsystem:console /machine:I386 /libpath:"..\..\..\ace"

+

+!ELSEIF  "$(CFG)" == "Queue_Based_Workers - Win32 Debug"

+

+# PROP BASE Use_MFC 0

+# PROP BASE Use_Debug_Libraries 1

+# PROP BASE Output_Dir "Debug"

+# PROP BASE Intermediate_Dir "Debug"

+# PROP BASE Target_Dir ""

+# PROP Use_MFC 0

+# PROP Use_Debug_Libraries 1

+# PROP Output_Dir ""

+# PROP Intermediate_Dir "Debug"

+# PROP Ignore_Export_Lib 0

+# PROP Target_Dir ""

+# ADD BASE CPP /nologo /W3 /Gm /GX /Zi /Od /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD  /c

+# ADD CPP /nologo /MDd /W3 /Gm /GX /Zi /Od /I "..\..\.." /D "WIN32" /D "_DEBUG" /D "_CONSOLE" /D "_MBCS" /YX /FD  /c

+# ADD BASE RSC /l 0x409 /d "_DEBUG"

+# ADD RSC /l 0x409 /d "_DEBUG"

+BSC32=bscmake.exe

+# ADD BASE BSC32 /nologo

+# ADD BSC32 /nologo

+LINK32=link.exe

+# ADD BASE LINK32 kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib kernel32.lib user32.lib gdi32.lib winspool.lib comdlg32.lib advapi32.lib shell32.lib ole32.lib oleaut32.lib uuid.lib odbc32.lib odbccp32.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept

+# ADD LINK32 aced.lib /nologo /subsystem:console /debug /machine:I386 /pdbtype:sept /libpath:"..\..\..\ace"

+

+!ENDIF 

+

+# Begin Target

+

+# Name "Queue_Based_Workers - Win32 Release"

+# Name "Queue_Based_Workers - Win32 Debug"

+# Begin Group "Source Files"

+

+# PROP Default_Filter "cpp;c;cxx;rc;def;r;odl;idl;hpj;bat"

+# Begin Source File

+

+SOURCE=.\workers.cpp

+# End Source File

+# End Group

+# Begin Group "Header Files"

+

+# PROP Default_Filter "h;hpp;hxx;hm;inl"

+# End Group

+# Begin Group "Resource Files"

+

+# PROP Default_Filter "ico;cur;bmp;dlg;rc2;rct;bin;rgs;gif;jpg;jpeg;jpe"

+# End Group

+# End Target

+# End Project

diff --git a/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsw b/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsw
new file mode 100644
index 00000000000..43794eb03f9
--- /dev/null
+++ b/performance-tests/Server_Concurrency/Queue_Based_Workers/Queue_Based_Workers.dsw
@@ -0,0 +1,29 @@
+Microsoft Developer Studio Workspace File, Format Version 6.00

+# WARNING: DO NOT EDIT OR DELETE THIS WORKSPACE FILE!

+

+###############################################################################

+

+Project: "Queue_Based_Workers"=".\Queue_Based_Workers.dsp" - Package Owner=<4>

+

+Package=<5>

+{{{

+}}}

+

+Package=<4>

+{{{

+}}}

+

+###############################################################################

+

+Global:

+

+Package=<5>

+{{{

+}}}

+

+Package=<3>

+{{{

+}}}

+

+###############################################################################

+

diff --git a/performance-tests/Server_Concurrency/Queue_Based_Workers/workers.cpp b/performance-tests/Server_Concurrency/Queue_Based_Workers/workers.cpp
new file mode 100644
index 00000000000..4ff06e020bf
--- /dev/null
+++ b/performance-tests/Server_Concurrency/Queue_Based_Workers/workers.cpp
@@ -0,0 +1,249 @@
+// $Id$
+
+#include "ace/Task.h"
+#include "ace/Get_Opt.h"
+
+static size_t number_of_messages = 100;
+static size_t message_size = 100;
+static size_t number_of_workers = 10;
+static size_t burst_size = 10;
+static size_t timeout_between_bursts = 1;
+static size_t debug = 0;
+
+typedef ACE_Task<ACE_MT_SYNCH> TASK;
+
+class Worker_Task : public TASK
+{
+public:
+  Worker_Task (ACE_Message_Queue<ACE_MT_SYNCH> *mq);
+  int svc (void);
+
+  size_t messages_dequeued_;
+
+};
+
+class IO_Task : public TASK
+{
+public:
+  IO_Task (ACE_Message_Queue<ACE_MT_SYNCH> *mq);
+  int svc (void);
+};
+
+Worker_Task::Worker_Task (ACE_Message_Queue<ACE_MT_SYNCH> *mq)
+  : TASK (0, mq),
+    messages_dequeued_ (0)
+{
+}
+
+int
+Worker_Task::svc (void)
+{
+  for (;;)
+    {
+      ACE_Message_Block *message_block = 0;
+      int result = this->getq (message_block);
+      if (result == -1)
+        {
+          ACE_ERROR_RETURN ((LM_ERROR,
+                             "Worker_Task::svc (%t) -> %p\n",
+                             "getq error"),
+                            -1);
+        }
+
+      ACE_Message_Block::ACE_Message_Type message_type =
+        message_block->msg_type ();
+
+      message_block->release ();
+
+      // If STOP message, break loop and end the task.
+      if (message_type == ACE_Message_Block::MB_STOP)
+        {
+          if (debug)
+            {
+              ACE_DEBUG ((LM_DEBUG,
+                          "(%t) stop message dequeued after %d data messages\n",
+                          this->messages_dequeued_));
+            }
+
+          break;
+        }
+      else
+        {
+          // Counter.
+          ++this->messages_dequeued_;
+
+          if (debug)
+            {
+              ACE_DEBUG ((LM_DEBUG,
+                          "(%t) dequeued its %d message\n",
+                          this->messages_dequeued_));
+            }
+        }
+    }
+
+  return 0;
+}
+
+IO_Task::IO_Task (ACE_Message_Queue<ACE_MT_SYNCH> *mq)
+  : TASK (0, mq)
+{
+}
+
+int
+IO_Task::svc (void)
+{
+  size_t i = 0;
+  size_t messages_queued = 1;
+  size_t burst = 1;
+
+  // Data messages.
+  while (number_of_messages > 0)
+    {
+      for (i = 1;
+           i <= burst_size && number_of_messages > 0;
+           ++i, --number_of_messages, ++messages_queued)
+        {
+          ACE_Message_Block *message_block = 0;
+          ACE_NEW_RETURN (message_block,
+                          ACE_Message_Block (message_size),
+                          -1);
+
+          int result = this->putq (message_block);
+          if (result == -1)
+            {
+              ACE_ERROR_RETURN ((LM_ERROR,
+                                 "IO::svc (%t) -> %p\n",
+                                 "putq error"),
+                                -1);
+            }
+
+          if (debug)
+            {
+              ACE_DEBUG ((LM_DEBUG,
+                          "(%t) IO thread -> burst %d: message %d; overall message %d\n",
+                          burst,
+                          i,
+                          messages_queued));
+            }
+        }
+
+      ++burst;
+      ACE_OS::sleep (timeout_between_bursts);
+    }
+
+  // Terminate messages.
+  for (i = 0; i < number_of_workers; ++i)
+    {
+      ACE_Message_Block *message_block = 0;
+      ACE_NEW_RETURN (message_block,
+                      ACE_Message_Block (0,
+                                         ACE_Message_Block::MB_STOP),
+                      -1);
+
+      int result = this->putq (message_block);
+      if (result == -1)
+        {
+          ACE_ERROR_RETURN ((LM_ERROR,
+                             "IO::svc (%t) -> %p\n",
+                             "putq error"),
+                            -1);
+        }
+    }
+
+  return 0;
+}
+
+static int
+parse_args (int argc, ASYS_TCHAR *argv[])
+{
+  ACE_Get_Opt get_opt (argc, argv, ASYS_TEXT ("m:s:w:b:t:d"));
+  int c;
+
+  while ((c = get_opt ()) != -1)
+    {
+      switch (c)
+        {
+        case 'm':
+          number_of_messages = ACE_OS::atoi (get_opt.optarg);
+          break;
+        case 's':
+          message_size = ACE_OS::atoi (get_opt.optarg);
+          break;
+        case 'w':
+          number_of_workers = ACE_OS::atoi (get_opt.optarg);
+          break;
+        case 'b':
+          burst_size = ACE_OS::atoi (get_opt.optarg);
+          break;
+        case 't':
+          timeout_between_bursts = ACE_OS::atoi (get_opt.optarg);
+          break;
+        case 'd':
+          debug = 1;
+          break;
+        default:
+          ACE_ERROR_RETURN ((LM_ERROR,
+                             "usage: %s\n"
+                             "\t[-m number of messages]\n"
+                             "\t[-s message size]\n"
+                             "\t[-w number of workers]\n"
+                             "\t[-b burst size]\n"
+                             "\t[-t timeout between bursts]\n"
+                             "\t[-d debug]\n",
+                             argv[0]),
+                            -1);
+        }
+    }
+
+  return 0;
+}
+
+int
+main (int argc, ASYS_TCHAR *argv[])
+{
+  int result = parse_args (argc, argv);
+  if (result != 0)
+    {
+      return result;
+    }
+
+  ACE_Message_Queue<ACE_MT_SYNCH> message_queue;
+
+  // Workers.
+  Worker_Task **workers = 0;
+  ACE_NEW_RETURN (workers,
+                  Worker_Task *[number_of_workers],
+                  -1);
+
+  // Create and activate them.
+  for (size_t i = 0; i < number_of_workers; ++i)
+    {
+      ACE_NEW_RETURN (workers[i],
+                      Worker_Task (&message_queue),
+                      -1);
+
+      // Activate the workers.
+      result = workers[i]->activate (THR_BOUND);
+      if (result != 0)
+        {
+          return result;
+        }
+    }
+
+  // IO Task.
+  IO_Task io (&message_queue);
+
+  // Activate the workers.
+  result = io.activate (THR_BOUND);
+  if (result != 0)
+    {
+      return result;
+    }
+
+  // Wait for all threads to terminate.
+  result = ACE_Thread_Manager::instance ()->wait ();
+
+  delete[] workers;
+
+  return result;
+}