OpenDDS::DCPS::TcpTransport Class Reference

#include <TcpTransport.h>

Inheritance diagram for OpenDDS::DCPS::TcpTransport:

[legend]Collaboration diagram for OpenDDS::DCPS::TcpTransport:

[legend]List of all members.

Public Member Functions
	TcpTransport (const TransportInst_rch &inst)
virtual	~TcpTransport ()
TcpInst *	get_configuration ()
int	fresh_link (TcpConnection_rch connection)
virtual void	unbind_link (DataLink *link)
	Remove any pending_release mappings.
Private Types
typedef ACE_Hash_Map_Manager_Ex< PriorityKey, TcpDataLink_rch, ACE_Hash< PriorityKey >, ACE_Equal_To< PriorityKey >, ACE_Null_Mutex >	AddrLinkMap
	Map Type: (key) PriorityKey to (value) TcpDataLink_rch.
typedef ACE_SYNCH_MUTEX	LockType
typedef ACE_Guard< LockType >	GuardType
typedef ACE_Condition< LockType >	ConditionType
Private Member Functions
virtual AcceptConnectResult	connect_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, TransportClient *client)
virtual AcceptConnectResult	accept_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, TransportClient *client)
virtual void	stop_accepting_or_connecting (TransportClient *client, const RepoId &remote_id)
virtual bool	configure_i (TransportInst *config)
virtual void	shutdown_i ()
virtual void	pre_shutdown_i ()
virtual bool	connection_info_i (TransportLocator &local_info) const
virtual void	release_datalink (DataLink *link)
	Called by the DataLink to release itself.
virtual std::string	transport_type () const
void	async_connect_failed (const PriorityKey &key)
void	passive_connection (const ACE_INET_Addr &remote_address, const TcpConnection_rch &connection)
bool	find_datalink_i (const PriorityKey &key, TcpDataLink_rch &link, TransportClient *client, const RepoId &remote_id)
int	connect_tcp_datalink (const TcpDataLink_rch &link, const TcpConnection_rch &connection)
	Common code used by accept_datalink(), passive_connection(), and active completion.
PriorityKey	blob_to_key (const TransportBLOB &remote, Priority priority, bool active)
typedef	OPENDDS_MAP (PriorityKey, TcpDataLink_rch) LinkMap
typedef	OPENDDS_MAP (PriorityKey, TcpConnection_rch) ConnectionMap
Private Attributes
TcpAcceptor *	acceptor_
	Used to accept passive connections on our local_address_.
ACE_Connector< TcpConnection, ACE_SOCK_Connector >	connector_
	Open TcpConnections using non-blocking connect.
TcpInst_rch	tcp_config_
	Our configuration object, supplied to us in config_i().
AddrLinkMap	links_
	This is the map of connected DataLinks.
AddrLinkMap	pending_release_links_
LockType	links_lock_
	This lock is used to protect the links_ data member.
ConnectionMap	connections_
LockType	connections_lock_
TransportReactorTask_rch	reactor_task_
	We need the reactor for our Acceptor.
TcpConnectionReplaceTask *	con_checker_
Friends
class	TcpConnection
class	TcpDataLink

---

Detailed Description

This class provides the "Tcp" transport specific implementation. It creates the acceptor for listening the incoming requests using TCP and maintains a collection of TCP specific connections/datalinks.

Notes about object ownership: 1) Own the datalink objects, passive connection objects, acceptor object and TcpConnectionReplaceTask object(used during reconnecting). 2) Reference to TransportReactorTask object owned by base class.

Definition at line 44 of file TcpTransport.h.

---

Member Typedef Documentation

typedef ACE_Hash_Map_Manager_Ex<PriorityKey, TcpDataLink_rch, ACE_Hash<PriorityKey>, ACE_Equal_To<PriorityKey>, ACE_Null_Mutex> OpenDDS::DCPS::TcpTransport::AddrLinkMap [private]

Map Type: (key) PriorityKey to (value) TcpDataLink_rch.

Definition at line 115 of file TcpTransport.h.

typedef ACE_Condition<LockType> OpenDDS::DCPS::TcpTransport::ConditionType [private]

Definition at line 122 of file TcpTransport.h.

typedef ACE_Guard<LockType> OpenDDS::DCPS::TcpTransport::GuardType [private]

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 121 of file TcpTransport.h.

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::TcpTransport::LockType [private]

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 120 of file TcpTransport.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::TcpTransport::TcpTransport

(

const TransportInst_rch &

inst

)

[explicit]

Definition at line 32 of file TcpTransport.cpp.

References acceptor_, con_checker_, OpenDDS::DCPS::TransportImpl::configure(), DBG_ENTRY_LVL, OpenDDS::DCPS::RcHandle< T >::in(), and OpenDDS::DCPS::RcHandle< T >::is_nil().

  : acceptor_(new TcpAcceptor(this)),
    con_checker_(new TcpConnectionReplaceTask(this))
{
  DBG_ENTRY_LVL("TcpTransport","TcpTransport",6);

  if (!inst.is_nil()) {
    if (!configure(inst.in())) {
      delete con_checker_;
      delete acceptor_;
      throw Transport::UnableToCreate();
    }
  }
}

