ACE/examples/Web_Crawler/Mem_Map_Stream.cpp

   1 #include "ace/FILE_Addr.h"
   2 #include <memory>
   3 #include "ace/Truncate.h"
   4 #include "Options.h"
   5 #include "Mem_Map_Stream.h"
   6
   7
   8 ACE_SOCK_Stream &
   9 Mem_Map_Stream::stream ()
  10 {
  11   return svc_handler_->peer ();
  12 }
  13
  14 ssize_t
  15 Mem_Map_Stream::send_n (const void *buf, size_t size, ACE_Time_Value *tv)
  16 {
  17   return svc_handler_->peer ().send_n (buf, size, 0, tv);
  18 }
  19
  20 int
  21 Mem_Map_Stream::eof () const
  22 {
  23   return this->get_pos_ >= this->end_of_mapping_plus1_;
  24 }
  25
  26 int
  27 Mem_Map_Stream::get_char ()
  28 {
  29   if (this->eof () && this->grow_file_and_remap () == -1)
  30     return EOF;
  31
  32   return *this->get_pos_++;
  33 }
  34
  35 int
  36 Mem_Map_Stream::rewind ()
  37 {
  38   this->recv_pos_ =
  39     reinterpret_cast<char *> (this->mem_map_.addr ());
  40   this->get_pos_ = this->recv_pos_;
  41   this->end_of_mapping_plus1_ =
  42     this->recv_pos_ + this->mem_map_.size ();
  43   return 0;
  44 }
  45
  46 int
  47 Mem_Map_Stream::peek_char (size_t offset)
  48 {
  49   // We may need to iterate if the size of <n> is large.
  50   while (this->get_pos_ + offset >= this->end_of_mapping_plus1_)
  51     if (this->grow_file_and_remap () == -1)
  52       return EOF;
  53
  54   return this->get_pos_[offset];
  55 }
  56
  57 const char *
  58 Mem_Map_Stream::recv () const
  59 {
  60   return this->recv_pos_;
  61 }
  62
  63 const char *
  64 Mem_Map_Stream::recv (size_t &len)
  65 {
  66   if (this->eof () && this->grow_file_and_remap () == -1)
  67     return 0;
  68
  69
  70   const char *s = this->recv_pos_;
  71   ACE_OFF_T olen = static_cast <ACE_OFF_T> (len);
  72   this->seek (olen, SEEK_CUR);
  73   len = this->get_pos_ - s;
  74   return s;
  75 }
  76
  77 size_t
  78 Mem_Map_Stream::recv_len () const
  79 {
  80   return this->get_pos_ - this->recv_pos_;
  81 }
  82
  83 const char *
  84 Mem_Map_Stream::peek_str (size_t offset,
  85                           size_t size)
  86 {
  87   // We will iterate if the size of <offset> is large.
  88   while (this->get_pos_ + (offset + size) > this->end_of_mapping_plus1_)
  89     if (this->grow_file_and_remap () == -1)
  90       return 0;
  91
  92   return &this->get_pos_[offset];
  93 }
  94
  95 ACE_OFF_T
  96 Mem_Map_Stream::seek (ACE_OFF_T offset, int whence)
  97 {
  98   switch (whence)
  99     {
 100     case SEEK_SET:
 101       this->get_pos_ =
 102         reinterpret_cast<char *> (this->mem_map_.addr ())
 103         + offset;
 104       break;
 105
 106     case SEEK_CUR:
 107       this->get_pos_ += offset;
 108       break;
 109
 110     case SEEK_END:
 111       this->get_pos_ =
 112         this->end_of_mapping_plus1_ + offset;
 113       // @@ Not sure how to implement this (yet).
 114       ACE_NOTSUP_RETURN (-1);
 115     }
 116
 117   // Make sure that the backing store will cover this.
 118   while (this->get_pos_ > this->end_of_mapping_plus1_)
 119     if (this->grow_file_and_remap () == -1)
 120       return (ACE_OFF_T) -1;
 121
 122   this->recv_pos_ = this->get_pos_;
 123   return
 124     ACE_Utils::truncate_cast<ACE_OFF_T> (
 125       this->recv_pos_ - reinterpret_cast<char *> (this->mem_map_.addr ()));
 126 }
 127
 128 Mem_Map_Stream::Svc_Handler *
 129 Mem_Map_Stream::svc_handler ()
 130 {
 131   return this->svc_handler_;
 132 }
 133
 134
 135 int
 136 Mem_Map_Stream::open (STRAT_CONNECTOR *connector,
 137                       const ACE_INET_Addr &addr)
 138 {
 139   svc_handler_ = 0;
 140
 141   // Connect to the server at <addr>. If the handler has to be
 142   // connected to the server again, the Caching strategy takes care
 143   // and uses the same connection.
 144   if (connector->connect (svc_handler_,
 145                           addr) == -1)
 146     {
 147       ACE_ERROR_RETURN ((LM_ERROR,
 148                          "%p %s %d\n",
 149                          "Connect failed",
 150                          addr.get_host_name (),
 151                          addr.get_port_number ()),
 152                         -1);
 153     }
 154   // Create a temporary filename.
 155   ACE_FILE_Addr file (ACE_sap_any_cast (ACE_FILE_Addr &));
 156
 157   // Create the temporary file via the <ACE_Mem_Map> class API.
 158   if (this->mem_map_.open (file.get_path_name (),
 159                           O_RDWR | O_CREAT | O_APPEND,
 160                           ACE_DEFAULT_FILE_PERMS) == -1)
 161     ACE_ERROR_RETURN ((LM_ERROR,
 162                        "%p\n",
 163                        "open"),
 164                       -1);
 165   // Make sure to unlink this right away so that if this process
 166   // crashes these files will be removed automatically.
 167 #if 0
 168   else if (ACE_OS::unlink (file.get_path_name ()) == -1)
 169   ACE_ERROR_RETURN ((LM_ERROR,
 170                        "%p\n",
 171                        "unlink"),
 172                       -1);
 173   else
 174 #endif
 175     // Initialize all the position pointers to 0.
 176     this->rewind ();
 177
 178   return 0;
 179 }
 180
 181 int
 182 Mem_Map_Stream::grow_file_and_remap ()
 183 {
 184   char buf[BUFSIZ + 1];
 185
 186   // Copy the next chunk of bytes from the socket into the temporary
 187   // file.
 188   ACE_Time_Value tv (*OPTIONS::instance ()->timeout ());
 189
 190   ssize_t n = this->svc_handler_->peer ().recv_n (buf,
 191                                                   sizeof buf,
 192                                                   0,
 193                                                   &tv);
 194   if (n == -1)
 195     {
 196       if (OPTIONS::instance ()->debug ())
 197         ACE_ERROR ((LM_ERROR,
 198                     "%p\n",
 199                     "recv"));
 200       return -1;
 201     }
 202   else if (n == 0)
 203     return -1;
 204   else if (ACE::write_n (this->mem_map_.handle (), buf, n) != n)
 205     ACE_ERROR_RETURN ((LM_ERROR,
 206                        "%p\n",
 207                        "write_n"),
 208                       -1);
 209
 210   // Grow the memory-mapping to encompass the entire temporary file.
 211   if (this->mem_map_.map (static_cast<size_t> (-1),
 212                           PROT_RDWR,
 213                           ACE_MAP_PRIVATE | ACE_MAP_FIXED,
 214                           ACE_DEFAULT_BASE_ADDR) == -1)
 215     ACE_ERROR_RETURN ((LM_ERROR,
 216                        "%p\n",
 217                        "map"),
 218                       -1);
 219   // MAP_FAILED is used as a "first time in" flag.
 220   if (this->recv_pos_ == MAP_FAILED)
 221     {
 222       this->recv_pos_ = reinterpret_cast<char *> (this->mem_map_.addr ());
 223       this->get_pos_ = this->recv_pos_;
 224     }
 225
 226   this->end_of_mapping_plus1_ =
 227     reinterpret_cast<char *> (this->mem_map_.addr ())
 228     + this->mem_map_.size ();
 229
 230   return 0;
 231 }
 232
 233 Mem_Map_Stream::~Mem_Map_Stream ()
 234 {
 235   // Remove the mapping and the file.
 236   this->mem_map_.remove ();
 237 }
 238