ACE/protocols/tests/INet/MT_Get/Main.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "ace/Task.h"
   3 #include "ace/Containers_T.h"
   4 #include "ace/Reactor.h"
   5 #include "ace/INet/HTTP_URL.h"
   6 #include "ace/INet/HTTP_ClientRequestHandler.h"
   7 #include "ace/INet/String_IOStream.h"
   8 #include "ace/Truncate.h"
   9 #include <iostream>
  10
  11 int n_threads =  3;
  12
  13 #if defined (ACE_HAS_THREADS)
  14 class Get_Task : public ACE_Task<ACE_MT_SYNCH>
  15 {
  16 public:
  17   Get_Task (ACE_Thread_Manager *thr_mgr,
  18             int n_threads);
  19
  20   virtual int svc ();
  21
  22   const ACE_Array<ACE_CString>& results ();
  23
  24 private:
  25   void shutdown ();
  26
  27   int n_threads_;
  28   ACE_Array<ACE_CString>  results_;
  29   ACE_SYNCH_MUTEX lock_;
  30 };
  31
  32 Get_Task::Get_Task (ACE_Thread_Manager *thr_mgr,
  33                     int n_threads)
  34   : ACE_Task<ACE_MT_SYNCH> (thr_mgr),
  35     n_threads_ (n_threads)
  36 {
  37   // Create worker threads.
  38   if (this->activate (THR_NEW_LWP, n_threads_) == -1)
  39     ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("activate failed")));
  40   else
  41     ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) started %d threads\n"), n_threads_));
  42 }
  43
  44 void Get_Task::shutdown ()
  45 {
  46   ACE_GUARD (ACE_SYNCH_MUTEX,
  47              guard_,
  48              lock_);
  49
  50   --n_threads_;
  51   if (n_threads_ <= 0)
  52     {
  53       ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) ending event loop\n")));
  54       ACE_Reactor::instance ()->end_event_loop ();
  55     }
  56 }
  57
  58 int Get_Task::svc ()
  59 {
  60   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) task started\n")));
  61
  62   ACE::HTTP::URL http_url;
  63   if (http_url.parse ("http://www.remedy.nl"))
  64     {
  65       ACE::HTTP::ClientRequestHandler rh;
  66       ACE::INet::URLStream urlin = http_url.open (rh);
  67       if (urlin)
  68         {
  69           ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) opened URL %C\n"), http_url.to_string ().c_str ()));
  70
  71           ACE::IOS::CString_OStream sos;
  72           sos << urlin->rdbuf ();
  73           ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) result downloaded\n")));
  74           if (!sos.bad ())
  75             {
  76               ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
  77                                 guard_,
  78                                 lock_,
  79                                 0);
  80               ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) storing result\n")));
  81               results_.size (results_.size () + 1);
  82               results_[results_.size ()-1] = sos.str ();
  83             }
  84         }
  85       else
  86           ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) failed to open URL %C\n"), http_url.to_string ().c_str ()));
  87       shutdown ();
  88     }
  89   else
  90     {
  91       ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) failed to parse URL : result = %C\n"),
  92                             http_url.to_string ().c_str ()));
  93     }
  94
  95   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) task finished\n")));
  96
  97   return 0;
  98 }
  99
 100 const ACE_Array<ACE_CString>& Get_Task::results ()
 101 {
 102   return results_;
 103 }
 104
 105 class Get_MultiTask : public ACE_Task<ACE_MT_SYNCH>
 106 {
 107 public:
 108   Get_MultiTask (ACE_Thread_Manager *thr_mgr);
 109
 110   virtual int svc ();
 111
 112   int failures ();
 113
 114 private:
 115   const char* get_url ();
 116
 117   void wait_for_all ();
 118
 119   void shutdown ();
 120
 121   int n_threads_;
 122   int n_thread_starts_;
 123   int n_open_urls_;
 124   int n_fails_;
 125   ACE_SYNCH_MUTEX lock_;
 126   ACE_SYNCH_CONDITION signal_;
 127 };
 128
 129 const char* urls_[] = {
 130   "http://www.theaceorb.nl",
 131   "http://www.remedy.nl",
 132   "http://www.google.com"
 133 };
 134
 135 Get_MultiTask::Get_MultiTask (ACE_Thread_Manager *thr_mgr)
 136   : ACE_Task<ACE_MT_SYNCH> (thr_mgr),
 137     n_threads_ (sizeof (urls_)/sizeof (const char*)),
 138     n_thread_starts_ (0),
 139     n_open_urls_ (0),
 140     n_fails_ (n_threads_),
 141     signal_ (lock_)
 142 {
 143   // Create worker threads.
 144   if (this->activate (THR_NEW_LWP, n_threads_) == -1)
 145     ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("activate failed")));
 146 }
 147
 148 const char* Get_MultiTask::get_url ()
 149 {
 150   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
 151                     guard_,
 152                     lock_,
 153                     0);
 154
 155   const char* url = urls_[n_thread_starts_];
 156   ++n_thread_starts_;
 157   return url;
 158 }
 159
 160 void Get_MultiTask::wait_for_all ()
 161 {
 162   ACE_GUARD (ACE_SYNCH_MUTEX,
 163              guard_,
 164              lock_);
 165
 166   ++n_open_urls_;
 167
 168   if (n_open_urls_ == n_threads_)
 169     signal_.broadcast (); // signal all threads that all urls are opened
 170   else
 171     signal_.wait (); // wait for all urls to be opened
 172 }
 173
 174 void Get_MultiTask::shutdown ()
 175 {
 176   ACE_GUARD (ACE_SYNCH_MUTEX,
 177              guard_,
 178              lock_);
 179
 180   --n_threads_;
 181   if (n_threads_ <= 0)
 182     {
 183       ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) ending event loop\n")));
 184       ACE_Reactor::instance ()->end_event_loop ();
 185     }
 186 }
 187
 188 int Get_MultiTask::svc ()
 189 {
 190   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) multitask started\n")));
 191
 192   ACE::HTTP::URL http_url;
 193   const char* url = get_url ();
 194
 195   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) parsing URL %C\n"), url));
 196
 197   if (http_url.parse (url))
 198     {
 199       ACE::HTTP::ClientRequestHandler rh;
 200       http_url.open (rh);
 201       if (rh.response_stream ())
 202           ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) opened URL %C\n"), http_url.to_string ().c_str ()));
 203       else
 204           ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) failed to open URL %C\n"), http_url.to_string ().c_str ()));
 205       wait_for_all ();
 206       if (rh.response_stream ())
 207         {
 208           ACE::IOS::CString_OStream sos;
 209           sos << rh.response_stream ().rdbuf ();
 210           ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) result downloaded\n")));
 211           if (!sos.bad ())
 212             {
 213               ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
 214                                 guard_,
 215                                 lock_,
 216                                 0);
 217               ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) marking result\n")));
 218               --n_fails_;
 219             }
 220         }
 221       shutdown ();
 222     }
 223
 224   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) task finished\n")));
 225
 226   return 0;
 227 }
 228
 229 int Get_MultiTask::failures ()
 230 {
 231   return n_fails_;
 232 }
 233
 234 #endif
 235
 236 int
 237 ACE_TMAIN (int, ACE_TCHAR *[])
 238 {
 239 #if defined (ACE_HAS_THREADS)
 240   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) MGet_Test started\n")));
 241
 242   ACE_Reactor* reactor = ACE_Reactor::instance ();
 243
 244 #if 1
 245   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) starting serialized test\n")));
 246
 247   Get_Task get_task (ACE_Thread_Manager::instance (),
 248                      n_threads);
 249
 250   // run event loop until ended by last task thread
 251   reactor->run_event_loop ();
 252
 253   // all threads really ended?
 254   ACE_Thread_Manager::instance ()->wait ();
 255
 256   // All threads should have resulted in the same data...
 257   if (get_task.results ().size () != ACE_Utils::truncate_cast<size_t> (n_threads))
 258     {
 259       ACE_ERROR ((LM_ERROR,
 260                   ACE_TEXT ("%d results found; ")
 261                   ACE_TEXT ("should be %d\n"),
 262                   get_task.results ().size (),
 263                   n_threads));
 264     }
 265   else
 266     {
 267       const ACE_Array<ACE_CString>& results = get_task.results ();
 268       for (int i=1;
 269            i < n_threads;
 270            ++i)
 271         {
 272           if (results[0] != results[i])
 273             {
 274               ACE_ERROR ((LM_ERROR,
 275                           ACE_TEXT ("result %d does not match other results\n"),
 276                           i));
 277             }
 278         }
 279     }
 280 #endif
 281
 282   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) starting parallelized test\n")));
 283
 284   reactor->reset_event_loop ();
 285
 286   Get_MultiTask get_multi_task (ACE_Thread_Manager::instance ());
 287
 288   // run event loop until ended by last task thread
 289   reactor->run_event_loop ();
 290
 291   // all threads really ended?
 292   ACE_Thread_Manager::instance ()->wait ();
 293
 294   // All threads should have resulted in data...
 295   if (get_multi_task.failures () > 0)
 296     {
 297       ACE_ERROR ((LM_ERROR,
 298                   ACE_TEXT ("%d failures found"),
 299                   get_multi_task.failures ()));
 300     }
 301
 302   ACE_DEBUG ((LM_INFO, ACE_TEXT ("(%P|%t) MGet_Test finished\n")));
 303 #else
 304   ACE_ERROR ((LM_INFO,
 305               ACE_TEXT ("threads not supported on this platform\n")));
 306 #endif /* ACE_HAS_THREADS */
 307   return 0;
 308 }