ACE/examples/APG/Shared_Memory/Pool_Growth.cpp

   1 #include "ace/OS_NS_stdio.h"
   2 #include "ace/OS_NS_unistd.h"
   3 #include "ace/Malloc_T.h"
   4 #include "ace/PI_Malloc.h"
   5 #include "ace/Process_Mutex.h"
   6 #include "ace/Process.h"
   7 #include "ace/Unbounded_Queue.h"
   8 #include "ace/MMAP_Memory_Pool.h"
   9
  10 #define BACKING_STORE "queue.dat"
  11 #define QUEUE_NAME "queue.db"
  12
  13 typedef ACE_Allocator_Adapter<ACE_Malloc <ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex> > ALLOCATOR;
  14
  15 ACE_Process_Mutex coordMutex("Coord-Mutex");
  16
  17 // Listing 1 code/ch17
  18 template <class T>
  19 class Unbounded_Queue : public ACE_Unbounded_Queue<T>
  20 {
  21 public:
  22   typedef ACE_Unbounded_Queue<T> BASE;
  23
  24   Unbounded_Queue(ACE_Allocator* allocator)
  25     : ACE_Unbounded_Queue<T> (allocator)
  26   { }
  27
  28   int enqueue_tail (const T &new_item, ACE_Allocator* allocator)
  29   {
  30     this->allocator_ = allocator;
  31     return BASE::enqueue_tail (new_item);
  32   }
  33
  34   int dequeue_head (T &item, ACE_Allocator* allocator)
  35   {
  36     this->allocator_ = allocator;
  37     return BASE::dequeue_head (item);
  38   }
  39
  40   void delete_nodes (ACE_Allocator* allocator)
  41   {
  42     this->allocator_ = allocator;
  43     delete_nodes ();
  44   }
  45 };
  46 // Listing 1
  47
  48 #include "Record.h"
  49
  50 typedef Unbounded_Queue<Record> QUEUE;
  51
  52 QUEUE* squeue(ALLOCATOR* shmem_allocator)
  53 {
  54   void *queue = 0;
  55
  56   // This is the easy case since if we find hash table in the
  57   // memory-mapped file we know it's already initialized.
  58   if (shmem_allocator->find (QUEUE_NAME, queue) == 0)
  59     return (QUEUE *) queue;
  60
  61   // Create a new map (because we've just created a new
  62   // memory-mapped file).
  63   size_t queue_size = sizeof (QUEUE);
  64
  65   queue = shmem_allocator->malloc (queue_size);
  66
  67   // If allocation failed ...
  68   if (queue == 0)
  69     return 0;
  70
  71   new (queue) QUEUE (shmem_allocator);
  72
  73   if (shmem_allocator->bind (QUEUE_NAME, queue) == -1)
  74     {
  75       // Attempt to clean up.
  76       ACE_ERROR ((LM_ERROR,
  77                   ACE_TEXT ("squeue bind\n")));
  78       shmem_allocator->remove();
  79
  80       return 0;
  81     }
  82
  83   return (QUEUE*)queue;
  84 }
  85
  86 static ALLOCATOR * g_shmem_allocator = 0;
  87
  88 // Listing 4 code/ch17
  89 int processRecord (ALLOCATOR *shmem_allocator)
  90 {
  91   ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1);
  92
  93   QUEUE* queue = squeue (shmem_allocator);
  94   if (queue == 0)
  95     {
  96       delete shmem_allocator;
  97       ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
  98                          ACE_TEXT ("Could not obtain queue")),
  99                         -1);
 100     }
 101
 102   if (queue->is_empty ())  // Check for anything to process.
 103     return 0;
 104
 105   Record record;
 106   if (queue->dequeue_head (record, shmem_allocator) == -1)
 107     {
 108       ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 109                          ACE_TEXT ("dequeue_head\n")),
 110                         -1);
 111     }
 112
 113   ACE_DEBUG ((LM_DEBUG,
 114               ACE_TEXT ("(%P|%t) Processing record|name: %C")
 115               ACE_TEXT ("|Record id1:%d|Record id2:%d\n"),
 116               record.name (), record.id1 (), record.id2 ()));
 117   if (record.id1 () == -1)
 118     queue->enqueue_tail (record, shmem_allocator);
 119   return record.id1 ();
 120 }
 121 // Listing 4
 122 // Listing 5 code/ch17
 123 #if defined(WIN32)
 124
 125 int handle_remap (EXCEPTION_POINTERS *ep)
 126 {
 127   ACE_DEBUG ((LM_INFO, ACE_TEXT ("Handle a remap\n")));
 128
 129   DWORD ecode = ep->ExceptionRecord->ExceptionCode;
 130   if (ecode != EXCEPTION_ACCESS_VIOLATION)
 131     return EXCEPTION_CONTINUE_SEARCH;
 132
 133   void *addr =
 134     (void *) ep->ExceptionRecord->ExceptionInformation[1];
 135   if (g_shmem_allocator->alloc().memory_pool().remap (addr) == -1)
 136     return EXCEPTION_CONTINUE_SEARCH;
 137 #if __X86__
 138   // This is 80x86-specific.
 139   ep->ContextRecord->Edi = (DWORD) addr;
 140 #elif __MIPS__
 141   ep->ContextRecord->IntA0 =
 142     ep->ContextRecord->IntV0 = (DWORD) addr;
 143   ep->ContextRecord->IntT5 =
 144     ep->ContextRecord->IntA0 + 3;
 145 #endif /* __X86__ */
 146
 147   return EXCEPTION_CONTINUE_EXECUTION;
 148 }
 149
 150 int processWin32Record (ALLOCATOR *shmem_allocator)
 151 {
 152   ACE_SEH_TRY
 153   {
 154     return processRecord (shmem_allocator);
 155   }
 156   ACE_SEH_EXCEPT (handle_remap (GetExceptionInformation ()))
 157   {
 158   }
 159
 160   return 0;
 161 }
 162 #endif /*WIN32*/
 163 // Listing 5
 164
 165 int sendRecord (int recordId,  ALLOCATOR *shmem_allocator)
 166 {
 167   ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1);
 168
 169   ACE_DEBUG ((LM_DEBUG,
 170               ACE_TEXT ("(%P|%t) Sending record %d\n"),
 171               recordId));
 172
 173   QUEUE * queue = squeue (shmem_allocator);
 174   char buf[128];
 175   ACE_OS::sprintf (buf, "%s:%d", "Record", recordId);
 176   Record newRecord (recordId, recordId+1, buf);
 177
 178   int result = queue->enqueue_tail (newRecord, shmem_allocator);
 179   if (result == -1)
 180     ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
 181                        ACE_TEXT ("enqueue failed\n")),
 182                       -1);
 183   return 0;
 184 }
 185
 186 // Listing 2 code/ch17
 187 int handle_parent (ACE_TCHAR *cmdLine)
 188 {
 189   ALLOCATOR *shmem_allocator = 0;
 190   ACE_MMAP_Memory_Pool_Options options
 191     (ACE_DEFAULT_BASE_ADDR,
 192      ACE_MMAP_Memory_Pool_Options::ALWAYS_FIXED);
 193
 194   // Create the allocator.
 195   ACE_NEW_RETURN (shmem_allocator,
 196                   ALLOCATOR (BACKING_STORE,
 197                              BACKING_STORE,
 198                              &options),
 199                   -1);
 200
 201   ACE_Process processa, processb;
 202   ACE_Process_Options poptions;
 203   const ACE_TCHAR *args[3];
 204   args[0] = cmdLine;
 205   args[1] = ACE_TEXT ("a");
 206   args[2] = 0;
 207   poptions.command_line (args);
 208   processa.spawn (poptions);
 209   processb.spawn (poptions);
 210
 211   // Make sure the child does map a partial pool in memory.
 212   ACE_OS::sleep (2);
 213
 214   for (int i = 0; i < 100; i++)
 215     sendRecord (i, shmem_allocator);
 216   sendRecord (-1, shmem_allocator);
 217
 218   processa.wait ();
 219   processb.wait ();
 220   shmem_allocator->remove ();
 221   return 0;
 222 }
 223 // Listing 2
 224
 225 // Listing 3 code/ch17
 226 int handle_child ()
 227 {
 228   ALLOCATOR *shmem_allocator = 0;
 229   ACE_MMAP_Memory_Pool_Options options
 230     (ACE_DEFAULT_BASE_ADDR,
 231      ACE_MMAP_Memory_Pool_Options::ALWAYS_FIXED);
 232   ACE_NEW_RETURN (shmem_allocator,
 233                   ALLOCATOR (BACKING_STORE,
 234                              BACKING_STORE,
 235                              &options),
 236                   -1);
 237   g_shmem_allocator = shmem_allocator;
 238
 239 #if defined (WIN32)
 240   while (processWin32Record (shmem_allocator) != -1)
 241     ;
 242 #else
 243   while (processRecord (shmem_allocator) != -1)
 244     ;
 245 #endif
 246   return 0;
 247 }
 248 // Listing 3
 249
 250 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 251 {
 252   if (argc == 1)
 253     handle_parent(argv[0]);
 254   else
 255     handle_child();
 256
 257   return 0;
 258 }
 259