ACE/tests/Reader_Writer_Test.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    Reader_Writer_Test.cpp
   5  *
   6  *    This test program verifies the functionality of the ACE_OS
   7  *    implementation of readers/writer locks on Win32 and Posix
   8  *    pthreads.
   9  *
  10  *  @author Prashant Jain <pjain@cs.wustl.edu> and Doug C. Schmidt <d.schmidt@vanderbilt.edu>
  11  */
  12 //=============================================================================
  13
  14 #include "test_config.h"
  15 #include "ace/Thread.h"
  16 #include "ace/Thread_Manager.h"
  17 #include "ace/Get_Opt.h"
  18 #include "ace/Atomic_Op.h"
  19 #include "ace/OS_NS_unistd.h"
  20 #include "ace/Guard_T.h"
  21 #include "ace/RW_Thread_Mutex.h"
  22 #include "ace/Time_Value.h"
  23
  24 #if defined (ACE_HAS_THREADS)
  25
  26 // Default number of iterations.
  27 #if defined (VXWORKS)
  28   // So the test doesn't run for too long . . .
  29   static size_t n_iterations = 25;
  30 #else
  31   static size_t n_iterations = 50;
  32 #endif /* VXWORKS */
  33
  34 // Default number of loops.
  35 #if defined (VXWORKS)
  36   // thr_yield () and/or thr_equal () are expensive on VxWorks, apparently.
  37   static size_t n_loops = 10;
  38 #else
  39   static size_t n_loops = 100;
  40 #endif /* VXWORKS */
  41
  42 // Default number of readers.
  43 static size_t n_readers = 6;
  44
  45 // Default number of writers.
  46 static size_t n_writers = 4;
  47
  48 // Thread id of last writer.
  49 static ACE_thread_t shared_data;
  50
  51 // Lock for shared_data.
  52 static ACE_RW_Thread_Mutex rw_mutex;
  53
  54 // Count of the number of readers and writers.
  55 static ACE_Atomic_Op<ACE_Thread_Mutex, long> current_readers;
  56 static ACE_Atomic_Op<ACE_Thread_Mutex, long> current_writers;
  57
  58 // Explain usage and exit.
  59 static void
  60 print_usage_and_die (void)
  61 {
  62   ACE_DEBUG ((LM_DEBUG,
  63               ACE_TEXT ("usage: %n [-r n_readers] [-w n_writers] [-n iteration_count]\n")));
  64   ACE_OS::exit (1);
  65 }
  66
  67 static void
  68 parse_args (int argc, ACE_TCHAR *argv[])
  69 {
  70   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("r:w:n:"));
  71
  72   int c;
  73
  74   while ((c = get_opt ()) != -1)
  75     switch (c)
  76     {
  77     case 'r':
  78       n_readers = ACE_OS::atoi (get_opt.opt_arg ());
  79       break;
  80     case 'w':
  81       n_writers = ACE_OS::atoi (get_opt.opt_arg ());
  82       break;
  83     case 'n':
  84       n_iterations = ACE_OS::atoi (get_opt.opt_arg ());
  85       break;
  86     default:
  87       print_usage_and_die ();
  88       break;
  89   }
  90 }
  91
  92 // Iterate <n_iterations> each time checking that nobody modifies the data
  93 // while we have a read lock.
  94
  95 static void *
  96 reader (void *)
  97 {
  98   ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) reader starting\n")));
  99
 100   // We use a random pause, around 2msec with 1msec jittering.
 101   int usecs = 1000 + ACE_OS::rand() % 2000;
 102   ACE_Time_Value pause(0, usecs);
 103
 104   for (size_t iterations = 1; iterations <= n_iterations; iterations++)
 105     {
 106       ACE_OS::sleep (pause);
 107       ACE_READ_GUARD_RETURN (ACE_RW_Thread_Mutex, g, rw_mutex, 0);
 108       // int n = ++current_readers;
 109       // ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) I'm reader number %d\n"), n));
 110
 111       if (current_writers > 0)
 112         ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) writers found!!!\n")));
 113
 114       ACE_thread_t data = shared_data;
 115
 116       for (size_t loop = 1; loop <= n_loops; loop++)
 117         {
 118           ACE_Thread::yield ();
 119
 120           if (!ACE_OS::thr_equal (shared_data, data))
 121             ACE_DEBUG ((LM_DEBUG,
 122                         ACE_TEXT (" (%t) somebody changed %d to %d\n"),
 123                         data, shared_data));
 124         }
 125
 126       int result = rw_mutex.tryacquire_write_upgrade ();
 127
 128       if (result == 0)
 129         {
 130           --current_readers;
 131           ++current_writers;
 132
 133           ACE_DEBUG ((LM_DEBUG,
 134                       ACE_TEXT ("(%t) upgraded to write lock!\n")));
 135
 136           ACE_thread_t self = ACE_Thread::self ();
 137
 138           shared_data = self;
 139           data = self;
 140
 141           for (size_t loop = 1; loop <= n_loops; loop++)
 142             {
 143               if (ACE_OS::thr_equal (shared_data, data) == 0)
 144                 ACE_DEBUG ((LM_DEBUG,
 145                             ACE_TEXT ("(%t) upgraded writer error: somebody changed %d to %d\n"),
 146                             data,
 147                             shared_data));
 148             }
 149
 150           --current_writers;
 151           // we were a writer
 152         }
 153       else if (result == -1 && errno == EBUSY)
 154         {
 155           --current_readers;
 156           // we were still a reader
 157
 158           ACE_DEBUG ((LM_DEBUG,
 159                       ACE_TEXT ("(%t) could not upgrade to write lock!\n")));
 160
 161         }
 162       else // result == -1
 163         {
 164           // These #ifs should generally match the logic in OS_NS_Thread.inl.
 165
 166 # if defined (ACE_HAS_PTHREADS_UNIX98_EXT) || !defined (ACE_LACKS_RWLOCK_T)
 167           // In this case we have native RW locks support, but native RW
 168           // locks may not support upgrading!
 169           ACE_DEBUG ((LM_DEBUG,
 170                       ACE_TEXT ("(%t) %p\n"),
 171                       ACE_TEXT ("can't upgrade write lock"),
 172                       1));
 173 # else
 174           // In this case we do not have native RW locks, but therefore the
 175           // emulation, which supports upgradable write locks.
 176           ACE_ERROR ((LM_ERROR,
 177                       ACE_TEXT ("(%t) %p\n"),
 178                       ACE_TEXT ("failure in upgrading to write lock"),
 179                       1));
 180 # endif /* ACE_HAS_PTHREADS_UNIX98_EXT || !ACE_LACKS_RWLOCK_T */
 181         }
 182       //ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) done with reading guarded data\n")));
 183
 184       ACE_DEBUG ((LM_DEBUG,
 185                   ACE_TEXT ("(%t) reader finished %d iterations at %T\n"),
 186                   iterations));
 187     }
 188   return 0;
 189 }
 190
 191 // Iterate <n_iterations> each time modifying the global data
 192 // and checking that nobody steps on it while we can write it.
 193
 194 static void *
 195 writer (void *)
 196 {
 197   ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) writer starting\n")));
 198
 199   // We use a random pause, around 2msec with 1msec jittering.
 200   int usecs = 1000 + ACE_OS::rand() % 2000;
 201   ACE_Time_Value pause(0, usecs);
 202
 203   for (size_t iterations = 1; iterations <= n_iterations; iterations++)
 204     {
 205       ACE_OS::sleep (pause);
 206
 207       ACE_WRITE_GUARD_RETURN (ACE_RW_Thread_Mutex, g, rw_mutex, 0);
 208
 209       ++current_writers;
 210
 211       if (current_writers > 1)
 212         ACE_DEBUG ((LM_DEBUG,
 213                     ACE_TEXT (" (%t) other writers found!!!\n")));
 214
 215       if (current_readers > 0)
 216         ACE_DEBUG ((LM_DEBUG,
 217                     ACE_TEXT (" (%t) readers found!!!\n")));
 218
 219       ACE_thread_t self = ACE_Thread::self ();
 220
 221       shared_data = self;
 222
 223       for (size_t loop = 1; loop <= n_loops; loop++)
 224         {
 225           ACE_Thread::yield ();
 226
 227           if (!ACE_OS::thr_equal (shared_data, self))
 228             ACE_DEBUG ((LM_DEBUG,
 229                         ACE_TEXT (" (%t) somebody wrote on my data %d\n"),
 230                         shared_data));
 231         }
 232
 233       --current_writers;
 234
 235       ACE_DEBUG((LM_DEBUG, ACE_TEXT (" (%t) write %d done at %T\n"), iterations));
 236     }
 237   return 0;
 238 }
 239
 240 #endif /* ACE_HAS_THREADS */
 241
 242 // Spawn off threads.
 243
 244 int run_main (int argc, ACE_TCHAR *argv[])
 245 {
 246   ACE_START_TEST (ACE_TEXT ("Reader_Writer_Test"));
 247
 248 #if defined (ACE_HAS_THREADS)
 249   parse_args (argc, argv);
 250
 251   current_readers = 0; // Possibly already done
 252   current_writers = 0; // Possibly already done
 253
 254   ACE_DEBUG ((LM_DEBUG,
 255               ACE_TEXT (" (%t) main thread starting\n")));
 256
 257   if (ACE_Thread_Manager::instance ()->spawn_n (n_readers,
 258                                                ACE_THR_FUNC (reader),
 259                                                0,
 260                                                THR_NEW_LWP) == -1)
 261     ACE_ERROR_RETURN ((LM_ERROR,
 262                        ACE_TEXT ("%p\n"),
 263                        ACE_TEXT ("spawn_n")), 1);
 264   else if (ACE_Thread_Manager::instance ()->spawn_n (n_writers,
 265                                                     ACE_THR_FUNC (writer),
 266                                                     0,
 267                                                     THR_NEW_LWP) == -1)
 268     ACE_ERROR_RETURN ((LM_ERROR,
 269                        ACE_TEXT ("%p\n"),
 270                        ACE_TEXT ("spawn_n")), 1);
 271
 272   ACE_Thread_Manager::instance ()->wait ();
 273
 274   ACE_DEBUG ((LM_DEBUG,
 275               ACE_TEXT (" (%t) exiting main thread\n")));
 276 #else
 277   ACE_UNUSED_ARG (argc);
 278   ACE_UNUSED_ARG (argv);
 279   ACE_ERROR ((LM_INFO,
 280               ACE_TEXT ("threads not supported on this platform\n")));
 281 #endif /* ACE_HAS_THREADS */
 282   ACE_END_TEST;
 283   return 0;
 284 }