ACE/protocols/ace/RMCast/Link.cpp

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2 #include "ace/Time_Value.h"        // ACE_Time_Value
   3 #include "ace/OS_NS_stdio.h"
   4 #include "ace/OS_NS_stdlib.h"
   5 #include "ace/OS_NS_time.h"
   6 #include "ace/OS_NS_sys_socket.h"
   7
   8 #include "Link.h"
   9
  10 namespace ACE_RMCast
  11 {
  12   Link::
  13   ~Link ()
  14   {
  15     ssock_.close ();
  16     rsock_.close ();
  17   }
  18
  19   Link::
  20   Link (Address const& addr, Parameters const& params)
  21       : params_ (params),
  22         addr_ (addr),
  23         ssock_ (Address (static_cast<unsigned short> (0),
  24                          static_cast<ACE_UINT32> (INADDR_ANY)),
  25                 AF_INET,
  26                 IPPROTO_UDP,
  27                 1),
  28         stop_ (false)
  29
  30   {
  31     ACE_OS::srand ((unsigned int) ACE_OS::time (0));
  32
  33
  34     rsock_.set_option (IP_MULTICAST_LOOP, 0);
  35     // rsock_.set_option (IP_MULTICAST_TTL, 0);
  36
  37     // Set recv/send buffers.
  38     //
  39     {
  40       int r (131070);
  41       int s (sizeof (r));
  42
  43       static_cast<ACE_SOCK&> (rsock_).set_option (
  44         SOL_SOCKET, SO_RCVBUF, &r, s);
  45
  46       static_cast<ACE_SOCK&> (ssock_).set_option (
  47         SOL_SOCKET, SO_RCVBUF, &r, s);
  48
  49       rsock_.get_option (SOL_SOCKET, SO_RCVBUF, &r, &s);
  50       //cerr << 5 << "recv buffer size: " << r << endl;
  51
  52       ssock_.get_option (SOL_SOCKET, SO_RCVBUF, &r, &s);
  53       //cerr << 5 << "send buffer size: " << r << endl;
  54     }
  55
  56     // Bind address and port.
  57     //
  58     if (ACE_OS::connect (ssock_.get_handle (),
  59                          reinterpret_cast<sockaddr*> (addr_.get_addr ()),
  60                          addr_.get_addr_size ()) == -1)
  61     {
  62       ACE_OS::perror ("connect: ");
  63       ACE_OS::abort ();
  64     }
  65
  66
  67     ssock_.get_local_addr (self_);
  68
  69     //cerr << 5 << "self: " << self_ << endl;
  70   }
  71
  72   void Link::
  73   in_start (In_Element* in)
  74   {
  75     Element::in_start (in);
  76
  77     rsock_.join (addr_);
  78
  79     // Start receiving thread.
  80     //
  81     recv_mgr_.spawn (recv_thunk, this);
  82   }
  83
  84   void Link::
  85   out_start (Out_Element* out)
  86   {
  87     Element::out_start (out);
  88   }
  89
  90   void Link::
  91   in_stop ()
  92   {
  93     // Stop receiving thread.
  94     //
  95     {
  96       Lock l (mutex_);
  97       stop_ = true;
  98     }
  99     recv_mgr_.wait ();
 100
 101     Element::in_stop ();
 102   }
 103
 104   void Link::send (Message_ptr m)
 105   {
 106     // Simulate message loss and reordering.
 107     //
 108     if (params_.simulator ())
 109     {
 110       if ((ACE_OS::rand () % 17) != 0)
 111       {
 112         Lock l (mutex_);
 113
 114         if (hold_.get ())
 115         {
 116           send_ (m);
 117           send_ (hold_);
 118           hold_ = Message_ptr (0);
 119         }
 120         else
 121         {
 122           if ((ACE_OS::rand () % 17) != 0)
 123           {
 124             send_ (m);
 125           }
 126           else
 127           {
 128             hold_ = m;
 129
 130             // Make a copy in M so that the reliable loop below
 131             // won't add FROM and TO to HOLD_.
 132             //
 133             m = hold_->clone ();
 134           }
 135         }
 136       }
 137     }
 138     else
 139       send_ (m);
 140
 141     // Reliable loop.
 142     //
 143     m->add (Profile_ptr (new From (self_)));
 144     m->add (Profile_ptr (new To (self_)));
 145
 146     in_->recv (m);
 147   }
 148
 149   void Link::
 150   send_ (Message_ptr m)
 151   {
 152     ostream os (m->size (), 1); // Always little-endian.
 153
 154     os << *m;
 155
 156     if (os.length () > size_t (params_.max_packet_size ()))
 157     {
 158       ACE_ERROR ((LM_ERROR,
 159                   "packet length (%d) exceeds max_poacket_size (%d)\n",
 160                   os.length (), params_.max_packet_size ()));
 161
 162       for (Message::ProfileIterator i (m->begin ()); !i.done (); i.advance ())
 163       {
 164         ACE_ERROR ((LM_ERROR,
 165                     "profile id: %d; size: %d\n",
 166                     (*i).ext_id_, (*i).int_id_->size ()));
 167       }
 168
 169       ACE_OS::abort ();
 170     }
 171
 172     ssock_.send (os.buffer (), os.length (), addr_);
 173
 174     /*
 175       if (m->find (nrtm::id))
 176       {
 177         ACE_OS::write (1, os.buffer (), os.length ());
 178         ACE_OS::exit (1);
 179       }
 180     */
 181   }
 182
 183   void Link::recv ()
 184   {
 185     size_t max_packet_size (params_.max_packet_size ());
 186
 187     std::unique_ptr<char[]> holder (new char[max_packet_size + ACE_CDR::MAX_ALIGNMENT]);
 188
 189     char* data = ACE_ptr_align_binary (holder.get (), ACE_CDR::MAX_ALIGNMENT);
 190
 191     size_t size (0);
 192
 193     while (true)
 194     {
 195       //@@ Should I lock here?
 196       //
 197
 198       Address addr;
 199
 200       // Block for up to one tick waiting for an incomming message.
 201       //
 202       for (;;)
 203       {
 204         ACE_Time_Value t (params_.tick ());
 205         ssize_t r = rsock_.recv (data, 4, addr, MSG_PEEK, &t);
 206
 207
 208         // Check for cancellation request.
 209         //
 210         {
 211           Lock l (mutex_);
 212           if (stop_)
 213             return;
 214         }
 215
 216         if (r == -1)
 217         {
 218           if (errno != ETIME)
 219             ACE_OS::abort ();
 220         }
 221         else
 222         {
 223           size = static_cast<size_t> (r);
 224           break;
 225         }
 226       }
 227
 228
 229       if (size != 4 || addr == self_)
 230       {
 231         // Discard bad messages and ones from ourselvs since
 232         // we are using reliable loopback.
 233         //
 234         rsock_.recv (data, 0, addr);
 235         continue;
 236       }
 237
 238       u32 msg_size;
 239       {
 240         istream is (data, size, 1); // Always little-endian.
 241         is >> msg_size;
 242       }
 243
 244       if (msg_size <= 4 || msg_size > max_packet_size)
 245       {
 246         // Bad message.
 247         //
 248         rsock_.recv (data, 0, addr);
 249         continue;
 250       }
 251
 252       size = rsock_.recv (data, max_packet_size, addr);
 253
 254       if (msg_size != size)
 255       {
 256         // Bad message.
 257         //
 258         continue;
 259       }
 260
 261       //cerr << 6 << "from: " << addr << endl;
 262
 263       Message_ptr m (new Message ());
 264
 265       m->add (Profile_ptr (new From (addr)));
 266       m->add (Profile_ptr (new To (self_)));
 267
 268       istream is (data, size, 1); // Always little-endian.
 269
 270       is >> msg_size;
 271
 272       while (true)
 273       {
 274         u16 id, size;
 275
 276         if (!((is >> id) && (is >> size))) break;
 277
 278         //cerr << 6 << "reading profile with id " << id << " "
 279         //     << size << " bytes long" << endl;
 280
 281         Profile::Header hdr (id, size);
 282
 283         if (id == SN::id)
 284           {
 285             m->add (Profile_ptr (new SN (hdr, is)));
 286           }
 287         else if (id == Data::id)
 288           {
 289             m->add (Profile_ptr (new Data (hdr, is)));
 290           }
 291         else if (id == NAK::id)
 292           {
 293             m->add (Profile_ptr (new NAK (hdr, is)));
 294           }
 295         else if (id == NRTM::id)
 296           {
 297             m->add (Profile_ptr (new NRTM (hdr, is)));
 298           }
 299         else if (id == NoData::id)
 300           {
 301             m->add (Profile_ptr (new NoData (hdr, is)));
 302           }
 303         else if (id == Part::id)
 304           {
 305             m->add (Profile_ptr (new Part (hdr, is)));
 306           }
 307         else
 308           {
 309             //cerr << 0 << "unknown profile id " << hdr.id () << endl;
 310             ACE_OS::abort ();
 311           }
 312       }
 313
 314       in_->recv (m);
 315     }
 316   }
 317
 318   ACE_THR_FUNC_RETURN Link::
 319   recv_thunk (void* obj)
 320   {
 321     reinterpret_cast<Link*> (obj)->recv ();
 322     return 0;
 323   }
 324
 325   void Link::recv (Message_ptr)
 326   {
 327     ACE_OS::abort ();
 328   }
 329 }