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 (TcpInst &inst)
virtual	~TcpTransport ()
int	fresh_link (TcpConnection_rch connection)
virtual void	unbind_link (DataLink *link)
	Remove any pending_release mappings.
TcpInst &	config () const
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, const TransportClient_rch &client)
virtual AcceptConnectResult	accept_datalink (const RemoteTransport &remote, const ConnectionAttribs &attribs, const TransportClient_rch &client)
virtual void	stop_accepting_or_connecting (const TransportClient_wrch &client, const RepoId &remote_id)
virtual bool	configure_i (TcpInst &config)
virtual void	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, const TransportClient_rch &client, const RepoId &remote_id)
int	connect_tcp_datalink (TcpDataLink &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
unique_ptr< TcpAcceptor >	acceptor_
	Used to accept passive connections on our local_address_.
ACE_Connector< TcpConnection, ACE_SOCK_Connector >	connector_
	Open TcpConnections using non-blocking connect.
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_
unique_ptr < 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 47 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 117 of file TcpTransport.h.

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

Definition at line 124 of file TcpTransport.h.

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

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 123 of file TcpTransport.h.

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

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 122 of file TcpTransport.h.

---

Constructor & Destructor Documentation

OpenDDS::DCPS::TcpTransport::TcpTransport

(

TcpInst &

inst

)

[explicit]

Definition at line 35 of file TcpTransport.cpp.

References configure_i(), DBG_ENTRY_LVL, OpenDDS::DCPS::TransportImpl::open(), and OpenDDS::DCPS::TransportImpl::shutdown().

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

  if (!(configure_i(inst) && open())) {
    this->shutdown();
    throw Transport::UnableToCreate();
  }

}

Here is the call graph for this function:

OpenDDS::DCPS::TcpTransport::~TcpTransport

(

)

[virtual]

Definition at line 49 of file TcpTransport.cpp.

References con_checker_, and DBG_ENTRY_LVL.

{
  DBG_ENTRY_LVL("TcpTransport","~TcpTransport",6);
  con_checker_->close(1);  // This could potentially fix a race condition
}

---

Member Function Documentation

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

accept_datalink() is called from TransportClient to initiate an association as the passive peer. A DataLink may be returned if one is already connected and ready to use, otherwise passively wait for a physical connection from the active side (either in the form of a connection event or handshaking message). Upon completion of the physical connection, the transport calls back to TransportClient::use_datalink().

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 232 of file TcpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::AcceptConnectResult::ACR_SUCCESS, OpenDDS::DCPS::TransportImpl::add_pending_connection(), OpenDDS::DCPS::PriorityKey::address(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::TransportImpl::RemoteTransport::blob_, blob_to_key(), connect_tcp_datalink(), connections_, connections_lock_, find_datalink_i(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TransportImpl::ConnectionAttribs::local_id_, OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportImpl::ConnectionAttribs::priority_, OpenDDS::DCPS::ref(), ACE_Guard< ACE_LOCK >::release(), OpenDDS::DCPS::TransportImpl::RemoteTransport::repo_id_, OpenDDS::DCPS::RcHandle< T >::reset(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), 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);

    } else {
      link = make_rch<TcpDataLink>(key.address(), ref(*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);
    }

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

    add_pending_connection(client, link);

    // 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.reset();
  }

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

Here is the call graph for this function:

void OpenDDS::DCPS::TcpTransport::async_connect_failed

(

const PriorityKey &

key

)

[private]

Definition at line 170 of file TcpTransport.cpp.

References ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::in(), links_, links_lock_, LM_ERROR, ACE_Guard< ACE_LOCK >::release(), and ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind().

Referenced by OpenDDS::DCPS::TcpConnection::open().

{

  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);
  }

}

Here is the call graph for this function:

Here is the caller graph for this function:

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

Definition at line 63 of file TcpTransport.cpp.

References config(), OpenDDS::DCPS::AssociationData::get_remote_address(), and OpenDDS::DCPS::TcpInst::local_address().

Referenced by accept_datalink(), and connect_datalink().

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

Here is the call graph for this function:

Here is the caller graph for this function:

TcpInst & OpenDDS::DCPS::TcpTransport::config

(

)

const

Expose the configuration information so others can see what we can do.

Reimplemented from OpenDDS::DCPS::TransportImpl.

Definition at line 57 of file TcpTransport.cpp.

Referenced by blob_to_key(), connect_datalink(), connect_tcp_datalink(), connection_info_i(), fresh_link(), OpenDDS::DCPS::TcpAcceptor::get_configuration(), and passive_connection().

{
  return static_cast<TcpInst&>(TransportImpl::config());
}

Here is the caller graph for this function:

bool OpenDDS::DCPS::TcpTransport::configure_i

(

TcpInst &

config

)

[private, virtual]

Definition at line 335 of file TcpTransport.cpp.

