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 int Consumer::open (void*)
  47 {
  48   if(this->activate (THR_NEW_LWP | THR_JOINABLE,
  49                      CONSUMER_THREADS_NO)==-1)
  50   {
  51     ACE_ERROR_RETURN((LM_ERROR,
  52                       ACE_TEXT("Consumer::open Error spanwing thread %p\n"),
  53                       "err="),
  54                      -1);
  55   }
  56   return 0;
  57 }
  58
  59 int Consumer::svc ()
  60 {
  61   User_Defined_Msg* pMsg=0;
  62   while(this->getq (pMsg)!=-1)
  63   {
  64     ACE_TEST_ASSERT (pMsg!=0);
  65     std::unique_ptr<User_Defined_Msg> pAuto(pMsg);
  66     ACE_DEBUG((LM_DEBUG,
  67                ACE_TEXT("Consumer::svc got msg id=%d\n"),
  68                pMsg->msg_id ()));
  69     if(pMsg->msg_id ()==NUMBER_OF_MSGS-1)
  70       break;
  71   }
  72
  73   ACE_DEBUG((LM_INFO,
  74              ACE_TEXT("Consumer::svc ended thread %t\n")));
  75
  76   return 0;
  77 }
  78
  79
  80 /// producer function produces user defined messages.
  81 ACE_THR_FUNC_RETURN producer (void *arg)
  82 {
  83   Consumer* c = static_cast<Consumer*> (arg);
  84   ACE_TEST_ASSERT(c!=0);
  85   if (c==0)
  86   {
  87     ACE_ERROR((LM_ERROR,
  88                ACE_TEXT("producer Error casting to consumer\n")));
  89     return (ACE_THR_FUNC_RETURN)-1;
  90   }
  91   for (int i=0;i!=NUMBER_OF_MSGS;++i)
  92   {
  93     User_Defined_Msg* pMsg=0;
  94     ACE_NEW_NORETURN(pMsg, User_Defined_Msg(i));
  95     if (pMsg==0)
  96     {
  97       ACE_ERROR((LM_ERROR,
  98                  ACE_TEXT("producer Error allocating data %p\n"),
  99                  "err="));
 100       return (ACE_THR_FUNC_RETURN)-1;
 101     }
 102     if(c->putq (pMsg)==-1)
 103     {
 104       ACE_ERROR((LM_ERROR,
 105                  ACE_TEXT("producer Error putq data %p\n"),
 106                  "err="));
 107       return (ACE_THR_FUNC_RETURN)-1;
 108     }
 109   }
 110   return 0;
 111 }
 112
 113 #endif /* ACE_HAS_THREADS */
 114
 115 int
 116 run_main (int, ACE_TCHAR *[])
 117 {
 118   ACE_START_TEST (ACE_TEXT ("Task_Ex_Test"));
 119
 120 #if defined (ACE_HAS_THREADS)
 121
 122   Consumer c;
 123   if(c.open (0)==-1)
 124     ACE_ERROR_RETURN((LM_ERROR,
 125                       ACE_TEXT ("main Error opening consumer\n")),-1);
 126
 127
 128   int result=ACE_Thread_Manager::instance()->spawn_n (PRODUCER_THREADS_NO,
 129                                                       ACE_THR_FUNC(producer),
 130                                                       static_cast<void*> (&c));
 131   if (result==-1)
 132   {
 133     ACE_ERROR_RETURN((LM_ERROR,
 134                       ACE_TEXT ("main Error spawning threads %p\n"),
 135                       "err="),-1);
 136   }
 137
 138   // wait all threads
 139   int wait_result=ACE_Thread_Manager::instance()->wait();
 140   if (wait_result==-1)
 141   {
 142     ACE_ERROR((LM_ERROR,
 143                ACE_TEXT("main Error Thread_Manager->wait %p\n"),
 144                "err="));
 145     return -1;
 146   }
 147 #else
 148   ACE_ERROR ((LM_INFO,
 149               ACE_TEXT ("threads not supported on this platform\n")));
 150 #endif /* ACE_HAS_THREADS */
 151   ACE_END_TEST;
 152   return 0;
 153 }