ACE/ace/MEM_IO.cpp

   1 // MEM_IO.cpp
   2 #include "ace/MEM_IO.h"
   3 #include "ace/Handle_Set.h"
   4
   5 #if (ACE_HAS_POSITION_INDEPENDENT_POINTERS == 1)
   6
   7 #if !defined (__ACE_INLINE__)
   8 #include "ace/MEM_IO.inl"
   9 #endif /* __ACE_INLINE__ */
  10
  11
  12
  13 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  14
  15 ACE_ALLOC_HOOK_DEFINE(ACE_MEM_IO)
  16
  17 ACE_Reactive_MEM_IO::~ACE_Reactive_MEM_IO (void)
  18 {
  19 }
  20
  21 int
  22 ACE_Reactive_MEM_IO::init (ACE_HANDLE handle,
  23                            const ACE_TCHAR *name,
  24                            MALLOC_OPTIONS *options)
  25 {
  26   ACE_TRACE ("ACE_Reactive_MEM_IO::init");
  27   this->handle_ = handle;
  28   return this->create_shm_malloc (name,
  29                                   options);
  30 }
  31
  32 ssize_t
  33 ACE_Reactive_MEM_IO::recv_buf (ACE_MEM_SAP_Node *&buf,
  34                                int flags,
  35                                const ACE_Time_Value *timeout)
  36 {
  37   ACE_TRACE ("ACE_Reactive_MEM_IO::recv_buf");
  38
  39   if (this->shm_malloc_ == 0 || this->handle_ == ACE_INVALID_HANDLE)
  40     return -1;
  41
  42   ACE_OFF_T new_offset = 0;
  43   ssize_t retv = ACE::recv (this->handle_,
  44                             (char *) &new_offset,
  45                             sizeof (ACE_OFF_T),
  46                             flags,
  47                             timeout);
  48
  49   if (retv == 0)
  50     {
  51       //      ACELIB_DEBUG ((LM_INFO, "MEM_Stream closed\n"));
  52       buf = 0;
  53       return 0;
  54     }
  55   else if (retv != static_cast <ssize_t> (sizeof (ACE_OFF_T)))
  56     {
  57       //  Nothing available or we are really screwed.
  58       buf = 0;
  59       return -1;
  60     }
  61
  62   return this->get_buf_len (new_offset, buf);
  63 }
  64
  65 ssize_t
  66 ACE_Reactive_MEM_IO::send_buf (ACE_MEM_SAP_Node *buf,
  67                                int flags,
  68                                const ACE_Time_Value *timeout)
  69 {
  70   ACE_TRACE ("ACE_Reactive_MEM_IO::send_buf");
  71
  72   if (this->shm_malloc_ == 0 || this->handle_ == ACE_INVALID_HANDLE)
  73     {
  74       return -1;
  75     }
  76
  77   // The offset.
  78   ACE_OFF_T offset =
  79     ACE_Utils::truncate_cast<ACE_OFF_T> (
  80       reinterpret_cast<char *> (buf)
  81       - static_cast<char *> (this->shm_malloc_->base_addr ()));
  82
  83   // Send the offset value over the socket.
  84   if (ACE::send (this->handle_,
  85                  (const char *) &offset,
  86                  sizeof (offset),
  87                  flags,
  88                  timeout) != static_cast <ssize_t> (sizeof (offset)))
  89     {
  90       // unsuccessful send, release the memory in the shared-memory.
  91       this->release_buffer (buf);
  92
  93       return -1;
  94     }
  95
  96   return ACE_Utils::truncate_cast<ssize_t> (buf->size ());
  97 }
  98
  99 #if defined (ACE_WIN32) || !defined (_ACE_USE_SV_SEM)
 100 int
 101 ACE_MT_MEM_IO::Simple_Queue::write (ACE_MEM_SAP_Node *new_node)
 102 {
 103   if (this->mq_ == 0)
 104     return -1;
 105
 106   // Here, we assume we already have acquired the lock necessary.
 107   // And we are allowed to write.
 108   if (this->mq_->tail_.addr () == 0)     // nothing in the queue.
 109     {
 110       this->mq_->head_ = new_node;
 111       this->mq_->tail_ = new_node;
 112       new_node->next_ = 0;
 113     }
 114   else
 115     {
 116       this->mq_->tail_->next_ = new_node;
 117       new_node->next_ = 0;
 118       this->mq_->tail_ = new_node;
 119     }
 120   return 0;
 121 }
 122
 123 ACE_MEM_SAP_Node *
 124 ACE_MT_MEM_IO::Simple_Queue::read ()
 125 {
 126   if (this->mq_ == 0)
 127     return 0;
 128
 129   ACE_MEM_SAP_Node *retv = 0;
 130
 131   ACE_SEH_TRY
 132     {
 133       retv = this->mq_->head_;
 134       // Here, we assume we already have acquired the lock necessary
 135       // and there are soemthing in the queue.
 136       if (this->mq_->head_ == this->mq_->tail_)
 137         {
 138           // Last message in the queue.
 139           this->mq_->head_ = 0;
 140           this->mq_->tail_ = 0;
 141         }
 142       else
 143         this->mq_->head_ = retv->next_;
 144     }
 145   ACE_SEH_EXCEPT (this->malloc_->memory_pool ().seh_selector (GetExceptionInformation ()))
 146     {
 147     }
 148
 149   return retv;
 150 }
 151
 152 ACE_MT_MEM_IO::~ACE_MT_MEM_IO ()
 153 {
 154   delete this->recv_channel_.sema_;
 155   delete this->recv_channel_.lock_;
 156   delete this->send_channel_.sema_;
 157   delete this->send_channel_.lock_;
 158 }
 159
 160 int
 161 ACE_MT_MEM_IO::init (ACE_HANDLE handle,
 162                      const ACE_TCHAR *name,
 163                      MALLOC_OPTIONS *options)
 164 {
 165   ACE_TRACE ("ACE_MT_MEM_IO::init");
 166   ACE_UNUSED_ARG (handle);
 167
 168   // @@ Give me a rule on naming and how the queue should
 169   //    be kept in the shared memory and we are done
 170   //    with this.
 171   if (this->create_shm_malloc (name, options) == -1)
 172     return -1;
 173
 174   ACE_TCHAR server_sema [MAXPATHLEN];
 175   ACE_TCHAR client_sema [MAXPATHLEN];
 176   ACE_TCHAR server_lock [MAXPATHLEN];
 177   ACE_TCHAR client_lock [MAXPATHLEN];
 178   const ACE_TCHAR *basename = ACE::basename (name);
 179   //  size_t baselen = ACE_OS::strlen (basename);
 180
 181   // Building names.  @@ Check buffer overflow?
 182   ACE_OS::strcpy (server_sema, basename);
 183   ACE_OS::strcat (server_sema, ACE_TEXT ("_sema_to_server"));
 184   ACE_OS::strcpy (client_sema, basename);
 185   ACE_OS::strcat (client_sema, ACE_TEXT ("_sema_to_client"));
 186   ACE_OS::strcpy (server_lock, basename);
 187   ACE_OS::strcat (server_lock, ACE_TEXT ("_lock_to_server"));
 188   ACE_OS::strcpy (client_lock, basename);
 189   ACE_OS::strcat (client_lock, ACE_TEXT ("_lock_to_client"));
 190
 191   void *to_server_ptr = 0;
 192   // @@ Here, we assume the shared memory fill will never be resued.
 193   //    So we can determine whether we are server or client by examining
 194   //    if the simple message queues have already been set up in
 195   //    the Malloc object or not.
 196   if (this->shm_malloc_->find ("to_server", to_server_ptr) == -1)
 197     {
 198       void *ptr = 0;
 199       // We are server.
 200       ACE_ALLOCATOR_RETURN (ptr,
 201                             this->shm_malloc_->malloc (2 * sizeof (MQ_Struct)),
 202                             -1);
 203
 204       MQ_Struct *mymq = reinterpret_cast<MQ_Struct *> (ptr);
 205       mymq->tail_ = 0;
 206       mymq->head_ = 0;
 207       (mymq + 1)->tail_ = 0;
 208       (mymq + 1)->head_ = 0;
 209       if (this->shm_malloc_->bind ("to_server", mymq) == -1)
 210         return -1;
 211
 212       if (this->shm_malloc_->bind ("to_client", mymq + 1) == -1)
 213         return -1;
 214
 215       this->recv_channel_.queue_.init (mymq, this->shm_malloc_);
 216       ACE_NEW_RETURN (this->recv_channel_.sema_,
 217                       ACE_SYNCH_PROCESS_SEMAPHORE (0, server_sema),
 218                       -1);
 219       ACE_NEW_RETURN (this->recv_channel_.lock_,
 220                       ACE_SYNCH_PROCESS_MUTEX (server_lock),
 221                       -1);
 222
 223       this->send_channel_.queue_.init (mymq + 1, this->shm_malloc_);
 224       ACE_NEW_RETURN (this->send_channel_.sema_,
 225                       ACE_SYNCH_PROCESS_SEMAPHORE (0, client_sema),
 226                       -1);
 227       ACE_NEW_RETURN (this->send_channel_.lock_,
 228                       ACE_SYNCH_PROCESS_MUTEX (client_lock),
 229                       -1);
 230     }
 231   else
 232     {
 233       // we are client.
 234       MQ_Struct *mymq = reinterpret_cast<MQ_Struct *> (to_server_ptr);
 235       this->recv_channel_.queue_.init (mymq +1, this->shm_malloc_);
 236       ACE_NEW_RETURN (this->recv_channel_.sema_,
 237                       ACE_SYNCH_PROCESS_SEMAPHORE (0, client_sema),
 238                       -1);
 239       ACE_NEW_RETURN (this->recv_channel_.lock_,
 240                       ACE_SYNCH_PROCESS_MUTEX (client_lock),
 241                       -1);
 242
 243       this->send_channel_.queue_.init (mymq, this->shm_malloc_);
 244       ACE_NEW_RETURN (this->send_channel_.sema_,
 245                       ACE_SYNCH_PROCESS_SEMAPHORE (0, server_sema),
 246                       -1);
 247       ACE_NEW_RETURN (this->send_channel_.lock_,
 248                       ACE_SYNCH_PROCESS_MUTEX (server_lock),
 249                       -1);
 250     }
 251   return 0;
 252 }
 253
 254 int
 255 ACE_MT_MEM_IO::fini ()
 256 {
 257   const int ret = ACE_MEM_SAP::fini ();
 258   ACE_Process_Mutex::unlink (this->recv_channel_.lock_->name ());
 259   ACE_Process_Mutex::unlink (this->send_channel_.lock_->name ());
 260   return ret;
 261 }
 262
 263 ssize_t
 264 ACE_MT_MEM_IO::recv_buf (ACE_MEM_SAP_Node *&buf,
 265                          int flags,
 266                          const ACE_Time_Value *timeout)
 267 {
 268   ACE_TRACE ("ACE_MT_MEM_IO::recv_buf");
 269
 270   // @@ Don't know how to handle timeout yet.
 271   ACE_UNUSED_ARG (timeout);
 272   ACE_UNUSED_ARG (flags);
 273
 274   if (this->shm_malloc_ == 0)
 275     {
 276       return -1;
 277     }
 278
 279   // Need to handle timeout here.
 280   if (this->recv_channel_.sema_->acquire () == -1)
 281     {
 282       return -1;
 283     }
 284
 285   {
 286     // @@ We can probably skip the lock in certain circumstance.
 287     ACE_GUARD_RETURN (ACE_SYNCH_PROCESS_MUTEX, ace_mon, *this->recv_channel_.lock_, -1);
 288
 289     buf = this->recv_channel_.queue_.read ();
 290
 291     if (buf != 0)
 292       {
 293         return ACE_Utils::truncate_cast<ssize_t> (buf->size ());
 294       }
 295
 296     return -1;
 297   }
 298 }
 299
 300 ssize_t
 301 ACE_MT_MEM_IO::send_buf (ACE_MEM_SAP_Node *buf,
 302                          int flags,
 303                          const ACE_Time_Value *timeout)
 304 {
 305   ACE_TRACE ("ACE_MT_MEM_IO::send_buf");
 306
 307   // @@ Don't know how to handle timeout yet.
 308   ACE_UNUSED_ARG (timeout);
 309   ACE_UNUSED_ARG (flags);
 310
 311   if (this->shm_malloc_ == 0)
 312     {
 313       return -1;
 314     }
 315
 316   {
 317     // @@ We can probably skip the lock in certain curcumstances.
 318     ACE_GUARD_RETURN (ACE_SYNCH_PROCESS_MUTEX, ace_mon, *this->send_channel_.lock_, -1);
 319
 320     if (this->send_channel_.queue_.write (buf) == -1)
 321       {
 322         this->release_buffer (buf);
 323         return -1;
 324       }
 325   }
 326
 327   if (this->send_channel_.sema_->release () == -1)
 328     {
 329       return -1;
 330     }
 331
 332   return ACE_Utils::truncate_cast<ssize_t> (buf->size ());
 333 }
 334 #endif /* ACE_WIN32 || !_ACE_USE_SV_SEM */
 335
 336 void
 337 ACE_MEM_IO::dump (void) const
 338 {
 339 #if defined (ACE_HAS_DUMP)
 340   ACE_TRACE ("ACE_MEM_IO::dump");
 341 #endif /* ACE_HAS_DUMP */
 342 }
 343
 344 int
 345 ACE_MEM_IO::init (const ACE_TCHAR *name,
 346                   ACE_MEM_IO::Signal_Strategy type,
 347                   ACE_MEM_SAP::MALLOC_OPTIONS *options)
 348 {
 349   ACE_UNUSED_ARG (type);
 350
 351   delete this->deliver_strategy_;
 352   this->deliver_strategy_ = 0;
 353   switch (type)
 354     {
 355     case ACE_MEM_IO::Reactive:
 356       ACE_NEW_RETURN (this->deliver_strategy_,
 357                       ACE_Reactive_MEM_IO (),
 358                       -1);
 359       break;
 360 #if defined (ACE_WIN32) || !defined (_ACE_USE_SV_SEM)
 361     case ACE_MEM_IO::MT:
 362       ACE_NEW_RETURN (this->deliver_strategy_,
 363                       ACE_MT_MEM_IO (),
 364                       -1);
 365       break;
 366 #endif /* ACE_WIN32 || !_ACE_USE_SV_SEM */
 367     default:
 368       return -1;
 369     }
 370
 371   return this->deliver_strategy_->init (this->get_handle (),
 372                                         name,
 373                                         options);
 374 }
 375
 376 int
 377 ACE_MEM_IO::fini (void)
 378 {
 379   if (this->deliver_strategy_ != 0)
 380     {
 381       return this->deliver_strategy_->fini ();
 382     }
 383   else
 384     {
 385       return -1;
 386     }
 387 }
 388
 389 // Allows a client to read from a socket without having to provide
 390 // a buffer to read.  This method determines how much data is in the
 391 // socket, allocates a buffer of this size, reads in the data, and
 392 // returns the number of bytes read.
 393
 394 ssize_t
 395 ACE_MEM_IO::send (const ACE_Message_Block *message_block,
 396                   const ACE_Time_Value *timeout)
 397 {
 398   ACE_TRACE ("ACE_MEM_IO::send");
 399
 400   if (this->deliver_strategy_ == 0)
 401     {
 402       return -1;                  // Something went seriously wrong.
 403     }
 404
 405   size_t len = message_block->total_length ();
 406
 407   if (len != 0)
 408     {
 409       ACE_MEM_SAP_Node *buf =
 410         reinterpret_cast<ACE_MEM_SAP_Node *> (
 411           this->deliver_strategy_->acquire_buffer (
 412             ACE_Utils::truncate_cast<ssize_t> (len)));
 413
 414       size_t n = 0;
 415
 416       while (message_block != 0)
 417         {
 418           ACE_OS::memcpy (static_cast<char *> (buf->data ()) + n,
 419                           message_block->rd_ptr (),
 420                           message_block->length ());
 421           n += message_block->length ();
 422
 423           if (message_block->cont ())
 424             {
 425               message_block = message_block->cont ();
 426             }
 427           else
 428             {
 429               message_block = message_block->next ();
 430             }
 431         }
 432
 433       buf->size_ = len;
 434
 435       return this->deliver_strategy_->send_buf (buf,
 436                                                 0,
 437                                                 timeout);
 438     }
 439
 440   return 0;
 441 }
 442
 443 ACE_END_VERSIONED_NAMESPACE_DECL
 444
 445 #endif /* ACE_HAS_POSITION_INDEPENDENT_POINTERS == 1 */