XrdPfc::ResourceMonitor Class Reference

#include <XrdPfcResourceMonitor.hh>

Collaboration diagram for XrdPfc::ResourceMonitor:

Public Member Functions
	ResourceMonitor (XrdOss &oss)
	~ResourceMonitor ()
void	CrossCheckIfScanIsInProgress (const std::string &lfn, XrdSysCondVar &cond)
void	fill_pshot_vec_children (const DirState &parent_ds, int parent_idx, std::vector< DirPurgeElement > &vec, int max_depth)
void	fill_sshot_vec_children (const DirState &parent_ds, int parent_idx, std::vector< DirStateElement > &vec, int max_depth)
void	heart_beat ()
void	init_before_main ()
void	main_thread_function ()
bool	perform_initial_scan ()
void	perform_purge_check (bool purge_cold_files, int tl)
void	perform_purge_task (DataFsPurgeshot &ps)
void	perform_purge_task_cleanup ()
int	process_queues ()
void	register_file_close (int token_id, time_t close_timestamp, const Stats &full_stats)
int	register_file_open (const std::string &filename, time_t open_timestamp, bool existing_file)
void	register_file_purge (const std::string &filename, long long size_in_st_blocks)
void	register_file_purge (DirState *target, long long size_in_st_blocks)
void	register_file_update_stats (int token_id, const Stats &stats)
void	register_multi_file_purge (const std::string &target, long long size_in_st_blocks, int n_files)
void	register_multi_file_purge (DirState *target, long long size_in_st_blocks, int n_files)
void	scan_dir_and_recurse (FsTraversal &fst)
AccessToken &	token (int i)
void	update_vs_and_file_usage_info ()

Public Attributes
bool	m_purge_task_active {false}
bool	m_purge_task_complete {false}
XrdSysCondVar	m_purge_task_cond {0}
time_t	m_purge_task_end {0}
time_t	m_purge_task_start {0}

Detailed Description

Definition at line 38 of file XrdPfcResourceMonitor.hh.

Constructor & Destructor Documentation

ResourceMonitor()

ResourceMonitor::ResourceMonitor

(

XrdOss &

oss

)

Definition at line 32 of file XrdPfcResourceMonitor.cc.

                                            :
   m_fs_state(* new DataFsState),
   m_oss(oss)
{}

~ResourceMonitor()

ResourceMonitor::~ResourceMonitor

(

)

Definition at line 37 of file XrdPfcResourceMonitor.cc.

{
   delete &m_fs_state;
}

Member Function Documentation

CrossCheckIfScanIsInProgress()

void ResourceMonitor::CrossCheckIfScanIsInProgress	(	const std::string &	lfn,
		XrdSysCondVar &	cond )

Definition at line 46 of file XrdPfcResourceMonitor.cc.

{
   m_dir_scan_mutex.Lock();
   if (m_dir_scan_in_progress) {
      bool dir_checked = false;
      m_dir_scan_open_requests.push_back({lfn, cond, dir_checked});
      cond.Lock();
      m_dir_scan_mutex.UnLock();
      while ( ! dir_checked)
         cond.Wait();
      cond.UnLock();
   } else {
      m_dir_scan_mutex.UnLock();
   }
}

References XrdSysCondVar::Lock(), XrdSysCondVar::UnLock(), and XrdSysCondVar::Wait().

Here is the call graph for this function:

fill_pshot_vec_children()

void ResourceMonitor::fill_pshot_vec_children	(	const DirState &	parent_ds,
		int	parent_idx,
		std::vector< DirPurgeElement > &	vec,
		int	max_depth )

Definition at line 535 of file XrdPfcResourceMonitor.cc.

{
   const int pos0       = (int) vec.size();
   const int n_children = (int) parent_ds.m_subdirs.size();
 
   // maxdepth 0 is the root alone: a node at max_depth is emitted, its children are not.
   if (parent_ds.m_depth >= max_depth)
   {
      // Truncated here: keep the daughter range at (-1, -1). No children: empty range.
      if (n_children == 0) {
         DirPurgeElement &dse = vec[parent_idx];
         dse.m_daughters_begin = dse.m_daughters_end = pos0;
      }
      return;
   }
 
   {  // scoped: the emplace_back loop below can reallocate vec
      DirPurgeElement &parent_dse = vec[parent_idx];
      parent_dse.m_daughters_begin = pos0;
      parent_dse.m_daughters_end   = pos0 + n_children;
   }
 
   if (n_children == 0) return;
 
   for (auto const & [name, child] : parent_ds.m_subdirs)
   {
      vec.emplace_back( DirPurgeElement(child, child.m_here_usage, child.m_recursive_subdir_usage, parent_idx) );
   }
 
   int pos = pos0;
   for (auto const & [name, child] : parent_ds.m_subdirs)
   {
      fill_pshot_vec_children(child, pos, vec, max_depth);
      ++pos;
   }
}

References child(), fill_pshot_vec_children(), XrdPfc::DirPurgeElement::m_daughters_begin, XrdPfc::DirPurgeElement::m_daughters_end, XrdPfc::DirState::m_depth, and XrdPfc::DirState::m_subdirs.

Referenced by fill_pshot_vec_children(), and perform_purge_check().

Here is the call graph for this function:

Here is the caller graph for this function:

fill_sshot_vec_children()

void ResourceMonitor::fill_sshot_vec_children	(	const DirState &	parent_ds,
		int	parent_idx,
		std::vector< DirStateElement > &	vec,
		int	max_depth )

