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 ()
  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       else // result == -1
 162         {
 163           // These #ifs should generally match the logic in OS_NS_Thread.inl.
 164
 165 # if defined (ACE_HAS_PTHREADS_UNIX98_EXT) || !defined (ACE_LACKS_RWLOCK_T)
 166           // In this case we have native RW locks support, but native RW
 167           // locks may not support upgrading!
 168           ACE_DEBUG ((LM_DEBUG,
 169                       ACE_TEXT ("(%t) %p\n"),
 170                       ACE_TEXT ("can't upgrade write lock"),
 171                       1));
 172 # else
 173           // In this case we do not have native RW locks, but therefore the
 174           // emulation, which supports upgradable write locks.
 175           ACE_ERROR ((LM_ERROR,
 176                       ACE_TEXT ("(%t) %p\n"),
 177                       ACE_TEXT ("failure in upgrading to write lock"),
 178                       1));
 179 # endif /* ACE_HAS_PTHREADS_UNIX98_EXT || !ACE_LACKS_RWLOCK_T */
 180         }
 181       //ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) done with reading guarded data\n")));
 182
 183       ACE_DEBUG ((LM_DEBUG,
 184                   ACE_TEXT ("(%t) reader finished %d iterations at %T\n"),
 185                   iterations));
 186     }
 187   return 0;
 188 }
 189
 190 // Iterate <n_iterations> each time modifying the global data
 191 // and checking that nobody steps on it while we can write it.
 192
 193 static void *
 194 writer (void *)
 195 {
 196   ACE_DEBUG ((LM_DEBUG, ACE_TEXT (" (%t) writer starting\n")));
 197
 198   // We use a random pause, around 2msec with 1msec jittering.
 199   int usecs = 1000 + ACE_OS::rand() % 2000;
 200   ACE_Time_Value pause(0, usecs);
 201
 202   for (size_t iterations = 1; iterations <= n_iterations; iterations++)
 203     {
 204       ACE_OS::sleep (pause);
 205
 206       ACE_WRITE_GUARD_RETURN (ACE_RW_Thread_Mutex, g, rw_mutex, 0);
 207
 208       ++current_writers;
 209
 210       if (current_writers > 1)
 211         ACE_DEBUG ((LM_DEBUG,
 212                     ACE_TEXT (" (%t) other writers found!!!\n")));
 213
 214       if (current_readers > 0)
 215         ACE_DEBUG ((LM_DEBUG,
 216                     ACE_TEXT (" (%t) readers found!!!\n")));
 217
 218       ACE_thread_t self = ACE_Thread::self ();
 219
 220       shared_data = self;
 221
 222       for (size_t loop = 1; loop <= n_loops; loop++)
 223         {
 224           ACE_Thread::yield ();
 225
 226           if (!ACE_OS::thr_equal (shared_data, self))
 227             ACE_DEBUG ((LM_DEBUG,
 228                         ACE_TEXT (" (%t) somebody wrote on my data %d\n"),
 229                         shared_data));
 230         }
 231
 232       --current_writers;
 233
 234       ACE_DEBUG((LM_DEBUG, ACE_TEXT (" (%t) write %d done at %T\n"), iterations));
 235     }
 236   return 0;
 237 }
 238
 239 #endif /* ACE_HAS_THREADS */
 240
 241 // Spawn off threads.
 242
 243 int run_main (int argc, ACE_TCHAR *argv[])
 244 {
 245   ACE_START_TEST (ACE_TEXT ("Reader_Writer_Test"));
 246
 247 #if defined (ACE_HAS_THREADS)
 248   parse_args (argc, argv);
 249
 250   current_readers = 0; // Possibly already done
 251   current_writers = 0; // Possibly already done
 252
 253   ACE_DEBUG ((LM_DEBUG,
 254               ACE_TEXT (" (%t) main thread starting\n")));
 255
 256   if (ACE_Thread_Manager::instance ()->spawn_n (n_readers,
 257                                                ACE_THR_FUNC (reader),
 258                                                0,
 259                                                THR_NEW_LWP) == -1)
 260     ACE_ERROR_RETURN ((LM_ERROR,
 261                        ACE_TEXT ("%p\n"),
 262                        ACE_TEXT ("spawn_n")), 1);
 263   else if (ACE_Thread_Manager::instance ()->spawn_n (n_writers,
 264                                                     ACE_THR_FUNC (writer),
 265                                                     0,
 266                                                     THR_NEW_LWP) == -1)
 267     ACE_ERROR_RETURN ((LM_ERROR,
 268                        ACE_TEXT ("%p\n"),
 269                        ACE_TEXT ("spawn_n")), 1);
 270
 271   ACE_Thread_Manager::instance ()->wait ();
 272
 273   ACE_DEBUG ((LM_DEBUG,
 274               ACE_TEXT (" (%t) exiting main thread\n")));
 275 #else
 276   ACE_UNUSED_ARG (argc);
 277   ACE_UNUSED_ARG (argv);
 278   ACE_ERROR ((LM_INFO,
 279               ACE_TEXT ("threads not supported on this platform\n")));
 280 #endif /* ACE_HAS_THREADS */
 281   ACE_END_TEST;
 282   return 0;
 283 }