ACE_RMCast::Link Class Reference

#include <Link.h>

Inheritance diagram for ACE_RMCast::Link:

[legend]Collaboration diagram for ACE_RMCast::Link:

[legend]List of all members.

Public Member Functions
	~Link ()
	Link (Address const &addr, Parameters const &params)
virtual void	in_start (In_Element *in)
virtual void	out_start (Out_Element *out)
virtual void	in_stop ()
virtual void	send (Message_ptr m)
Private Member Functions
virtual void	send_ (Message_ptr m)
void	recv ()
virtual void	recv (Message_ptr)
Static Private Member Functions
static ACE_THR_FUNC_RETURN	recv_thunk (void *obj)
Private Attributes
Parameters const &	params_
Address	addr_
Address	self_
ACE_SOCK_Dgram_Mcast	rsock_
ACE_SOCK_Dgram	ssock_
bool	stop_
ACE_Thread_Manager	recv_mgr_
Message_ptr	hold_
Mutex	mutex_

---

Detailed Description

Definition at line 19 of file Link.h.

---

Constructor & Destructor Documentation

ACE_RMCast::Link::~Link

(

)

Definition at line 16 of file Link.cpp.

References ACE_SOCK::close(), rsock_, and ssock_.

  {
    ssock_.close ();
    rsock_.close ();
  }

ACE_RMCast::Link::Link	(	Address const &	addr,
		Parameters const &	params
	)

Definition at line 23 of file Link.cpp.

References ACE_OS::abort(), addr_, ACE_OS::connect(), ACE_INET_Addr::get_addr(), ACE_INET_Addr::get_addr_size(), ACE_IPC_SAP::get_handle(), ACE_SOCK::get_local_addr(), ACE_SOCK::get_option(), IP_MULTICAST_LOOP, ACE_OS::perror(), rsock_, self_, ACE_SOCK_Dgram_Mcast::set_option(), SO_RCVBUF, SOL_SOCKET, ACE_OS::srand(), ssock_, and ACE_OS::time().

      : params_ (params),
        addr_ (addr),
        ssock_ (Address (static_cast<unsigned short> (0),
                         static_cast<ACE_UINT32> (INADDR_ANY)),
                AF_INET,
                IPPROTO_UDP,
                1),
        stop_ (false)

  {
    ACE_OS::srand ((unsigned int) ACE_OS::time (0));


    rsock_.set_option (IP_MULTICAST_LOOP, 0);
    // rsock_.set_option (IP_MULTICAST_TTL, 0);

    // Set recv/send buffers.
    //
    {
      int r (131070);
      int s (sizeof (r));

      static_cast<ACE_SOCK&> (rsock_).set_option (
        SOL_SOCKET, SO_RCVBUF, &r, s);

      static_cast<ACE_SOCK&> (ssock_).set_option (
        SOL_SOCKET, SO_RCVBUF, &r, s);

      rsock_.get_option (SOL_SOCKET, SO_RCVBUF, &r, &s);
      //cerr << 5 << "recv buffer size: " << r << endl;

      ssock_.get_option (SOL_SOCKET, SO_RCVBUF, &r, &s);
      //cerr << 5 << "send buffer size: " << r << endl;

    }

    // Bind address and port.
    //
    if (ACE_OS::connect (ssock_.get_handle (),
                         reinterpret_cast<sockaddr*> (addr_.get_addr ()),
                         addr_.get_addr_size ()) == -1)
    {
      ACE_OS::perror ("connect: ");
      ACE_OS::abort ();
    }


    ssock_.get_local_addr (self_);

    //cerr << 5 << "self: " << self_ << endl;
  }

---

Member Function Documentation

void ACE_RMCast::Link::in_start

(

In_Element *

in

)

[virtual]

Reimplemented from ACE_RMCast::In_Element.

Definition at line 77 of file Link.cpp.

References addr_, ACE_RMCast::In_Element::in_start(), ACE_SOCK_Dgram_Mcast::join(), recv_mgr_, recv_thunk(), rsock_, and ACE_Thread_Manager::spawn().

  {
    Element::in_start (in);

    rsock_.join (addr_);

    // Start receiving thread.
    //
    recv_mgr_.spawn (recv_thunk, this);
  }

void ACE_RMCast::Link::in_stop

(

)

[virtual]

Reimplemented from ACE_RMCast::In_Element.

Definition at line 95 of file Link.cpp.

