Line data    Source code

       1             : /*
       2             :  *
       3             :  *
       4             :  * Distributed under the OpenDDS License.
       5             :  * See: http://www.opendds.org/license.html
       6             :  */
       7             : 
       8             : #include <DCPS/DdsDcps_pch.h> // Only the _pch include should start with DCPS/
       9             : 
      10             : #include "ThreadStatusManager.h"
      11             : #include "SafetyProfileStreams.h"
      12             : #include "debug.h"
      13             : 
      14             : OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL
      15             : 
      16             : namespace OpenDDS {
      17             : namespace DCPS {
      18             : 
      19           0 : void ThreadStatusManager::Thread::update(const MonotonicTimePoint& m_now,
      20             :                                          const SystemTimePoint& s_now,
      21             :                                          ThreadStatus next_status,
      22             :                                          const TimeDuration& bucket_limit,
      23             :                                          bool nested)
      24             : {
      25           0 :   timestamp_ = s_now;
      26             : 
      27           0 :   if (nested) {
      28           0 :     switch(next_status) {
      29           0 :     case ThreadStatus_Active:
      30           0 :       ++nesting_depth_;
      31           0 :       break;
      32           0 :     case ThreadStatus_Idle:
      33           0 :       --nesting_depth_;
      34           0 :       break;
      35             :     }
      36             :   }
      37             : 
      38           0 :   if (!nested ||
      39           0 :       (next_status == ThreadStatus_Active && nesting_depth_ == 1) ||
      40           0 :       (next_status == ThreadStatus_Idle && nesting_depth_ == 0)) {
      41           0 :     if (bucket_[current_bucket_].active_time + bucket_[current_bucket_].idle_time > bucket_limit) {
      42           0 :       current_bucket_ = (current_bucket_ + 1) % bucket_count;
      43           0 :       total_.active_time -= bucket_[current_bucket_].active_time;
      44           0 :       bucket_[current_bucket_].active_time = 0;
      45           0 :       total_.idle_time -= bucket_[current_bucket_].idle_time;
      46           0 :       bucket_[current_bucket_].idle_time = 0;
      47             :     }
      48             : 
      49           0 :     const TimeDuration t = m_now - last_update_;
      50             : 
      51           0 :     switch (status_) {
      52           0 :     case ThreadStatus_Active:
      53           0 :       bucket_[current_bucket_].active_time += t;
      54           0 :       total_.active_time += t;
      55           0 :       break;
      56           0 :     case ThreadStatus_Idle:
      57           0 :       bucket_[current_bucket_].idle_time += t;
      58           0 :       total_.idle_time += t;
      59           0 :       break;
      60             :     }
      61             : 
      62           0 :     last_update_ = m_now;
      63           0 :     status_ = next_status;
      64           0 :   }
      65           0 : }
      66             : 
      67           0 : double ThreadStatusManager::Thread::utilization(const MonotonicTimePoint& now) const
      68             : {
      69           0 :   const TimeDuration active_bonus = (now > last_update_ && status_ == ThreadStatus_Active) ? (now - last_update_) : TimeDuration::zero_value;
      70           0 :   const TimeDuration idle_bonus = (now > last_update_ && status_ == ThreadStatus_Idle) ? (now - last_update_) : TimeDuration::zero_value;
      71           0 :   const TimeDuration denom = total_.active_time + active_bonus + total_.idle_time + idle_bonus;
      72             : 
      73           0 :   if (!denom.is_zero()) {
      74           0 :     return (total_.active_time + active_bonus) / denom;
      75             :   }
      76           0 :   return 0;
      77           0 : }
      78             : 
      79           0 : ThreadStatusManager::ThreadId ThreadStatusManager::get_thread_id()
      80             : {
      81             : #if defined (ACE_WIN32)
      82             :   return static_cast<unsigned>(ACE_Thread::self());
      83             : #else
      84             : #  ifdef ACE_HAS_GETTID
      85           0 :   return ACE_OS::thr_gettid();
      86             : #  else
      87             :   char buffer[32];
      88             :   const size_t len = ACE_OS::thr_id(buffer, 32);
      89             :   return String(buffer, len);
      90             : #  endif
      91             : #endif /* ACE_WIN32 */
      92             : }
      93             : 
      94           9 : void ThreadStatusManager::add_thread(const String& name)
      95             : {
      96           9 :   if (!update_thread_status()) {
      97           9 :     return;
      98             :   }
      99             : 
     100           0 :   const ThreadId thread_id = get_thread_id();
     101             : 
     102           0 :   String bit_key = to_dds_string(thread_id);
     103             : 
     104           0 :   if (name.length()) {
     105           0 :     bit_key += " (" + name + ")";
     106             :   }
     107             : 
     108           0 :   if (DCPS_debug_level > 4) {
     109           0 :     ACE_DEBUG((LM_DEBUG, "(%P|%t) ThreadStatusManager::add_thread: "
     110             :                "adding thread %C\n", bit_key.c_str()));
     111             :   }
     112             : 
     113           0 :   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
     114           0 :   map_.insert(std::make_pair(thread_id, Thread(bit_key)));
     115           0 : }
     116             : 
     117        2752 : void ThreadStatusManager::active(bool nested)
     118             : {
     119        2752 :   if (!update_thread_status()) {
     120        2752 :     return;
     121             :   }
     122             : 
     123           0 :   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
     124             : 
     125           0 :   const MonotonicTimePoint m_now = MonotonicTimePoint::now();
     126           0 :   const SystemTimePoint s_now = SystemTimePoint::now();
     127           0 :   const ThreadId thread_id = get_thread_id();
     128             : 
     129           0 :   const Map::iterator pos = map_.find(thread_id);
     130           0 :   if (pos != map_.end()) {
     131           0 :     pos->second.update(m_now, s_now, Thread::ThreadStatus_Active, bucket_limit_, nested);
     132             :   }
     133             : 
     134           0 :   cleanup(m_now);
     135           0 : }
     136             : 
     137        2752 : void ThreadStatusManager::idle(bool nested)
     138             : {
     139        2752 :   if (!update_thread_status()) {
     140        2752 :     return;
     141             :   }
     142             : 
     143           0 :   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
     144             : 
     145           0 :   const MonotonicTimePoint m_now = MonotonicTimePoint::now();
     146           0 :   const SystemTimePoint s_now = SystemTimePoint::now();
     147           0 :   const ThreadId thread_id = get_thread_id();
     148             : 
     149           0 :   const Map::iterator pos = map_.find(thread_id);
     150           0 :   if (pos != map_.end()) {
     151           0 :     pos->second.update(m_now, s_now, Thread::ThreadStatus_Idle, bucket_limit_, nested);
     152             :   }
     153             : 
     154           0 :   cleanup(m_now);
     155           0 : }
     156             : 
     157           9 : void ThreadStatusManager::finished()
     158             : {
     159           9 :   if (!update_thread_status()) {
     160           9 :     return;
     161             :   }
     162             : 
     163           0 :   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
     164             : 
     165           0 :   const MonotonicTimePoint m_now = MonotonicTimePoint::now();
     166           0 :   const SystemTimePoint s_now = SystemTimePoint::now();
     167           0 :   const ThreadId thread_id = get_thread_id();
     168             : 
     169           0 :   const Map::iterator pos = map_.find(thread_id);
     170           0 :   if (pos != map_.end()) {
     171           0 :     pos->second.update(m_now, s_now, Thread::ThreadStatus_Idle, bucket_limit_, false);
     172             : 
     173           0 :     list_.push_back(pos->second);
     174           0 :     map_.erase(pos);
     175             :   }
     176             : 
     177           0 :   cleanup(m_now);
     178           0 : }
     179             : 
     180           0 : void ThreadStatusManager::harvest(const MonotonicTimePoint& start,
     181             :                                   ThreadStatusManager::List& running,
     182             :                                   ThreadStatusManager::List& finished) const
     183             : {
     184           0 :   ACE_GUARD(ACE_Thread_Mutex, g, lock_);
     185             : 
     186           0 :   for (Map::const_iterator pos = map_.begin(), limit = map_.end(); pos != limit; ++pos) {
     187           0 :     if (pos->second.last_update() > start) {
     188           0 :       running.push_back(pos->second);
     189             :     }
     190             :   }
     191             : 
     192           0 :   for (List::const_iterator pos = list_.begin(), limit = list_.end(); pos != limit; ++pos) {
     193           0 :     if (pos->last_update() > start) {
     194           0 :       finished.push_back(*pos);
     195             :     }
     196             :   }
     197           0 : }
     198             : 
     199           0 : void ThreadStatusManager::cleanup(const MonotonicTimePoint& now)
     200             : {
     201           0 :   const MonotonicTimePoint cutoff = now - 10 * thread_status_interval_;
     202             : 
     203           0 :   while (!list_.empty() && list_.front().last_update() < cutoff) {
     204           0 :     list_.pop_front();
     205             :   }
     206           0 : }
     207             : 
     208             : 
     209             : } // namespace DCPS
     210             : } // namespace OpenDDS
     211             : 
     212             : OPENDDS_END_VERSIONED_NAMESPACE_DECL