ACE/tests/Task_Ex_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Task_Ex_Test.cpp
   5  *
   6  *  This test program illustrates the ACE_Task_Ex class which has the ACE_Message_Queue_Ex
   7  *  that has the capability to hold user-defined messages instead of ACE_Message_Block
   8  *
   9  *  @author Kobi Cohen-Arazi <kobi-co@barak-online.net>
  10  */
  11 //=============================================================================
  12
  13
  14 #include "test_config.h"
  15 #include "Task_Ex_Test.h"
  16 #include "ace/Task_Ex_T.h"
  17 #include "ace/Log_Msg.h"
  18 #include "ace/Auto_Ptr.h"
  19
  20
  21
  22 #if defined (ACE_HAS_THREADS)
  23
  24 /// default params
  25 #if defined (ACE_VXWORKS) || defined (ACE_LYNXOS_MAJOR)
  26 // this is a very expensive test on VxWorks so limit it otherwise it will never finish in time:-)
  27 const ACE_INT32 PRODUCER_THREADS_NO=10;
  28 const ACE_INT32 CONSUMER_THREADS_NO=10;
  29 const ACE_INT32 NUMBER_OF_MSGS=200;
  30 #else
  31 const ACE_INT32 PRODUCER_THREADS_NO=20;
  32 const ACE_INT32 CONSUMER_THREADS_NO=20;
  33 const ACE_INT32 NUMBER_OF_MSGS=2000;
  34 #endif
  35
  36 /// @class Consumer consumes user defined Msgs
  37 class Consumer : public ACE_Task_Ex<ACE_MT_SYNCH, User_Defined_Msg>
  38 {
  39 public:
  40   //FUZZ: disable check_for_lack_ACE_OS
  41   /// activate/spawn the threads.
  42   ///FUZZ: enable check_for_lack_ACE_OS
  43   int open (void*);
  44
  45   /// svc thread entry point
  46   virtual int svc (void);
  47 private:
  48
  49 };
  50
  51 int Consumer::open (void*)
  52 {
  53   if(this->activate (THR_NEW_LWP | THR_JOINABLE,
  54                      CONSUMER_THREADS_NO)==-1)
  55   {
  56     ACE_ERROR_RETURN((LM_ERROR,
  57                       ACE_TEXT("Consumer::open Error spanwing thread %p\n"),
  58                       "err="),
  59                      -1);
  60   }
  61   return 0;
  62 }
  63
  64 int Consumer::svc ()
  65 {
  66   User_Defined_Msg* pMsg=0;
  67   while(this->getq (pMsg)!=-1)
  68   {
  69     ACE_TEST_ASSERT (pMsg!=0);
  70 #if defined (ACE_HAS_CPP11)
  71     std::unique_ptr<User_Defined_Msg> pAuto(pMsg);
  72 #else
  73     auto_ptr<User_Defined_Msg> pAuto(pMsg);
  74 #endif /* ACE_HAS_CPP11 */
  75     ACE_DEBUG((LM_DEBUG,
  76                ACE_TEXT("Consumer::svc got msg id=%d\n"),
  77                pMsg->msg_id ()));
  78     if(pMsg->msg_id ()==NUMBER_OF_MSGS-1)
  79       break;
  80   }
  81
  82   ACE_DEBUG((LM_INFO,
  83              ACE_TEXT("Consumer::svc ended thread %t\n")));
  84
  85   return 0;
  86 }
  87
  88
  89 /// producer function produces user defined messages.
  90 ACE_THR_FUNC_RETURN producer (void *arg)
  91 {
  92   Consumer* c = static_cast<Consumer*> (arg);
  93   ACE_TEST_ASSERT(c!=0);
  94   if (c==0)
  95   {
  96     ACE_ERROR((LM_ERROR,
  97                ACE_TEXT("producer Error casting to consumer\n")));
  98     return (ACE_THR_FUNC_RETURN)-1;
  99   }
 100   for (int i=0;i!=NUMBER_OF_MSGS;++i)
 101   {
 102     User_Defined_Msg* pMsg=0;
 103     ACE_NEW_NORETURN(pMsg, User_Defined_Msg(i));
 104     if (pMsg==0)
 105     {
 106       ACE_ERROR((LM_ERROR,
 107                  ACE_TEXT("producer Error allocating data %p\n"),
 108                  "err="));
 109       return (ACE_THR_FUNC_RETURN)-1;
 110     }
 111     if(c->putq (pMsg)==-1)
 112     {
 113       ACE_ERROR((LM_ERROR,
 114                  ACE_TEXT("producer Error putq data %p\n"),
 115                  "err="));
 116       return (ACE_THR_FUNC_RETURN)-1;
 117     }
 118   }
 119   return 0;
 120 }
 121
 122 #endif /* ACE_HAS_THREADS */
 123
 124 int
 125 run_main (int, ACE_TCHAR *[])
 126 {
 127   ACE_START_TEST (ACE_TEXT ("Task_Ex_Test"));
 128
 129 #if defined (ACE_HAS_THREADS)
 130
 131   Consumer c;
 132   if(c.open (0)==-1)
 133     ACE_ERROR_RETURN((LM_ERROR,
 134                       ACE_TEXT ("main Error opening consumer\n")),-1);
 135
 136
 137   int result=ACE_Thread_Manager::instance()->spawn_n (PRODUCER_THREADS_NO,
 138                                                       ACE_THR_FUNC(producer),
 139                                                       static_cast<void*> (&c));
 140   if (result==-1)
 141   {
 142     ACE_ERROR_RETURN((LM_ERROR,
 143                       ACE_TEXT ("main Error spawning threads %p\n"),
 144                       "err="),-1);
 145   }
 146
 147   // wait all threads
 148   int wait_result=ACE_Thread_Manager::instance()->wait();
 149   if (wait_result==-1)
 150   {
 151     ACE_ERROR((LM_ERROR,
 152                ACE_TEXT("main Error Thread_Manager->wait %p\n"),
 153                "err="));
 154     return -1;
 155   }
 156 #else
 157   ACE_ERROR ((LM_INFO,
 158               ACE_TEXT ("threads not supported on this platform\n")));
 159 #endif /* ACE_HAS_THREADS */
 160   ACE_END_TEST;
 161   return 0;
 162 }