TAO_ECG_CDR_Message_Receiver Class Reference

Receives UDP and Multicast messages. More...

#include <ECG_CDR_Message_Receiver.h>

Collaboration diagram for TAO_ECG_CDR_Message_Receiver:

[legend]List of all members.

Public Member Functions
int	handle_input (ACE_SOCK_Dgram &dgram, TAO_ECG_CDR_Processor *cdr_processor)
	TAO_ECG_CDR_Message_Receiver (CORBA::Boolean check_crc)
	~TAO_ECG_CDR_Message_Receiver (void)
void	init (TAO_ECG_Refcounted_Endpoint ignore_from)
void	shutdown (void)
Static Public Attributes
TAO_ECG_UDP_Request_Entry	Request_Completed_
Private Types
typedef ACE_Hash_Map_Manager< ACE_INET_Addr, Requests *, ACE_Null_Mutex >	Request_Map
enum	{ ECG_DEFAULT_MAX_FRAGMENTED_REQUESTS = 1024, ECG_DEFAULT_FRAGMENTED_REQUESTS_MIN_PURGE_COUNT = 32 }
Private Member Functions
int	mark_received (const ACE_INET_Addr &from, CORBA::ULong request_id)
int	process_fragment (const ACE_INET_Addr &from, const Mcast_Header &header, char *data_buf, TAO_ECG_CDR_Processor *cdr_processor)
Request_Map::ENTRY *	get_source_entry (const ACE_INET_Addr &from)
Private Attributes
TAO_ECG_Refcounted_Endpoint	ignore_from_
	Ignore any events coming from this IP address.
Request_Map	request_map_
size_t	max_requests_
	Serializes use of .
size_t	min_purge_count_
CORBA::Boolean	check_crc_
	Flag to indicate whether CRC should be computed and checked.

---

Detailed Description

Receives UDP and Multicast messages.

Todo:

Update class documentation below.

5) Make status array size and purge_count configurable.

This class receives UDP and Multicast message fragments, assembles them (described in detail below), and passes complete messages in the form of cdr streams to the calling classes.

This class is used by various Gateway classes (Senders/Receivers) responsible for federating Event Channels with UDP/Mcast.

= REASSEMBLY Fragmentation is described in ECG_CDR_Message_Sender.h Whenever an incomplete fragment is received (one with fragment_count > 1) we allocate an entry for the message in an map indexed by (host,port,request_id). The entry contains the buffer, a bit vector to keep track of the fragments received so far, and a timeout counter. This timeout counter is set to 0 on each (new) fragment arrival, and incremented on a regular basis. If the counter reaches a maximum value the message is dropped. Once all the fragments have been received the message is sent up to the calling classes, and the memory reclaimed.

Definition at line 144 of file ECG_CDR_Message_Receiver.h.

---

Member Typedef Documentation

typedef ACE_Hash_Map_Manager<ACE_INET_Addr, Requests*, ACE_Null_Mutex> TAO_ECG_CDR_Message_Receiver::Request_Map [private]

Definition at line 193 of file ECG_CDR_Message_Receiver.h.

---

Member Enumeration Documentation

anonymous enum [private]

Enumeration values: ECG_DEFAULT_MAX_FRAGMENTED_REQUESTS  ECG_DEFAULT_FRAGMENTED_REQUESTS_MIN_PURGE_COUNT  Definition at line 183 of file ECG_CDR_Message_Receiver.h. { ECG_DEFAULT_MAX_FRAGMENTED_REQUESTS = 1024, ECG_DEFAULT_FRAGMENTED_REQUESTS_MIN_PURGE_COUNT = 32 };

---

Constructor & Destructor Documentation

ACE_INLINE TAO_ECG_CDR_Message_Receiver::TAO_ECG_CDR_Message_Receiver (  CORBA::Boolean  check_crc  )