Definition at line 495 of file XrdPfcResourceMonitor.cc.

{
   const int pos0       = (int) vec.size();
   const int n_children = (int) parent_ds.m_subdirs.size();
 
   // maxdepth 0 is the root alone: a node at max_depth is emitted, its children are not.
   if (parent_ds.m_depth >= max_depth)
   {
      // Truncated here: keep the daughter range at (-1, -1). No children: empty range.
      if (n_children == 0) {
         DirStateElement &dse = vec[parent_idx];
         dse.m_daughters_begin = dse.m_daughters_end = pos0;
      }
      return;
   }
 
   {  // scoped: the emplace_back loop below can reallocate vec
      DirStateElement &parent_dse = vec[parent_idx];
      parent_dse.m_daughters_begin = pos0;
      parent_dse.m_daughters_end   = pos0 + n_children;
   }
 
   if (n_children == 0) return;
 
   for (auto const & [name, child] : parent_ds.m_subdirs)
   {
      vec.emplace_back( DirStateElement(child, parent_idx) );
   }
 
   int pos = pos0;
   for (auto const & [name, child] : parent_ds.m_subdirs)
   {
      fill_sshot_vec_children(child, pos, vec, max_depth);
      ++pos;
   }
}

References child(), fill_sshot_vec_children(), XrdPfc::DirStateElement::m_daughters_begin, XrdPfc::DirStateElement::m_daughters_end, XrdPfc::DirState::m_depth, and XrdPfc::DirState::m_subdirs.

Referenced by fill_sshot_vec_children(), and heart_beat().

Here is the call graph for this function:

Here is the caller graph for this function:

heart_beat()

void ResourceMonitor::heart_beat

(

)

Definition at line 343 of file XrdPfcResourceMonitor.cc.

{
   static const char *tpfx = "heart_beat() ";
 
   const Configuration &conf = Cache::Conf();
 
   const int s_queue_proc_interval   = 10;
   const int s_sshot_report_interval = conf.m_dirStatsInterval; // one of 60, 120, 300,
                                                                //  600, 900, 1200, 1800, 3600 s
   const int s_purge_check_interval  = 60;
   const int s_purge_report_interval = conf.m_purgeInterval;
   const int s_purge_cold_files_interval = conf.m_purgeInterval * conf.m_purgeAgeBasedPeriod;
 
   // initial scan performed as part of config
 
   time_t now = time(0);
   time_t next_queue_proc_time       = now + s_queue_proc_interval;
   time_t next_sshot_report_time     = (now / 60) * 60 + 60; // at next full minute
   time_t next_purge_check_time      = now + s_purge_check_interval;
   time_t next_purge_report_time     = now + s_purge_report_interval;
   time_t next_purge_cold_files_time = now + s_purge_cold_files_interval;
 
   long long n_reported_token_failures = 0;
 
   while (true)
   {
      time_t start = time(0);
      time_t next_event = std::min({ next_queue_proc_time, next_sshot_report_time,
                                     next_purge_check_time, next_purge_report_time, next_purge_cold_files_time });
 
      if (next_event > start)
      {
         unsigned int t_sleep = next_event - start;
         TRACE(Dump, tpfx << "sleeping for " << t_sleep << " seconds until the next beat.");
         sleep(t_sleep);
      }
 
      // Check if purge has been running and has completed yet.
      // For now this is only used to prevent removal of empty leaf directories
      // during stat propagation so we do not need to wait for the condition in
      // the above sleep.
      if (m_purge_task_active) {
         MutexHolder _lck(m_purge_task_cond);
         if (m_purge_task_complete) {
            m_purge_task_active = m_purge_task_complete = false;
         }
      }
 
      time_t queue_swap_time = time(0);
 
      // Always process the queues.
      int n_processed = process_queues();
      next_queue_proc_time = queue_swap_time + s_queue_proc_interval;
      TRACE(Dump, tpfx << "process_queues -- n_records=" << n_processed);
 
      {  // Access-token space exhausted -- should be unreachable, so say so loudly.
         long long taf;
         { XrdSysMutexHelper _lock(&m_queue_mutex); taf = m_token_alloc_failures; }
         if (taf > n_reported_token_failures) {
            TRACE(Error, tpfx << "access-token space exhausted "
                  << taf - n_reported_token_failures << " time(s) (total " << taf
                  << "); those files are open but not accounted for. Raise MaxChunks "
                     "on ResourceMonitor::m_access_tokens.");
            n_reported_token_failures = taf;
         }
      }
 
      // Always update basic info on m_fs_state (space, usage, file_usage).
      update_vs_and_file_usage_info();
 
      now = time(0);
 
      // Make planning for fs_state_update, sshot dump and purge task.
      // Second two require the first, so figure out what is going to happen.
      bool do_sshot_report     = next_sshot_report_time <= now;
      bool do_purge_check      = next_purge_check_time <= now;
      bool do_purge_report     = next_purge_report_time <= now;
      bool do_purge_cold_files = next_purge_cold_files_time <= now;
 
      // Update stats in usages if any secondary activity will happen.
      if (do_sshot_report || do_purge_check || do_purge_report || do_purge_cold_files)
      {
         unlink_func unlink_foo = [&](const std::string &dp)->int {
            int ret = m_oss.Unlink(dp.c_str());
            if (ret != 0) {
               TRACE(Info, tpfx << "Empty dir unlink error: " << ret << " at " << dp);
            } else {
               TRACE(Debug, tpfx << "Empty dir unlink success: " << dp);
            }
            return ret;
         };
 
         // Potentially prune the empty leaf dirs even less frequently, once per hour, maybe?
         bool purge_leaf_dirs = do_sshot_report && ! m_purge_task_active;
         m_fs_state.update_stats_and_usages(queue_swap_time, purge_leaf_dirs, unlink_foo);
 
         // This reporting into log/stdout is to be removed.
         // Meaning of conf.is_dir_stat_reporting_on() etc is to be clarified / improved.
         if (do_sshot_report && conf.is_dir_stat_reporting_on())
         {
            const int store_depth  =  conf.m_dirStatsStoreDepth;
         #ifdef RM_DEBUG
            const DirState      &root_ds = *m_fs_state.get_root();
            dprintf("Snapshot n_dirs=%d, total n_dirs=%d\n", root_ds.count_dirs_to_level(store_depth),
                  root_ds.m_here_usage.m_NDirectories + root_ds.m_recursive_subdir_usage.m_NDirectories + 1);
         #endif
            m_fs_state.dump_recursively(store_depth);
         }
 
         m_fs_state.reset_stats(queue_swap_time);
      }
 
      if (do_sshot_report)
      {
         // Sshot reports are equidistant, at "full" reporting interval.
         next_sshot_report_time = ((now + 1) / s_sshot_report_interval) * s_sshot_report_interval + s_sshot_report_interval;
 
         // This should dump out binary snapshot into /pfc-stats/, if so configured.
 
         // json dump to std::out for debug purpose
         DataFsSnapshot ss(m_fs_state, m_fs_state.m_sshot_stats_reset_time);
         const DirState &root_ds = *m_fs_state.get_root();
         const int store_depth  =  conf.m_dirStatsStoreDepth;
         const int n_sshot_dirs = root_ds.count_dirs_to_level(store_depth);
         ss.m_dir_states.reserve(n_sshot_dirs);
         ss.m_dir_states.emplace_back( DirStateElement(root_ds, -1) );
         fill_sshot_vec_children(root_ds, 0, ss.m_dir_states, store_depth);
 
         // This should really be export to a file (preferably binary, but then bin->json command is needed, too).
         // ss.dump();
 
         const char* dumpfile = "/pfc-stats/DirStat.json";
         ss.write_json_file(dumpfile, m_oss, false);
         m_fs_state.reset_sshot_stats(queue_swap_time);
      }
 
      if (do_purge_check || do_purge_report || do_purge_cold_files)
      {
         perform_purge_check(do_purge_cold_files, do_purge_report ? TRACE_Info : TRACE_Debug);
 
         next_purge_check_time = now + s_purge_check_interval;
         if (do_purge_report) next_purge_report_time = now + s_purge_report_interval;
         if (do_purge_cold_files) next_purge_cold_files_time = now + s_purge_cold_files_interval;
      }
 
   } // end while forever
}

References XrdPfc::Cache::Conf(), XrdPfc::DirState::count_dirs_to_level(), Debug, dprintf, Error, fill_sshot_vec_children(), XrdPfc::Configuration::is_dir_stat_reporting_on(), XrdPfc::DataFsSnapshot::m_dir_states, XrdPfc::Configuration::m_dirStatsInterval, XrdPfc::Configuration::m_dirStatsStoreDepth, XrdPfc::DirState::m_here_usage, XrdPfc::DirUsage::m_NDirectories, m_purge_task_active, m_purge_task_complete, m_purge_task_cond, XrdPfc::Configuration::m_purgeAgeBasedPeriod, XrdPfc::Configuration::m_purgeInterval, XrdPfc::DirState::m_recursive_subdir_usage, perform_purge_check(), process_queues(), TRACE, TRACE_Debug, TRACE_Info, update_vs_and_file_usage_info(), and XrdPfc::DataFsSnapshot::write_json_file().

Referenced by main_thread_function().

Here is the call graph for this function:

Here is the caller graph for this function:

init_before_main()

void ResourceMonitor::init_before_main

(

)

Definition at line 870 of file XrdPfcResourceMonitor.cc.

{
   // setup for in-scan -- this is called from initial setup.
   MutexHolder _lck(m_dir_scan_mutex);
   m_dir_scan_in_progress = true;
}

main_thread_function()

void ResourceMonitor::main_thread_function

(

)

Definition at line 877 of file XrdPfcResourceMonitor.cc.

{
   const char *tpfx = "main_thread_function ";
   {
      time_t is_start = time(0);
      m_fs_state.init_stat_reset_times(is_start);
      TRACE(Info, tpfx << "Stating initial directory scan.");
 
      if ( ! perform_initial_scan()) {
         TRACE(Error, tpfx << "Initial directory scan has failed. This is a terminal error, aborting.")
         _exit(1);
      }
      // Reset of m_dir_scan_in_progress is done in perform_initial_scan()
 
      time_t is_duration = time(0) - is_start;
      TRACE(Info, tpfx << "Initial directory scan complete, duration=" << is_duration <<"s");
 
      // run first process queues
      int n_proc_is = process_queues();
      TRACE(Info, tpfx << "First process_queues finished, n_records=" << n_proc_is);
 
      // shrink queues if scan time was longer than 30s.
      if (is_duration > 30 || n_proc_is > 3000)
      {
         m_file_open_q.shrink_read_queue();
         m_file_update_stats_q.shrink_read_queue();
         m_file_close_q.shrink_read_queue();
         m_file_purge_q1.shrink_read_queue();
         m_file_purge_q2.shrink_read_queue();
         m_file_purge_q3.shrink_read_queue();
      }
   }
   heart_beat();
}

References Error, heart_beat(), perform_initial_scan(), process_queues(), and TRACE.

Referenced by ResourceMonitorThread().

Here is the call graph for this function:

Here is the caller graph for this function:

perform_initial_scan()

bool ResourceMonitor::perform_initial_scan

(

)

Definition at line 174 of file XrdPfcResourceMonitor.cc.

{
   // Called after PFC configuration is complete, but before full startup of the daemon.
   // Base line usages are accumulated as part of the file-system, traversal.
 
   update_vs_and_file_usage_info();
 
   DirState   *root_ds = m_fs_state.get_root();
   FsTraversal fst(m_oss);
   fst.m_protected_top_dirs.insert("pfc-stats"); // XXXX This should come from config. Also: N2N?
 
   if ( ! fst.begin_traversal(root_ds, "/"))
      return false;
 
   // The following are initialized in ResourceMonitor.hh to avoid a race at startup:
   //   m_dir_scan_in_progress = true;
   //   m_dir_scan_check_counter = 0;
 
   scan_dir_and_recurse(fst);
 
   fst.end_traversal();
 
   // We have all directories scanned, available in DirState tree, let all remaining files go
   // and then we shall do the upward propagation of usages.
   {
      XrdSysMutexHelper _lock(m_dir_scan_mutex);
      m_dir_scan_in_progress = false;
      m_dir_scan_check_counter = 0;
   }
   // m_dir_scan_open_requests should now be final, ie, no new entries will be added.
   while ( ! m_dir_scan_open_requests.empty())
   {
      LfnCondRecord &lcr = m_dir_scan_open_requests.front();
      lcr.f_cond.Lock();
      lcr.f_checked = true;
      lcr.f_cond.Signal();
      lcr.f_cond.UnLock();
 
      m_dir_scan_open_requests.pop_front();
   }
 
   // Do upward propagation of usages.
   root_ds->upward_propagate_initial_scan_usages();
   m_current_usage_in_st_blocks = root_ds->m_here_usage.m_StBlocks + 
                                  root_ds->m_recursive_subdir_usage.m_StBlocks;
   update_vs_and_file_usage_info();
 
   return true;
}

References XrdPfc::FsTraversal::begin_traversal(), XrdPfc::FsTraversal::end_traversal(), XrdSysCondVar::Lock(), XrdPfc::DirState::m_here_usage, XrdPfc::FsTraversal::m_protected_top_dirs, XrdPfc::DirState::m_recursive_subdir_usage, XrdPfc::DirUsage::m_StBlocks, scan_dir_and_recurse(), XrdSysCondVar::Signal(), XrdSysCondVar::UnLock(), update_vs_and_file_usage_info(), and XrdPfc::DirState::upward_propagate_initial_scan_usages().

Referenced by main_thread_function().

Here is the call graph for this function:

Here is the caller graph for this function:

perform_purge_check()

void ResourceMonitor::perform_purge_check	(	bool	purge_cold_files,
		int	tl )

Definition at line 711 of file XrdPfcResourceMonitor.cc.

