diff options
Diffstat (limited to 'examples/APG/Shared_Memory/Pool_Growth.cpp')
| -rw-r--r-- | examples/APG/Shared_Memory/Pool_Growth.cpp | 268 |
1 files changed, 268 insertions, 0 deletions
diff --git a/examples/APG/Shared_Memory/Pool_Growth.cpp b/examples/APG/Shared_Memory/Pool_Growth.cpp new file mode 100644 index 00000000000..0835413dd9d --- /dev/null +++ b/examples/APG/Shared_Memory/Pool_Growth.cpp @@ -0,0 +1,268 @@ +// $Id$ + +#include "ace/OS_NS_stdio.h" +#include "ace/Malloc.h" +#include "ace/PI_Malloc.h" +#include "ace/Process_Mutex.h" +#include "ace/Process.h" +#include "ace/Unbounded_Queue.h" + +#define BACKING_STORE "queue.dat" +#define QUEUE_NAME "queue.db" + +typedef ACE_Allocator_Adapter<ACE_Malloc <ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex> > ALLOCATOR; + +ACE_Process_Mutex coordMutex("Coord-Mutex"); + +// Listing 1 code/ch17 +template <class T> +class Unbounded_Queue : public ACE_Unbounded_Queue<T> +{ +public: + typedef ACE_Unbounded_Queue<T> BASE; + + Unbounded_Queue(ACE_Allocator* allocator) + : ACE_Unbounded_Queue<T> (allocator) + { } + + int enqueue_tail (const T &new_item, ACE_Allocator* allocator) + { + this->allocator_ = allocator; + return BASE::enqueue_tail (new_item); + } + + int dequeue_head (T &item, ACE_Allocator* allocator) + { + this->allocator_ = allocator; + return BASE::dequeue_head (item); + } + + void delete_nodes (ACE_Allocator* allocator) + { + this->allocator_ = allocator; + delete_nodes (); + } +}; +// Listing 1 + +#include "Record.h" + +typedef Unbounded_Queue<Record> QUEUE; + +QUEUE* squeue(ALLOCATOR* shmem_allocator) +{ + void *queue = 0; + + // This is the easy case since if we find hash table in the + // memory-mapped file we know it's already initialized. + if (shmem_allocator->find (QUEUE_NAME, queue) == 0) + return (QUEUE *) queue; + + // Create a new map (because we've just created a new + // memory-mapped file). + size_t queue_size = sizeof (QUEUE); + + queue = shmem_allocator->malloc (queue_size); + + // If allocation failed ... + if (queue == 0) + return 0; + + new (queue) QUEUE (shmem_allocator); + + if (shmem_allocator->bind (QUEUE_NAME, queue) == -1) + { + // Attempt to clean up. + ACE_ERROR ((LM_ERROR, + ACE_TEXT ("squeue bind\n"))); + shmem_allocator->remove(); + + return 0; + } + + return (QUEUE*)queue; +} + +static ALLOCATOR * g_shmem_allocator = 0; + +// Listing 4 code/ch17 +int processRecord (ALLOCATOR *shmem_allocator) +{ + ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1); + + QUEUE* queue = squeue (shmem_allocator); + if (queue == 0) + { + delete shmem_allocator; + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("Could not obtain queue")), + -1); + } + + if (queue->is_empty ()) // Check for anything to process. + return 0; + + Record record; + if (queue->dequeue_head (record, shmem_allocator) == -1) + { + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("dequeue_head\n")), + -1); + } + + ACE_DEBUG ((LM_DEBUG, + ACE_TEXT ("(%P|%t) Processing record|name: %C") + ACE_TEXT ("|Record id1:%d|Record id2:%d\n"), + record.name (), record.id1 (), record.id2 ())); + if (record.id1 () == -1) + queue->enqueue_tail (record, shmem_allocator); + return record.id1 (); +} +// Listing 4 +// Listing 5 code/ch17 +#if defined(WIN32) + +int handle_remap (EXCEPTION_POINTERS *ep) +{ + ACE_DEBUG ((LM_INFO, ACE_TEXT ("Handle a remap\n"))); + + DWORD ecode = ep->ExceptionRecord->ExceptionCode; + if (ecode != EXCEPTION_ACCESS_VIOLATION) + return EXCEPTION_CONTINUE_SEARCH; + + void *addr = + (void *) ep->ExceptionRecord->ExceptionInformation[1]; + if (g_shmem_allocator->alloc().memory_pool().remap (addr) == -1) + return EXCEPTION_CONTINUE_SEARCH; +#if __X86__ + // This is 80x86-specific. + ep->ContextRecord->Edi = (DWORD) addr; +#elif __MIPS__ + ep->ContextRecord->IntA0 = + ep->ContextRecord->IntV0 = (DWORD) addr; + ep->ContextRecord->IntT5 = + ep->ContextRecord->IntA0 + 3; +#endif /* __X86__ */ + + return EXCEPTION_CONTINUE_EXECUTION; +} + +int processWin32Record (ALLOCATOR *shmem_allocator) +{ + ACE_SEH_TRY + { + return processRecord (shmem_allocator); + } + + ACE_SEH_EXCEPT (handle_remap (GetExceptionInformation ())) + { } + + return 0; +} +#endif /*WIN32*/ +// Listing 5 + +int sendRecord (int recordId, ALLOCATOR *shmem_allocator) +{ + ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1); + + ACE_DEBUG ((LM_DEBUG, + ACE_TEXT ("(%P|%t) Sending record %d\n"), + recordId)); + + QUEUE * queue = squeue (shmem_allocator); + char buf[128]; + ACE_OS::sprintf (buf, "%s:%d", "Record", recordId); + Record newRecord (recordId, recordId+1, buf); + + int result = queue->enqueue_tail (newRecord, shmem_allocator); + if (result == -1) + ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), + ACE_TEXT ("enqueue failed\n")), + -1); + return 0; +} + +// Listing 2 code/ch17 +int handle_parent (char *cmdLine) +{ + ALLOCATOR *shmem_allocator = 0; + ACE_MMAP_Memory_Pool_Options options + (ACE_DEFAULT_BASE_ADDR, + ACE_MMAP_Memory_Pool_Options::ALWAYS_FIXED); + + // Create the allocator. + ACE_NEW_RETURN (shmem_allocator, + ALLOCATOR (BACKING_STORE, + BACKING_STORE, + &options), + -1); + + ACE_Process processa, processb; + ACE_Process_Options poptions; + poptions.command_line ("%s a", ACE_TEXT_ALWAYS_CHAR (cmdLine)); + processa.spawn (poptions); + processb.spawn (poptions); + + // Make sure the child does map a partial pool in memory. + ACE_OS::sleep (2); + + for (int i = 0; i < 100; i++) + sendRecord (i, shmem_allocator); + sendRecord (-1, shmem_allocator); + + processa.wait (); + processb.wait (); + shmem_allocator->remove (); + return 0; +} +// Listing 2 + +// Listing 3 code/ch17 +int handle_child (void) +{ + ALLOCATOR *shmem_allocator = 0; + ACE_MMAP_Memory_Pool_Options options + (ACE_DEFAULT_BASE_ADDR, + ACE_MMAP_Memory_Pool_Options::ALWAYS_FIXED); + ACE_NEW_RETURN (shmem_allocator, + ALLOCATOR (BACKING_STORE, + BACKING_STORE, + &options), + -1); + g_shmem_allocator = shmem_allocator; + +#if defined (WIN32) + while (processWin32Record (shmem_allocator) != -1) + ; +#else + while (processRecord (shmem_allocator) != -1) + ; +#endif + return 0; +} +// Listing 3 + +int ACE_TMAIN (int argc, ACE_TCHAR *argv[]) +{ + if (argc == 1) + handle_parent(argv[0]); + else + handle_child(); + + return 0; +} + +#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION) +template class ACE_Allocator_Adapter<ACE_Malloc<ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex> >; +template class ACE_Malloc_T<ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex, ACE_Control_Block>; +template class ACE_Node<Record>; +template class ACE_Unbounded_Queue<Record>; +template class ACE_Unbounded_Queue_Iterator<Record>; +#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA) +#pragma instantiate ACE_Allocator_Adapter<ACE_Malloc<ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex> > +#pragma instantiate ACE_Malloc_T<ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex, ACE_Control_Block> +#pragma instantiate ACE_Node<Record> +#pragma instantiate ACE_Unbounded_Queue<Record> +#pragma instantiate ACE_Unbounded_Queue_Iterator<Record> +#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */ |