diff options
Diffstat (limited to 'ACE/ace/MEM_IO.cpp')
| -rw-r--r-- | ACE/ace/MEM_IO.cpp | 438 |
1 files changed, 438 insertions, 0 deletions
diff --git a/ACE/ace/MEM_IO.cpp b/ACE/ace/MEM_IO.cpp new file mode 100644 index 00000000000..b51b5a53c8c --- /dev/null +++ b/ACE/ace/MEM_IO.cpp @@ -0,0 +1,438 @@ +// MEM_IO.cpp +// $Id$ + +#include "ace/MEM_IO.h" +#include "ace/Handle_Set.h" + +#if (ACE_HAS_POSITION_INDEPENDENT_POINTERS == 1) + +#if !defined (__ACE_INLINE__) +#include "ace/MEM_IO.inl" +#endif /* __ACE_INLINE__ */ + + + +ACE_BEGIN_VERSIONED_NAMESPACE_DECL + +ACE_ALLOC_HOOK_DEFINE(ACE_MEM_IO) + +ACE_Reactive_MEM_IO::~ACE_Reactive_MEM_IO (void) +{ +} + +int +ACE_Reactive_MEM_IO::init (ACE_HANDLE handle, + const ACE_TCHAR *name, + MALLOC_OPTIONS *options) +{ + ACE_TRACE ("ACE_Reactive_MEM_IO::init"); + this->handle_ = handle; + return this->create_shm_malloc (name, + options); +} + +ssize_t +ACE_Reactive_MEM_IO::recv_buf (ACE_MEM_SAP_Node *&buf, + int flags, + const ACE_Time_Value *timeout) +{ + ACE_TRACE ("ACE_Reactive_MEM_IO::recv_buf"); + + if (this->shm_malloc_ == 0 || this->handle_ == ACE_INVALID_HANDLE) + return -1; + + ACE_OFF_T new_offset = 0; + ssize_t retv = ACE::recv (this->handle_, + (char *) &new_offset, + sizeof (ACE_OFF_T), + flags, + timeout); + + if (retv == 0) + { + // ACE_DEBUG ((LM_INFO, "MEM_Stream closed\n")); + buf = 0; + return 0; + } + else if (retv != static_cast <ssize_t> (sizeof (ACE_OFF_T))) + { + // Nothing available or we are really screwed. + buf = 0; + return -1; + } + + return this->get_buf_len (new_offset, buf); +} + +ssize_t +ACE_Reactive_MEM_IO::send_buf (ACE_MEM_SAP_Node *buf, + int flags, + const ACE_Time_Value *timeout) +{ + ACE_TRACE ("ACE_Reactive_MEM_IO::send_buf"); + + if (this->shm_malloc_ == 0 || this->handle_ == ACE_INVALID_HANDLE) + { + return -1; + } + + // The offset. + ACE_OFF_T offset = + ACE_Utils::truncate_cast<ACE_OFF_T> ( + reinterpret_cast<char *> (buf) + - static_cast<char *> (this->shm_malloc_->base_addr ())); + + // Send the offset value over the socket. + if (ACE::send (this->handle_, + (const char *) &offset, + sizeof (offset), + flags, + timeout) != static_cast <ssize_t> (sizeof (offset))) + { + // unsuccessful send, release the memory in the shared-memory. + this->release_buffer (buf); + + return -1; + } + + return ACE_Utils::truncate_cast<ssize_t> (buf->size ()); +} + +#if defined (ACE_WIN32) || !defined (_ACE_USE_SV_SEM) +int +ACE_MT_MEM_IO::Simple_Queue::write (ACE_MEM_SAP_Node *new_node) +{ + if (this->mq_ == 0) + return -1; + + // Here, we assume we already have acquired the lock necessary. + // And we are allowed to write. + if (this->mq_->tail_.addr () == 0) // nothing in the queue. + { + this->mq_->head_ = new_node; + this->mq_->tail_ = new_node; + new_node->next_ = 0; + } + else + { + this->mq_->tail_->next_ = new_node; + new_node->next_ = 0; + this->mq_->tail_ = new_node; + } + return 0; +} + +ACE_MEM_SAP_Node * +ACE_MT_MEM_IO::Simple_Queue::read () +{ + if (this->mq_ == 0) + return 0; + + ACE_MEM_SAP_Node *retv = 0; + + ACE_SEH_TRY + { + retv = this->mq_->head_; + // Here, we assume we already have acquired the lock necessary + // and there are soemthing in the queue. + if (this->mq_->head_ == this->mq_->tail_) + { + // Last message in the queue. + this->mq_->head_ = 0; + this->mq_->tail_ = 0; + } + else + this->mq_->head_ = retv->next_; + } + ACE_SEH_EXCEPT (this->malloc_->memory_pool ().seh_selector (GetExceptionInformation ())) + { + } + + return retv; +} + +ACE_MT_MEM_IO::~ACE_MT_MEM_IO () +{ + delete this->recv_channel_.sema_; + delete this->recv_channel_.lock_; + delete this->send_channel_.sema_; + delete this->send_channel_.lock_; +} + +int +ACE_MT_MEM_IO::init (ACE_HANDLE handle, + const ACE_TCHAR *name, + MALLOC_OPTIONS *options) +{ + ACE_TRACE ("ACE_MT_MEM_IO::init"); + ACE_UNUSED_ARG (handle); + + // @@ Give me a rule on naming and how the queue should + // be kept in the shared memory and we are done + // with this. + if (this->create_shm_malloc (name, options) == -1) + return -1; + + ACE_TCHAR server_sema [MAXPATHLEN]; + ACE_TCHAR client_sema [MAXPATHLEN]; + ACE_TCHAR server_lock [MAXPATHLEN]; + ACE_TCHAR client_lock [MAXPATHLEN]; + const ACE_TCHAR *basename = ACE::basename (name); + // size_t baselen = ACE_OS::strlen (basename); + + // Building names. @@ Check buffer overflow? + ACE_OS::strcpy (server_sema, basename); + ACE_OS::strcat (server_sema, ACE_TEXT ("_sema_to_server")); + ACE_OS::strcpy (client_sema, basename); + ACE_OS::strcat (client_sema, ACE_TEXT ("_sema_to_client")); + ACE_OS::strcpy (server_lock, basename); + ACE_OS::strcat (server_lock, ACE_TEXT ("_lock_to_server")); + ACE_OS::strcpy (client_lock, basename); + ACE_OS::strcat (client_lock, ACE_TEXT ("_lock_to_client")); + + void *to_server_ptr = 0; + // @@ Here, we assume the shared memory fill will never be resued. + // So we can determine whether we are server or client by examining + // if the simple message queues have already been set up in + // the Malloc object or not. + if (this->shm_malloc_->find ("to_server", to_server_ptr) == -1) + { + void *ptr = 0; + // We are server. + ACE_ALLOCATOR_RETURN (ptr, + this->shm_malloc_->malloc (2 * sizeof (MQ_Struct)), + -1); + + MQ_Struct *mymq = reinterpret_cast<MQ_Struct *> (ptr); + mymq->tail_ = 0; + mymq->head_ = 0; + (mymq + 1)->tail_ = 0; + (mymq + 1)->head_ = 0; + if (this->shm_malloc_->bind ("to_server", mymq) == -1) + return -1; + + if (this->shm_malloc_->bind ("to_client", mymq + 1) == -1) + return -1; + + this->recv_channel_.queue_.init (mymq, this->shm_malloc_); + ACE_NEW_RETURN (this->recv_channel_.sema_, + ACE_SYNCH_PROCESS_SEMAPHORE (0, server_sema), + -1); + ACE_NEW_RETURN (this->recv_channel_.lock_, + ACE_SYNCH_PROCESS_MUTEX (server_lock), + -1); + + this->send_channel_.queue_.init (mymq + 1, this->shm_malloc_); + ACE_NEW_RETURN (this->send_channel_.sema_, + ACE_SYNCH_PROCESS_SEMAPHORE (0, client_sema), + -1); + ACE_NEW_RETURN (this->send_channel_.lock_, + ACE_SYNCH_PROCESS_MUTEX (client_lock), + -1); + } + else + { + // we are client. + MQ_Struct *mymq = reinterpret_cast<MQ_Struct *> (to_server_ptr); + this->recv_channel_.queue_.init (mymq +1, this->shm_malloc_); + ACE_NEW_RETURN (this->recv_channel_.sema_, + ACE_SYNCH_PROCESS_SEMAPHORE (0, client_sema), + -1); + ACE_NEW_RETURN (this->recv_channel_.lock_, + ACE_SYNCH_PROCESS_MUTEX (client_lock), + -1); + + this->send_channel_.queue_.init (mymq, this->shm_malloc_); + ACE_NEW_RETURN (this->send_channel_.sema_, + ACE_SYNCH_PROCESS_SEMAPHORE (0, server_sema), + -1); + ACE_NEW_RETURN (this->send_channel_.lock_, + ACE_SYNCH_PROCESS_MUTEX (server_lock), + -1); + } + return 0; +} + +ssize_t +ACE_MT_MEM_IO::recv_buf (ACE_MEM_SAP_Node *&buf, + int flags, + const ACE_Time_Value *timeout) +{ + ACE_TRACE ("ACE_MT_MEM_IO::recv_buf"); + + // @@ Don't know how to handle timeout yet. + ACE_UNUSED_ARG (timeout); + ACE_UNUSED_ARG (flags); + + if (this->shm_malloc_ == 0) + { + return -1; + } + + // Need to handle timeout here. + if (this->recv_channel_.sema_->acquire () == -1) + { + return -1; + } + + { + // @@ We can probably skip the lock in certain circumstance. + ACE_GUARD_RETURN (ACE_SYNCH_PROCESS_MUTEX, ace_mon, *this->recv_channel_.lock_, -1); + + buf = this->recv_channel_.queue_.read (); + + if (buf != 0) + { + return ACE_Utils::truncate_cast<ssize_t> (buf->size ()); + } + + return -1; + } +} + +ssize_t +ACE_MT_MEM_IO::send_buf (ACE_MEM_SAP_Node *buf, + int flags, + const ACE_Time_Value *timeout) +{ + ACE_TRACE ("ACE_MT_MEM_IO::send_buf"); + + // @@ Don't know how to handle timeout yet. + ACE_UNUSED_ARG (timeout); + ACE_UNUSED_ARG (flags); + + if (this->shm_malloc_ == 0) + { + return -1; + } + + { + // @@ We can probably skip the lock in certain curcumstances. + ACE_GUARD_RETURN (ACE_SYNCH_PROCESS_MUTEX, ace_mon, *this->send_channel_.lock_, -1); + + if (this->send_channel_.queue_.write (buf) == -1) + { + this->release_buffer (buf); + return -1; + } + } + + if (this->send_channel_.sema_->release () == -1) + { + return -1; + } + + return ACE_Utils::truncate_cast<ssize_t> (buf->size ()); +} +#endif /* ACE_WIN32 || !_ACE_USE_SV_SEM */ + +void +ACE_MEM_IO::dump (void) const +{ +#if defined (ACE_HAS_DUMP) + ACE_TRACE ("ACE_MEM_IO::dump"); +#endif /* ACE_HAS_DUMP */ +} + +int +ACE_MEM_IO::init (const ACE_TCHAR *name, + ACE_MEM_IO::Signal_Strategy type, + ACE_MEM_SAP::MALLOC_OPTIONS *options) +{ + ACE_UNUSED_ARG (type); + + delete this->deliver_strategy_; + this->deliver_strategy_ = 0; + switch (type) + { + case ACE_MEM_IO::Reactive: + ACE_NEW_RETURN (this->deliver_strategy_, + ACE_Reactive_MEM_IO (), + -1); + break; +#if defined (ACE_WIN32) || !defined (_ACE_USE_SV_SEM) + case ACE_MEM_IO::MT: + ACE_NEW_RETURN (this->deliver_strategy_, + ACE_MT_MEM_IO (), + -1); + break; +#endif /* ACE_WIN32 || !_ACE_USE_SV_SEM */ + default: + return -1; + } + + return this->deliver_strategy_->init (this->get_handle (), + name, + options); +} + +int +ACE_MEM_IO::fini (void) +{ + if (this->deliver_strategy_ != 0) + { + return this->deliver_strategy_->fini (); + } + else + { + return -1; + } +} + +// Allows a client to read from a socket without having to provide +// a buffer to read. This method determines how much data is in the +// socket, allocates a buffer of this size, reads in the data, and +// returns the number of bytes read. + +ssize_t +ACE_MEM_IO::send (const ACE_Message_Block *message_block, + const ACE_Time_Value *timeout) +{ + ACE_TRACE ("ACE_MEM_IO::send"); + + if (this->deliver_strategy_ == 0) + { + return -1; // Something went seriously wrong. + } + + size_t len = message_block->total_length (); + + if (len != 0) + { + ACE_MEM_SAP_Node *buf = + reinterpret_cast<ACE_MEM_SAP_Node *> ( + this->deliver_strategy_->acquire_buffer ( + ACE_Utils::truncate_cast<ssize_t> (len))); + + size_t n = 0; + + while (message_block != 0) + { + ACE_OS::memcpy (static_cast<char *> (buf->data ()) + n, + message_block->rd_ptr (), + message_block->length ()); + n += message_block->length (); + + if (message_block->cont ()) + { + message_block = message_block->cont (); + } + else + { + message_block = message_block->next (); + } + } + + buf->size_ = len; + + return this->deliver_strategy_->send_buf (buf, + 0, + timeout); + } + + return 0; +} + +ACE_END_VERSIONED_NAMESPACE_DECL + +#endif /* ACE_HAS_POSITION_INDEPENDENT_POINTERS == 1 */ |