{
   static const char *trc_pfx = "perform_purge_check() ";
   const Configuration &conf = Cache::Conf();
 
   std::unique_ptr<DataFsPurgeshot> psp( new DataFsPurgeshot(m_fs_state) );
   DataFsPurgeshot &ps = *psp;
 
   ps.m_file_usage = 512ll * m_current_usage_in_st_blocks;
   // These are potentially wrong as cache might be writing over preallocated byte ranges.
   ps.m_estimated_writes_from_writeq = Cache::GetInstance().WritesSinceLastCall();
   // Can have another estimate based on eiter writes or st-blocks from purge-stats, once we have them.
   
   TRACE_INT(tl, trc_pfx << "Purge check:");
 
   ps.m_bytes_to_remove = 0;
   if (conf.are_file_usage_limits_set())
   {
      ps.m_bytes_to_remove = get_file_usage_bytes_to_remove(ps, ps.m_estimated_writes_from_writeq, tl);
   }
   else
   {
      if (ps.m_disk_used > conf.m_diskUsageHWM)
      {
         TRACE_INT(tl, "Disk usage: " <<  ps.m_disk_used << " exceed highWatermark.");
         ps.m_bytes_to_remove = ps.m_disk_used - conf.m_diskUsageLWM;
      }
   }
 
   ps.m_space_based_purge = ps.m_bytes_to_remove ? 1 : 0;
 
   // Purge precheck -- check if age-based purge is required
   // We ignore uvkeep time, it requires reading of cinfo files and it is enforced in File::Open() anyway.
 
   if (purge_cold_files && conf.is_age_based_purge_in_effect()) // || conf.is_uvkeep_purge_in_effect())
   {
      ps.m_age_based_purge = true;
   }
 
   TRACE_INT(tl, "\tbytes_to_remove   = " << ps.m_bytes_to_remove   << " B");
   TRACE_INT(tl, "\tspace_based_purge = " << ps.m_space_based_purge);
   TRACE_INT(tl, "\tage_based_purge   = " << ps.m_age_based_purge);
 
   bool periodic = Cache::GetInstance().GetPurgePin() ? 
                   Cache::GetInstance().GetPurgePin()->CallPeriodically() : false;
 
   if ( ! ps.m_space_based_purge && ! ps.m_age_based_purge && !periodic ) {
      TRACE(Info, trc_pfx << "purge not required.");
      Cache::GetInstance().ClearPurgeProtectedSet();
      return;
   }
   if (m_purge_task_active) {
      TRACE(Warning, trc_pfx << "purge required but previous purge task is still active!");
      return;
   }
 
   TRACE(Info, trc_pfx << "scheduling purge task.");
 
   // At this point we have all the information: report, decide on action.
   // There is still some missing infrastructure, especially as regards to purge-plugin:
   // - at what point do we start bugging the pu-pin to start coughing up purge lists?
   //   - have a new parameter or just do it "one cycle before full"?
   //   - what if it doesn't -- when do we do the old-stlye scan & purge?
   // - how do we do age-based purge and uvkeep purge?
   //   - they are really quite different -- and could run separately, registering
   //     files into a purge-candidate list. This has to be rechecked before the actual
   //     deletion -- eg, by comparing stat time of cinfo + doing the is-active / is-purge-protected.
 
   const DirState &root_ds = *m_fs_state.get_root();
 
   // Size from the tree itself; the usage counters can be off (#2808).
   const int n_pshot_dirs = root_ds.count_dirs_to_level(9999);
 
   // Report a counters-vs-tree mismatch when it appears or changes.
   const int n_calc_dirs = 1 + root_ds.m_here_usage.m_NDirectories
                             + root_ds.m_recursive_subdir_usage.m_NDirectories;
   dprintf("purge dir count recursive=%d vs from_usage=%d\n", n_pshot_dirs, n_calc_dirs);
   const int discrepancy = n_calc_dirs - n_pshot_dirs;
   if (discrepancy != m_dir_count_discrepancy)
   {
      if (discrepancy != 0) {
         TRACE(Error, trc_pfx << "directory count mismatch: usage counters " << n_calc_dirs
               << ", tree " << n_pshot_dirs << ", previous difference " << m_dir_count_discrepancy);
      } else {
         TRACE(Info, trc_pfx << "directory count consistent again: " << n_pshot_dirs);
      }
      m_dir_count_discrepancy = discrepancy;
   }
 
   ps.m_dir_vec.reserve(n_pshot_dirs);
   ps.m_dir_vec.emplace_back( DirPurgeElement(root_ds, root_ds.m_here_usage, root_ds.m_recursive_subdir_usage, -1) );
   fill_pshot_vec_children(root_ds, 0, ps.m_dir_vec, 9999);
 
   m_purge_task_active = true;
 
   struct PurgeDriverJob : public XrdJob
   {
      DataFsPurgeshot  *m_purge_shot_ptr;
 
      PurgeDriverJob(DataFsPurgeshot *psp) :
         XrdJob("XrdPfc::ResourceMonitor::PurgeDriver"),
         m_purge_shot_ptr(psp)
      {}
 
      void DoIt() override
      {
         Cache::ResMon().perform_purge_task(*m_purge_shot_ptr);
         Cache::ResMon().perform_purge_task_cleanup();
 
         delete m_purge_shot_ptr;
         delete this;
      }
   };
 
   Cache::schedP->Schedule( new PurgeDriverJob(psp.release()) );
}

References XrdPfc::Configuration::are_file_usage_limits_set(), XrdPfc::PurgePin::CallPeriodically(), XrdPfc::Cache::ClearPurgeProtectedSet(), XrdPfc::Cache::Conf(), XrdPfc::DirState::count_dirs_to_level(), DoIt(), dprintf, Error, fill_pshot_vec_children(), XrdPfc::Cache::GetInstance(), XrdPfc::Cache::GetPurgePin(), XrdPfc::Configuration::is_age_based_purge_in_effect(), XrdPfc::DataFsPurgeshot::m_age_based_purge, XrdPfc::DataFsPurgeshot::m_bytes_to_remove, XrdPfc::DataFsPurgeshot::m_dir_vec, XrdPfc::DataFsStateBase::m_disk_used, XrdPfc::Configuration::m_diskUsageHWM, XrdPfc::Configuration::m_diskUsageLWM, XrdPfc::DataFsPurgeshot::m_estimated_writes_from_writeq, XrdPfc::DataFsStateBase::m_file_usage, XrdPfc::DirState::m_here_usage, XrdPfc::DirUsage::m_NDirectories, m_purge_task_active, XrdPfc::DirState::m_recursive_subdir_usage, XrdPfc::DataFsPurgeshot::m_space_based_purge, perform_purge_task(), perform_purge_task_cleanup(), XrdPfc::Cache::ResMon(), XrdPfc::Cache::schedP, TRACE, TRACE_INT, and XrdPfc::Cache::WritesSinceLastCall().

Referenced by heart_beat().

Here is the call graph for this function:

Here is the caller graph for this function:

perform_purge_task()

void ResourceMonitor::perform_purge_task

(

DataFsPurgeshot &

ps

)

Definition at line 833 of file XrdPfcResourceMonitor.cc.

{
   // BEWARE: Runs in a dedicated thread - is only to communicate back to the
   // hear_beat() / data structs via the purge queues and condition variable.
 
   // const char *tpfx = "perform_purge_task ";
 
   {
      MutexHolder _lck(m_purge_task_cond);
      m_purge_task_start = time(0);
   }
 
   // For now, fall back to the old purge ... to be improved with:
   // - new scan, following the DataFsPurgeshot;
   // - usage of cinfo stat mtime for time of last access (touch already done at output);
   // - use DirState* to report back purged files.
   // Already changed to report back purged files --- but using the string / path variant.
   OldStylePurgeDriver(ps); // In XrdPfcPurge.cc
}

References m_purge_task_cond, m_purge_task_start, and XrdPfc::OldStylePurgeDriver().

Referenced by perform_purge_check().

