ACE/tests/Task_Ex_Test.cpp

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