OpenDDS::DCPS::TcpTransport::~TcpTransport

(

)

[virtual]

Definition at line 47 of file TcpTransport.cpp.

References acceptor_, OpenDDS::DCPS::QueueTaskBase< T >::close(), con_checker_, and DBG_ENTRY_LVL.

{
  DBG_ENTRY_LVL("TcpTransport","~TcpTransport",6);
  delete acceptor_;

  con_checker_->close(1);  // This could potentially fix a race condition
  delete con_checker_;
}

---

Member Function Documentation

TransportImpl::AcceptConnectResult OpenDDS::DCPS::TcpTransport::accept_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		TransportClient *	client
	)			[private, virtual]

Definition at line 231 of file TcpTransport.cpp.

References OpenDDS::DCPS::RcHandle< T >::_retn(), OpenDDS::DCPS::TransportImpl::add_pending_connection(), OpenDDS::DCPS::PriorityKey::address(), blob_to_key(), connect_tcp_datalink(), connections_, connections_lock_, find_datalink_i(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportClient::repo_id_, TcpDataLink, and VDBG_LVL.

{
  GuidConverter remote_conv(remote.repo_id_);
  GuidConverter local_conv(attribs.local_id_);

  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink local %C "
            "accepting connection from remote %C\n",
            std::string(local_conv).c_str(),
            std::string(remote_conv).c_str()), 5);

  GuardType guard(connections_lock_);
  const PriorityKey key =
    blob_to_key(remote.blob_, attribs.priority_, false /* !active */);

  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink PriorityKey "
            "prio=%d, addr=%C:%hu, is_loopback=%d, is_active=%d\n", attribs.priority_,
            key.address().get_host_addr(), key.address().get_port_number(),
            key.is_loopback(), key.is_active()), 2);

  TcpDataLink_rch link;
  {
    GuardType guard(links_lock_);

    if (find_datalink_i(key, link, client, remote.repo_id_)) {
      return link.is_nil()
        ? AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS)
        : AcceptConnectResult(link._retn());

    } else {
      link = new TcpDataLink(key.address(), this, key.priority(),
                             key.is_loopback(), key.is_active());

      if (links_.bind(key, link) != 0 /*OK*/) {
        ACE_ERROR((LM_ERROR,
                   "(%P|%t) ERROR: TcpTransport::accept_datalink "
                   "Unable to bind new TcpDataLink to "
                   "TcpTransport in links_ map.\n"));
        return AcceptConnectResult();
      }
    }
  }

  TcpConnection_rch connection;
  const ConnectionMap::iterator iter = connections_.find(key);

  if (iter != connections_.end()) {
    connection = iter->second;
    connections_.erase(iter);
  }

  if (connection.is_nil()) {
    if (!link->add_on_start_callback(client, remote.repo_id_)) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink "
                "got started link %@.\n", link.in()), 0);
      return AcceptConnectResult(link._retn());
    }

    VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink "
              "no existing TcpConnection.\n"), 0);

    add_pending_connection(client, link.in());

    // no link ready, passive_connection will complete later
    return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
  }

  guard.release(); // connect_tcp_datalink() isn't called with connections_lock_

  if (connect_tcp_datalink(link, connection) == -1) {
    GuardType guard(links_lock_);
    links_.unbind(key);
    link = 0;
  }

  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::accept_datalink "
            "connected link %@.\n", link.in()), 2);
  return AcceptConnectResult(link._retn());
}

void OpenDDS::DCPS::TcpTransport::async_connect_failed

(

const PriorityKey &

key

)

[private]

Definition at line 169 of file TcpTransport.cpp.

References OpenDDS::DCPS::RcHandle< T >::in(), links_, and links_lock_.

{

  ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: Failed to make active connection.\n"));
  GuardType guard(links_lock_);
  TcpDataLink_rch link;
  links_.find(key, link);
  links_.unbind(key);
  guard.release();

  if (link.in()) {
    link->invoke_on_start_callbacks(false);
  }

}

PriorityKey OpenDDS::DCPS::TcpTransport::blob_to_key	(	const TransportBLOB &	remote,
		Priority	priority,
		bool	active
	)			[private]

Definition at line 57 of file TcpTransport.cpp.

References OpenDDS::DCPS::AssociationData::get_remote_address(), and tcp_config_.

Referenced by accept_datalink(), and connect_datalink().

{
  const ACE_INET_Addr remote_address =
    AssociationData::get_remote_address(remote);
  const bool is_loopback = remote_address == tcp_config_->local_address();
  return PriorityKey(priority, remote_address, is_loopback, active);
}

bool OpenDDS::DCPS::TcpTransport::configure_i

(

TransportInst *

config

)

[private, virtual]

