自学教程：C++ DBG_ENTRY_LVL函数代码示例

/// This method is called when the (remote) subscriber is being/// released.  This method will return a 0 if the subscriber_id is/// successfully disassociated with the publisher_id *and* there/// are still other subscribers associated with the publisher_id./// This method will return 1 if, after the disassociation, the/// publisher_id is no longer associated with any subscribers (which/// also means it's element was removed from our map_).intOpenDDS::DCPS::ReceiveListenerSetMap::release_subscriber(RepoId publisher_id,                                                         RepoId subscriber_id){  DBG_ENTRY_LVL("ReceiveListenerSetMap","release_subscriber",6);  ReceiveListenerSet_rch listener_set;  if (OpenDDS::DCPS::find(map_, publisher_id, listener_set) != 0) {    GuidConverter converter(publisher_id);    ACE_ERROR((LM_ERROR,               ACE_TEXT("(%P|%t) ERROR: ReciveListenerSetMap::release_subscriber: ")               ACE_TEXT("publisher %C not found in map_./n"),               std::string(converter).c_str()));    // Return 1 to indicate that the publisher_id is no longer associated    // with any subscribers at all.    return 1;  }  int result = listener_set->remove(subscriber_id);  // Ignore the result  ACE_UNUSED_ARG(result);  if (listener_set->size() == 0) {    if (unbind(map_, publisher_id) != 0) {      GuidConverter converter(publisher_id);      ACE_ERROR((LM_ERROR,                 ACE_TEXT("(%P|%t) ERROR: ReceiveListenerSetMap::release_subscriber: ")                 ACE_TEXT("failed to remove empty ReceiveListenerSet for ")                 ACE_TEXT("publisher %C./n"),                 std::string(converter).c_str()));    }    // We always return 1 if we know the publisher_id is no longer    // associated with any ReceiveListeners.    return 1;  }  // There are still ReceiveListeners associated with the publisher_id.  // We return a 0 in this case.  return 0;}