Here is the call graph for this function:

Here is the caller graph for this function:

perform_purge_task_cleanup()

void ResourceMonitor::perform_purge_task_cleanup

(

)

Definition at line 853 of file XrdPfcResourceMonitor.cc.

{
   // Separated out so the purge_task can exit without post-checks.
 
   {
      MutexHolder _lck(m_purge_task_cond);
      m_purge_task_end = time(0);
      m_purge_task_complete = true;
      m_purge_task_cond.Signal();
   }
   Cache::GetInstance().ClearPurgeProtectedSet();
}

References XrdPfc::Cache::ClearPurgeProtectedSet(), XrdPfc::Cache::GetInstance(), m_purge_task_complete, m_purge_task_cond, and m_purge_task_end.

Referenced by perform_purge_check().

Here is the call graph for this function:

Here is the caller graph for this function:

process_queues()

int ResourceMonitor::process_queues

(

)

Definition at line 228 of file XrdPfcResourceMonitor.cc.

{
   static const char *trc_pfx = "process_queues() ";
 
   // Assure that we pick up only entries that are present now.
   // We really want all open records to be processed before file-stats updates
   // and all those before the close records.
   // Purges are sort of tangential as they really just modify bytes / number
   // of files in a directory and do not deal with any persistent file id tokens.
 
   int n_records = 0;
   {
      XrdSysMutexHelper _lock(&m_queue_mutex);
      n_records += m_file_open_q.swap_queues();
      n_records += m_file_update_stats_q.swap_queues();
      n_records += m_file_close_q.swap_queues();
      n_records += m_file_purge_q1.swap_queues();
      n_records += m_file_purge_q2.swap_queues();
      n_records += m_file_purge_q3.swap_queues();
      ++m_queue_swap_u1;
   }
 
   for (auto &i : m_file_open_q.read_queue())
   {
      // i.id: LFN, i.record: OpenRecord
      AccessToken &at = token(i.id);
      dprintf("process file open for token %d, time %ld -- %s\n",
              i.id, i.record.m_open_time, at.m_filename.c_str());
 
      // Resolve fname into DirState.
      // We could clear the filename after this ... or keep it, should we need it later on.
      // For now it is just used for debug printouts.
      DirState *last_existing_ds = nullptr;
      DirState *ds = m_fs_state.find_dirstate_for_lfn(at.m_filename, &last_existing_ds);
      at.m_dir_state = ds;
      note_file_open(ds, last_existing_ds, i.record.m_existing_file);
 
      ds->m_here_usage.m_LastOpenTime = i.record.m_open_time;
   }
 
   for (auto &i : m_file_update_stats_q.read_queue())
   {
      // i.id: token, i.record: Stats
      AccessToken &at = token(i.id);
      // Stats
      DirState *ds = at.m_dir_state;
      dprintf("process file update for token %d, %p -- %s\n",
             i.id, ds, at.m_filename.c_str());
 
      ds->m_here_stats.AddUp(i.record);
      m_current_usage_in_st_blocks += i.record.m_StBlocksAdded;
   }
 
   for (auto &i : m_file_close_q.read_queue())
   {
      // i.id: token, i.record: CloseRecord
      AccessToken &at = token(i.id);
      dprintf("process file close for token %d, time %ld -- %s\n",
              i.id, i.record.m_close_time, at.m_filename.c_str());
 
      DirState *ds = at.m_dir_state;
      ds->m_here_stats.m_NFilesClosed += 1;
      ds->m_here_usage.m_LastCloseTime = i.record.m_close_time;
 
   }
   { // Under the lock: clear() writes fields register_file_update_stats() uses.
      XrdSysMutexHelper _lock(&m_queue_mutex);
      for (auto &i : m_file_close_q.read_queue())
         m_access_tokens.release(i.id);
   }
 
   for (auto &i : m_file_purge_q1.read_queue())
   {
      // i.id: DirState*, i.record: PurgeRecord
      DirState *ds = i.id;
      ds->m_here_stats.m_StBlocksRemoved += i.record.m_size_in_st_blocks;
      ds->m_here_stats.m_NFilesRemoved   += i.record.m_n_files;
      m_current_usage_in_st_blocks       -= i.record.m_size_in_st_blocks;
   }
   for (auto &i : m_file_purge_q2.read_queue())
   {
      // i.id: directory-path, i.record: PurgeRecord
      DirState *ds = m_fs_state.get_root()->find_path(i.id, -1, false, false);
      if ( ! ds) {
         TRACE(Error, trc_pfx << "DirState not found for directory path '" << i.id << "'.");
         // find_path can return the last dir found ... but this clearly isn't a valid purge record.
         continue;
      }
      ds->m_here_stats.m_StBlocksRemoved += i.record.m_size_in_st_blocks;
      ds->m_here_stats.m_NFilesRemoved   += i.record.m_n_files;
      m_current_usage_in_st_blocks       -= i.record.m_size_in_st_blocks;
   }
   for (auto &i : m_file_purge_q3.read_queue())
   {
      // i.id: LFN, i.record: size of file in st_blocks
      DirState *ds = m_fs_state.get_root()->find_path(i.id, -1, true, false);
      if ( ! ds) {
         TRACE(Error, trc_pfx << "DirState not found for LFN path '" << i.id << "'.");
         continue;
      }
      ds->m_here_stats.m_StBlocksRemoved += i.record;
      ds->m_here_stats.m_NFilesRemoved   += 1;
      m_current_usage_in_st_blocks       -= i.record;
   }
 
   // Read queues / vectors are cleared at swap time.
   // We might consider reducing their capacity by half if, say, their usage is below 25%.
 
   return n_records;
}