Concrete subclass gets a shot at the config object. The subclass will likely downcast the TransportInst object to a subclass type that it expects/requires.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 334 of file TcpTransport.cpp.

References OpenDDS::DCPS::RcObject< T >::_add_ref(), OpenDDS::DCPS::RcHandle< T >::_retn(), OpenDDS::DCPS::TransportImpl::config(), connector_, OpenDDS::DCPS::TransportImpl::create_reactor_task(), DBG_ENTRY_LVL, OpenDDS::DCPS::get_fully_qualified_hostname(), OpenDDS::DCPS::TcpInst::local_address(), OPENDDS_STRING, OpenDDS::DCPS::TransportImpl::reactor_task(), reactor_task_, tcp_config_, TheServiceParticipant, and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpTransport", "configure_i", 6);

  // Downcast the config argument to a TcpInst*
  TcpInst* tcp_config =
    static_cast<TcpInst*>(config);

  if (tcp_config == 0) {
    // The downcast failed.
    ACE_ERROR_RETURN((LM_ERROR,
                      "(%P|%t) ERROR: Failed downcast from TransportInst "
                      "to TcpInst.\n"),
                     false);
  }

  this->create_reactor_task();

  // Ask our base class for a "copy" of the reference to the reactor task.
  this->reactor_task_ = reactor_task();

  connector_.open(reactor_task_->get_reactor());

  // Make a "copy" of the reference for ourselves.
  tcp_config->_add_ref();
  this->tcp_config_ = tcp_config;

  // Open the reconnect task
  if (this->con_checker_->open()) {
    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: connection checker failed to open : %p\n"),
                      ACE_TEXT("open")),
                     false);
  }

  // Override with DCPSDefaultAddress.
  if (this->tcp_config_->local_address() == ACE_INET_Addr () &&
      !TheServiceParticipant->default_address ().empty ()) {
    this->tcp_config_->local_address(0, TheServiceParticipant->default_address ().c_str ());
  }

  // Open our acceptor object so that we can accept passive connections
  // on our this->tcp_config_->local_address_.

  if (this->acceptor_->open(this->tcp_config_->local_address(),
                            this->reactor_task_->get_reactor()) != 0) {
    // Remember to drop our reference to the tcp_config_ object since
    // we are about to return -1 here, which means we are supposed to
    // keep a copy after all.
    TcpInst_rch cfg = this->tcp_config_._retn();

    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: Acceptor failed to open %C:%d: %p\n"),
                      cfg->local_address().get_host_addr(),
                      cfg->local_address().get_port_number(),
                      ACE_TEXT("open")),
                     false);
  }

  // update the port number (incase port zero was given).
  ACE_INET_Addr address;

  if (this->acceptor_->acceptor().get_local_addr(address) != 0) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: TcpTransport::configure_i ")
               ACE_TEXT("- %p"),
               ACE_TEXT("cannot get local addr\n")));
  }

  OPENDDS_STRING listening_addr(address.get_host_addr());
  VDBG_LVL((LM_DEBUG,
            ACE_TEXT("(%P|%t) TcpTransport::configure_i listening on %C:%hu\n"),
            listening_addr.c_str(), address.get_port_number()), 2);

  unsigned short port = address.get_port_number();

  // As default, the acceptor will be listening on INADDR_ANY but advertise with the fully
  // qualified hostname and actual listening port number.
  if (tcp_config_->local_address().is_any()) {
    std::string hostname = get_fully_qualified_hostname();

    this->tcp_config_->local_address(port, hostname.c_str());
  }

  // Now we got the actual listening port. Update the port number in the configuration
  // if it's 0 originally.
  else if (tcp_config_->local_address().get_port_number() == 0) {
    tcp_config_->local_address_set_port(port);
  }

  // Ahhh...  The sweet smell of success!
  return true;
}

TransportImpl::AcceptConnectResult OpenDDS::DCPS::TcpTransport::connect_datalink	(	const RemoteTransport &	remote,
		const ConnectionAttribs &	attribs,
		TransportClient *	client
	)			[private, virtual]

Definition at line 68 of file TcpTransport.cpp.

