Merge branch 'master' of github.com:OpenDDS/OpenDDS into dcs-recorder-replayer

Author: sonndinh

cmake/opendds_utils.cmake

@@ -372,17 +372,28 @@ endfunction()
 
 macro(_opendds_save_cache name type value)
   list(APPEND _opendds_save_cache_vars ${name})
-  set(_opendds_save_cache_${name}_type ${type})
-  set(_opendds_save_cache_${name}_value "${${name}}")
+  set(_prefix _opendds_save_cache_${name})
+  if(DEFINED ${name})
+    set(${_prefix}_defined TRUE)
+    set(${_prefix}_type ${type})
+    set(${_prefix}_value "${${name}}")
+  else()
+    set(${_prefix}_defined FALSE)
+  endif()
   set(${name} "${value}" CACHE ${type} "" FORCE)
 endmacro()
 
 macro(_opendds_restore_cache)
-  foreach(name ${_opendds_save_cache_vars})
-    set(${name} "${_opendds_save_cache_${name}_value}" CACHE
-      "${_opendds_save_cache_${name}_type}" "" FORCE)
-    unset(_opendds_save_cache_${name}_type)
-    unset(_opendds_save_cache_${name}_value)
+  foreach(_name ${_opendds_save_cache_vars})
+    set(_prefix _opendds_save_cache_${_name})
+    if(${_prefix}_defined)
+      set(${_name} "${${_prefix}_value}" CACHE "${${_prefix}_type}" "" FORCE)
+      unset(${_prefix}_type)
+      unset(${_prefix}_value)
+    else()
+      unset(${_name} CACHE)
+    endif()
+    unset(${_prefix}_defined)
   endforeach()
   unset(_opendds_save_cache_vars)
 endmacro()

dds/DCPS/AddressCache.h

@@ -232,6 +232,12 @@ class AddressCache {
     return map_.empty() && id_map_.empty();
   }
 
+  size_t size() const
+  {
+    ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
+    return map_.size() + id_map_.size();
+  }
+
 private:
 
   void insert_ids(const Key& key)

dds/DCPS/Dynamic_Cached_Allocator_With_Overflow_T.h

@@ -51,29 +51,22 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
     allocs_from_pool_(0),
     frees_to_heap_(0),
     frees_to_pool_(0),