References ACE_RMCast::In_Element::in_stop(), mutex_, recv_mgr_, stop_, and ACE_Thread_Manager::wait().

  {
    // Stop receiving thread.
    //
    {
      Lock l (mutex_);
      stop_ = true;
    }
    recv_mgr_.wait ();

    Element::in_stop ();
  }

void ACE_RMCast::Link::out_start

(

Out_Element *

out

)

[virtual]

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 89 of file Link.cpp.

References ACE_RMCast::Out_Element::out_start().

  {
    Element::out_start (out);
  }

void ACE_RMCast::Link::recv

(

Message_ptr

)

[private, virtual]

Reimplemented from ACE_RMCast::In_Element.

Definition at line 333 of file Link.cpp.

References ACE_OS::abort().

  {
    ACE_OS::abort ();
  }

void ACE_RMCast::Link::recv

(

)

[private]

Definition at line 187 of file Link.cpp.

References ACE_OS::abort(), ACE_ptr_align_binary(), ACE_RMCast::Profile::id(), ACE_RMCast::In_Element::in_, ACE_CDR::MAX_ALIGNMENT, ACE_RMCast::Parameters::max_packet_size(), MSG_PEEK, mutex_, params_, ACE_RMCast::In_Element::recv(), ACE_SOCK_Dgram::recv(), rsock_, self_, stop_, and ACE_RMCast::Parameters::tick().

Referenced by recv_thunk().

  {
    size_t max_packet_size (params_.max_packet_size ());

    // This is wicked.
    //
    ACE_Auto_Ptr<char> holder (
      reinterpret_cast<char*> (
        operator new (max_packet_size + ACE_CDR::MAX_ALIGNMENT)));

    char* data = ACE_ptr_align_binary (holder.get (), ACE_CDR::MAX_ALIGNMENT);

    size_t size (0);

    while (true)
    {
      //@@ Should I lock here?
      //

      Address addr;

      // Block for up to one tick waiting for an incomming message.
      //
      for (;;)
      {
        ACE_Time_Value t (params_.tick ());
        ssize_t r = rsock_.recv (data, 4, addr, MSG_PEEK, &t);


        // Check for cancellation request.
        //
        {
          Lock l (mutex_);
          if (stop_)
            return;
        }

        if (r == -1)
        {
          if (errno != ETIME)
            ACE_OS::abort ();
        }
        else
        {
          size = static_cast<size_t> (r);
          break;
        }
      }


      if (size != 4 || addr == self_)
      {
        // Discard bad messages and ones from ourselvs since
        // we are using reliable loopback.
        //
        rsock_.recv (data, 0, addr);
        continue;
      }

      u32 msg_size;
      {
        istream is (data, size, 1); // Always little-endian.
        is >> msg_size;
      }

      if (msg_size <= 4 || msg_size > max_packet_size)
      {
        // Bad message.
        //
        rsock_.recv (data, 0, addr);
        continue;
      }

      size = rsock_.recv (data, max_packet_size, addr);

      if (msg_size != size)
      {
        // Bad message.
        //
        continue;
      }

      //cerr << 6 << "from: " << addr << endl;

      Message_ptr m (new Message ());

      m->add (Profile_ptr (new From (addr)));
      m->add (Profile_ptr (new To (self_)));

      istream is (data, size, 1); // Always little-endian.

      is >> msg_size;

      while (true)
      {
        u16 id, size;

        if (!((is >> id) && (is >> size))) break;

        //cerr << 6 << "reading profile with id " << id << " "
        //     << size << " bytes long" << endl;

        Profile::Header hdr (id, size);

        if (id == SN::id)
          {
            m->add (Profile_ptr (new SN (hdr, is)));
          }
        else if (id == Data::id)
          {
            m->add (Profile_ptr (new Data (hdr, is)));
          }
        else if (id == NAK::id)
          {
            m->add (Profile_ptr (new NAK (hdr, is)));
          }
        else if (id == NRTM::id)
          {
            m->add (Profile_ptr (new NRTM (hdr, is)));
          }
        else if (id == NoData::id)
          {
            m->add (Profile_ptr (new NoData (hdr, is)));
          }
        else if (id == Part::id)
          {
            m->add (Profile_ptr (new Part (hdr, is)));
          }
        else
          {
            //cerr << 0 << "unknown profile id " << hdr.id () << endl;
            ACE_OS::abort ();
          }
      }

      in_->recv (m);
    }
  }