References OpenDDS::DCPS::RcHandle< T >::_retn(), OpenDDS::DCPS::TransportImpl::add_pending_connection(), OpenDDS::DCPS::PriorityKey::address(), blob_to_key(), connections_lock_, connector_, DBG_ENTRY_LVL, find_datalink_i(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportClient::repo_id_, tcp_config_, TcpConnection, TcpDataLink, and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpTransport", "connect_datalink", 6);

  const PriorityKey key =
    blob_to_key(remote.blob_, attribs.priority_, true /*active*/);

  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink PriorityKey "
            "prio=%d, addr=%C:%hu, is_loopback=%d, is_active=%d\n",
            key.priority(), key.address().get_host_addr(),
            key.address().get_port_number(), key.is_loopback(),
            key.is_active()), 0);

  TcpDataLink_rch link;
  {
    GuardType guard(links_lock_);

    if (find_datalink_i(key, link, client, remote.repo_id_)) {
      VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink found datalink link[%@]\n", link.in()), 0);
      return link.is_nil()
        ? AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS)
        : AcceptConnectResult(link._retn());
    }

    link = new TcpDataLink(key.address(), this, attribs.priority_,
                           key.is_loopback(), true /*active*/);
    VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink create new link[%@]\n", link.in()), 0);
    if (links_.bind(key, link) != 0 /*OK*/) {
      ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TcpTransport::connect_datalink "
                 "Unable to bind new TcpDataLink[%@] to "
                 "TcpTransport in links_ map.\n", link.in()));
      return AcceptConnectResult();
    }
  }

  TcpConnection_rch connection =
    new TcpConnection(key.address(), link->transport_priority(), tcp_config_);
  connection->set_datalink(link.in());

  TcpConnection* pConn = connection.in();
  TcpConnection_rch reactor_refcount(connection); // increment for reactor callback

  ACE_TCHAR str[64];
  key.address().addr_to_string(str,sizeof(str)/sizeof(str[0]));

  // Can't make this call while holding onto TransportClient::lock_
  const int ret =
    connector_.connect(pConn, key.address(), ACE_Synch_Options::asynch);

  if (ret == -1 && errno != EWOULDBLOCK) {

    VDBG_LVL((LM_ERROR, "(%P|%t) TcpTransport::connect_datalink error %m.\n"), 2);
    ACE_DEBUG((LM_ERROR, "(%P|%t) TcpTransport::connect_datalink error %m.\n"));
    //If the connection fails and, in the interim between releasing
    //lock and re-acquiring to remove the failed link, another association may have found
    //the datalink in links_ (always using find_datalink_i) so must allow the other
    //association to either try to connect again (might succeed for it)
    //or try another transport.  If find_datalink_i was called for this datalink, an
    //on_start_callback will be registered and can be invoked.
    VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink connect failed, remove link[%@]\n", link.in()), 0);
    {
      GuardType guard(links_lock_);
      if (links_.unbind(key, link) != 0 /*OK*/) {
        ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: TcpTransport::connect_datalink "
                   "Unable to unbind failed TcpDataLink[%@] from "
                   "TcpTransport links_ map.\n", link.in()));
      }
    }
    link->invoke_on_start_callbacks(false);

    return AcceptConnectResult();
  }

  // Don't decrement count when reactor_refcount goes out of scope, see
  // TcpConnection::open()
  (void) reactor_refcount._retn();

  if (ret == 0) {
    // connect() completed synchronously and called TcpConnection::active_open().
    VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink "
              "completed synchronously.\n"), 0);
    return AcceptConnectResult(link._retn());
  }

  if (!link->add_on_start_callback(client, remote.repo_id_)) {
    // link was started by the reactor thread before we could add a callback

    VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink got link.\n"), 0);
    return AcceptConnectResult(link._retn());
  }

  GuardType connections_guard(connections_lock_);

  add_pending_connection(client, link.in());
  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::connect_datalink pending.\n"), 0);
  return AcceptConnectResult(AcceptConnectResult::ACR_SUCCESS);
}

int OpenDDS::DCPS::TcpTransport::connect_tcp_datalink	(	const TcpDataLink_rch &	link,
		const TcpConnection_rch &	connection
	)			[private]

Common code used by accept_datalink(), passive_connection(), and active completion.

Code common to make_active_connection() and make_passive_connection().

Definition at line 695 of file TcpTransport.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, and OpenDDS::DCPS::TransportImpl::last_link_.

Referenced by accept_datalink(), and passive_connection().

{
  DBG_ENTRY_LVL("TcpTransport", "connect_tcp_datalink", 6);

  if (link->reuse_existing_connection(connection) == 0) {
    return 0;
  }

  ++last_link_;

  if (DCPS_debug_level > 4) {
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) TcpTransport::connect_tcp_datalink() [%d] - ")
               ACE_TEXT("creating send strategy with priority %d.\n"),
               last_link_, link->transport_priority()));
  }

  connection->id() = last_link_;

  TransportSendStrategy_rch send_strategy =
    new TcpSendStrategy(last_link_, link, this->tcp_config_, connection,
                        new TcpSynchResource(connection,
                                             this->tcp_config_->max_output_pause_period_),
                        this->reactor_task_, link->transport_priority());

  TransportStrategy_rch receive_strategy =
    new TcpReceiveStrategy(link, connection, this->reactor_task_);

  if (link->connect(connection, send_strategy, receive_strategy) != 0) {
    return -1;
  }

  return 0;
}

bool OpenDDS::DCPS::TcpTransport::connection_info_i

(

TransportLocator &

local_info

)

const [private, virtual]

Called by our connection_info() method to allow the concrete TransportImpl subclass to do the dirty work since it really is the one that knows how to populate the supplied TransportLocator object.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 497 of file TcpTransport.cpp.

