diff options
Diffstat (limited to 'protocols/ace/RMCast/RMCast_IO_UDP.cpp')
| -rw-r--r-- | protocols/ace/RMCast/RMCast_IO_UDP.cpp | 458 |
1 files changed, 0 insertions, 458 deletions
diff --git a/protocols/ace/RMCast/RMCast_IO_UDP.cpp b/protocols/ace/RMCast/RMCast_IO_UDP.cpp deleted file mode 100644 index 315d5c11f82..00000000000 --- a/protocols/ace/RMCast/RMCast_IO_UDP.cpp +++ /dev/null @@ -1,458 +0,0 @@ -// $Id$ - -#include "RMCast_IO_UDP.h" -#include "RMCast_UDP_Proxy.h" -#include "RMCast_Module_Factory.h" - -#include "ace/Handle_Set.h" -#include "ace/Reactor.h" -#include "ace/Message_Block.h" - -#if !defined (__ACE_INLINE__) -# include "RMCast_IO_UDP.i" -#endif /* ! __ACE_INLINE__ */ - -ACE_RCSID(ace, RMCast_IO_UDP, "RMCast_IO_UDP.cpp,v 1.12 2000/12/20 22:00:33 oci Exp") - -ACE_RMCast_IO_UDP::~ACE_RMCast_IO_UDP (void) -{ -} - -int -ACE_RMCast_IO_UDP::init (const ACE_INET_Addr &mcast_group, - const ACE_Addr &local, - int protocol_family, - int protocol, - int reuse_addr) -{ - this->mcast_group_ = mcast_group; - - ACE_SOCK_Dgram &dgram = this->dgram_; - return dgram.open (local, protocol_family, protocol, reuse_addr); -} - -int -ACE_RMCast_IO_UDP::subscribe (const ACE_INET_Addr &mcast_addr, - int reuse_addr, - const ACE_TCHAR *net_if, - int protocol_family, - int protocol) -{ - this->mcast_group_ = mcast_addr; - return this->dgram_.subscribe (mcast_addr, - reuse_addr, - net_if, - protocol_family, - protocol); -} - -int -ACE_RMCast_IO_UDP::handle_events (ACE_Time_Value *tv) -{ - ACE_HANDLE h = this->dgram_.get_handle (); - if (h == ACE_INVALID_HANDLE) - return -1; - - ACE_Handle_Set handle_set; - handle_set.set_bit (h); - - ACE_Countdown_Time countdown (tv); - - int r = ACE_OS::select (int(h) + 1, - handle_set, 0, 0, - tv); - if (r == -1) - { - if (errno == EINTR) - return 0; - else - return -1; - } - else if (r == 0) - { - return 0; - } - - return this->handle_input (h); -} - -int -ACE_RMCast_IO_UDP::handle_input (ACE_HANDLE) -{ - // @@ We should use a system constant instead of this literal - const int max_udp_packet_size = 65536; - char buffer[max_udp_packet_size]; - - ACE_INET_Addr from_address; - ssize_t r = - this->dgram_.recv (buffer, sizeof(buffer), from_address); - - if (r == -1) - { - // @@ LOG?? - ACE_ERROR ((LM_ERROR, - "RMCast_IO_UDP::handle_input () - error in recv %p\n", - ACE_TEXT (""))); - return -1; - } - - // ACE_HEX_DUMP ((LM_DEBUG, buffer, 16, "Receiver::handle_input")); - - // @@ Locking! - - int type = buffer[0]; - - if (type < 0 || type >= ACE_RMCast::MT_LAST) - { - // @@ Log: invalid message type!! - // @@ TODO: should we return -1? The socket is still valid, it - // makes little sense to destroy it just because one remote - // sender is sending invalid messages. Maybe we should - // strategize this too, and report the problem to the - // application, this could indicate a misconfiguration or - // something worse... - - // In any case the proxy should be destroyed, its peer is making - // something really wrong. - ACE_RMCast_UDP_Proxy *proxy; - if (this->map_.unbind (from_address, proxy) == 0) - { - this->factory_->destroy (proxy->next ()); - delete proxy; - } - return 0; - } - - ACE_RMCast_UDP_Proxy *proxy; - if (this->map_.find (from_address, proxy) != 0) - { - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::handle_input - new proxy from <%s:%d>\n", - // from_address.get_host_addr (), - // from_address.get_port_number ())); - - // @@ We should validate the message *before* creating the - // object, all we need is some sort of validation strategy, a - // different one for the receiver and another one for the - // sender. - -#if 0 - if (type == ACE_RMCast::MT_ACK - || type == ACE_RMCast::MT_JOIN - || type == ACE_RMCast::MT_LEAVE - || type == ACE_RMCast::MT_ACK_LEAVE) - { - // All these message types indicate a problem, the should be - // generated by receivers, not received by them. - return 0; - } -#endif /* 0 */ - ACE_RMCast_Module *module = this->factory_->create (); - if (module == 0) - { - // @@ LOG?? - // Try to continue working, maybe the module can be created - // later. - return 0; - } - // This is necessary to satisfy the xgcc for Lynx on Solaris - // by including the code directly causes : - // RMCast_IO_UDP.cpp:202: error: internal error--unrecognizable insn: - // (insn 1510 1507 524 (set (mem:SI (plus:SI (reg:SI 28 r28) - // (const_int 65536))) - // (reg:SI 0 r0)) -1 (insn_list 528 (insn_list 1507 (nil))) - // (nil)) - // /usr/lynx/home2/jose/98r2/src/gcc/toplev.c:1489: Internal compiler error in function fatal_insn - // to be thrown at the end of the function. - if ((proxy = allocate_and_bind_proxy(module,from_address)) == 0) - return 0; - } - - // Have the proxy process the message and do the right thing. - if (proxy->receive_message (buffer, r) != 0) - { - (void) this->map_.unbind (from_address); - this->factory_->destroy (proxy->next ()); - delete proxy; - } - - return 0; -} - -ACE_HANDLE -ACE_RMCast_IO_UDP::get_handle (void) const -{ - return this->dgram_.get_handle (); -} - -int -ACE_RMCast_IO_UDP::data (ACE_RMCast::Data &data) -{ - return this->send_data (data, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::poll (ACE_RMCast::Poll &poll) -{ - return this->send_poll (poll, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::ack_join (ACE_RMCast::Ack_Join &ack_join) -{ - return this->send_ack_join (ack_join, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::ack_leave (ACE_RMCast::Ack_Leave &ack_leave) -{ - return this->send_ack_leave (ack_leave, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::ack (ACE_RMCast::Ack &ack) -{ - return this->send_ack (ack, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::join (ACE_RMCast::Join &join) -{ - return this->send_join (join, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::leave (ACE_RMCast::Leave &leave) -{ - return this->send_leave (leave, this->mcast_group_); -} - -int -ACE_RMCast_IO_UDP::send_data (ACE_RMCast::Data &data, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_data - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // The first message block contains the header - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - ACE_UINT32 tmp; - char header[1 + 3 * sizeof(ACE_UINT32)]; - header[0] = ACE_RMCast::MT_DATA; - - tmp = ACE_HTONL (data.sequence_number); - ACE_OS::memcpy (header + 1, - &tmp, sizeof(ACE_UINT32)); - tmp = ACE_HTONL (data.total_size); - ACE_OS::memcpy (header + 1 + sizeof(ACE_UINT32), - &tmp, sizeof(ACE_UINT32)); - tmp = ACE_HTONL (data.fragment_offset); - ACE_OS::memcpy (header + 1 + 2 * sizeof(ACE_UINT32), - &tmp, sizeof(ACE_UINT32)); - - iovec iov[ACE_IOV_MAX]; - int iovcnt = 1; - - iov[0].iov_base = header; - iov[0].iov_len = sizeof(header); - - ACE_Message_Block *mb = data.payload; - - for (const ACE_Message_Block *i = mb; i != 0; i = i->cont ()) - { - iov[iovcnt].iov_base = i->rd_ptr (); - iov[iovcnt].iov_len = i->length (); - iovcnt++; - if (iovcnt >= IOV_MAX) - return -1; - } - - // @@ This pacing stuff here reduced the number of packet lost in - // loopback tests, but it should be taken out for real applications - // (or at least made configurable!) - ACE_Time_Value tv (0, 10000); - ACE_OS::sleep (tv); - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (iov, iovcnt, to) == -1) - return -1; - -#if 0 - ACE_HEX_DUMP ((LM_DEBUG, - (char*)iov[0].iov_base, - iov[0].iov_len, - "Sending")); -#endif - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_poll (ACE_RMCast::Poll &, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_poll - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_POLL; - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1, to) == -1) - return -1; - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_ack_join (ACE_RMCast::Ack_Join &ack_join, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_ack_join - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_ACK_JOIN; - - ACE_UINT32 tmp = ACE_HTONL (ack_join.next_sequence_number); - ACE_OS::memcpy (header + 1, - &tmp, sizeof(ACE_UINT32)); - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1 + sizeof(ACE_UINT32), to) == -1) - return -1; - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_ack_leave (ACE_RMCast::Ack_Leave &, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_ack_leave - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_ACK_LEAVE; - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1, to) == -1) - return -1; - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_ack (ACE_RMCast::Ack &ack, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_ack - pushing (%d:%d) out to <%s:%d>\n", - // ack.next_expected, - // ack.highest_received, - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_ACK; - - ACE_UINT32 tmp = ACE_HTONL (ack.next_expected); - ACE_OS::memcpy (header + 1, - &tmp, sizeof(ACE_UINT32)); - tmp = ACE_HTONL (ack.highest_received); - ACE_OS::memcpy (header + 1 + sizeof(ACE_UINT32), - &tmp, sizeof(ACE_UINT32)); - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1 + 2*sizeof(ACE_UINT32), to) == -1) - return -1; - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_join (ACE_RMCast::Join &, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_join - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_JOIN; - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1, to) == -1) - return -1; - - return 0; -} - -int -ACE_RMCast_IO_UDP::send_leave (ACE_RMCast::Leave &, - const ACE_INET_Addr &to) -{ - //ACE_DEBUG ((LM_DEBUG, - // "IO_UDP::send_leave - pushing out to <%s:%d>\n", - // to.get_host_addr (), - // to.get_port_number ())); - - // @@ TODO: We could keep the header pre-initialized, and only - // update the portions that do change... - char header[16]; - header[0] = ACE_RMCast::MT_LEAVE; - - // ACE_SOCK_MCast_Dgram disallows sending, but it actually works. - ACE_SOCK_Dgram &dgram = this->dgram_; - - if (dgram.send (header, 1, to) == -1) - return -1; - - return 0; -} - - -#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION) - -template class ACE_Hash_Map_Manager<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Null_Mutex>; -template class ACE_Hash_Map_Manager_Ex<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Hash<ACE_INET_Addr>,ACE_Equal_To<ACE_INET_Addr>,ACE_Null_Mutex>; -template class ACE_Hash_Map_Iterator<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Null_Mutex>; -template class ACE_Hash_Map_Iterator_Ex<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Hash<ACE_INET_Addr>,ACE_Equal_To<ACE_INET_Addr>,ACE_Null_Mutex>; -template class ACE_Hash_Map_Reverse_Iterator_Ex<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Hash<ACE_INET_Addr>,ACE_Equal_To<ACE_INET_Addr>,ACE_Null_Mutex>; -template class ACE_Hash_Map_Iterator_Base_Ex<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*,ACE_Hash<ACE_INET_Addr>,ACE_Equal_To<ACE_INET_Addr>,ACE_Null_Mutex>; -template class ACE_Hash_Map_Entry<ACE_INET_Addr,ACE_RMCast_UDP_Proxy*>; -template class ACE_Equal_To<ACE_INET_Addr>; -template class ACE_Hash<ACE_INET_Addr>; - -#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */ |