ACE/FAQ/APG/Shared Memory
Материал из Wiki.crossplatform.ru
Содержание |
Record.h
/** * $Id: Record.h 80826 2008-03-04 14:51:23Z wotte $ * * Sample code from The ACE Programmer's Guide, * copyright 2003 Addison-Wesley. All Rights Reserved. */ #ifndef __RECORD_H_ #define __RECORD_H_ #include "ace/OS_NS_string.h" #include "ace/Based_Pointer_T.h" // Listing 11 code/ch17 class Record { public: Record () { } ~Record () { } Record (const Record& rec) : id1_(rec.id1_), id2_(rec.id2_) { ACE_OS::strcpy (recName_, rec.name_); this->name_ = recName_; } Record (int id1, int id2, char *name) : id1_(id1), id2_(id2) { ACE_OS::strcpy (recName_, name); this->name_ = recName_; } char *name (void) { return recName_; } int id1 (void) { return id1_; } int id2 (void) { return id2_; } private: int id1_; int id2_; char recName_[128]; ACE_Based_Pointer_Basic<char> name_; }; // Listing 11 #endif /* __RECORD_H_ */
Hash Map
// $Id: Hash_Map.cpp 80826 2008-03-04 14:51:23Z wotte $ #include "ace/OS_NS_stdio.h" // Listing 1 code/ch17 #include "ace/MMAP_Memory_Pool.h" #include "ace/Hash_Map_With_Allocator_T.h" #include "ace/Malloc_T.h" #include "ace/PI_Malloc.h" #include "ace/Process_Mutex.h" #include "ace/Process.h" #define BACKING_STORE "map.store" #define MAP_NAME "records.db" #include "Record.h" typedef ACE_Allocator_Adapter<ACE_Malloc_T <ACE_MMAP_MEMORY_POOL, ACE_Process_Mutex, ACE_Control_Block> > ALLOCATOR; typedef ACE_Hash_Map_With_Allocator<int, Record> HASH_MAP; ACE_Process_Mutex coordMutex("Coord-Mutex"); // Listing 1 // Listing 2 code/ch17 HASH_MAP* smap (ALLOCATOR *shmem_allocator) { void *db = 0; if (shmem_allocator->find (MAP_NAME, db) == 0) return (HASH_MAP *) db; size_t hash_table_size = sizeof (HASH_MAP); void *hash_map = shmem_allocator->malloc (hash_table_size); if (hash_map == 0) return 0; new (hash_map) HASH_MAP (hash_table_size, shmem_allocator); if (shmem_allocator->bind (MAP_NAME, hash_map) == -1) { ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("allocate_map"))); shmem_allocator->remove (); return 0; } return (HASH_MAP*)hash_map; } // Listing 2 // Listing 6 code/ch17 int processRecords (HASH_MAP *map, ALLOCATOR *shmem_allocator) { ACE_TRACE ("processRecords"); size_t mapLength = map->current_size (); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Found %d records\n\n"), mapLength)); int *todelete = new int[mapLength]; int i = 0; for (HASH_MAP::iterator iter = map->begin (); iter != map->end (); iter++) { int key = (*iter).ext_id_; ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) [%d] Preprocessing %d:%@\n"), i+1, key, &(*iter).ext_id_)); todelete[i++] = key; // Mark message for deletion. // Illustrate the find feature of the map. Record record; int result = map->find (key, record, shmem_allocator); if (result == -1) ACE_DEBUG ((LM_ERROR, ACE_TEXT ("Could not find record for %d\n"), key)); else ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Record name: %C|id1:%d|id2:%d\n"), record.name (), record.id1(), record.id2())); } // Delete everything we processed. for (int j = 0; j < i ; j++) { int result = map->unbind (todelete[j], shmem_allocator); if (result == -1) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("Failed on key %d: %p\n"), ACE_TEXT ("unbind"), todelete[j]), -1); else ACE_DEBUG ((LM_INFO, ACE_TEXT ("Fully processed and removed %d\n"), j)); } delete [] todelete; return 0; } // Listing 6 // Listing 4 code/ch17 int addRecords(HASH_MAP *map, ALLOCATOR *shmem_allocator) { ACE_TRACE ("addRecords"); char buf[32]; int mapLength = static_cast<int> (map->current_size ()); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Map has %d entries; adding 20 more\n"), mapLength)); for (int i = mapLength ; i < mapLength + 20; i++) { ACE_OS::sprintf (buf, "%s:%d", "Record", i); // Allocate new record on stack; Record newRecord (i, i+1, buf); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Adding a record for %d\n"), i)); int result = map->bind (i, newRecord, shmem_allocator); if (result == -1) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("bind failed")), -1); } return 0; } // Listing 4 // Listing 5 code/ch17 int handle_child (void) { ACE_TRACE ("handle_child"); ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1); 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); HASH_MAP *map = smap (shmem_allocator); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Map has %d entries\n"), map->current_size ())); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("In child, map is located at %@\n"), map)); processRecords (map, shmem_allocator); shmem_allocator->sync (); delete shmem_allocator; return 0; } // Listing 5 // Listing 3 code/ch17 int handle_parent (ACE_TCHAR *cmdLine) { ACE_TRACE ("handle_parent"); 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); HASH_MAP *map = smap (shmem_allocator); ACE_Process processa, processb; ACE_Process_Options poptions; const ACE_TCHAR *args[3]; args[0] = cmdLine; args[1] = ACE_TEXT ("a"); args[2] = 0; poptions.command_line (args); { ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Map has %d entries\n"), map->current_size ())); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("In parent, map is located at %@\n"), map)); // Then have the child show and eat them up. processa.spawn (poptions); // First append a few records. addRecords (map, shmem_allocator); } { ACE_GUARD_RETURN (ACE_Process_Mutex, ace_mon, coordMutex, -1); // Add a few more records.. addRecords (map, shmem_allocator); // Let's see what's left. ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Parent finished adding, ") ACE_TEXT ("map has %d entries\n"), map->current_size ())); // Have another child try to eat them up. processb.spawn (poptions); } processa.wait (); processb.wait (); // No processes are left and we don't want to keep the data // around anymore; it's now safe to remove it. // !!This will remove the backing store.!! shmem_allocator->remove (); delete shmem_allocator; return 0; } int ACE_TMAIN (int argc, ACE_TCHAR *argv[]) { if (argc == 1) // parent ACE_ASSERT (handle_parent (argv[0]) == 0); else ACE_ASSERT (handle_child () == 0); ACE_UNUSED_ARG (argv); return 0; } // Listing 3
Malloc
// $Id: Malloc.cpp 80826 2008-03-04 14:51:23Z wotte $ #include "ace/OS_NS_stdio.h" #include "ace/OS_NS_string.h" // Listing 1 code/ch17 #include "ace/MMAP_Memory_Pool.h" #include "ace/Malloc_T.h" #include "ace/Null_Mutex.h" typedef ACE_Malloc<ACE_MMAP_MEMORY_POOL, ACE_Null_Mutex> ALLOCATOR; typedef ACE_Malloc_LIFO_Iterator <ACE_MMAP_MEMORY_POOL, ACE_Null_Mutex> MALLOC_LIFO_ITERATOR; ALLOCATOR *g_allocator; // Listing 1 // Listing 2 code/ch17 class Record { public: Record (int id1, int id2, char *name) : id1_(id1), id2_(id2), name_(0) { size_t len = ACE_OS::strlen (name) + 1; this->name_ = reinterpret_cast<char *> (g_allocator->malloc (len)); ACE_OS::strcpy (this->name_, name); } ~Record () { g_allocator->free (name_); } char* name(void) { return name_; } int id1 (void) { return id1_; } int id2 (void) { return id2_; } private: int id1_; int id2_; char *name_; }; // Listing 2 // Listing 5 code/ch17 void showRecords () { ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("The following records were found:\n"))); { MALLOC_LIFO_ITERATOR iter (*g_allocator); for (void *temp = 0; iter.next (temp) != 0; iter.advance ()) { Record *record = reinterpret_cast<Record *> (temp); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Record name: %C|id1:%d|id2:%d\n"), record->name (), record->id1 (), record->id2 ())); } } } // Listing 5 // Listing 3 code/ch17 int addRecords () { char buf[32]; for (int i = 0; i < 10; i++) { ACE_OS::sprintf (buf, "%s:%d", "Record", i); void *memory = g_allocator->malloc (sizeof (Record)); if (memory == 0) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("Unable to malloc")), -1); // Allocate and place record Record* newRecord = new (memory) Record (i, i+1, buf); if (g_allocator->bind (buf, newRecord) == -1) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("bind failed")), -1); } return 0; } // Listing 3 // Listing 4 code/ch17 // Backing file where the data is kept. #define BACKING_STORE ACE_TEXT("backing.store") int ACE_TMAIN (int argc, ACE_TCHAR *[]) { ACE_NEW_RETURN (g_allocator, ALLOCATOR (BACKING_STORE), -1); if (argc > 1) { showRecords (); } else { addRecords (); } g_allocator->sync (); delete g_allocator; return 0; } // Listing 4
Mem Map
// $Id: Mem_Map.cpp 80826 2008-03-04 14:51:23Z wotte $ #include "ace/OS_NS_fcntl.h" #include "ace/OS_NS_string.h" #include "ace/Mem_Map.h" #include "ace/Log_Msg.h" // Listing 1 code/ch17 int ACE_TMAIN (int, ACE_TCHAR *argv[]) { ACE_HANDLE srcHandle = ACE_OS::open (argv[1], O_RDONLY); ACE_ASSERT(srcHandle != ACE_INVALID_HANDLE); ACE_Mem_Map srcMap (srcHandle, static_cast<size_t> (-1), PROT_READ, ACE_MAP_PRIVATE); ACE_ASSERT(srcMap.addr () != 0); ACE_Mem_Map destMap (argv[2], srcMap.size (), O_RDWR | O_CREAT, ACE_DEFAULT_FILE_PERMS, PROT_RDWR, ACE_MAP_SHARED); ACE_ASSERT(destMap.addr () != 0); ACE_OS::memcpy (destMap.addr (), srcMap.addr (), srcMap.size ()); destMap.sync (); srcMap.close (); destMap.close (); return 0; } // Listing 1
PI Malloc
// $Id: PI_Malloc.cpp 80826 2008-03-04 14:51:23Z wotte $ #include "ace/OS_NS_stdio.h" #include "ace/OS_NS_string.h" // Listing 1 code/ch17 #include "ace/MMAP_Memory_Pool.h" #include "ace/Malloc_T.h" #include "ace/Null_Mutex.h" #include "ace/PI_Malloc.h" typedef ACE_Malloc_T <ACE_MMAP_MEMORY_POOL, ACE_Null_Mutex, ACE_PI_Control_Block> ALLOCATOR; typedef ACE_Malloc_LIFO_Iterator_T<ACE_MMAP_MEMORY_POOL, ACE_Null_Mutex, ACE_PI_Control_Block> MALLOC_LIFO_ITERATOR; ALLOCATOR *g_allocator; // Listing 1 // Listing 2 code/ch17 class Record { public: Record (int id1, int id2, char *name) : id1_(id1), id2_(id2) { size_t len = ACE_OS::strlen (name) + 1; char *buf = reinterpret_cast<char *> (g_allocator->malloc (len)); ACE_OS::strcpy (buf, name); name_ = buf; } ~Record() { g_allocator->free (name_.addr ()); } char *name (void) { return name_; } int id1 (void) { return id1_; } int id2 (void) { return id2_; } private: int id1_; int id2_; ACE_Based_Pointer_Basic<char> name_; }; // Listing 2 void showRecords (void) { ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("The following records were found:\n"))); { MALLOC_LIFO_ITERATOR iter (*g_allocator); for (void *temp = 0; iter.next (temp) != 0; iter.advance ()) { Record *record = reinterpret_cast<Record *> (temp); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Record name: %C|id1:%d|id2:%d\n"), record->name(), record->id1(), record->id2())); } } } int addRecords (void) { char buf[32]; for (int i = 0; i < 10; i++) { ACE_OS::sprintf (buf, "%s:%d", "Record", i); void *memory = g_allocator->malloc (sizeof (Record)); if (memory == 0) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("Unable to malloc")), -1); // Allocate and place record Record* newRecord = new (memory) Record (i, i+1, buf); if (g_allocator->bind (buf, newRecord) == -1) ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("bind failed")), -1); } return 0; } // Listing 3 code/ch17 // Backing file where the data is kept. #define BACKING_STORE ACE_TEXT("backing2.store") int ACE_TMAIN (int argc, ACE_TCHAR *[]) { if (argc > 1) { ACE_MMAP_Memory_Pool_Options options (ACE_DEFAULT_BASE_ADDR, ACE_MMAP_Memory_Pool_Options::ALWAYS_FIXED); ACE_NEW_RETURN (g_allocator, ALLOCATOR (BACKING_STORE, BACKING_STORE, &options), -1); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Mapped to base address %@\n"), g_allocator->base_addr ())); showRecords (); } else { ACE_MMAP_Memory_Pool_Options options (0, ACE_MMAP_Memory_Pool_Options::NEVER_FIXED); ACE_NEW_RETURN (g_allocator, ALLOCATOR (BACKING_STORE, BACKING_STORE, &options), -1); ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Mapped to base address %@\n"), g_allocator->base_addr ())); addRecords(); } g_allocator->sync (); delete g_allocator; return 0; } // Listing 3
Pool Growth
// $Id: Pool_Growth.cpp 80826 2008-03-04 14:51:23Z wotte $ #include "ace/OS_NS_stdio.h" #include "ace/OS_NS_unistd.h" #include "ace/Malloc_T.h" #include "ace/PI_Malloc.h" #include "ace/Process_Mutex.h" #include "ace/Process.h" #include "ace/Unbounded_Queue.h" #include "ace/MMAP_Memory_Pool.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 (ACE_TCHAR *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; const ACE_TCHAR *args[3]; args[0] = cmdLine; args[1] = ACE_TEXT ("a"); args[2] = 0; poptions.command_line (args); 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; }