References DBG_ENTRY_LVL, tcp_config_, and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpTransport", "connection_info_i", 6);

  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport public address str %C\n",
            this->tcp_config_->get_public_address().c_str()), 2);

  this->tcp_config_->populate_locator(local_info);

  return true;
}

bool OpenDDS::DCPS::TcpTransport::find_datalink_i	(	const PriorityKey &	key,
		TcpDataLink_rch &	link,
		TransportClient *	client,
		const RepoId &	remote_id
	)			[private]

Definition at line 187 of file TcpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::add_pending_connection(), DBG_ENTRY_LVL, OpenDDS::DCPS::RcHandle< T >::in(), links_, pending_release_links_, and VDBG_LVL.

Referenced by accept_datalink(), and connect_datalink().

{
  DBG_ENTRY_LVL("TcpTransport", "find_datalink_i", 6);

  if (links_.find(key, link) == 0 /*OK*/) {
    if (!link->add_on_start_callback(client, remote_id)) {
      VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                ACE_TEXT("link[%@] found, already started.\n"), link.in()), 0);
      // Since the link was already started, we won't get an "on start"
      // callback, and the link is immediately usable.
      return true;
    }

    VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
              ACE_TEXT("link[%@] found, add to pending connections.\n"), link.in()), 0);
    add_pending_connection(client, link.in());
    link = 0; // don't return link to TransportClient
    return true;

  } else if (pending_release_links_.find(key, link) == 0 /*OK*/) {
    if (link->cancel_release()) {
      link->set_release_pending(false);

      if (pending_release_links_.unbind(key, link) == 0 /*OK*/
          && links_.bind(key, link) == 0 /*OK*/) {
        VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                  ACE_TEXT("found link[%@] in pending release list, cancelled release and moved back to links_.\n"), link.in()), 0);
        return true;
      }
      VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                ACE_TEXT("found link[%@] in pending release list but was unable to shift back to links_.\n"), link.in()), 0);
    } else {
      VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::find_datalink_i ")
                ACE_TEXT("found link[%@] in pending release list but was unable to cancel release.\n"), link.in()), 0);
    }
    link = 0; // don't return link to TransportClient
    return false;
  }

  return false;
}

int OpenDDS::DCPS::TcpTransport::fresh_link

(

TcpConnection_rch

connection

)

This function is called by the TcpReconnectTask thread to check if the passively accepted connection is the re-established connection. If it is, then the "old" connection object in the datalink is replaced by the "new" connection object.

Definition at line 735 of file TcpTransport.cpp.

References DBG_ENTRY_LVL, OpenDDS::DCPS::RcHandle< T >::in(), and tcp_config_.

Referenced by OpenDDS::DCPS::TcpConnectionReplaceTask::execute().

{
  DBG_ENTRY_LVL("TcpTransport","fresh_link",6);

  TcpDataLink_rch link;
  GuardType guard(this->links_lock_);

  PriorityKey key(connection->transport_priority(),
                  connection->get_remote_address(),
                  connection->get_remote_address() == this->tcp_config_->local_address(),
                  connection->is_connector());

  if (this->links_.find(key, link) == 0) {
    TcpConnection_rch old_con = link->get_connection();

    // The connection is accepted but may not be associated with the datalink
    // at this point. The thread calling add_associations() will associate
    // the datalink with the connection in make_passive_connection().
    if (old_con.is_nil()) {
      return 0;
    }

    if (old_con.in() != connection.in())
      // Replace the "old" connection object with the "new" connection object.
    {
      return link->reconnect(connection.in());
    }
  }

  return 0;
}

TcpInst * OpenDDS::DCPS::TcpTransport::get_configuration

(

)

Definition at line 622 of file TcpTransport.cpp.

References OpenDDS::DCPS::RcHandle< T >::in(), and tcp_config_.

{
  return this->tcp_config_.in();
}

typedef OpenDDS::DCPS::TcpTransport::OPENDDS_MAP	(	PriorityKey	,
		TcpConnection_rch
	)			[private]

typedef OpenDDS::DCPS::TcpTransport::OPENDDS_MAP	(	PriorityKey	,
		TcpDataLink_rch
	)			[private]

void OpenDDS::DCPS::TcpTransport::passive_connection	(	const ACE_INET_Addr &	remote_address,
		const TcpConnection_rch &	connection
	)			[private]

Called by the TcpConnection object when it has been created by the acceptor and needs to be attached to a DataLink. The DataLink may or may not already be created and waiting for this passive connection to appear.

Definition at line 633 of file TcpTransport.cpp.