voidDataLink::transport_shutdown(){  DBG_ENTRY_LVL("DataLink", "transport_shutdown", 6);  if (!this->send_strategy_.is_nil()) {    this->send_strategy_->transport_shutdown();  }  //this->cancel_release();  this->set_scheduling_release(false);  this->scheduled_to_stop_at_ = ACE_Time_Value::zero;  ACE_Reactor_Timer_Interface* reactor = this->impl_->timer();  reactor->cancel_timer(this);  this->stop();  // Drop our reference to the TransportImpl object  this->impl_ = 0;}

void OpenDDS::DCPS::ThreadPerConnectionSendTask::execute(SendRequest& req){  DBG_ENTRY_LVL("ThreadPerConnectionSendTask","execute",6);  switch (req.op_) {  case SEND_START:    this->link_->send_start_i();    break;  case SEND:    this->link_->send_i(req.element_);    break;  case SEND_STOP:    this->link_->send_stop_i();    break;  default:    ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: ThreadPerConnectionSendTask::execute unknown command %d/n",               req.op_));    break;  }}

// This version removes an entire RepoIdSet from the map_.OpenDDS::DCPS::RepoIdSet*OpenDDS::DCPS::RepoIdSetMap::remove_set(RepoId key){  DBG_ENTRY_LVL("RepoIdSetMap","remove_set",6);  RepoIdSet_rch value;  if (unbind(map_, key, value) != 0) {    if (DCPS_debug_level > 4) {      RepoIdConverter converter(key);      ACE_DEBUG((LM_DEBUG,                 ACE_TEXT("(%P|%t) RepeIdSetMap::remove_set: ")                 ACE_TEXT("RepoId %C not found in map./n"),                 std::string(converter).c_str()));    }    return 0;  }  return value._retn();}

OpenDDS::DCPS::DataLink*OpenDDS::DCPS::TransportImpl::reserve_datalink(  RepoId                  local_id,  const AssociationData*  remote_association,  CORBA::Long             priority,  TransportSendListener*  send_listener){  DBG_ENTRY_LVL("TransportImpl","reserve_datalink",6);  // Ask our concrete subclass to find or create a (concrete) DataLink  // that matches the supplied criterea.  // Note that we pass-in true as the third argument.  This means that  // if a new DataLink needs to be created (ie, the find operation fails),  // then the connection establishment logic will treat the local endpoint  // as the publisher.  This knowledge dictates whether a passive or active  // connection establishment procedure should be followed.  DataLink_rch link =    this->find_or_create_datalink(local_id,                                  remote_association,                                  priority,                                  true);  if (link.is_nil()) {    OpenDDS::DCPS::RepoIdConverter pub_converter(local_id);    OpenDDS::DCPS::RepoIdConverter sub_converter(remote_association->remote_id_);    ACE_ERROR_RETURN((LM_ERROR,                      ACE_TEXT("(%P|%t) ERROR: TransportImpl::reserve_datalink: ")                      ACE_TEXT("subclass was unable to find ")                      ACE_TEXT("or create a DataLink for local publisher_id %C ")                      ACE_TEXT("to remote subscriber_id %C./n"),                      std::string(pub_converter).c_str(),                      std::string(sub_converter).c_str()),0);  }  link->make_reservation(remote_association->remote_id_,  // subscription_id                         local_id,                        // publication_id                         send_listener);  return link._retn();}

/// The TcpTransport calls this method when it has an established/// connection object for us.  This call puts this TcpDataLink into/// the "connected" state.intOpenDDS::DCPS::TcpDataLink::connect(  const TcpConnection_rch& connection,  const TransportSendStrategy_rch& send_strategy,  const TransportStrategy_rch& receive_strategy){  DBG_ENTRY_LVL("TcpDataLink","connect",6);  // Sanity check - cannot connect() if we are already connected.  if (!this->connection_.is_nil()) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: TcpDataLink already connected./n"),                     -1);  }  this->connection_ = connection;  if (this->connection_->peer().enable(ACE_NONBLOCK) == -1) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: TcpDataLink::connect failed to set "                      "ACE_NONBLOCK %p/n", ACE_TEXT("enable")), -1);  }  // Let connection know the datalink for callbacks upon reconnect failure.  this->connection_->set_datalink(this);  // And lastly, inform our base class (DataLink) that we are now "connected",  // and it should start the strategy objects.  if (this->start(send_strategy, receive_strategy) != 0) {    // Our base (DataLink) class failed to start the strategy objects.    // We need to "undo" some things here before we return -1 to indicate    // that an error has taken place.    // Drop our reference to the connection object.    this->connection_ = 0;    return -1;  }  return 0;}

// This method is called on acceptor side when the lost connection is detected.// A timer is scheduled to check if a new connection is created within the// passive_reconnect_duration_ period.intOpenDDS::DCPS::TcpConnection::passive_reconnect_i(){  DBG_ENTRY_LVL("TcpConnection","passive_reconnect_i",6);  GuardType guard(this->reconnect_lock_);  // The passive_reconnect_timer_id_ is used as flag to allow the timer scheduled just once.  if (this->reconnect_state_ == INIT_STATE) {    // Mark the connection lost since the recv/send just failed.    this->connected_ = false;    if (this->tcp_config_->passive_reconnect_duration_ == 0)      return -1;    ACE_Time_Value timeout(this->tcp_config_->passive_reconnect_duration_/1000,                           this->tcp_config_->passive_reconnect_duration_%1000 * 1000);    this->reconnect_state_ = PASSIVE_WAITING_STATE;    this->link_->notify(DataLink::DISCONNECTED);    // It is possible that the passive reconnect is called after the new connection    // is accepted and the receive_strategy of this old connection is reset to nil.    if (!this->receive_strategy_.is_nil()) {      TcpReceiveStrategy* rs        = dynamic_cast <TcpReceiveStrategy*>(this->receive_strategy_.in());      // Give a copy to reactor.      this->_add_ref();      this->passive_reconnect_timer_id_ = rs->get_reactor()->schedule_timer(this, 0, timeout);      if (this->passive_reconnect_timer_id_ == -1) {        this->_remove_ref();        ACE_ERROR_RETURN((LM_ERROR,                          ACE_TEXT("(%P|%t) ERROR: TcpConnection::passive_reconnect_i")                          ACE_TEXT(", %p./n"), ACE_TEXT("schedule_timer")),                         -1);      }    }  }  return 0;}

voidOpenDDS::DCPS::TcpReceiveStrategy::deliver_sample  (ReceivedDataSample& sample, const ACE_INET_Addr&){  DBG_ENTRY_LVL("TcpReceiveStrategy","deliver_sample",6);  if (sample.header_.message_id_ == GRACEFUL_DISCONNECT) {    VDBG((LM_DEBUG, "(%P|%t) DBG:  received GRACEFUL_DISCONNECT /n"));    this->gracefully_disconnected_ = true;  }  else if (sample.header_.message_id_ == REQUEST_ACK) {    VDBG((LM_DEBUG, "(%P|%t) DBG:  received REQUEST_ACK /n"));    this->link_->request_ack_received(sample);  }  else if (sample.header_.message_id_ == SAMPLE_ACK) {    VDBG((LM_DEBUG, "(%P|%t) DBG:  received SAMPLE_ACK /n"));    this->link_->ack_received(sample);  }  else {    this->link_->data_received(sample);  }}

boolTcpTransport::connection_info_i(TransportLocator& local_info) const{  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);  // Get the public address string from the inst (usually the local address)  NetworkAddress network_order_address(this->tcp_config_->get_public_address());  ACE_OutputCDR cdr;  cdr << network_order_address;  const CORBA::ULong len = static_cast<CORBA::ULong>(cdr.total_length());  char* buffer = const_cast<char*>(cdr.buffer()); // safe  local_info.transport_type = "tcp";  local_info.data = TransportBLOB(len, len,                                  reinterpret_cast<CORBA::Octet*>(buffer));  return true;}

OpenDDS::DCPS::TcpSendStrategy::TcpSendStrategy(  std::size_t id,  const TcpDataLink_rch& link,  const TcpInst_rch& config,  const TcpConnection_rch& connection,  TcpSynchResource* synch_resource,  const TransportReactorTask_rch& task,  Priority priority)  : TransportSendStrategy(id, static_rchandle_cast<TransportInst>(config),                          synch_resource, priority,                          new ReactorSynchStrategy(                                this,                                task->get_reactor()))  , connection_(connection)  , link_(link)  , reactor_task_(task){  DBG_ENTRY_LVL("TcpSendStrategy","TcpSendStrategy",6);  connection->set_send_strategy(this);}

int OpenDDS::DCPS::ThreadPerConnectionSendTask::svc(){  DBG_ENTRY_LVL("ThreadPerConnectionSendTask","svc",6);  this->thr_id_ = ACE_OS::thr_self ();  // Start the "GetWork-And-PerformWork" loop for the current worker thread.  while (! this->shutdown_initiated_)  {    SendRequest* req;    {      GuardType guard(this->lock_);      if (this->queue_.size() == 0)      {        this->work_available_.wait ();      }      if (this->shutdown_initiated_)        break;      req = queue_.get ();      if(req == 0)      {        //I'm not sure why this thread got more signals than actual signals        //when using thread_per_connection and the user application thread        //send requests without interval. We just need ignore the dequeue        //failure.        //ACE_ERROR ((LM_ERROR, "(%P|%t)ERROR: ThreadPerConnectionSendTask::svc  %p/n",        //  "dequeue_head"));        continue;      }    }    this->execute(*req);    delete req;  }  // This will never get executed.  return 0;}

voidOpenDDS::DCPS::TcpSendStrategy::schedule_output(){  DBG_ENTRY_LVL("TcpSendStrategy","schedule_output",6);  // Notify the reactor to adjust its processing policy according to mode_.  synch()->work_available();  if (DCPS_debug_level > 4) {    const char* action = "";    if( mode() == MODE_DIRECT) {      action = "canceling";    } else if( (mode() == MODE_QUEUE)            || (mode() == MODE_SUSPEND)) {      action = "starting";    }    ACE_DEBUG((LM_DEBUG,               ACE_TEXT("(%P|%t) TcpSendStrategy::schedule_output() [%d] - ")               ACE_TEXT("%C data queueing for handle %d./n"),               id(),action,get_handle()));  }}

OpenDDS::DCPS::TcpConnection::TcpConnection()  : connected_(false)  , is_connector_(false)  , passive_reconnect_timer_id_(-1)  , reconnect_task_(this)  , reconnect_state_(INIT_STATE)  , last_reconnect_attempted_(ACE_Time_Value::zero)  , transport_priority_(0)  // TRANSPORT_PRIORITY.value default value - 0.  , shutdown_(false)  , passive_setup_(false)  , passive_setup_buffer_(sizeof(ACE_UINT32))  , id_(0){  DBG_ENTRY_LVL("TcpConnection","TcpConnection",6);  if (this->reconnect_task_.open()) {    ACE_ERROR((LM_ERROR,               ACE_TEXT("(%P|%t) ERROR: Reconnect task failed to open : %p/n"),               ACE_TEXT("open")));  }}

/// A timer is scheduled on acceptor side to check if a new connection/// is accepted after the connection is lost.intOpenDDS::DCPS::SimpleTcpConnection::handle_timeout (const ACE_Time_Value &,                                                const void *){  DBG_ENTRY_LVL("SimpleTcpConnection","handle_timeout",6);  this->reconnect_state_ = PASSIVE_TIMEOUT_CALLED_STATE;  GuardType guard (this->reconnect_lock_);  switch (this->reconnect_state_)  {  case PASSIVE_TIMEOUT_CALLED_STATE:    {      // We stay in PASSIVE_TIMEOUT_CALLED_STATE indicates there is no new connection.      // Now we need declare the connection is lost.      this->link_->notify (DataLink::LOST);      // The handle_timeout may be called after the connection is re-established       // and the send strategy of this old connection is reset to nil.      if (! this->send_strategy_.is_nil ())        this->send_strategy_->terminate_send ();      this->reconnect_state_ = LOST_STATE;    }    break;  case RECONNECTED_STATE:    // reconnected successfully.    break;  default :    ACE_ERROR((LM_ERROR,      ACE_TEXT("(%P|%t) ERROR: SimpleTcpConnection::handle_timeout, ")      ACE_TEXT(" unknown state or it should not be in state=%d /n"), this->reconnect_state_));    break;  }  // Take back the "copy" we gave to reactor when we schedule the timer.  this->_remove_ref ();  return 0;}

/// The client application calls this to attach this TransportInterface/// object to the supplied TransportImpl object.OpenDDS::DCPS::AttachStatusOpenDDS::DCPS::TransportInterface::attach_transport(TransportImpl* impl){  DBG_ENTRY_LVL("TransportInterface","attach_transport",6);  { // guard scope    GuardType guard(this->lock_);    if (!this->impl_.is_nil()) {      // This TransportInterface object is currently attached to      // a TransportImpl.      return ATTACH_BAD_TRANSPORT;    }    // Ask the impl for the swap_bytes() value, and cache the answer in    // a data member (this->swap_bytes_) so that we don't have to ask    // the impl for it anymore.    this->swap_bytes_ = impl->swap_bytes();    // Ask the impl for the connection_info() value, and cache the answer in    // a data member (this->connection_info_) so that we don't have to ask    // the impl for it anymore.    impl->connection_info(this->connection_info_);    // Attempt to attach ourselves to the TransportImpl object now.    if (impl->attach_interface(this) == ATTACH_BAD_TRANSPORT) {      // The TransportImpl didn't accept our attachment for some reason.      return ATTACH_BAD_TRANSPORT;    }    // Now the TransportImpl object knows about us.  Let's remember the    // TransportImpl object by saving a "copy" of the reference to our    // impl_ data member.    impl->_add_ref();    this->impl_ = impl;  }  return ATTACH_OK;}

OpenDDS::DCPS::DataLinkSet*OpenDDS::DCPS::DataLinkSet::select_links(const RepoId* remoteIds,                                         const CORBA::ULong num_targets){  DBG_ENTRY_LVL("DataLinkSet","select_links",6);  DataLinkSet_rch selected_links = new DataLinkSet();  GuardType guard(this->lock_);  for (MapType::iterator itr = map_.begin();       itr != map_.end();       ++itr) {    for (CORBA::ULong i = 0; i < num_targets; ++i) {      if (itr->second->is_target(remoteIds[i])) {        OpenDDS::DCPS::bind(selected_links->map_,             itr->second->id(), itr->second);        break;      }    }  }  return selected_links._retn();}

intOpenDDS::DCPS::TransportReactorTask::open(void*){  DBG_ENTRY_LVL("TransportReactorTask","open",6);  GuardType guard(this->lock_);  // Reset our state.  this->state_ = STATE_NOT_RUNNING;  // For now, we only support one thread to run the reactor event loop.  // Parts of the logic in this class would need to change to support  // more than one thread running the reactor.  // Attempt to activate ourselves.  If successful, a new thread will be  // started and it will invoke our svc() method.  Note that we still have  // a hold on our lock while we do this.  if (this->activate(THR_NEW_LWP | THR_JOINABLE,1) != 0) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: TransportReactorTask Failed to activate "                      "itself./n"),                     -1);  }  this->wait_for_startup();  // Here we need to wait until a condition is triggered by the new thread(s)  // that we created.  Note that this will cause us to release the lock and  // wait until the condition_ is signal()'ed.  When it is signaled, the  // condition will attempt to obtain the lock again, and then return to us  // here.  We can then go on.  if (this->state_ == STATE_NOT_RUNNING) {    this->state_ = STATE_OPENING;    this->condition_.wait();  }  return 0;}

voidOpenDDS::DCPS::RepoIdSetMap::dump(){  DBG_ENTRY_LVL("RepoIdSetMap","dump",6);  for (MapType::iterator itr = map_.begin();       itr != map_.end();       ++itr) {    RepoIdSet_rch set = itr->second;    for (RepoIdSet::MapType::iterator it = set->map().begin();         it != set->map().end(); ++it) {      OPENDDS_STRING buffer;      buffer + "key  ";      buffer += to_string(itr->first);      buffer + " - value ";      buffer += to_string(it->first);      ACE_DEBUG((LM_DEBUG,                 ACE_TEXT("(%P|%t)   %C /n"),                 buffer.c_str()));    }  }}

/// The SimpleTcpTransport calls this method when it has an established/// connection object for us.  This call puts this SimpleTcpDataLink into/// the "connected" state.intOpenDDS::DCPS::SimpleTcpDataLink::connect(SimpleTcpConnection*      connection, TransportSendStrategy*    send_strategy, TransportReceiveStrategy* receive_strategy){  DBG_ENTRY_LVL("SimpleTcpDataLink","connect",6);  // Sanity check - cannot connect() if we are already connected.  if (!this->connection_.is_nil()) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: SimpleTcpDataLink already connected./n"),                     -1);  }  // Keep a "copy" of the reference to the connection object for ourselves.  connection->_add_ref();  this->connection_ = connection;  // Let connection know the datalink for callbacks upon reconnect failure.  this->connection_->set_datalink(this);  // And lastly, inform our base class (DataLink) that we are now "connected",  // and it should start the strategy objects.  if (this->start(send_strategy,receive_strategy) != 0) {    // Our base (DataLink) class failed to start the strategy objects.    // We need to "undo" some things here before we return -1 to indicate    // that an error has taken place.    // Drop our reference to the connection object.    this->connection_ = 0;    return -1;  }  return 0;}

voidRecorderImpl::remove_all_associations(){  DBG_ENTRY_LVL("RecorderImpl","remove_all_associations",6);  OpenDDS::DCPS::WriterIdSeq writers;  int size;  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->publication_handle_lock_);  {    ACE_READ_GUARD(ACE_RW_Thread_Mutex, read_guard, this->writers_lock_);    size = static_cast<int>(writers_.size());    writers.length(size);    WriterMapType::iterator curr_writer = writers_.begin();    WriterMapType::iterator end_writer = writers_.end();    int i = 0;    while (curr_writer != end_writer) {      writers[i++] = curr_writer->first;      ++curr_writer;    }  }  try {    CORBA::Boolean dont_notify_lost = 0;    if (0 < size) {      remove_associations(writers, dont_notify_lost);    }  } catch (const CORBA::Exception&) {  }}

intOpenDDS::DCPS::TcpSendStrategy::reset(TcpConnection* connection){  DBG_ENTRY_LVL("TcpSendStrategy","reset",6);  // Sanity check - this connection is passed in from the constructor and  // it should not be nil.  if (this->connection_.is_nil()) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: TcpSendStrategy::reset  previous connection "                      "should not be nil./n"),                     -1);  }  if (this->connection_.in() == connection) {    ACE_ERROR_RETURN((LM_ERROR,                      "(%P|%t) ERROR: TcpSendStrategy::reset should not be called"                      " to replace the same connection./n"),                     -1);  }  // This will cause the connection_ object to drop its reference to this  // TransportSendStrategy object.  this->connection_->remove_send_strategy();  // Replace with a new connection.  connection->_add_ref();  this->connection_ = connection;  // Tell the TcpConnection that we are the object that it should  // call when it receives a handle_input() "event", and we will carry  // it out.  The TcpConnection object will make a "copy" of the  // reference (to this object) that we pass-in here.  this->connection_->set_send_strategy(this);  return 0;}