References XrdPfc::DirStats::AddUp(), dprintf, Error, XrdPfc::DirState::find_path(), XrdPfc::DirState::m_here_stats, XrdPfc::DirState::m_here_usage, XrdPfc::DirUsage::m_LastCloseTime, XrdPfc::DirUsage::m_LastOpenTime, XrdPfc::DirStats::m_NFilesClosed, XrdPfc::DirStats::m_NFilesRemoved, XrdPfc::DirStats::m_StBlocksRemoved, XrdPfc::note_file_open(), token(), and TRACE.

Referenced by heart_beat(), and main_thread_function().

Here is the call graph for this function:

Here is the caller graph for this function:

register_file_close()

void XrdPfc::ResourceMonitor::register_file_close ( int token_id, time_t close_timestamp, const Stats & full_stats )

inline

Definition at line 190 of file XrdPfcResourceMonitor.hh.

                                                                                           {
      if (token_id < 0) return;   // register_file_open() could not get a token
      XrdSysMutexHelper _lock(&m_queue_mutex);
      m_file_close_q.push(token_id, {close_timestamp, full_stats});
   }

register_file_open()

int XrdPfc::ResourceMonitor::register_file_open ( const std::string & filename, time_t open_timestamp, bool existing_file )

inline

Definition at line 154 of file XrdPfcResourceMonitor.hh.

                                                                                                {
      // Simply return a token, we will resolve it in the actual processing of the queue.
      XrdSysMutexHelper _lock(&m_queue_mutex);
      int token_id = m_access_tokens.acquire();
      if (token_id < 0) {
         // Unreachable in practice; heart_beat() reports the count.
         ++m_token_alloc_failures;
         return -1;
      }
      AccessToken &at = m_access_tokens[token_id];
      at.m_filename           = filename;
      at.m_last_queue_swap_u1 = m_queue_swap_u1 - 1;
 
      m_file_open_q.push(token_id, {open_timestamp, existing_file});
      return token_id;
   }

register_file_purge() [1/2]

void XrdPfc::ResourceMonitor::register_file_purge ( const std::string & filename, long long size_in_st_blocks )

inline

Definition at line 210 of file XrdPfcResourceMonitor.hh.

                                                                                    {
      XrdSysMutexHelper _lock(&m_queue_mutex);
      m_file_purge_q3.push(filename, size_in_st_blocks);
   }

register_file_purge() [2/2]

void XrdPfc::ResourceMonitor::register_file_purge ( DirState * target, long long size_in_st_blocks )

inline

Definition at line 198 of file XrdPfcResourceMonitor.hh.

                                                                           {
      XrdSysMutexHelper _lock(&m_queue_mutex);
      m_file_purge_q1.push(target, {size_in_st_blocks, 1});
   }

register_file_update_stats()

void XrdPfc::ResourceMonitor::register_file_update_stats ( int token_id, const Stats & stats )

inline

Definition at line 171 of file XrdPfcResourceMonitor.hh.

                                                                     {
      if (token_id < 0) return;   // register_file_open() could not get a token
      XrdSysMutexHelper _lock(&m_queue_mutex);
      AccessToken &at = token(token_id);
      // Check if this is the first update within this queue swap cycle.
      if (at.m_last_queue_swap_u1 != m_queue_swap_u1) {
         m_file_update_stats_q.push(token_id, stats);
         at.m_last_queue_swap_u1 = m_queue_swap_u1;
         at.m_last_write_queue_pos = m_file_update_stats_q.write_queue_size() - 1;
      } else {
         Stats &existing_stats = m_file_update_stats_q.write_record(at.m_last_write_queue_pos);
         existing_stats.AddUp(stats);
      }
      // Optionally, one could return "scaler" to moodify stat-reporting
      // frequency in the file ... if it comes too often or too rarely.
      // See also the logic for determining reporting interval (in N_bytes_read)
      // in File::Open().
   }

References XrdPfc::Stats::AddUp(), and token().

Here is the call graph for this function:

register_multi_file_purge() [1/2]

void XrdPfc::ResourceMonitor::register_multi_file_purge ( const std::string & target, long long size_in_st_blocks, int n_files )

inline

Definition at line 206 of file XrdPfcResourceMonitor.hh.

                                                                                                     {
      XrdSysMutexHelper _lock(&m_queue_mutex);
      m_file_purge_q2.push(target, {size_in_st_blocks, n_files});
   }

register_multi_file_purge() [2/2]

void XrdPfc::ResourceMonitor::register_multi_file_purge ( DirState * target, long long size_in_st_blocks, int n_files )

inline

Definition at line 202 of file XrdPfcResourceMonitor.hh.

                                                                                              {
      XrdSysMutexHelper _lock(&m_queue_mutex);
      m_file_purge_q1.push(target, {size_in_st_blocks, n_files});
   }

scan_dir_and_recurse()

void ResourceMonitor::scan_dir_and_recurse

(

FsTraversal &

fst

)

Definition at line 121 of file XrdPfcResourceMonitor.cc.