References OpenDDS::DCPS::QueueTaskBase< T >::add(), con_checker_, connect_tcp_datalink(), connections_, connections_lock_, DBG_ENTRY_LVL, OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, tcp_config_, and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpTransport", "passive_connection", 6);

  const PriorityKey key(connection->transport_priority(),
                        remote_address,
                        remote_address == tcp_config_->local_address(),
                        connection->is_connector());

  VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) TcpTransport::passive_connection() - ")
            ACE_TEXT("established with %C:%d.\n"),
            remote_address.get_host_name(),
            remote_address.get_port_number()), 2);

  GuardType connection_guard(connections_lock_);
  TcpDataLink_rch link;
  {
    GuardType guard(links_lock_);
    links_.find(key, link);
  }

  if (!link.is_nil()) {
    connection_guard.release();

    if (connect_tcp_datalink(link, connection) == -1) {
      VDBG_LVL((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: connect_tcp_datalink failed\n")), 5);
      GuardType guard(links_lock_);
      links_.unbind(key);

    } else {
      con_checker_->add(connection);
    }

    return;
  }

  // If we reach this point, this link was not in links_, so the
  // accept_datalink() call hasn't happened yet.  Store in connections_ for the
  // accept_datalink() method to find.
  VDBG_LVL((LM_DEBUG, "(%P|%t) # of bef connections: %d\n", connections_.size()), 5);
  const ConnectionMap::iterator where = connections_.find(key);

  if (where != connections_.end()) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: TcpTransport::passive_connection() - ")
               ACE_TEXT("connection with %C:%d at priority %d already exists, ")
               ACE_TEXT("overwriting previously established connection.\n"),
               remote_address.get_host_name(),
               remote_address.get_port_number(),
               connection->transport_priority()));
  }

  connections_[key] = connection;
  VDBG_LVL((LM_DEBUG, "(%P|%t) # of after connections: %d\n", connections_.size()), 5);

  con_checker_->add(connection);
}

void OpenDDS::DCPS::TcpTransport::pre_shutdown_i

(

)

[private, virtual]

Called before transport is shutdown to let the concrete transport to do anything necessary.

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 429 of file TcpTransport.cpp.

References DBG_ENTRY_LVL.

{
  DBG_ENTRY_LVL("TcpTransport","pre_shutdown_i",6);

  GuardType guard(this->links_lock_);

  AddrLinkMap::ENTRY* entry;

  for (AddrLinkMap::ITERATOR itr(this->links_);
       itr.next(entry);
       itr.advance()) {
    entry->int_id_->pre_stop_i();
  }
}

void OpenDDS::DCPS::TcpTransport::release_datalink

(

DataLink *

link

)

[private, virtual]

Called by the DataLink to release itself.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 510 of file TcpTransport.cpp.