+    chunk_size_(ACE_MALLOC_ROUNDUP(chunk_size, ACE_MALLOC_ALIGN)),
+    begin_(static_cast<unsigned char*>(ACE_Allocator::instance()->malloc(n_chunks * chunk_size_))),
+    end_(begin_ + n_chunks * chunk_size_),
     free_list_(ACE_PURE_FREE_LIST)
   {
-    chunk_size_ = ACE_MALLOC_ROUNDUP(chunk_size, ACE_MALLOC_ALIGN);
-    begin_ = static_cast<unsigned char*> (ACE_Allocator::instance()->malloc(n_chunks * chunk_size_));
-    // Remember end of the pool.
-    end_ = begin_ + n_chunks * chunk_size_;
-
     // Put into free list using placement contructor, no real memory
     // allocation in the <new> below.
-    for (size_t c = 0;
-         c < n_chunks;
-         c++) {
-      void* placement = begin_ + c * chunk_size_;
-
+    for (size_t c = 0; c < n_chunks; ++c) {
+      void* const placement = begin_ + c * chunk_size_;
       free_list_.add(new(placement) ACE_Cached_Mem_Pool_Node<char>);
     }
   }
 
-  /// Clear things up.
-  ~Dynamic_Cached_Allocator_With_Overflow() {
+  ~Dynamic_Cached_Allocator_With_Overflow()
+  {
     ACE_Allocator::instance()->free(begin_);
-    begin_ = 0;
-    chunk_size_ = 0;
   }
 
   /**
@@ -82,7 +75,8 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
   * and is otherwise ignored since malloc() always returns a pointer to an
   * item of @a chunk_size size.
   */
-  void *malloc(size_t nbytes = 0) {
+  void* malloc(size_t nbytes = 0)
+  {
     // Check if size requested fits within pre-determined size.
     if (nbytes > chunk_size_)
       return 0;
@@ -93,7 +87,8 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
 
     if (0 == rtn) {
       rtn = ACE_Allocator::instance()->malloc(chunk_size_);
-      allocs_from_heap_++;
+      ++allocs_from_heap_;
+      heap_allocated_ += chunk_size_;
 
       if (DCPS_debug_level >= 2) {
         if (allocs_from_heap_ == 1 && DCPS_debug_level >= 2)
@@ -113,7 +108,7 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
       }
 
     } else {
-      allocs_from_pool_++;
+      ++allocs_from_pool_;
 
       if (DCPS_debug_level >= 6)
         if (allocs_from_pool_ % 500 == 0)
@@ -134,26 +129,28 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
   * that it's less or equal to @a chunk_size, and is otherwise ignored
   * since calloc() always returns a pointer to an item of @a chunk_size.
   */
-  virtual void *calloc(size_t /* nbytes */,
-                       char /* initial_value */ = '\0') {
+  virtual void* calloc(size_t /* nbytes */,
+                       char /* initial_value */ = '\0')
+  {
     ACE_NOTSUP_RETURN(0);
   }
 
   /// This method is a no-op and just returns 0 since the free list
   /// only works with fixed sized entities.
-  virtual void *calloc(size_t /* n_elem */,
+  virtual void* calloc(size_t /* n_elem */,
                        size_t /* elem_size */,
                        char /* initial_value */ = '\0') {
     ACE_NOTSUP_RETURN(0);
   }
 
   /// Return a chunk of memory back to free list cache.
-  void free(void * ptr) {
-    unsigned char* tmp = static_cast<unsigned char*> (ptr);
-    if (tmp < begin_ ||
-        tmp >= end_) {
+  void free(void* ptr)
+  {
+    unsigned char* tmp = static_cast<unsigned char*>(ptr);
+    if (tmp < begin_ || tmp >= end_) {
       ACE_Allocator::instance()->free(tmp);
-      frees_to_heap_ ++;
+      ++frees_to_heap_;
+      heap_allocated_ -= chunk_size_;
 
       if (frees_to_heap_ > allocs_from_heap_) {
         ACE_ERROR((LM_ERROR,
@@ -177,7 +174,7 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
       return;
 
     } else if (ptr != 0) {
-      frees_to_pool_ ++;
+      ++frees_to_pool_;
 
       if (frees_to_pool_ > allocs_from_pool_) {
         ACE_ERROR((LM_ERROR,
@@ -188,7 +185,7 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
                    allocs_from_pool_.load()));
       }
 
-      free_list_.add((ACE_Cached_Mem_Pool_Node<char> *) ptr) ;
+      free_list_.add((ACE_Cached_Mem_Pool_Node<char>*) ptr);
 
       if (DCPS_debug_level >= 6)
         if (available() % 500 == 0)
@@ -201,41 +198,42 @@ class Dynamic_Cached_Allocator_With_Overflow : public ACE_New_Allocator, public
   }
 
   /// Return the number of chunks available in the cache.
-  size_t pool_depth() {
-    return free_list_.size() ;
-  }
+  size_t pool_depth() { return free_list_.size(); }
 
   // -- for debug
 
   /** How many chunks are available at this time.
   */
-  size_t available() {
-    return free_list_.size();
-  };
+  size_t available() { return free_list_.size(); }
+
+  size_t bytes_heap_allocated() const { return heap_allocated_.value(); }
 
   /// number of allocations from the heap.
   Atomic<unsigned long> allocs_from_heap_;
   /// number of allocations from the pool.
   Atomic<unsigned long> allocs_from_pool_;
   /// number of frees returned to the heap
-  Atomic<unsigned long> frees_to_heap_ ;
+  Atomic<unsigned long> frees_to_heap_;
   /// number of frees returned to the pool
   Atomic<unsigned long> frees_to_pool_;
+
 private:
+  /// Remember the size of our chunks.
+  const size_t chunk_size_;
+
   /// Remember how we allocate the memory in the first place so
   /// we can clear things up later.
-  unsigned char* begin_;
+  unsigned char* const begin_;
   /// The end of the pool.
-  unsigned char* end_;
+  unsigned char* const end_;
 
   /// Maintain a cached memory free list. We use @c char as template
   /// parameter, although sizeof(char) is usually less than
   /// sizeof(void*). Really important is that @a chunk_size
   /// must be greater or equal to sizeof(void*).
   ACE_Locked_Free_List<ACE_Cached_Mem_Pool_Node<char>, ACE_LOCK> free_list_;
 
-  /// Remember the size of our chunks.
-  size_t chunk_size_;
+  ACE_Atomic_Op<ACE_LOCK, size_t> heap_allocated_;
 };
 
 } // namespace DCPS

dds/DCPS/ReactorTask.cpp

@@ -367,6 +367,12 @@ void ReactorTask::process_command_queue_i(ACE_Guard<ACE_Thread_Mutex>& guard,
   }
 }
 
+size_t ReactorTask::command_queue_size() const
+{
+  GuardType guard(lock_);
+  return command_queue_.size();
+}
+
 bool ReactorWrapper::open(ACE_Reactor* reactor)
 {
   reactor_ = reactor;

dds/DCPS/ReactorTask.h

@@ -129,6 +129,8 @@ class OpenDDS_Dcps_Export ReactorTask
   CommandPtr execute_or_enqueue(CommandPtr command);
   void wait_until_empty();
 
+  size_t command_queue_size() const;
+
   OPENDDS_POOL_ALLOCATION_FWD
 
 private:

dds/DCPS/transport/framework/DataLink.cpp

@@ -1288,6 +1288,42 @@ DataLink::terminate_send_if_suspended()
   }
 }
 
+StatisticSeq DataLink::stats_template()
+{
+  static const DDS::UInt32 num_local_stats = 9;
+  StatisticSeq stats(num_local_stats);
+  stats.length(num_local_stats);
+  stats[0].name = "DataLinkSendListeners";
+  stats[1].name = "DataLinkRecvListeners";
+  stats[2].name = "DataLinkAssociationsByRemote";
+  stats[3].name = "DataLinkAssociationsByLocal";
+  stats[4].name = "DataLinkAssociationsReleasing";
+  stats[5].name = "DataLinkOnStartCallbacks";
+  stats[6].name = "DataLinkPendingOnStarts";
+  stats[7].name = "DataLinkMessageBlocks";
+  stats[8].name = "DataLinkDataBlocks";
+  return stats;
+}
+
+void DataLink::fill_stats(StatisticSeq& stats, DDS::UInt32& idx) const
+{
+  {
+    GuardType guard(pub_sub_maps_lock_);
+    stats[idx++].value = send_listeners_.size();
+    stats[idx++].value = recv_listeners_.size();
+    stats[idx++].value = assoc_by_remote_.size();
+    stats[idx++].value = assoc_by_local_.size();
+    stats[idx++].value = assoc_releasing_.size();
+  }
+  {
+    GuardType guard(strategy_lock_);
+    stats[idx++].value = on_start_callbacks_.size();
+    stats[idx++].value = pending_on_starts_.size();
+  }
+  stats[idx++].value = mb_allocator_ ? mb_allocator_->bytes_heap_allocated() : 0;
+  stats[idx++].value = db_allocator_ ? db_allocator_->bytes_heap_allocated() : 0;
+}
+
 }
 }
 

dds/DCPS/transport/framework/DataLink.h

@@ -436,7 +436,7 @@ class OpenDDS_Dcps_Export DataLink
 
   /// The transport send strategy object for this DataLink.
   TransportSendStrategy_rch send_strategy_;
-  LockType strategy_lock_;
+  mutable LockType strategy_lock_;
 
   TransportSendStrategy_rch get_send_strategy();
 
@@ -463,6 +463,9 @@ class OpenDDS_Dcps_Export DataLink
 
   /// Listener for TransportSendControlElements created in send_control
   SendResponseListener send_response_listener_;
+
+  static StatisticSeq stats_template();
+  void fill_stats(StatisticSeq& stats, DDS::UInt32& idx) const;
 };
 
 } // namespace DCPS

dds/DCPS/transport/framework/TransportImpl.cpp

@@ -154,6 +154,25 @@ TransportImpl::dump_to_str()
   return cfg ? cfg->dump_to_str(domain_) : OPENDDS_STRING();
 }
 
+StatisticSeq TransportImpl::stats_template()
+{
+  static const DDS::UInt32 num_local_stats = 2;
+  StatisticSeq stats(num_local_stats);
+  stats.length(num_local_stats);
+  stats[0].name = "TransportImplPendingConnections";
+  stats[1].name = "TransportImplReactorTaskCmdQueue";
+  return stats;
+}
+
+void TransportImpl::fill_stats(StatisticSeq& stats, DDS::UInt32& idx) const
+{
+  {
+    GuardType guard(pending_connections_lock_);
+    stats[idx++].value = pending_connections_.size();
+  }
+  stats[idx++].value = reactor_task_ ? reactor_task_->command_queue_size() : 0;
+}
+
 }
 }
 

dds/DCPS/transport/framework/TransportImpl.h

@@ -308,6 +308,9 @@ class OpenDDS_Dcps_Export TransportImpl : public virtual RcObject {
   /// Id of the last link established.
   AtomicBool is_shut_down_;
   DDS::DomainId_t domain_;
+
+  static StatisticSeq stats_template();
+  void fill_stats(StatisticSeq& stats, DDS::UInt32& idx) const;
 };
 
 } // namespace DCPS

dds/DCPS/transport/framework/TransportReassembly.cpp

@@ -308,6 +308,15 @@ void TransportReassembly::check_expirations(const MonotonicTimePoint& now)
   }
 }
 
+size_t TransportReassembly::total_frags() const
+{
+  size_t total = 0;
+  for (FragInfoMap::const_iterator iter = fragments_.begin(); iter != fragments_.end(); ++iter) {
+    total += iter->second.total_frags_;
+  }
+  return total;
+}
+
 TransportReassembly::FragInfo::FragInfo()
   : have_first_(false)
   , total_frags_(0)

dds/DCPS/transport/framework/TransportReassembly.h

@@ -105,6 +105,11 @@ class OpenDDS_Dcps_Export TransportReassembly : public RcObject {
                         CORBA::Long bitmap[], CORBA::ULong length,
                         CORBA::ULong& numBits) const;
 
+  size_t fragments_size() const { return fragments_.size(); }
+  size_t queue_size() const { return expiration_queue_.size(); }
+  size_t completed_size() const { return completed_.size(); }
+  size_t total_frags() const;
+
 private:
 
   bool reassemble_i(const FragmentRange& fragRange, bool firstFrag,