References acceptor_, ACE_TEXT(), con_checker_, connector_, OpenDDS::DCPS::TransportImpl::create_reactor_task(), DBG_ENTRY_LVL, OpenDDS::DCPS::get_fully_qualified_hostname(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), hostname(), ACE_INET_Addr::is_any(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TcpInst::local_address(), OpenDDS::DCPS::TcpInst::local_address_set_port(), ACE_Connector< class, class >::open(), OPENDDS_STRING, OpenDDS::DCPS::TransportImpl::reactor_task(), TheServiceParticipant, and VDBG_LVL.

Referenced by TcpTransport().

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

  this->create_reactor_task();

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

  // 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 (config.local_address() == ACE_INET_Addr () &&
      !TheServiceParticipant->default_address ().empty ()) {
    config.local_address(0, TheServiceParticipant->default_address ().c_str ());
  }

  // Open our acceptor object so that we can accept passive connections
  // on our config.local_address_.

  if (this->acceptor_->open(config.local_address(),
                            this->reactor_task()->get_reactor()) != 0) {

    ACE_ERROR_RETURN((LM_ERROR,
                      ACE_TEXT("(%P|%t) ERROR: Acceptor failed to open %C:%d: %p\n"),
                      config.local_address().get_host_addr(),
                      config.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 (config.local_address().is_any()) {
    std::string hostname = get_fully_qualified_hostname();

    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 (config.local_address().get_port_number() == 0) {
    config.local_address_set_port(port);
  }

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

connect_datalink() is called from TransportClient to initiate an association as the active peer. A DataLink may be returned if one is already connected and ready to use, otherwise initiate a connection to the passive side and return from this method. Upon completion of the physical connection, the transport calls back to TransportClient::use_datalink().

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 74 of file TcpTransport.cpp.

References OpenDDS::DCPS::TransportImpl::AcceptConnectResult::ACR_SUCCESS, OpenDDS::DCPS::TransportImpl::add_pending_connection(), ACE_INET_Addr::addr_to_string(), OpenDDS::DCPS::PriorityKey::address(), ACE_Synch_Options::asynch, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::TransportImpl::RemoteTransport::blob_, blob_to_key(), config(), ACE_Connector< class, class >::connect(), connections_lock_, connector_, DBG_ENTRY_LVL, find_datalink_i(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::PriorityKey::is_active(), OpenDDS::DCPS::PriorityKey::is_loopback(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::PriorityKey::priority(), OpenDDS::DCPS::TransportImpl::ConnectionAttribs::priority_, OpenDDS::DCPS::ref(), OpenDDS::DCPS::TransportImpl::RemoteTransport::repo_id_, str, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), 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);
    }

    link = make_rch<TcpDataLink>(key.address(), ref(*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(
    make_rch<TcpConnection>(key.address(), link->transport_priority(), this->config()));
  connection->set_datalink(link);

  TcpConnection* pConn = connection.in();

  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_ERROR((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();
  }

  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);
  }

  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);
  }

  GuardType connections_guard(connections_lock_);

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

Here is the call graph for this function:

int OpenDDS::DCPS::TcpTransport::connect_tcp_datalink	(	TcpDataLink &	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 658 of file TcpTransport.cpp.

References ACE_TEXT(), config(), OpenDDS::DCPS::TcpDataLink::connect(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, OpenDDS::DCPS::TransportImpl::last_link_, LM_DEBUG, OpenDDS::DCPS::TransportImpl::reactor_task(), OpenDDS::DCPS::ref(), OpenDDS::DCPS::TcpDataLink::reuse_existing_connection(), and OpenDDS::DCPS::DataLink::transport_priority().

Referenced by accept_datalink(), OpenDDS::DCPS::TcpConnection::open(), 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_;

  TcpSendStrategy_rch send_strategy (
    make_rch<TcpSendStrategy>(last_link_, ref(link),
                             new TcpSynchResource(link,
                                                  this->config().max_output_pause_period_),
                             this->reactor_task(), link.transport_priority()));

  TcpReceiveStrategy_rch receive_strategy(
    make_rch<TcpReceiveStrategy>(ref(link), this->reactor_task()));

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

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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 466 of file TcpTransport.cpp.

References config(), DBG_ENTRY_LVL, LM_DEBUG, OpenDDS::DCPS::TcpInst::populate_locator(), and VDBG_LVL.

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

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

  this->config().populate_locator(local_info);

  return true;
}

Here is the call graph for this function:

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

Definition at line 188 of file TcpTransport.cpp.

References ACE_TEXT(), OpenDDS::DCPS::TransportImpl::add_pending_connection(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::in(), links_, LM_DEBUG, pending_release_links_, OpenDDS::DCPS::RcHandle< T >::reset(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), 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);
    link.reset(); // 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.reset(); // don't return link to TransportClient
    return false;
  }

  return false;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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 698 of file TcpTransport.cpp.

References config(), DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, and OpenDDS::DCPS::TcpInst::local_address().

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->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);
    }
  }

  return 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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.