References OpenDDS::DCPS::DataLink::datalink_release_delay(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::schedule_delayed_release(), OpenDDS::DCPS::DataLink::schedule_stop(), OpenDDS::DCPS::DataLink::set_scheduling_release(), OpenDDS::DCPS::DataLink::transport_priority(), and VDBG_LVL.

{
  DBG_ENTRY_LVL("TcpTransport", "release_datalink", 6);

  TcpDataLink* tcp_link = static_cast<TcpDataLink*>(link);

  if (tcp_link == 0) {
    // Really an assertion failure
    ACE_ERROR((LM_ERROR,
               "(%P|%t) INTERNAL ERROR - Failed to downcast DataLink to "
               "TcpDataLink.\n"));
    return;
  }

  TcpDataLink_rch released_link;

  // Possible actions that will be taken to release the link.
  enum LinkAction { None, StopLink, ScheduleLinkRelease };
  LinkAction linkAction = None;

  // Scope for locking to protect the links (and pending_release) containers.
  GuardType guard(this->links_lock_);

  // Attempt to remove the TcpDataLink from our links_ map.
  PriorityKey key(
    tcp_link->transport_priority(),
    tcp_link->remote_address(),
    tcp_link->is_loopback(),
    tcp_link->is_active());

  VDBG_LVL((LM_DEBUG,
            "(%P|%t) TcpTransport::release_datalink link[%@] PriorityKey "
            "prio=%d, addr=%C:%hu, is_loopback=%d, is_active=%d\n",
            link,
            tcp_link->transport_priority(),
            tcp_link->remote_address().get_host_addr(),
            tcp_link->remote_address().get_port_number(),
            (int)tcp_link->is_loopback(),
            (int)tcp_link->is_active()), 2);

  if (this->links_.unbind(key, released_link) != 0) {
    //No op
  } else if (link->datalink_release_delay() > ACE_Time_Value::zero) {
    link->set_scheduling_release(true);

    VDBG_LVL((LM_DEBUG,
              "(%P|%t) TcpTransport::release_datalink datalink_release_delay "
              "is %: sec %d usec\n",
              link->datalink_release_delay().sec(),
              link->datalink_release_delay().usec()), 4);

    // Atomic value update, safe to perform here.
    released_link->set_release_pending(true);

    switch (this->pending_release_links_.bind(key, released_link)) {
    case -1:
      ACE_ERROR((LM_ERROR,
                 "(%P|%t) ERROR: Unable to bind released TcpDataLink[%@] to "
                 "pending_release_links_ map: %p\n", released_link.in(), ACE_TEXT("bind")));
      linkAction = StopLink;
      break;

    case 1:
      ACE_ERROR((LM_ERROR,
                 "(%P|%t) ERROR: Unable to bind released TcpDataLink[%@] to "
                 "pending_release_links_ map: already bound\n", released_link.in()));
      linkAction = StopLink;
      break;

    case 0:
      linkAction = ScheduleLinkRelease;
      break;

    default:
      break;
    }

  } else { // datalink_release_delay_ is 0
    link->set_scheduling_release(true);

    linkAction = StopLink;
  }

  // Actions are executed outside of the lock scope.
  ACE_Time_Value cancel_now = ACE_OS::gettimeofday();
  switch (linkAction) {
  case StopLink:
    link->schedule_stop(cancel_now);
    break;

  case ScheduleLinkRelease:

    link->schedule_delayed_release();
    break;

  case None:
    break;
  }

  if (DCPS_debug_level > 9) {
    std::stringstream buffer;
    buffer << *link;
    ACE_DEBUG((LM_DEBUG,
               ACE_TEXT("(%P|%t) TcpTransport::release_datalink() - ")
               ACE_TEXT("link[%@] with priority %d released.\n%C"),
               link,
               link->transport_priority(),
               buffer.str().c_str()));
  }
}

void OpenDDS::DCPS::TcpTransport::shutdown_i

(

)

[private, virtual]

Called during the shutdown() method in order to give the concrete TransportImpl subclass a chance to do something when the shutdown "event" occurs.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 445 of file TcpTransport.cpp.

References acceptor_, OpenDDS::DCPS::QueueTaskBase< T >::close(), con_checker_, connections_, DBG_ENTRY_LVL, links_, pending_release_links_, reactor_task_, tcp_config_, and OpenDDS::DCPS::TcpAcceptor::transport_shutdown().

{
  DBG_ENTRY_LVL("TcpTransport","shutdown_i",6);

  // Don't accept any more connections.
  this->acceptor_->close();
  this->acceptor_->transport_shutdown();

  this->con_checker_->close(1);

  {
    GuardType guard(this->connections_lock_);

    this->connections_.clear();
    this->pending_connections_.clear();
  }

  // Disconnect all of our DataLinks, and clear our links_ collection.
  {
    GuardType guard(this->links_lock_);

    AddrLinkMap::ENTRY* entry;

    for (AddrLinkMap::ITERATOR itr(this->links_);
         itr.next(entry);
         itr.advance()) {
      entry->int_id_->transport_shutdown();
    }

    this->links_.unbind_all();

    for (AddrLinkMap::ITERATOR itr(this->pending_release_links_);
         itr.next(entry);
         itr.advance()) {
      entry->int_id_->transport_shutdown();
    }

    this->pending_release_links_.unbind_all();
  }

  // Drop our reference to the TcpInst object.
  this->tcp_config_ = 0;

  // Drop our reference to the TransportReactorTask
  this->reactor_task_ = 0;

  // Tell our acceptor about this event so that it can drop its reference
  // it holds to this TcpTransport object (via smart-pointer).
  this->acceptor_->transport_shutdown();
}

void OpenDDS::DCPS::TcpTransport::stop_accepting_or_connecting	(	TransportClient *	client,
		const RepoId &	remote_id
	)			[private, virtual]

stop_accepting_or_connecting() is called from TransportClient to terminate the accepting process begun by accept_datalink() or connect_datalink(). This allows the TransportImpl to clean up any resources associated with this pending connection. The TransportClient* passed in to accept or connect is not valid after this method is called.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 313 of file TcpTransport.cpp.

References connections_lock_, and VDBG_LVL.

{
  GuidConverter remote_converted(remote_id);
  VDBG_LVL((LM_DEBUG, "(%P|%t) TcpTransport::stop_accepting_or_connecting "
            "stop connecting to remote: %C\n",
            std::string(remote_converted).c_str()), 5);

  GuardType guard(connections_lock_);
  typedef std::multimap<TransportClient*, DataLink_rch>::iterator iter_t;
  const std::pair<iter_t, iter_t> range =
    pending_connections_.equal_range(client);

  for (iter_t iter = range.first; iter != range.second; ++iter) {
    iter->second->remove_on_start_callback(client, remote_id);
  }

  pending_connections_.erase(range.first, range.second);
}

virtual std::string OpenDDS::DCPS::TcpTransport::transport_type

(

)

const [inline, private, virtual]

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 79 of file TcpTransport.h.

{ return "tcp"; }

void OpenDDS::DCPS::TcpTransport::unbind_link

(

DataLink *

link

)

[virtual]

Remove any pending_release mappings.

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 768 of file TcpTransport.cpp.

References OpenDDS::DCPS::DataLink::datalink_release_delay(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::transport_priority(), and VDBG_LVL.

{
  TcpDataLink* tcp_link = static_cast<TcpDataLink*>(link);

  if (tcp_link == 0) {
    // Really an assertion failure
    ACE_ERROR((LM_ERROR,
               "(%P|%t) TcpTransport::unbind_link INTERNAL ERROR - "
               "Failed to downcast DataLink to TcpDataLink.\n"));
    return;
  }

  // Attempt to remove the TcpDataLink from our links_ map.
  PriorityKey key(
    tcp_link->transport_priority(),
    tcp_link->remote_address(),
    tcp_link->is_loopback(),
    tcp_link->is_active());

  VDBG_LVL((LM_DEBUG,
            "(%P|%t) TcpTransport::unbind_link link %@ PriorityKey "
            "prio=%d, addr=%C:%hu, is_loopback=%d, is_active=%d\n",
            link,
            tcp_link->transport_priority(),
            tcp_link->remote_address().get_host_addr(),
            tcp_link->remote_address().get_port_number(),
            (int)tcp_link->is_loopback(),
            (int)tcp_link->is_active()), 2);

  GuardType guard(this->links_lock_);

  if (this->pending_release_links_.unbind(key) != 0 &&
      link->datalink_release_delay() > ACE_Time_Value::zero) {
    ACE_ERROR((LM_ERROR,
               "(%P|%t) TcpTransport::unbind_link INTERNAL ERROR - "
               "Failed to find link %@ tcp_link %@ PriorityKey "
               "prio=%d, addr=%C:%hu, is_loopback=%d, is_active=%d\n",
               link,
               tcp_link,
               tcp_link->transport_priority(),
               tcp_link->remote_address().get_host_addr(),
               tcp_link->remote_address().get_port_number(),
               (int)tcp_link->is_loopback(),
               (int)tcp_link->is_active()));
  }
}

---

Friends And Related Function Documentation

friend class TcpConnection [friend]

The TcpConnection is our friend. It tells us when it has been created (by our acceptor_), and is seeking the DataLink that should be (or will be) expecting the passive connection.

Definition at line 87 of file TcpTransport.h.

Referenced by connect_datalink().

friend class TcpDataLink [friend]

Definition at line 88 of file TcpTransport.h.

Referenced by accept_datalink(), and connect_datalink().

---

Member Data Documentation

TcpAcceptor* OpenDDS::DCPS::TcpTransport::acceptor_ [private]

Used to accept passive connections on our local_address_.

Definition at line 138 of file TcpTransport.h.

Referenced by shutdown_i(), TcpTransport(), and ~TcpTransport().

TcpConnectionReplaceTask* OpenDDS::DCPS::TcpTransport::con_checker_ [private]

This task is used to resolve some deadlock situation during reconnecting. TODO: reuse the reconnect_task in the TcpConnection for new connection checking.

Definition at line 168 of file TcpTransport.h.

Referenced by passive_connection(), shutdown_i(), TcpTransport(), and ~TcpTransport().

ConnectionMap OpenDDS::DCPS::TcpTransport::connections_ [private]

Map of passive connection objects that need to be paired with a DataLink.

Definition at line 155 of file TcpTransport.h.

Referenced by accept_datalink(), passive_connection(), and shutdown_i().

LockType OpenDDS::DCPS::TcpTransport::connections_lock_ [private]

This protects the connections_ and the pending_connections_ data members.

Definition at line 159 of file TcpTransport.h.

Referenced by accept_datalink(), connect_datalink(), passive_connection(), and stop_accepting_or_connecting().

ACE_Connector<TcpConnection, ACE_SOCK_Connector> OpenDDS::DCPS::TcpTransport::connector_ [private]

Open TcpConnections using non-blocking connect.

Definition at line 141 of file TcpTransport.h.

Referenced by configure_i(), and connect_datalink().

AddrLinkMap OpenDDS::DCPS::TcpTransport::links_ [private]

This is the map of connected DataLinks.

Definition at line 147 of file TcpTransport.h.

Referenced by accept_datalink(), async_connect_failed(), connect_datalink(), find_datalink_i(), passive_connection(), and shutdown_i().

LockType OpenDDS::DCPS::TcpTransport::links_lock_ [private]

This lock is used to protect the links_ data member.

Definition at line 151 of file TcpTransport.h.

Referenced by accept_datalink(), async_connect_failed(), connect_datalink(), and passive_connection().

AddrLinkMap OpenDDS::DCPS::TcpTransport::pending_release_links_ [private]

Definition at line 148 of file TcpTransport.h.

Referenced by find_datalink_i(), and shutdown_i().

TransportReactorTask_rch OpenDDS::DCPS::TcpTransport::reactor_task_ [private]

We need the reactor for our Acceptor.

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 162 of file TcpTransport.h.

Referenced by configure_i(), and shutdown_i().

TcpInst_rch OpenDDS::DCPS::TcpTransport::tcp_config_ [private]

Our configuration object, supplied to us in config_i().

Definition at line 144 of file TcpTransport.h.

Referenced by blob_to_key(), configure_i(), connect_datalink(), connection_info_i(), fresh_link(), get_configuration(), passive_connection(), and shutdown_i().

---

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

• TcpTransport.h
• TcpTransport.cpp

---