Definition at line 20 of file ECG_CDR_Message_Receiver.i. References check_crc_, ECG_DEFAULT_FRAGMENTED_REQUESTS_MIN_PURGE_COUNT, ECG_DEFAULT_MAX_FRAGMENTED_REQUESTS, ignore_from_, max_requests_, min_purge_count_, and request_map_. : ignore_from_ () , request_map_ () /* , lock_ (0) */ , max_requests_ (ECG_DEFAULT_MAX_FRAGMENTED_REQUESTS) , min_purge_count_ (ECG_DEFAULT_FRAGMENTED_REQUESTS_MIN_PURGE_COUNT) , check_crc_ (crc) { // ACE_NEW (this->lock_, // ACE_Lock_Adapter<ACE_Null_Mutex>); }

ACE_INLINE TAO_ECG_CDR_Message_Receiver::~TAO_ECG_CDR_Message_Receiver (  void   )

Definition at line 33 of file ECG_CDR_Message_Receiver.i. References shutdown(). { this->shutdown (); }

---

Member Function Documentation

TAO_ECG_CDR_Message_Receiver::Request_Map::ENTRY * TAO_ECG_CDR_Message_Receiver::get_source_entry (  const ACE_INET_Addr &  from  )  [private]

Definition at line 485 of file ECG_CDR_Message_Receiver.cpp. References ACE_ERROR_RETURN, ACE_NEW_RETURN, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_LOCK >::bind(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_LOCK >::find(), TAO_ECG_CDR_Message_Receiver::Requests::init(), LM_ERROR, ACE_Auto_Basic_Ptr< X >::release(), and request_map_. Referenced by mark_received(), and process_fragment(). { // Get the entry for <from> from the <request_map_>. Request_Map::ENTRY * entry; if (this->request_map_.find (from, entry) == -1) { // Create an entry if one doesn't exist. Requests *requests = 0; ACE_NEW_RETURN (requests, Requests, 0); auto_ptr<Requests> requests_aptr (requests); if (requests->init (this->max_requests_, this->min_purge_count_) == -1 || this->request_map_.bind (from, requests, entry) == -1) { ACE_ERROR_RETURN ((LM_ERROR, "Unable to create hash map " "entry for a new request.\n"), 0); } requests_aptr.release (); } return entry; }

int TAO_ECG_CDR_Message_Receiver::handle_input (  ACE_SOCK_Dgram &  dgram, TAO_ECG_CDR_Processor *  cdr_processor )

Returns 1 if data was read successfully and accepted by without errors. Returns 0 if there were no errors, but no data has been passed to , either due to request being incomplete (not all fragments received), or it being a duplicate. Returns -1 if there were errors. Definition at line 241 of file ECG_CDR_Message_Receiver.cpp. References ACE_ERROR, ACE_ERROR_RETURN, ACE_MAX_DGRAM_SIZE, ACE_ptr_align_binary, TAO_ECG_CDR_Message_Receiver::Mcast_Header::byte_order, check_crc_, TAO_ECG_CDR_Message_Receiver::Mcast_Header::crc, ACE::crc32(), TAO_ECG_CDR_Processor::decode(), EWOULDBLOCK, TAO_ECG_CDR_Message_Receiver::Mcast_Header::fragment_count, ACE_Refcounted_Auto_Ptr< X, ACE_LOCK >::get(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_host_name(), ACE_INET_Addr::get_port_number(), ignore_from_, iovec::iov_base, iovec::iov_len, LM_DEBUG, LM_ERROR, mark_received(), process_fragment(), TAO_ECG_CDR_Message_Receiver::Mcast_Header::read(), ACE_SOCK_Dgram::recv(), TAO_ECG_CDR_Message_Receiver::Mcast_Header::request_id, TAO_ECG_CDR_Message_Receiver::Mcast_Header::request_size, and ssize_t. Referenced by TAO_ECG_UDP_Receiver::handle_input(). { char nonaligned_header[TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE + ACE_CDR::MAX_ALIGNMENT]; char *header_buf = ACE_ptr_align_binary (nonaligned_header, ACE_CDR::MAX_ALIGNMENT); char nonaligned_data[ACE_MAX_DGRAM_SIZE + ACE_CDR::MAX_ALIGNMENT]; char *data_buf = ACE_ptr_align_binary (nonaligned_data, ACE_CDR::MAX_ALIGNMENT); // Read the message from dgram. const int iovcnt = 2; iovec iov[iovcnt]; iov[0].iov_base = header_buf; iov[0].iov_len = TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE; iov[1].iov_base = data_buf; iov[1].iov_len = ACE_MAX_DGRAM_SIZE; ACE_INET_Addr from; ssize_t n = dgram.recv (iov, iovcnt, from); if (n == -1) { if (errno == EWOULDBLOCK) return 0; ACE_ERROR_RETURN ((LM_ERROR, "Error reading mcast fragment (%m).\n"), -1); } if (n == 0) { ACE_ERROR_RETURN ((LM_ERROR, "Trying to read mcast fragment: " "read 0 bytes from socket.\n"), 0); } if (n < TAO_ECG_CDR_Message_Sender::ECG_HEADER_SIZE) { ACE_ERROR_RETURN ((LM_ERROR, "Trying to read mcast fragment: " "# of bytes read < mcast header size.\n"), -1); } u_int crc = 0; if (this->check_crc_) { iov[1].iov_len = n - iov[0].iov_len; iov[0].iov_len -= 4; // don't include crc crc = ACE::crc32 (iov, 2); } // Check whether the message is a loopback message. if (this->ignore_from_.get () != 0 && this->ignore_from_->is_loopback (from)) { return 0; } // Decode and validate mcast header. Mcast_Header header; if (header.read (header_buf, n, this->check_crc_) == -1) return -1; if ( this->check_crc_ && header.crc != crc) { static unsigned int err_count = 0; ACE_ERROR ((LM_DEBUG, "******************************\n")); ACE_ERROR ((LM_DEBUG, "ERROR DETECTED \n")); if (crc == 0) { ACE_ERROR ((LM_DEBUG, "Sending process may not have computed CRC \n")); } else { ACE_ERROR ((LM_DEBUG, " NETWORK CRC CHECKSUM FAILED\n")); } ACE_ERROR ((LM_ERROR, "Message was received from [%s:%s:%d] \n", from.get_host_name (), from.get_host_addr (), from.get_port_number())); ACE_ERROR ((LM_ERROR, "Num errors = %d \n", ++err_count)); ACE_ERROR ((LM_ERROR, "This is a bad thing. Attempting to ignore ..\n")); return 0; } // Process received data. if (header.fragment_count == 1) { // Update <request_map_> to mark this request as completed. (Not // needed if we don't care about duplicates.) int result = this->mark_received (from, header.request_id); if (result != 1) return result; TAO_InputCDR cdr (data_buf, header.request_size, header.byte_order); if (cdr_processor->decode (cdr) == -1) return -1; else return 1; } return this->process_fragment (from, header, data_buf, cdr_processor); }

ACE_INLINE void TAO_ECG_CDR_Message_Receiver::init (  TAO_ECG_Refcounted_Endpoint  ignore_from  )

Parameters: ignore_from  Endpoint used to remove events generated by the same process. Definition at line 39 of file ECG_CDR_Message_Receiver.i. References ignore_from_, and TAO_ECG_Refcounted_Endpoint. Referenced by TAO_ECG_UDP_Receiver::init(). { this->ignore_from_ = ignore_from; // if (lock) // { // delete this->lock_; // this->lock_ = lock; // } }

int TAO_ECG_CDR_Message_Receiver::mark_received (  const ACE_INET_Addr &  from, CORBA::ULong  request_id )  [private]

Returns 1 on success, 0 if has already been received or is below current request range, and -1 on error. Definition at line 365 of file ECG_CDR_Message_Receiver.cpp. References ACE_DEBUG, ACE_ERROR_RETURN, get_source_entry(), LM_ERROR, LM_INFO, LM_WARNING, and Request_Completed_. Referenced by handle_input(). { // ACE_GUARD_RETURN (ACE_Lock, guard, *this->lock_, -1); Request_Map::ENTRY * entry = this->get_source_entry (from); if (!entry) return -1; TAO_ECG_UDP_Request_Entry ** request = entry->int_id_->get_request (request_id); if (request == 0) { ACE_DEBUG ((LM_WARNING, "Received mcast request with sequence" "below currently expected range.\n")); return 0; } if (*request == &Request_Completed_) { ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment. " "(Request already complete).\n")); return 0; } if (*request != 0) { ACE_ERROR_RETURN ((LM_ERROR, "Inconsistent fragments for " "mcast request.\n"), -1); } *request = &Request_Completed_; return 1; }

int TAO_ECG_CDR_Message_Receiver::process_fragment (  const ACE_INET_Addr &  from, const Mcast_Header &  header, char *  data_buf, TAO_ECG_CDR_Processor *  cdr_processor )  [private]

Returns 1 if complete request is received and is populated, 0 if request has only partially been received or is a duplicate, and -1 on error. Definition at line 401 of file ECG_CDR_Message_Receiver.cpp. References ACE_DEBUG, ACE_ERROR_RETURN, ACE_NEW_RETURN, TAO_ECG_CDR_Message_Receiver::Mcast_Header::byte_order, TAO_ECG_CDR_Processor::decode(), TAO_ECG_CDR_Message_Receiver::Mcast_Header::fragment_count, TAO_ECG_CDR_Message_Receiver::Mcast_Header::fragment_id, TAO_ECG_CDR_Message_Receiver::Mcast_Header::fragment_offset, TAO_ECG_CDR_Message_Receiver::Mcast_Header::fragment_size, get_source_entry(), LM_ERROR, LM_INFO, LM_WARNING, ACE_OS::memcpy(), Request_Completed_, TAO_ECG_CDR_Message_Receiver::Mcast_Header::request_id, and TAO_ECG_CDR_Message_Receiver::Mcast_Header::request_size. Referenced by handle_input(). { // ACE_GUARD_RETURN (ACE_Lock, guard, *this->lock_, -1); Request_Map::ENTRY * source_entry = this->get_source_entry (from); if (!source_entry) return -1; TAO_ECG_UDP_Request_Entry ** request = source_entry->int_id_->get_request (header.request_id); if (request == 0) { ACE_DEBUG ((LM_WARNING, "Received mcast request with sequence " "below currently expected range.\n")); return 0; } if (*request == &Request_Completed_) { ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment. " "(Request already complete).\n")); return 0; } if (*request == 0) // Entry for this request has not yet been allocated. { ACE_NEW_RETURN (*request, TAO_ECG_UDP_Request_Entry (header.byte_order, header.request_id, header.request_size, header.fragment_count), -1); } // Validate the fragment. if ((*request)->validate_fragment (header.byte_order, header.request_size, header.fragment_size, header.fragment_offset, header.fragment_id, header.fragment_count) == 0) { ACE_ERROR_RETURN ((LM_ERROR, "Received invalid mcast fragment.\n"), -1); } // Check whether this fragment was already received. if ((*request)->test_received (header.fragment_id) == 1) { ACE_DEBUG ((LM_INFO, "Received duplicate mcast fragment.\n")); return 0; } // Add the fragment to the request entry. (*request)->mark_received (header.fragment_id); ACE_OS::memcpy ((*request)->fragment_buffer (header.fragment_offset), data_buf, header.fragment_size); // The request is not yet complete. if (!(*request)->complete ()) { return 0; } // The request is complete - decode it. TAO_InputCDR cdr ((*request)->fragment_buffer (0), header.request_size, header.byte_order); if (cdr_processor->decode (cdr) == -1) return -1; delete *request; *request = &Request_Completed_; return 1; }

void TAO_ECG_CDR_Message_Receiver::shutdown (  void   )

Definition at line 513 of file ECG_CDR_Message_Receiver.cpp. References ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_LOCK >::begin(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, ACE_Hash< EXT_ID >, ACE_Equal_To< EXT_ID >, ACE_LOCK >::end(), ignore_from_, request_map_, and ACE_Refcounted_Auto_Ptr< X, ACE_LOCK >::reset(). Referenced by TAO_ECG_UDP_Receiver::shutdown(), and ~TAO_ECG_CDR_Message_Receiver(). { // ACE_GUARD (ACE_Lock, guard, *this->lock_); Request_Map::iterator end = this->request_map_.end (); for (Request_Map::iterator i = this->request_map_.begin (); i != end; ++i) { delete (*i).int_id_; (*i).int_id_ = 0; } this->ignore_from_.reset (); }

---

Member Data Documentation

CORBA::Boolean TAO_ECG_CDR_Message_Receiver::check_crc_ [private]

Flag to indicate whether CRC should be computed and checked. Definition at line 235 of file ECG_CDR_Message_Receiver.h. Referenced by handle_input(), and TAO_ECG_CDR_Message_Receiver().

TAO_ECG_Refcounted_Endpoint TAO_ECG_CDR_Message_Receiver::ignore_from_ [private]

Ignore any events coming from this IP address. Definition at line 216 of file ECG_CDR_Message_Receiver.h. Referenced by handle_input(), init(), shutdown(), and TAO_ECG_CDR_Message_Receiver().

size_t TAO_ECG_CDR_Message_Receiver::max_requests_ [private]

Serializes use of . Size of a fragmented requests array, i.e., max number of partially received requests kept at any given time per source. Definition at line 227 of file ECG_CDR_Message_Receiver.h. Referenced by TAO_ECG_CDR_Message_Receiver().

size_t TAO_ECG_CDR_Message_Receiver::min_purge_count_ [private]

Minimum number of requests purged from a fragmented requests array when the range of requests represented there needs to be shifted. Definition at line 232 of file ECG_CDR_Message_Receiver.h. Referenced by TAO_ECG_CDR_Message_Receiver().

TAO_ECG_UDP_Request_Entry TAO_ECG_CDR_Message_Receiver::Request_Completed_ [static]

Represents any request that has been fully received and serviced, to simplify the internal logic. Referenced by mark_received(), process_fragment(), TAO_ECG_CDR_Message_Receiver::Requests::purge_requests(), and TAO_ECG_CDR_Message_Receiver::Requests::~Requests().

Request_Map TAO_ECG_CDR_Message_Receiver::request_map_ [private]

The map containing all the incoming requests which have been partially received. Definition at line 220 of file ECG_CDR_Message_Receiver.h. Referenced by get_source_entry(), shutdown(), and TAO_ECG_CDR_Message_Receiver().

---

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

• ECG_CDR_Message_Receiver.h
• ECG_CDR_Message_Receiver.cpp
• ECG_CDR_Message_Receiver.i

---