{
   dprintf("In scan_dir_and_recurse for '%s', size of dir_vec = %d, file_stat_map = %d\n",
           fst.m_current_path.c_str(),
          (int)fst.m_current_dirs.size(), (int)fst.m_current_files.size());
 
   // Breadth first, accumulate into "here", unless it was already scanned via an
   // OOB open file request.
   if ( ! fst.m_dir_state->m_scanned)
   {
      DirUsage &here = fst.m_dir_state->m_here_usage;
      for (auto it = fst.m_current_files.begin(); it != fst.m_current_files.end(); ++it)
      {
         dprintf("would be doing something with %s ... has_data=%d, has_cinfo=%d\n",
               it->first.c_str(), it->second.has_data, it->second.has_cinfo);
 
         // XXX Make some of these optional?
         // Remove files that do not have both cinfo and data?
         // Remove empty directories before even descending?
         // Leave this for some consistency pass?
         // Note that FsTraversal supports ignored paths ... some details (config, N2N) to be clarified.
 
         if (it->second.has_data && it->second.has_cinfo) {
            here.m_StBlocks += it->second.stat_data.st_blocks;
            here.m_NFiles   += 1;
         }
      }
      fst.m_dir_state->m_scanned = true;
   }
 
   // Swap-out directories as inter_dir_scan can use the FsTraversal.
   std::vector<std::string> dirs;
   dirs.swap(fst.m_current_dirs);
 
   if (++m_dir_scan_check_counter >= 100)
   {
      process_inter_dir_scan_open_requests(fst);
      m_dir_scan_check_counter = 0;
   }
 
   // Descend into sub-dirs, do not accumulate into recursive_subdir_usage yet. This is done
   // in a separate pass to allow for proper accounting of files being opened during the initial scan.
   for (auto &dname : dirs)
   {
      if (fst.cd_down(dname))
      {
         scan_dir_and_recurse(fst);
         fst.cd_up();
      }
      // XXX else try to remove it?
   }
}

References XrdPfc::FsTraversal::cd_down(), XrdPfc::FsTraversal::cd_up(), dprintf, XrdPfc::FsTraversal::m_current_dirs, XrdPfc::FsTraversal::m_current_files, XrdPfc::FsTraversal::m_current_path, XrdPfc::FsTraversal::m_dir_state, XrdPfc::DirState::m_here_usage, XrdPfc::DirUsage::m_NFiles, XrdPfc::DirState::m_scanned, XrdPfc::DirUsage::m_StBlocks, and scan_dir_and_recurse().

Referenced by perform_initial_scan(), and scan_dir_and_recurse().

Here is the call graph for this function:

Here is the caller graph for this function:

token()

AccessToken & XrdPfc::ResourceMonitor::token ( int i )

inline

Definition at line 223 of file XrdPfcResourceMonitor.hh.

{ return m_access_tokens[i]; }

Referenced by process_queues(), and register_file_update_stats().

Here is the caller graph for this function:

update_vs_and_file_usage_info()

void ResourceMonitor::update_vs_and_file_usage_info

(

)

Definition at line 579 of file XrdPfcResourceMonitor.cc.

{
   static const char *trc_pfx = "update_vs_and_file_usage_info() ";
 
   const auto &conf = Cache::Conf();
   XrdOssVSInfo vsi;
 
   // StatVS error (after it succeeded in config) implies a memory corruption (according to Mr. H).
   if (m_oss.StatVS(&vsi, conf.m_data_space.c_str(), 1) < 0) {
      TRACE(Error, trc_pfx << "can't get StatVS for oss space '" << conf.m_data_space << "'. This is a fatal error.");
      _exit(1);
   }
   m_fs_state.m_disk_total = vsi.Total;
   m_fs_state.m_disk_used  = vsi.Total - vsi.Free;
   m_fs_state.m_file_usage = 512ll * m_current_usage_in_st_blocks;
   if (m_oss.StatVS(&vsi, conf.m_meta_space.c_str(), 1) < 0) {
      TRACE(Error, trc_pfx << "can't get StatVS for oss space '" << conf.m_meta_space << "'. This is a fatal error.");
      _exit(1);
   }
   m_fs_state.m_meta_total = vsi.Total;
   m_fs_state.m_meta_used  = vsi.Total - vsi.Free;
}

References XrdPfc::Cache::Conf(), Error, XrdOssVSInfo::Free, XrdOssVSInfo::Total, and TRACE.

Referenced by heart_beat(), and perform_initial_scan().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

m_purge_task_active

bool XrdPfc::ResourceMonitor::m_purge_task_active {false}

Definition at line 254 of file XrdPfcResourceMonitor.hh.

{false}; // from the perspective of heart-beat, set only in heartbeat

Referenced by heart_beat(), and perform_purge_check().

m_purge_task_complete

bool XrdPfc::ResourceMonitor::m_purge_task_complete {false}

Definition at line 255 of file XrdPfcResourceMonitor.hh.

{false}; // from the perspective of the task, reset in heartbeat, set in task

Referenced by heart_beat(), and perform_purge_task_cleanup().

m_purge_task_cond

XrdSysCondVar XrdPfc::ResourceMonitor::m_purge_task_cond {0}

Definition at line 250 of file XrdPfcResourceMonitor.hh.

{0};

Referenced by heart_beat(), perform_purge_task(), and perform_purge_task_cleanup().

m_purge_task_end

time_t XrdPfc::ResourceMonitor::m_purge_task_end {0}

Definition at line 253 of file XrdPfcResourceMonitor.hh.

{0};

Referenced by perform_purge_task_cleanup().

m_purge_task_start

time_t XrdPfc::ResourceMonitor::m_purge_task_start {0}

Definition at line 252 of file XrdPfcResourceMonitor.hh.

{0};

Referenced by perform_purge_task().

---

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

• XrdPfcResourceMonitor.hh
• XrdPfcResourceMonitor.cc