ACE_THR_FUNC_RETURN ACE_RMCast::Link::recv_thunk

(

void *

obj

)

[static, private]

Definition at line 327 of file Link.cpp.

References recv().

Referenced by in_start().

  {
    reinterpret_cast<Link*> (obj)->recv ();
    return 0;
  }

void ACE_RMCast::Link::send

(

Message_ptr

m

)

[virtual]

Reimplemented from ACE_RMCast::Out_Element.

Definition at line 108 of file Link.cpp.

References ACE_Strong_Bound_Ptr< X, ACE_LOCK >::get(), hold_, ACE_RMCast::In_Element::in_, mutex_, params_, ACE_OS::rand(), ACE_RMCast::In_Element::recv(), self_, send_(), and ACE_RMCast::Parameters::simulator().

  {
    // Simulate message loss and reordering.
    //
    if (params_.simulator ())
    {
      if ((ACE_OS::rand () % 17) != 0)
      {
        Lock l (mutex_);

        if (hold_.get ())
        {
          send_ (m);
          send_ (hold_);
          hold_ = Message_ptr (0);
        }
        else
        {
          if ((ACE_OS::rand () % 17) != 0)
          {
            send_ (m);
          }
          else
          {
            hold_ = m;

            // Make a copy in M so that the reliable loop below
            // won't add FROM and TO to HOLD_.
            //
            m = hold_->clone ();
          }
        }
      }
    }
    else
      send_ (m);

    // Reliable loop.
    //
    m->add (Profile_ptr (new From (self_)));
    m->add (Profile_ptr (new To (self_)));

    in_->recv (m);
  }

void ACE_RMCast::Link::send_

(

Message_ptr

m

)

[private, virtual]

Definition at line 154 of file Link.cpp.

References ACE_OS::abort(), ACE_ERROR, addr_, LM_ERROR, ACE_RMCast::Parameters::max_packet_size(), params_, ACE_SOCK_Dgram::send(), and ssock_.

Referenced by send().

  {
    ostream os (m->size (), 1); // Always little-endian.

    os << *m;

    if (os.length () > size_t (params_.max_packet_size ()))
    {
      ACE_ERROR ((LM_ERROR,
                  "packet length (%d) exceeds max_poacket_size (%d)\n",
                  os.length (), params_.max_packet_size ()));

      for (Message::ProfileIterator i (m->begin ()); !i.done (); i.advance ())
      {
        ACE_ERROR ((LM_ERROR,
                    "profile id: %d; size: %d\n",
                    (*i).ext_id_, (*i).int_id_->size ()));
      }

      ACE_OS::abort ();
    }

    ssock_.send (os.buffer (), os.length (), addr_);

    /*
      if (m->find (nrtm::id))
      {
        ACE_OS::write (1, os.buffer (), os.length ());
        ACE_OS::exit (1);
      }
    */
  }

---

Member Data Documentation

Address ACE_RMCast::Link::addr_ [private]

Definition at line 57 of file Link.h.

Referenced by in_start(), Link(), and send_().

Message_ptr ACE_RMCast::Link::hold_ [private]

Definition at line 66 of file Link.h.

Referenced by send().

Mutex ACE_RMCast::Link::mutex_ [private]

Definition at line 68 of file Link.h.

Referenced by in_stop(), recv(), and send().

Parameters const& ACE_RMCast::Link::params_ [private]

Definition at line 55 of file Link.h.

Referenced by recv(), send(), and send_().

ACE_Thread_Manager ACE_RMCast::Link::recv_mgr_ [private]

Definition at line 62 of file Link.h.

Referenced by in_start(), and in_stop().

ACE_SOCK_Dgram_Mcast ACE_RMCast::Link::rsock_ [private]

Definition at line 58 of file Link.h.

Referenced by in_start(), Link(), recv(), and ~Link().

Address ACE_RMCast::Link::self_ [private]

Definition at line 57 of file Link.h.

Referenced by Link(), recv(), and send().

ACE_SOCK_Dgram ACE_RMCast::Link::ssock_ [private]

Definition at line 59 of file Link.h.

Referenced by Link(), send_(), and ~Link().

bool ACE_RMCast::Link::stop_ [private]

Definition at line 61 of file Link.h.

Referenced by in_stop(), and recv().

---

The documentation for this class was generated from the following files:

• Link.h
• Link.cpp

---