This method is called by a TcpConnection object that has been created and opened by our acceptor_ as a result of passively accepting a connection on our local address. Ultimately, the connection object needs to be paired with a DataLink object that is (or will be) expecting this passive connection to be established.

Definition at line 596 of file TcpTransport.cpp.

References ACE_TEXT(), con_checker_, config(), connect_tcp_datalink(), connections_, connections_lock_, DBG_ENTRY_LVL, ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::find(), OpenDDS::DCPS::RcHandle< T >::is_nil(), links_, links_lock_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TcpInst::local_address(), ACE_Guard< ACE_LOCK >::release(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), and VDBG_LVL.

Referenced by OpenDDS::DCPS::TcpConnection::handle_setup_input().

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

  const PriorityKey key(connection->transport_priority(),
                        remote_address,
                        remote_address == 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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void OpenDDS::DCPS::TcpTransport::release_datalink

(

DataLink *

link

)

[private, virtual]

Called by the DataLink to release itself.

Implements OpenDDS::DCPS::TransportImpl.

Definition at line 479 of file TcpTransport.cpp.

References ACE_TEXT(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::bind(), OpenDDS::DCPS::DataLink::datalink_release_delay(), DBG_ENTRY_LVL, OpenDDS::DCPS::DCPS_debug_level, ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), ACE_OS::gettimeofday(), OpenDDS::DCPS::RcHandle< T >::in(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), links_, links_lock_, LM_DEBUG, LM_ERROR, pending_release_links_, OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::schedule_delayed_release(), OpenDDS::DCPS::DataLink::schedule_stop(), ACE_Time_Value::sec(), OpenDDS::DCPS::DataLink::set_scheduling_release(), OpenDDS::DCPS::DataLink::transport_priority(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), ACE_Time_Value::usec(), VDBG_LVL, and ACE_Time_Value::zero.

{
  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()));
  }
}

Here is the call graph for this function:

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 408 of file TcpTransport.cpp.

References acceptor_, con_checker_, connections_, connections_lock_, DBG_ENTRY_LVL, ACE_Hash_Map_Entry< EXT_ID, INT_ID >::int_id_, links_, links_lock_, OpenDDS::DCPS::TransportImpl::pending_connections_, pending_release_links_, and ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind_all().

{
  DBG_ENTRY_LVL("TcpTransport","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();
    }
  }

  // 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();
  }

  // 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();
}

Here is the call graph for this function:

void OpenDDS::DCPS::TcpTransport::stop_accepting_or_connecting	(	const TransportClient_wrch &	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 314 of file TcpTransport.cpp.

References connections_lock_, LM_DEBUG, OpenDDS::DCPS::TransportImpl::pending_connections_, 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 PendConnMap::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 81 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 731 of file TcpTransport.cpp.

References OpenDDS::DCPS::DataLink::datalink_release_delay(), ACE_INET_Addr::get_host_addr(), ACE_INET_Addr::get_port_number(), OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataLink::is_loopback(), links_lock_, LM_DEBUG, LM_ERROR, pending_release_links_, OpenDDS::DCPS::TcpDataLink::remote_address(), OpenDDS::DCPS::DataLink::transport_priority(), ACE_Hash_Map_Manager_Ex< EXT_ID, INT_ID, HASH_KEY, COMPARE_KEYS, ACE_LOCK >::unbind(), VDBG_LVL, and ACE_Time_Value::zero.

{
  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()));
  }
}

Here is the call graph for this function:

---

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 89 of file TcpTransport.h.

friend class TcpDataLink [friend]

Definition at line 90 of file TcpTransport.h.

---

Member Data Documentation

unique_ptr<TcpAcceptor> OpenDDS::DCPS::TcpTransport::acceptor_ [private]

Used to accept passive connections on our local_address_.

Definition at line 140 of file TcpTransport.h.

Referenced by configure_i(), and shutdown_i().

unique_ptr<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 164 of file TcpTransport.h.

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

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

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

Definition at line 154 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 158 of file TcpTransport.h.

Referenced by accept_datalink(), connect_datalink(), passive_connection(), shutdown_i(), 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 143 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 146 of file TcpTransport.h.

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

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

This lock is used to protect the links_ data member.

Definition at line 150 of file TcpTransport.h.

Referenced by accept_datalink(), async_connect_failed(), connect_datalink(), fresh_link(), passive_connection(), release_datalink(), shutdown_i(), and unbind_link().

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

Definition at line 147 of file TcpTransport.h.

Referenced by find_datalink_i(), release_datalink(), shutdown_i(), and unbind_link().

---

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

• TcpTransport.h
• TcpTransport.cpp