XrdPfc::Cache Class Reference

Attaches/creates and detaches/deletes cache-io objects for disk based cache. More...

#include <XrdPfc.hh>

Inheritance diagram for XrdPfc::Cache:

Collaboration diagram for XrdPfc::Cache:

Public Member Functions
	Cache (XrdSysLogger *logger, XrdOucEnv *env)
	Constructor.
void	AddWriteTask (Block *b, bool from_read)
	Add downloaded block in write queue.
virtual XrdOucCacheIO *	Attach (XrdOucCacheIO *, int Options=0)
void	ClearPurgeProtectedSet ()
bool	Config (const char *config_filename, const char *parameters)
	Parse configuration file.
virtual int	ConsiderCached (const char *url)
bool	Decide (XrdOucCacheIO *)
	Makes decision if the original XrdOucCacheIO should be cached.
bool	DecideIfConsideredCached (long long file_size, long long bytes_on_disk)
void	DeRegisterPrefetchFile (File *)
long long	DetermineFullFileSize (const std::string &cinfo_fname)
void	ExecuteCommandUrl (const std::string &command_url)
void	FileSyncDone (File *, bool high_debug)
File *	GetFile (const std::string &, IO *, long long off=0, long long filesize=0)
XrdXrootdGStream *	GetGStream ()
XrdSysError *	GetLog ()
File *	GetNextFileToPrefetch ()
XrdOss *	GetOss () const
PurgePin *	GetPurgePin () const
XrdSysTrace *	GetTrace ()
bool	IsFileActiveOrPurgeProtected (const std::string &) const
virtual int	LocalFilePath (const char *url, char *buff=0, int blen=0, LFP_Reason why=ForAccess, bool forall=false)
void	Prefetch ()
virtual int	Prepare (const char *url, int oflags, mode_t mode)
void	ProcessWriteTasks ()
	Separate task which writes blocks from ram to disk.
const Configuration &	RefConfiguration () const
	Reference XrdPfc configuration.
ResourceMonitor &	RefResMon ()
void	RegisterPrefetchFile (File *)
void	ReleaseFile (File *, IO *)
void	ReleaseRAM (char *buf, long long size)
void	RemoveWriteQEntriesFor (File *f)
	Remove blocks from write queue which belong to given prefetch. This method is used at the time of File destruction.
char *	RequestRAM (long long size)
void	ScheduleFileSync (File *f)
virtual int	Stat (const char *url, struct stat &sbuff)
virtual int	Unlink (const char *url)
int	UnlinkFile (const std::string &f_name, bool fail_if_open)
	Remove cinfo and data files from cache.
void	WriteFileSizeXAttr (int cinfo_fd, long long file_size)
long long	WritesSinceLastCall ()
Public Member Functions inherited from XrdOucCache
	XrdOucCache (const char *ctype)
virtual	~XrdOucCache ()
	Destructor.
virtual int	Rename (const char *oldp, const char *newp)
virtual int	Rmdir (const char *dirp)
virtual int	Truncate (const char *path, off_t size)
virtual int	Xeq (XeqCmd cmd, char *arg, int arglen)

Static Public Member Functions
static const Configuration &	Conf ()
static Cache &	CreateInstance (XrdSysLogger *logger, XrdOucEnv *env)
	Singleton creation.
static Cache &	GetInstance ()
	Singleton access.
static ResourceMonitor &	ResMon ()
static const Cache &	TheOne ()
static bool	VCheck (XrdVersionInfo &urVersion)
	Version check.

Static Public Attributes
static XrdScheduler *	schedP = nullptr
Static Public Attributes inherited from XrdOucCache
static const int	optFIS = 0x0001
	File is structured (e.g. root file)
static const int	optNEW = 0x0014
	File is new -> optRW (o/w read or write)
static const int	optRW = 0x0004
	File is read/write (o/w read/only)
static const int	optWIN = 0x0024
	File is new -> optRW use write-in cache.

Additional Inherited Members
Public Types inherited from XrdOucCache
enum	LFP_Reason {   ForAccess =0 ,   ForInfo ,   ForPath }
enum	XeqCmd { xeqNoop = 0 }
Public Attributes inherited from XrdOucCache
const char	CacheType [8]
	A 1-to-7 character cache type identifier (usually pfc or rmc).
XrdOucCacheStats	Statistics

Detailed Description

Attaches/creates and detaches/deletes cache-io objects for disk based cache.

Definition at line 153 of file XrdPfc.hh.

Constructor & Destructor Documentation

Cache()

Cache::Cache	(	XrdSysLogger *	logger,
		XrdOucEnv *	env )

Constructor.

Definition at line 158 of file XrdPfc.cc.

                                                 :
   XrdOucCache("pfc"),
   m_env(env),
   m_log(logger, "XrdPfc_"),
   m_trace(new XrdSysTrace("XrdPfc", logger)),
   m_traceID("Cache"),
   m_oss(0),
   m_gstream(0),
   m_purge_pin(0),
   m_prefetch_condVar(0),
   m_prefetch_enabled(false),
   m_RAM_used(0),
   m_RAM_write_queue(0),
   m_RAM_std_size(0),
   m_isClient(false),
   m_active_cond(0)
{
   // Default log level is Warning.
   m_trace->What = 2;
}

References XrdOucCache::XrdOucCache().

Referenced by CreateInstance(), GetInstance(), and TheOne().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Function Documentation

AddWriteTask()

void Cache::AddWriteTask	(	Block *	b,
		bool	from_read )

Add downloaded block in write queue.

Definition at line 225 of file XrdPfc.cc.

{
   TRACE(Dump, "AddWriteTask() offset=" <<  b->m_offset << ". file " << b->get_file()->GetLocalPath());
 
   {
      XrdSysMutexHelper lock(&m_RAM_mutex);
      m_RAM_write_queue += b->get_size();
   }
 
   m_writeQ.condVar.Lock();
   if (fromRead)
      m_writeQ.queue.push_back(b);
   else
      m_writeQ.queue.push_front(b);
   m_writeQ.size++;
   m_writeQ.condVar.Signal();
   m_writeQ.condVar.UnLock();
}

References XrdPfc::Block::get_file(), XrdPfc::Block::get_size(), XrdPfc::File::GetLocalPath(), XrdPfc::Block::m_offset, and TRACE.

Here is the call graph for this function:

Attach()

XrdOucCacheIO * Cache::Attach ( XrdOucCacheIO * io, int Options = 0 )

virtual

Implements XrdOucCache.

Definition at line 179 of file XrdPfc.cc.

{
   const char* tpfx = "Attach() ";
 
   if (Options & XrdOucCache::optRW)
   {
      TRACE(Info, tpfx << "passing through write operation" << obfuscateAuth(io->Path()));
   }
   else if (Cache::GetInstance().Decide(io))
   {
      TRACE(Info, tpfx << obfuscateAuth(io->Path()));
 
      IO *cio;
 
      if (Cache::GetInstance().RefConfiguration().m_hdfsmode)
      {
         cio = new IOFileBlock(io, *this);
      }
      else
      {
         IOFile *iof = new IOFile(io, *this);
 
         if ( ! iof->HasFile())
         {
           delete iof;
           // TODO - redirect instead. But this is kind of an awkward place for it.
           // errno is set during IOFile construction.
           TRACE(Error, tpfx << "Failed opening local file, falling back to remote access " << io->Path());
           return io;
         }
 
         cio = iof;
      }
 
      TRACE_PC(Debug, const char* loc = io->Location(), tpfx << io->Path() << " location: " <<
               ((loc && loc[0] != 0) ? loc : "<deferred open>"));
 
      return cio;
   }
   else
   {
      TRACE(Info, tpfx << "decision decline " << io->Path());
   }
   return io;
}

References Debug, Decide(), Error, GetInstance(), XrdPfc::IOFile::HasFile(), XrdOucCacheIO::Location(), obfuscateAuth(), XrdOucCache::optRW, XrdOucCacheIO::Path(), RefConfiguration(), TRACE, and TRACE_PC.

Here is the call graph for this function:

ClearPurgeProtectedSet()

void Cache::ClearPurgeProtectedSet

(

)

Definition at line 688 of file XrdPfc.cc.

{
   XrdSysCondVarHelper lock(&m_active_cond);
   m_purge_delay_set.clear();
}

Referenced by XrdPfc::ResourceMonitor::perform_purge_check(), and XrdPfc::ResourceMonitor::perform_purge_task_cleanup().

Here is the caller graph for this function:

Conf()

const Configuration & Cache::Conf ( )

static

Definition at line 134 of file XrdPfc.cc.

{ return  m_instance->RefConfiguration(); }

Referenced by XrdPfc::ResourceMonitor::heart_beat(), XrdPfc::OldStylePurgeDriver(), XrdPfc::ResourceMonitor::perform_purge_check(), Proto_ResourceMonitorHeartBeat(), XrdPfc::ResourceMonitor::update_vs_and_file_usage_info(), and XrdPfc::DataFsSnapshot::write_json_file().

Here is the caller graph for this function:

Config()

bool Cache::Config	(	const char *	config_filename,
		const char *	parameters )

Parse configuration file.

Parameters

config_filename	path to configuration file
parameters	optional parameters to be passed

Returns

parse status

Definition at line 409 of file XrdPfcConfiguration.cc.

{
   // Indicate whether or not we are a client instance
   const char *theINS = getenv("XRDINSTANCE");
   m_isClient = (theINS != 0 && strncmp("*client ", theINS, 8) == 0);
 
   // Tell everyone else we are a caching proxy
   XrdOucEnv::Export("XRDPFC", 1);
 
   XrdOucEnv myEnv;
   XrdOucStream Config(&m_log, theINS, &myEnv, "=====> ");
 
   if (! config_filename || ! *config_filename)
   {
      TRACE(Error, "Config() configuration file not specified.");
      return false;
   }
 
   int fd;
   if ( (fd = open(config_filename, O_RDONLY, 0)) < 0)
   {
      TRACE( Error, "Config() can't open configuration file " << config_filename);
      return false;
   }
 
   Config.Attach(fd);
   static const char *cvec[] = { "*** pfc plugin config:", 0 };
   Config.Capture(cvec);
 
   // Obtain OFS configurator for OSS plugin.
   XrdOfsConfigPI *ofsCfg = XrdOfsConfigPI::New(config_filename,&Config,&m_log,
                                                &XrdVERSIONINFOVAR(XrdOucGetCache));
   if (! ofsCfg) return false;
 
   TmpConfiguration tmpc;
 
   // Adjust default parameters for client/serverless caching
   if (m_isClient)
   {
      m_configuration.m_bufferSize     = 128 * 1024; // same as normal.
      m_configuration.m_wqueue_blocks  = 8;
      m_configuration.m_wqueue_threads = 1;
   }
 
   // If network checksum processing is the default, indicate so.
   if (m_configuration.is_cschk_net()) m_env->Put("psx.CSNet", m_configuration.m_cs_ChkTLS ? "2" : "1");
 
   // Actual parsing of the config file.
   bool retval = true, aOK = true;
   char *var;
   while ((var = Config.GetMyFirstWord()))
   {
      if (! strcmp(var,"pfc.osslib"))
      {
         retval = ofsCfg->Parse(XrdOfsConfigPI::theOssLib);
      }
      else if (! strcmp(var,"pfc.cschk"))
      {
         retval = xcschk(Config);
      }
      else if (! strcmp(var,"pfc.decisionlib"))
      {
         retval = xdlib(Config);
      }
      else if (! strcmp(var,"pfc.purgelib"))
      {
         retval = xplib(Config);
      }
      else if (! strcmp(var,"pfc.trace"))
      {
         retval = xtrace(Config);
      }
      else if (! strcmp(var,"pfc.allow_xrdpfc_command"))
      {
         m_configuration.m_allow_xrdpfc_command = true;
      }
      else if (! strncmp(var,"pfc.", 4))
      {
         retval = ConfigParameters(std::string(var+4), Config, tmpc);
      }
 
      if ( ! retval)
      {
         TRACE(Error, "Config() error in parsing");
         aOK = false;
      }
   }
 
   Config.Close();
 
   // Load OSS plugin.
   myEnv.Put("oss.runmode", "pfc");
   if (m_configuration.is_cschk_cache())
   {
      char csi_conf[128];
      if (snprintf(csi_conf, 128, "space=%s nofill", m_configuration.m_meta_space.c_str()) < 128)
      {
         ofsCfg->Push(XrdOfsConfigPI::theOssLib, "libXrdOssCsi.so", csi_conf);
      } else {
         TRACE(Error, "Config() buffer too small for libXrdOssCsi params.");
         return false;
      }
   }
   if (ofsCfg->Load(XrdOfsConfigPI::theOssLib, &myEnv))
   {
      ofsCfg->Plugin(m_oss);
   }
   else
   {
      TRACE(Error, "Config() Unable to create an OSS object");
      return false;
   }
 
   // Test if OSS is operational, determine optional features.
   aOK &= test_oss_basics_and_features();
 
   // sets default value for disk usage
   XrdOssVSInfo sP;
   {
      if (m_configuration.m_meta_space != m_configuration.m_data_space &&
          m_oss->StatVS(&sP, m_configuration.m_meta_space.c_str(), 1) < 0)
      {
         m_log.Emsg("ConfigParameters()", "error obtaining stat info for meta space ", m_configuration.m_meta_space.c_str());
         return false;
      }
      if (m_configuration.m_meta_space != m_configuration.m_data_space && sP.Total < 10ll << 20)
      {
         m_log.Emsg("ConfigParameters()", "available data space is less than 10 MB (can be due to a mistake in oss.localroot directive) for space ",
                    m_configuration.m_meta_space.c_str());
                    return false;
      }
      if (m_oss->StatVS(&sP, m_configuration.m_data_space.c_str(), 1) < 0)
      {
         m_log.Emsg("ConfigParameters()", "error obtaining stat info for data space ", m_configuration.m_data_space.c_str());
         return false;
      }
      if (sP.Total < 10ll << 20)
      {
         m_log.Emsg("ConfigParameters()", "available data space is less than 10 MB (can be due to a mistake in oss.localroot directive) for space ",
                    m_configuration.m_data_space.c_str());
                    return false;
      }
 
      m_configuration.m_diskTotalSpace = sP.Total;
 
      if (cfg2bytes(tmpc.m_diskUsageLWM, m_configuration.m_diskUsageLWM, sP.Total, "lowWatermark") &&
          cfg2bytes(tmpc.m_diskUsageHWM, m_configuration.m_diskUsageHWM, sP.Total, "highWatermark"))
      {
         if (m_configuration.m_diskUsageLWM >= m_configuration.m_diskUsageHWM) {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage should have lowWatermark < highWatermark.");
            aOK = false;
         }
      }
      else aOK = false;
 
      if ( ! tmpc.m_fileUsageMax.empty())
      {
        if (cfg2bytes(tmpc.m_fileUsageBaseline, m_configuration.m_fileUsageBaseline, sP.Total, "files baseline") &&
            cfg2bytes(tmpc.m_fileUsageNominal,  m_configuration.m_fileUsageNominal,  sP.Total, "files nominal")  &&
            cfg2bytes(tmpc.m_fileUsageMax,      m_configuration.m_fileUsageMax,      sP.Total, "files max"))
        {
          if (m_configuration.m_fileUsageBaseline >= m_configuration.m_fileUsageNominal ||
              m_configuration.m_fileUsageBaseline >= m_configuration.m_fileUsageMax     ||
              m_configuration.m_fileUsageNominal  >= m_configuration.m_fileUsageMax)
          {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage files should have baseline < nominal < max.");
            aOK = false;
          }
 
 
         if (aOK && m_configuration.m_fileUsageMax >= m_configuration.m_diskUsageLWM)
         {
            m_log.Emsg("ConfigParameters()", "pfc.diskusage files values must be below lowWatermark");
            aOK = false;
         }
        }
        else aOK = false;
      }
   }
 
   // sets flush frequency
   if ( ! tmpc.m_flushRaw.empty())
   {
      if (::isalpha(*(tmpc.m_flushRaw.rbegin())))
      {
         if (XrdOuca2x::a2sz(m_log, "Error getting number of bytes written before flush",  tmpc.m_flushRaw.c_str(),
                             &m_configuration.m_flushCnt,
                             100 * m_configuration.m_bufferSize , 100000 * m_configuration.m_bufferSize))
         {
            return false;
         }
         m_configuration.m_flushCnt /= m_configuration.m_bufferSize;
      }
      else
      {
         if (XrdOuca2x::a2ll(m_log, "Error getting number of blocks written before flush", tmpc.m_flushRaw.c_str(),
                             &m_configuration.m_flushCnt, 100, 100000))
         {
            return false;
         }
      }
   }
 
   // set the write mode
   if ( ! tmpc.m_writemodeRaw.empty())
   {
      if (tmpc.m_writemodeRaw == "writethrough")
      {
         m_configuration.m_write_through = true;
      }
      else if (tmpc.m_writemodeRaw != "off")
      {
         m_log.Emsg("ConfigParameters()", "Unknown value for pfc.writemode (valid values are `writethrough` or `off`): %s",
                    tmpc.m_writemodeRaw.c_str());
         return false;
      }
   }
 
   // get number of available RAM blocks after process configuration
   if (m_configuration.m_RamAbsAvailable == 0)
   {
      m_configuration.m_RamAbsAvailable = m_isClient ? 256ll * 1024 * 1024 : 1024ll * 1024 * 1024;
      char buff[1024];
      snprintf(buff, sizeof(buff), "RAM usage pfc.ram is not specified. Default value %s is used.", m_isClient ? "256m" : "1g");
      m_log.Say("Config info: ", buff);
   }
   // Setup number of standard-size blocks not released back to the system to 5% of total RAM.
   m_configuration.m_RamKeepStdBlocks = (m_configuration.m_RamAbsAvailable / m_configuration.m_bufferSize + 1) * 5 / 100;
 
   // Set tracing to debug if this is set in environment
   char* cenv = getenv("XRDDEBUG");
   if (cenv && ! strcmp(cenv,"1") && m_trace->What < 4) m_trace->What = 4;
 
   if (aOK)
   {
      int  loff = 0;
//                         000    001            010
      const char *csc[] = {"off", "cache nonet", "nocache net notls",
//                         011
                           "cache net notls",
//                         100    101            110
                           "off", "cache nonet", "nocache net tls",
//                         111
                           "cache net tls"};
      char buff[8192], uvk[32];
      if (m_configuration.m_cs_UVKeep < 0)
         strcpy(uvk, "lru");
      else
         sprintf(uvk, "%lld", (long long) m_configuration.m_cs_UVKeep);
      float rg = (m_configuration.m_RamAbsAvailable) / float(1024*1024*1024);
      loff = snprintf(buff, sizeof(buff), "Config effective %s pfc configuration:\n"
                      "       pfc.cschk %s uvkeep %s\n"
                      "       pfc.blocksize %lld\n"
                      "       pfc.prefetch %d\n"
                      "       pfc.ram %.fg\n"
                      "       pfc.writequeue %d %d\n"
                      "       # Total available disk: %lld\n"
                      "       pfc.diskusage %lld %lld files %lld %lld %lld purgeinterval %d purgecoldfiles %d\n"
                      "       pfc.spaces %s %s\n"
                      "       pfc.trace %d\n"
                      "       pfc.flush %lld\n"
                      "       pfc.acchistorysize %d\n"
                      "       pfc.onlyIfCachedMinBytes %lld\n"
                      "       pfc.onlyIfCachedMinFrac %.2f\n",
                      config_filename,
                      csc[int(m_configuration.m_cs_Chk)], uvk,
                      m_configuration.m_bufferSize,
                      m_configuration.m_prefetch_max_blocks,
                      rg,
                      m_configuration.m_wqueue_blocks, m_configuration.m_wqueue_threads,
                      sP.Total,
                      m_configuration.m_diskUsageLWM, m_configuration.m_diskUsageHWM,
                      m_configuration.m_fileUsageBaseline, m_configuration.m_fileUsageNominal, m_configuration.m_fileUsageMax,
                      m_configuration.m_purgeInterval, m_configuration.m_purgeColdFilesAge,
                      m_configuration.m_data_space.c_str(),
                      m_configuration.m_meta_space.c_str(),
                      m_trace->What,
                      m_configuration.m_flushCnt,
                      m_configuration.m_accHistorySize,
                      m_configuration.m_onlyIfCachedMinSize,
                      m_configuration.m_onlyIfCachedMinFrac);
 
      if (m_configuration.is_dir_stat_reporting_on())
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff,
                          "       pfc.dirstats interval %d maxdepth %d ((internal: store_depth %d, size_of_dirlist %d, size_of_globlist %d))\n",
                          m_configuration.m_dirStatsInterval, m_configuration.m_dirStatsMaxDepth, m_configuration.m_dirStatsStoreDepth,
                          (int) m_configuration.m_dirStatsDirs.size(), (int) m_configuration.m_dirStatsDirGlobs.size());
         loff += snprintf(buff + loff, sizeof(buff) - loff, "           dirlist:\n");
         for (std::set<std::string>::iterator i = m_configuration.m_dirStatsDirs.begin(); i != m_configuration.m_dirStatsDirs.end(); ++i)
            loff += snprintf(buff + loff, sizeof(buff) - loff, "               %s\n", i->c_str());
         loff += snprintf(buff + loff, sizeof(buff) - loff, "           globlist:\n");
         for (std::set<std::string>::iterator i = m_configuration.m_dirStatsDirGlobs.begin(); i != m_configuration.m_dirStatsDirGlobs.end(); ++i)
            loff += snprintf(buff + loff, sizeof(buff) - loff, "               %s/*\n", i->c_str());
      }
 
      if (m_configuration.m_hdfsmode)
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff, "       pfc.hdfsmode hdfsbsize %lld\n", m_configuration.m_hdfsbsize);
      }
      loff += snprintf(buff + loff, sizeof(buff) - loff, "       pfc.writemode %s\n", m_configuration.m_write_through ? "writethrough" : "off");
 
      if (m_configuration.m_username.empty())
      {
         char unameBuff[256];
         XrdOucUtils::UserName(getuid(), unameBuff, sizeof(unameBuff));
         m_configuration.m_username = unameBuff;
      }
      else
      {
         loff += snprintf(buff + loff, sizeof(buff) - loff, "       pfc.user %s\n", m_configuration.m_username.c_str());
      }
 
      m_log.Say(buff);
 
      m_env->Put("XRDPFC.SEGSIZE", std::to_string(m_configuration.m_bufferSize).c_str());
   }
 
   // Derived settings
   m_prefetch_enabled   = m_configuration.m_prefetch_max_blocks > 0;
   Info::s_maxNumAccess = m_configuration.m_accHistorySize;
 
   m_gstream = (XrdXrootdGStream*) m_env->GetPtr("pfc.gStream*");
 
   m_log.Say("       pfc g-stream has", m_gstream ? "" : " NOT", " been configured via xrootd.monitor directive\n");
 
   // Create the ResourceMonitor and get it ready for starting the main thread function.
   if (aOK)
   {
      m_res_mon = new ResourceMonitor(*m_oss);
      m_res_mon->init_before_main();
   }
 
   m_log.Say("=====> Proxy file cache configuration parsing ", aOK ? "completed" : "failed");
 
   if (ofsCfg) delete ofsCfg;
 
   // XXXX-CKSUM Testing. To be removed after OssPgi is also merged and valildated.
   // Building of xrdpfc_print fails when this is enabled.
#ifdef XRDPFC_CKSUM_TEST
   {
      int xxx = m_configuration.m_cs_Chk;
 
      for (m_configuration.m_cs_Chk = CSChk_None; m_configuration.m_cs_Chk <= CSChk_Both; ++m_configuration.m_cs_Chk)
      {
         Info::TestCksumStuff();
      }
 
      m_configuration.m_cs_Chk = xxx;
   }
#endif
 
   return aOK;
}

References XrdOuca2x::a2ll(), XrdOuca2x::a2sz(), Config(), XrdPfc::CSChk_Both, XrdPfc::CSChk_None, Error, XrdOucEnv::Export(), XrdOfsConfigPI::Load(), XrdPfc::TmpConfiguration::m_diskUsageHWM, XrdPfc::TmpConfiguration::m_diskUsageLWM, XrdPfc::TmpConfiguration::m_fileUsageBaseline, XrdPfc::TmpConfiguration::m_fileUsageMax, XrdPfc::TmpConfiguration::m_fileUsageNominal, XrdPfc::TmpConfiguration::m_flushRaw, XrdPfc::TmpConfiguration::m_writemodeRaw, XrdOfsConfigPI::New(), open, XrdOfsConfigPI::Parse(), XrdOfsConfigPI::Plugin(), XrdOfsConfigPI::Push(), XrdOucEnv::Put(), XrdPfc::Info::s_maxNumAccess, XrdOfsConfigPI::theOssLib, XrdOssVSInfo::Total, TRACE, XrdOucUtils::UserName(), and XrdOucGetCache().

Referenced by Config(), and XrdOucGetCache().

Here is the call graph for this function:

Here is the caller graph for this function:

ConsiderCached()

int Cache::ConsiderCached ( const char * curl )

virtual

Returns

0 - the file is complete and the local path to the file is in the buffer, if it has been supllied.

<0 - the request could not be fulfilled. The return value is -errno describing why.

>0 - Reserved for future use.

Definition at line 1003 of file XrdPfc.cc.

{
   static const char* tpfx = "ConsiderCached ";
 
   TRACE(Debug, tpfx << curl);
 
   XrdCl::URL url(curl);
   std::string f_name = url.GetPath();
 
   File *file = nullptr;
   {
      XrdSysCondVarHelper lock(&m_active_cond);
      auto it = m_active.find(f_name);
      if (it != m_active.end()) {
         file = it->second;
         // If the file-open is in progress, `file` is a nullptr
         // so we cannot increase the reference count.  For now,
         // simply treat it as if the file open doesn't exist instead
         // of trying to wait and see if it succeeds.
         if (file) {
            inc_ref_cnt(file, false, false);
         }
      }
   }
   if (file) {
      struct stat sbuff;
      int res = file->Fstat(sbuff);
      dec_ref_cnt(file, false);
      if (res)
         return res;
      // DecideIfConsideredCached() already called in File::Fstat().
      return sbuff.st_atime > 0 ? 0 : -EREMOTE;
   }
 
   struct stat sbuff;
   int res = m_oss->Stat(f_name.c_str(), &sbuff);
   if (res != XrdOssOK) {
      TRACE(Debug, tpfx << curl << " -> " << res);
      return res;
   }
   if (S_ISDIR(sbuff.st_mode))
   {
      TRACE(Debug, tpfx << curl << " -> EISDIR");
      return -EISDIR;
   }
 
   long long file_size = DetermineFullFileSize(f_name + Info::s_infoExtension);
   if (file_size < 0) {
      TRACE(Debug, tpfx << curl << " -> " << file_size);
      return (int) file_size;
   }
   bool is_cached = DecideIfConsideredCached(file_size, sbuff.st_blocks * 512ll);
 
   return is_cached ? 0 : -EREMOTE;
}

References Debug, DecideIfConsideredCached(), DetermineFullFileSize(), XrdPfc::File::Fstat(), XrdCl::URL::GetPath(), XrdPfc::Info::s_infoExtension, stat, TRACE, and XrdOssOK.

Here is the call graph for this function:

CreateInstance()

Cache & Cache::CreateInstance ( XrdSysLogger * logger, XrdOucEnv * env )

static

Singleton creation.

Definition at line 125 of file XrdPfc.cc.

{
   assert (m_instance == 0);
   m_instance = new Cache(logger, env);
   return *m_instance;
}

References Cache().

Referenced by XrdOucGetCache().

Here is the call graph for this function:

Here is the caller graph for this function:

Decide()

bool Cache::Decide

(

XrdOucCacheIO *

io

)

Makes decision if the original XrdOucCacheIO should be cached.

Parameters

&	URL of file

Returns

decision if IO object will be cached.

Definition at line 137 of file XrdPfc.cc.

{
   if (! m_decisionpoints.empty())
   {
      XrdCl::URL url(io->Path());
      std::string filename = url.GetPath();
      std::vector<Decision*>::const_iterator it;
      for (it = m_decisionpoints.begin(); it != m_decisionpoints.end(); ++it)
      {
         XrdPfc::Decision *d = *it;
         if (! d) continue;
         if (! d->Decide(filename, *m_oss))
         {
            return false;
         }
      }
   }
 
   return true;
}

References XrdPfc::Decision::Decide(), XrdCl::URL::GetPath(), and XrdOucCacheIO::Path().

Referenced by Attach().

Here is the call graph for this function:

Here is the caller graph for this function:

DecideIfConsideredCached()

bool Cache::DecideIfConsideredCached	(	long long	file_size,
		long long	bytes_on_disk )

Definition at line 970 of file XrdPfc.cc.

{
   if (file_size == 0 || bytes_on_disk >= file_size)
      return true;
 
   double frac_on_disk = (double) bytes_on_disk / file_size;
 
   if (file_size <= m_configuration.m_onlyIfCachedMinSize)
   {
      if (frac_on_disk >= m_configuration.m_onlyIfCachedMinFrac)
         return true;
   }
   else
   {
      if (bytes_on_disk >= m_configuration.m_onlyIfCachedMinSize &&
          frac_on_disk  >= m_configuration.m_onlyIfCachedMinFrac)
         return true;
   }
   return false;
}

Referenced by ConsiderCached(), and Stat().

Here is the caller graph for this function:

DeRegisterPrefetchFile()

void Cache::DeRegisterPrefetchFile

(

File *

file

)

Definition at line 714 of file XrdPfc.cc.

{
   // Can be called with other locks held.
 
   if ( ! m_prefetch_enabled)
   {
      return;
   }
 
   m_prefetch_condVar.Lock();
   for (PrefetchList::iterator it = m_prefetchList.begin(); it != m_prefetchList.end(); ++it)
   {
      if (*it == file)
      {
         m_prefetchList.erase(it);
         break;
      }
   }
   m_prefetch_condVar.UnLock();
}

DetermineFullFileSize()

long long Cache::DetermineFullFileSize

(

const std::string &

cinfo_fname

)

Definition at line 929 of file XrdPfc.cc.

{
   if (m_metaXattr) {
      char pfn[4096];
      m_oss->Lfn2Pfn(cinfo_fname.c_str(), pfn, 4096);
      long long fsize = -1ll;
      int res = XrdSysXAttrActive->Get("pfc.fsize", &fsize, sizeof(long long), pfn);
      if (res == sizeof(long long))
      {
         return fsize;
      }
      else
      {
         TRACE(Debug, "DetermineFullFileSize error getting xattr " << res);
      }
   }
 
   XrdOssDF *infoFile = m_oss->newFile(m_configuration.m_username.c_str());
   XrdOucEnv env;
   long long ret;
   int res = infoFile->Open(cinfo_fname.c_str(), O_RDONLY, 0600, env);
   if (res < 0) {
      ret = res;
   } else {
      Info info(m_trace, 0);
      if ( ! info.Read(infoFile, cinfo_fname.c_str())) {
         ret = -EBADF;
      } else {
         ret = info.GetFileSize();
      }
      infoFile->Close();
   }
   delete infoFile;
   return ret;
}

References XrdOssDF::Close(), Debug, XrdPfc::Info::GetFileSize(), XrdOssDF::Open(), XrdPfc::Info::Read(), TRACE, and XrdSysXAttrActive.

Referenced by ConsiderCached(), and Stat().

Here is the call graph for this function:

Here is the caller graph for this function:

ExecuteCommandUrl()

void Cache::ExecuteCommandUrl

(

const std::string &

command_url

)

Definition at line 51 of file XrdPfcCommand.cc.

{
   static const char *top_epfx = "ExecuteCommandUrl ";
 
   SplitParser cp(command_url, "/");
 
   std::string token = cp.get_token();
 
   if (token != "xrdpfc_command")
   {
      TRACE(Error, top_epfx << "First token is NOT xrdpfc_command.");
      return;
   }
 
   // Get the command
   token = cp.get_token_as_string();
 
   auto get_opt = [](SplitParser &sp) -> char {
      const char *t = sp.get_token();
      if (t)
         return (t[0] == '-' && t[1] != 0) ? t[1] : 0;
      else
         return -1;
   };
 
   //================================================================
   // create_file
   //================================================================
 
   if (token == "create_file")
   {
      static const char* err_prefix = "ExecuteCommandUrl: /xrdpfc_command/create_file: ";
      static const char* usage =
         "Usage: create_file/ [-h] [-s filesize] [-b blocksize] [-t access_time] [-d access_duration]/<path>\n"
         "  Creates a cache file with given parameters. Data in file is random.\n"
         "  Useful for cache purge testing.\n"
         "Notes:\n"
         "  . If no options are needed one should still leave a space between / separators, ie., '/ /'\n"
         "  . Default filesize=1G, blocksize=<as configured>, access_time=-10, access_duration=10.\n"
         "  . -t and -d can be given multiple times to record several accesses.\n"
         "  . Negative arguments given to -t are interpreted as relative to now.\n";
 
      const Configuration &conf = m_configuration;
 
      token = cp.get_token_as_string();
      TRACE(Debug, err_prefix << "Entered with argument string '" << token <<"'.");
      if (token.empty()) {
         TRACE(Error, err_prefix << "Options section must not be empty, a single space character is OK.");
         return;
      }
      TRACE(Debug, err_prefix << "File path (reminder of URL) is '" << cp.get_reminder() <<"'.");
      if ( ! cp.has_reminder()) {
         TRACE(Error, err_prefix << "Path section must not be empty.");
         return;
      }
 
      long long   file_size    = ONE_GB;
      long long   block_size   = conf.m_bufferSize;
      int         access_time    [MAX_ACCESSES];
      int         access_duration[MAX_ACCESSES];
      int         at_count = 0, ad_count = 0;
 
      time_t time_now = time(0);
 
      SplitParser ap(token, " ");
      char theOpt;
 
      while ((theOpt = get_opt(ap)) != (char) -1)
      {
         switch (theOpt)
         {
            case 'h': {
               m_log.Say(err_prefix, " -- printing help, no action will be taken\n", usage);
               return;
            }
            case 's': {
               if (XrdOuca2x::a2sz(m_log, "Error getting filesize", ap.get_token(),
                                   &file_size, 0ll, 32 * ONE_GB))
                  return;
               break;
            }
            case 'b': {
               if (XrdOuca2x::a2sz(m_log, "Error getting blocksize", ap.get_token(),
                                   &block_size, 0ll, 64 * ONE_MB))
                  return;
               break;
            }
            case 't': {
               if (XrdOuca2x::a2i(m_log, "Error getting access time", ap.get_token(),
                                  &access_time[at_count++], INT_MIN, INT_MAX))
                  return;
               break;
            }
            case 'd': {
               if (XrdOuca2x::a2i(m_log, "Error getting access duration", ap.get_token(),
                                  &access_duration[ad_count++], 0, 24 * 3600))
                  return;
               break;
            }
            default: {
               TRACE(Error, err_prefix << "Unhandled command argument.");
               return;
            }
         }
      }
 
      if (at_count < 1) access_time    [at_count++] = time_now - 10;
      if (ad_count < 1) access_duration[ad_count++] = 10;
 
      if (at_count != ad_count)
      {
         TRACE(Error, err_prefix << "Options -t and -d must be given the same number of times.");
         return;
      }
 
      std::string file_path (cp.get_reminder_with_delim());
      std::string cinfo_path(file_path + Info::s_infoExtension);
 
      TRACE(Debug, err_prefix << "Command arguments parsed successfully. Proceeding to create file " << file_path);
 
      // Check if cinfo exists ... bail out if it does.
      {
         struct stat infoStat;
         if (GetOss()->Stat(cinfo_path.c_str(), &infoStat) == XrdOssOK)
         {
            TRACE(Error, err_prefix << "cinfo file already exists for '" << file_path << "'. Refusing to overwrite.");
            return;
         }
      }
 
      TRACE(Debug, err_prefix << "Command arguments parsed successfully, proceeding to execution.");
 
      {
         const char          *myUser = conf.m_username.c_str();
         XrdOucEnv            myEnv;
 
         // Create the data file.
 
         char size_str[32]; sprintf(size_str, "%lld", file_size);
         myEnv.Put("oss.asize",  size_str);
         myEnv.Put("oss.cgroup", conf.m_data_space.c_str());
         int cret;
         if ((cret = GetOss()->Create(myUser, file_path.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
         {
            TRACE(Error, err_prefix << "Create failed for data file " << file_path << ERRNO_AND_ERRSTR(-cret));
            return;
         }
 
         XrdOssDF *myFile = GetOss()->newFile(myUser);
         if ((cret = myFile->Open(file_path.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
         {
            TRACE(Error, err_prefix << "Open failed for data file " << file_path << ERRNO_AND_ERRSTR(-cret));
            delete myFile;
            return;
         }
 
         // Create the info file.
 
         myEnv.Put("oss.asize", "64k"); // TODO: Calculate? Get it from configuration? Do not know length of access lists ...
         myEnv.Put("oss.cgroup", conf.m_meta_space.c_str());
         if ((cret = GetOss()->Create(myUser, cinfo_path.c_str(), 0600, myEnv, XRDOSS_mkpath)) != XrdOssOK)
         {
            TRACE(Error, err_prefix << "Create failed for info file " << cinfo_path << ERRNO_AND_ERRSTR(-cret));
            myFile->Close(); delete myFile;
            return;
         }
 
         XrdOssDF *myInfoFile = GetOss()->newFile(myUser);
         if ((cret = myInfoFile->Open(cinfo_path.c_str(), O_RDWR, 0600, myEnv)) != XrdOssOK)
         {
            TRACE(Error, err_prefix << "Open failed for info file " << cinfo_path << ERRNO_AND_ERRSTR(-cret));
            delete myInfoFile;
            myFile->Close(); delete myFile;
            return;
         }
 
         // Allocate space for the data file.
 
         if ((cret = posix_fallocate(myFile->getFD(), 0, file_size)))
         {
            TRACE(Error, err_prefix << "posix_fallocate failed for data file " << file_path << ERRNO_AND_ERRSTR(cret));
         }
 
         // Fill up cinfo.
 
         Info myInfo(m_trace, false);
         myInfo.SetBufferSizeFileSizeAndCreationTime(block_size, file_size);
         myInfo.SetAllBitsSynced();
 
         for (int i = 0; i < at_count; ++i)
         {
            time_t att_time = access_time[i] >= 0 ? access_time[i] : time_now + access_time[i];
 
            myInfo.WriteIOStatSingle(file_size, att_time, att_time + access_duration[i]);
         }
 
         myInfo.Write(myInfoFile, cinfo_path.c_str());
 
         // Fake last modified time to the last access_time
         {
            time_t last_detach;
            myInfo.GetLatestDetachTime(last_detach);
            struct timespec acc_mod_time[2] = { {last_detach, UTIME_OMIT}, {last_detach, 0} };
 
            futimens(myInfoFile->getFD(), acc_mod_time);
         }
 
         myInfoFile->Close(); delete myInfoFile;
         myFile->Close();     delete myFile;
 
         struct stat dstat;
         GetOss()->Stat(file_path.c_str(), &dstat);
         TRACE(Info, err_prefix << "Created file '" << file_path << "', size=" << (file_size>>20) << "MB, "
                                << "st_blocks=" << dstat.st_blocks);
 
         {
            XrdSysCondVarHelper lock(&m_writeQ.condVar);
 
            m_writeQ.writes_between_purges += file_size;
         }
         {
            int token = m_res_mon->register_file_open(file_path, time_now, false);
            XrdPfc::Stats stats;
            stats.m_BytesWritten  = file_size;
            stats.m_StBlocksAdded = dstat.st_blocks;
            m_res_mon->register_file_update_stats(token, stats);
            m_res_mon->register_file_close(token, time(0), stats);
         }
      }
   }
 
   //================================================================
   // remove_file
   //================================================================
 
   else if (token == "remove_file")
   {
      static const char* err_prefix = "ExecuteCommandUrl: /xrdpfc_command/remove_file: ";
      static const char* usage =
         "Usage: remove_file/ [-h] /<path>\n"
         "  Removes given file from the cache unless it is currently open.\n"
         "  Useful for removal of stale files or duplicate files in a caching cluster.\n"
         "Notes:\n"
         "  . If no options are needed one should still leave a space between / separators, ie., '/ /'\n";
 
      token = cp.get_token_as_string();
      TRACE(Debug, err_prefix << "Entered with argument string '" << token <<"'.");
      if (token.empty()) {
         TRACE(Error, err_prefix << "Options section must not be empty, a single space character is OK.");
         return;
      }
      TRACE(Debug, err_prefix << "File path (reminder of URL) is '" << cp.get_reminder() <<"'.");
      if ( ! cp.has_reminder()) {
         TRACE(Error, err_prefix << "Path section must not be empty.");
         return;
      }
 
      SplitParser ap(token, " ");
      char theOpt;
 
      while ((theOpt = get_opt(ap)) != (char) -1)
      {
         switch (theOpt)
         {
            case 'h': {
               m_log.Say(err_prefix, " -- printing help, no action will be taken\n", usage);
               return;
            }
            default: {
               TRACE(Error, err_prefix << "Unhandled command argument.");
               return;
            }
         }
      }
 
      std::string f_name(cp.get_reminder_with_delim());
 
      TRACE(Debug, err_prefix << "file argument '" << f_name << "'.");
 
      int ret = UnlinkFile(f_name, true);
 
      TRACE(Info, err_prefix << "returned with status " << ret);
   }
 
   //================================================================
   // unknown command
   //================================================================
 
   else
   {
      TRACE(Error, top_epfx << "Unknown or empty command '" << token << "'");
   }
}

References XrdOuca2x::a2i(), XrdOuca2x::a2sz(), XrdOssDF::Close(), Create, Debug, ERRNO_AND_ERRSTR, Error, XrdPfc::SplitParser::get_reminder(), XrdPfc::SplitParser::get_reminder_with_delim(), XrdPfc::SplitParser::get_token(), XrdPfc::SplitParser::get_token_as_string(), XrdOssDF::getFD(), XrdPfc::Info::GetLatestDetachTime(), GetOss(), XrdPfc::SplitParser::has_reminder(), XrdPfc::Configuration::m_bufferSize, XrdPfc::Stats::m_BytesWritten, XrdPfc::Configuration::m_data_space, XrdPfc::Configuration::m_meta_space, XrdPfc::Stats::m_StBlocksAdded, XrdPfc::Configuration::m_username, XrdOss::newFile(), XrdOssDF::Open(), XrdOucEnv::Put(), XrdPfc::Info::s_infoExtension, XrdPfc::Info::SetAllBitsSynced(), XrdPfc::Info::SetBufferSizeFileSizeAndCreationTime(), Stat, XrdOss::Stat(), stat, TRACE, UnlinkFile(), usage(), XrdPfc::Info::Write(), XrdPfc::Info::WriteIOStatSingle(), XRDOSS_mkpath, and XrdOssOK.

Here is the call graph for this function:

FileSyncDone()

void Cache::FileSyncDone	(	File *	f,
		bool	high_debug )

Definition at line 546 of file XrdPfc.cc.

{
   dec_ref_cnt(f, high_debug);
}

GetFile()

File * Cache::GetFile	(	const std::string &	path,
		IO *	io,
		long long	off = 0,
		long long	filesize = 0 )

Definition at line 393 of file XrdPfc.cc.

{
   // Called from virtual IOFile constructor.
   
   TRACE(Debug, "GetFile " << path << ", io " << io);
 
   ActiveMap_i it;
 
   {
      XrdSysCondVarHelper lock(&m_active_cond);
 
      while (true)
      {
         it = m_active.find(path);
 
         // File is not open or being opened. Mark it as being opened and
         // proceed to opening it outside of while loop.
         if (it == m_active.end())
         {
            it = m_active.insert(std::make_pair(path, (File*) 0)).first;
            break;
         }
 
         if (it->second != 0)
         {
            it->second->AddIO(io);
            inc_ref_cnt(it->second, false, true);
 
            return it->second;
         }
         else
         {
            // Wait for some change in m_active, then recheck.
            m_active_cond.Wait();
         }
      }
   }
 
   // This is always true, now that IOFileBlock is unsupported.
   if (filesize == 0)
   {
      struct stat st;
      int res = io->Fstat(st);
      if (res < 0) {
         errno = res;
         TRACE(Error, "GetFile, could not get valid stat");
      } else if (res > 0) {
         errno = ENOTSUP;
         TRACE(Error, "GetFile, stat returned positive value, this should NOT happen here");
      } else {
         filesize = st.st_size;
      }
   }
 
   File *file = 0;
 
   if (filesize >= 0)
   {
      file = File::FileOpen(path, off, filesize);
   }
 
   {
      XrdSysCondVarHelper lock(&m_active_cond);
 
      if (file)
      {
         inc_ref_cnt(file, false, true);
         it->second = file;
 
         file->AddIO(io);
      }
      else
      {
         m_active.erase(it);
      }
 
      m_active_cond.Broadcast();
   }
 
   return file;
}

References XrdPfc::File::AddIO(), Debug, Error, XrdPfc::File::FileOpen(), XrdOucCacheIO::Fstat(), stat, and TRACE.

Referenced by XrdPfc::IOFile::IOFile().

Here is the call graph for this function:

Here is the caller graph for this function:

GetGStream()

XrdXrootdGStream * XrdPfc::Cache::GetGStream ( )

inline

Definition at line 288 of file XrdPfc.hh.

{ return m_gstream; }

GetInstance()

Cache & Cache::GetInstance ( )

static

Singleton access.

Definition at line 132 of file XrdPfc.cc.

{ return *m_instance; }

References Cache().

Referenced by XrdPfc::IOFile::IOFile(), XrdPfc::IOFileBlock::IOFileBlock(), Attach(), XrdPfc::IOFile::DetachFinalize(), XrdPfc::File::GetLog(), XrdPfc::File::GetTrace(), XrdPfc::ResourceMonitor::perform_purge_check(), XrdPfc::ResourceMonitor::perform_purge_task_cleanup(), PrefetchThread(), Prepare(), ProcessWriteTaskThread(), Proto_ResourceMonitorHeartBeat(), XrdPfc::File::Sync(), and XrdPfc::File::WriteBlockToDisk().

Here is the call graph for this function:

Here is the caller graph for this function:

GetLog()

XrdSysError * XrdPfc::Cache::GetLog ( )

inline

Definition at line 284 of file XrdPfc.hh.

{ return &m_log;  }

Referenced by XrdPfc::File::GetLog().

Here is the caller graph for this function:

GetNextFileToPrefetch()

File * Cache::GetNextFileToPrefetch

(

)

Definition at line 736 of file XrdPfc.cc.

{
   m_prefetch_condVar.Lock();
   while (m_prefetchList.empty())
   {
      m_prefetch_condVar.Wait();
   }
 
   //  std::sort(m_prefetchList.begin(), m_prefetchList.end(), myobject);
 
   size_t l = m_prefetchList.size();
   int idx = rand() % l;
   File* f = m_prefetchList[idx];
 
   m_prefetch_condVar.UnLock();
   return f;
}

Referenced by Prefetch().

Here is the caller graph for this function:

GetOss()

XrdOss * XrdPfc::Cache::GetOss ( ) const

inline

Definition at line 270 of file XrdPfc.hh.

{ return m_oss; }

Referenced by ExecuteCommandUrl().

Here is the caller graph for this function:

GetPurgePin()

PurgePin * XrdPfc::Cache::GetPurgePin ( ) const

inline

Definition at line 274 of file XrdPfc.hh.

{ return m_purge_pin; }

Referenced by XrdPfc::ResourceMonitor::perform_purge_check().

Here is the caller graph for this function:

GetTrace()

XrdSysTrace * XrdPfc::Cache::GetTrace ( )

inline

Definition at line 285 of file XrdPfc.hh.

{ return m_trace; }

Referenced by XrdPfc::File::GetTrace().

Here is the caller graph for this function:

IsFileActiveOrPurgeProtected()

bool Cache::IsFileActiveOrPurgeProtected

(

const std::string &

path

)

const

Definition at line 680 of file XrdPfc.cc.

{
   XrdSysCondVarHelper lock(&m_active_cond);
 
   return m_active.find(path)          != m_active.end() ||
          m_purge_delay_set.find(path) != m_purge_delay_set.end();
}

LocalFilePath()

int Cache::LocalFilePath ( const char * curl, char * buff = 0, int blen = 0, LFP_Reason why = ForAccess, bool forall = false )

virtual

Get the path to a file that is complete in the local cache. By default, the file must be complete in the cache (i.e. no blocks are missing). This can be overridden. This path can be used to access the file on the local node.

Returns

0 - the file is complete and the local path to the file is in the buffer, if it has been supllied.

<0 - the request could not be fulfilled. The return value is -errno describing why. If a buffer was supplied and a path could be generated it is returned only if "why" is ForCheck or ForInfo. Otherwise, a null path is returned.

>0 - Reserved for future use.

Reimplemented from XrdOucCache.

Definition at line 797 of file XrdPfc.cc.

{
   static const mode_t groupReadable = S_IRUSR | S_IWUSR | S_IRGRP;
   static const mode_t worldReadable = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
   static const char  *lfpReason[] = { "ForAccess", "ForInfo", "ForPath" };
 
   TRACE(Debug, "LocalFilePath '" << curl << "', why=" << lfpReason[why]);
 
   if (buff && blen > 0) buff[0] = 0;
 
   XrdCl::URL url(curl);
   std::string f_name = url.GetPath();
   std::string i_name = f_name + Info::s_infoExtension;
 
   if (why == ForPath)
   {
     int ret = m_oss->Lfn2Pfn(f_name.c_str(), buff, blen);
     TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> " << ret);
     return ret;
   }
 
   {
      XrdSysCondVarHelper lock(&m_active_cond);
      m_purge_delay_set.insert(f_name);
   }
 
   struct stat sbuff, sbuff2;
   if (m_oss->Stat(f_name.c_str(), &sbuff)  == XrdOssOK &&
       m_oss->Stat(i_name.c_str(), &sbuff2) == XrdOssOK)
   {
      if (S_ISDIR(sbuff.st_mode))
      {
         TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> EISDIR");
         return -EISDIR;
      }
      else
      {
         bool read_ok     = false;
         bool is_complete = false;
 
         // Lock and check if the file is active. If NOT, keep the lock
         // and add dummy access after successful reading of info file.
         // If it IS active, just release the lock, this ongoing access will
         // assure the file continues to exist.
 
         // XXXX How can I just loop over the cinfo file when active?
         // Can I not get is_complete from the existing file?
         // Do I still want to inject access record?
         // Oh, it writes only if not active .... still let's try to use existing File.
 
         m_active_cond.Lock();
 
         bool is_active = m_active.find(f_name) != m_active.end();
 
         if (is_active) m_active_cond.UnLock();
 
         XrdOssDF* infoFile = m_oss->newFile(m_configuration.m_username.c_str());
         XrdOucEnv myEnv;
         int res = infoFile->Open(i_name.c_str(), O_RDWR, 0600, myEnv);
         if (res >= 0)
         {
            Info info(m_trace, 0);
            if (info.Read(infoFile, i_name.c_str()))
            {
               read_ok = true;
 
               is_complete = info.IsComplete();
 
               // Add full-size access if reason is for access.
               if ( ! is_active && is_complete && why == ForAccess)
               {
                  info.WriteIOStatSingle(info.GetFileSize());
                  info.Write(infoFile, i_name.c_str());
               }
            }
            infoFile->Close();
         }
         delete infoFile;
 
         if ( ! is_active) m_active_cond.UnLock();
 
         if (read_ok)
         {
            if ((is_complete || why == ForInfo) && buff != 0)
            {
               int res2 = m_oss->Lfn2Pfn(f_name.c_str(), buff, blen);
               if (res2 < 0)
                  return res2;
 
               // Normally, files are owned by us but when direct cache access
               // is wanted and possible, make sure the file is world readable.
               if (why == ForAccess)
                  {mode_t mode = (forall ? worldReadable : groupReadable);
                   if (((sbuff.st_mode & worldReadable) != mode)
                   &&  (m_oss->Chmod(f_name.c_str(), mode) != XrdOssOK))
                      {is_complete = false;
                       *buff = 0;
                      }
                  }
            }
 
            TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] <<
                  (is_complete ? " -> FILE_COMPLETE_IN_CACHE" : " -> EREMOTE"));
 
            return is_complete ? 0 : -EREMOTE;
         }
      }
   }
 
   TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> ENOENT");
   return -ENOENT;
}

References XrdOssDF::Close(), Debug, XrdOucCache::ForAccess, XrdOucCache::ForInfo, XrdOucCache::ForPath, XrdPfc::Info::GetFileSize(), XrdCl::URL::GetPath(), XrdPfc::Info::IsComplete(), XrdOssDF::Open(), XrdPfc::Info::Read(), XrdPfc::Info::s_infoExtension, stat, TRACE, XrdPfc::Info::Write(), XrdPfc::Info::WriteIOStatSingle(), and XrdOssOK.

Here is the call graph for this function:

Prefetch()

void Cache::Prefetch

(

)

Definition at line 755 of file XrdPfc.cc.

{
   const long long limit_RAM = m_configuration.m_RamAbsAvailable * 7 / 10;
 
   while (true)
   {
      m_RAM_mutex.Lock();
      bool doPrefetch = (m_RAM_used < limit_RAM);
      m_RAM_mutex.UnLock();
 
      if (doPrefetch)
      {
         File* f = GetNextFileToPrefetch();
         f->Prefetch();
      }
      else
      {
         XrdSysTimer::Wait(5);
      }
   }
}

References GetNextFileToPrefetch(), XrdPfc::File::Prefetch(), and XrdSysTimer::Wait().

Referenced by PrefetchThread().

Here is the call graph for this function:

Here is the caller graph for this function:

Prepare()

int Cache::Prepare ( const char * curl, int oflags, mode_t mode )

virtual

Preapare the cache for a file open request. This method is called prior to actually opening a file. This method is meant to allow defering an open request or implementing the full I/O stack in the cache layer.

Returns

<0 Error has occurred, return value is -errno; fail open request. =0 Continue with open() request. >0 Defer open but treat the file as actually being open. Use the XrdOucCacheIO::Open() method to open the file at a later time.

Reimplemented from XrdOucCache.

Definition at line 1069 of file XrdPfc.cc.

{
   XrdCl::URL url(curl);
   std::string f_name = url.GetPath();
   std::string i_name = f_name + Info::s_infoExtension;
 
   // Do not allow write access.
   if ((oflags & O_ACCMODE) != O_RDONLY)
   {
      if (Cache::GetInstance().RefConfiguration().m_write_through)
      {
         return 0;
      }
      TRACE(Warning, "Prepare write access requested on file " << f_name << ". Denying access.");
      return -EROFS;
   }
 
   // Intercept xrdpfc_command requests.
   if (m_configuration.m_allow_xrdpfc_command && strncmp("/xrdpfc_command/", f_name.c_str(), 16) == 0)
   {
      // Schedule a job to process command request.
      {
         CommandExecutor *ce = new CommandExecutor(f_name, "CommandExecutor");
 
         schedP->Schedule(ce);
      }
 
      return -EAGAIN;
   }
 
   {
      XrdSysCondVarHelper lock(&m_active_cond);
      m_purge_delay_set.insert(f_name);
   }
 
   struct stat sbuff;
   if (m_oss->Stat(i_name.c_str(), &sbuff) == XrdOssOK)
   {
      TRACE(Dump, "Prepare defer open " << f_name);
      return 1;
   }
   else
   {
      return 0;
   }
}

References GetInstance(), XrdCl::URL::GetPath(), RefConfiguration(), XrdPfc::Info::s_infoExtension, schedP, stat, TRACE, and XrdOssOK.

Here is the call graph for this function:

ProcessWriteTasks()

void Cache::ProcessWriteTasks

(

)

Separate task which writes blocks from ram to disk.

Definition at line 277 of file XrdPfc.cc.

{
   std::vector<Block*> blks_to_write(m_configuration.m_wqueue_blocks);
 
   while (true)
   {
      m_writeQ.condVar.Lock();
      while (m_writeQ.size == 0)
      {
         m_writeQ.condVar.Wait();
      }
 
      // MT -- optimize to pop several blocks if they are available (or swap the list).
      // This makes sense especially for smallish block sizes.
 
      int       n_pushed = std::min(m_writeQ.size, m_configuration.m_wqueue_blocks);
      long long sum_size = 0;
 
      for (int bi = 0; bi < n_pushed; ++bi)
      {
         Block* block = m_writeQ.queue.front();
         m_writeQ.queue.pop_front();
         m_writeQ.writes_between_purges += block->get_size();
         sum_size += block->get_size();
 
         blks_to_write[bi] = block;
 
         TRACE(Dump, "ProcessWriteTasks for block " <<  (void*)(block) << " path " << block->m_file->lPath());
      }
      m_writeQ.size -= n_pushed;
 
      m_writeQ.condVar.UnLock();
 
      {
         XrdSysMutexHelper lock(&m_RAM_mutex);
         m_RAM_write_queue -= sum_size;
      }
 
      for (int bi = 0; bi < n_pushed; ++bi)
      {
         Block* block = blks_to_write[bi];
 
         block->m_file->WriteBlockToDisk(block);
      }
   }
}

References XrdPfc::Block::get_size(), XrdPfc::File::lPath(), XrdPfc::Block::m_file, TRACE, and XrdPfc::File::WriteBlockToDisk().

Referenced by ProcessWriteTaskThread().

Here is the call graph for this function:

Here is the caller graph for this function:

RefConfiguration()

const Configuration & XrdPfc::Cache::RefConfiguration ( ) const

inline

Reference XrdPfc configuration.

Definition at line 206 of file XrdPfc.hh.

{ return m_configuration; }

Referenced by XrdPfc::IOFileBlock::IOFileBlock(), Attach(), Prepare(), and XrdOucGetCache().

Here is the caller graph for this function:

RefResMon()

ResourceMonitor & XrdPfc::Cache::RefResMon ( )

inline

Definition at line 287 of file XrdPfc.hh.

{ return *m_res_mon; }

RegisterPrefetchFile()

void Cache::RegisterPrefetchFile

(

File *

file

)

Definition at line 698 of file XrdPfc.cc.

{
   // Can be called with other locks held.
 
   if ( ! m_prefetch_enabled)
   {
      return;
   }
 
   m_prefetch_condVar.Lock();
   m_prefetchList.push_back(file);
   m_prefetch_condVar.Signal();
   m_prefetch_condVar.UnLock();
}

ReleaseFile()

void Cache::ReleaseFile	(	File *	f,
		IO *	io )

Definition at line 475 of file XrdPfc.cc.

{
   // Called from virtual IO::DetachFinalize.
 
   TRACE(Debug, "ReleaseFile " << f->GetLocalPath() << ", io " << io);
 
   {
     XrdSysCondVarHelper lock(&m_active_cond);
 
     f->RemoveIO(io);
   }
   dec_ref_cnt(f, true);
}

References Debug, XrdPfc::File::GetLocalPath(), XrdPfc::File::RemoveIO(), and TRACE.

Referenced by XrdPfc::IOFile::DetachFinalize().

Here is the call graph for this function:

Here is the caller graph for this function:

ReleaseRAM()

void Cache::ReleaseRAM	(	char *	buf,
		long long	size )

Definition at line 375 of file XrdPfc.cc.

{
   bool std_size = (size == m_configuration.m_bufferSize);
   {
      XrdSysMutexHelper lock(&m_RAM_mutex);
 
      m_RAM_used -= size;
 
      if (std_size && m_RAM_std_size < m_configuration.m_RamKeepStdBlocks)
      {
         m_RAM_std_blocks.push_back(buf);
         ++m_RAM_std_size;
         return;
      }
   }
   free(buf);
}

RemoveWriteQEntriesFor()

void Cache::RemoveWriteQEntriesFor

(

File *

f

)

Remove blocks from write queue which belong to given prefetch. This method is used at the time of File destruction.

Definition at line 244 of file XrdPfc.cc.

{
   std::list<Block*> removed_blocks;
   long long         sum_size = 0;
 
   m_writeQ.condVar.Lock();
   std::list<Block*>::iterator i = m_writeQ.queue.begin();
   while (i != m_writeQ.queue.end())
   {
      if ((*i)->m_file == file)
      {
         TRACE(Dump, "Remove entries for " <<  (void*)(*i) << " path " <<  file->lPath());
         std::list<Block*>::iterator j = i++;
         removed_blocks.push_back(*j);
         sum_size += (*j)->get_size();
         m_writeQ.queue.erase(j);
         --m_writeQ.size;
      }
      else
      {
         ++i;
      }
   }
   m_writeQ.condVar.UnLock();
 
   {
      XrdSysMutexHelper lock(&m_RAM_mutex);
      m_RAM_write_queue -= sum_size;
   }
 
   file->BlocksRemovedFromWriteQ(removed_blocks);
}

References XrdPfc::File::BlocksRemovedFromWriteQ(), XrdPfc::File::lPath(), and TRACE.

Referenced by UnlinkFile().

Here is the call graph for this function:

Here is the caller graph for this function:

RequestRAM()

char * Cache::RequestRAM

(

long long

size

)

Definition at line 335 of file XrdPfc.cc.

{
   static const size_t s_block_align = sysconf(_SC_PAGESIZE);
 
   bool  std_size = (size == m_configuration.m_bufferSize);
 
   m_RAM_mutex.Lock();
 
   long long total = m_RAM_used + size;
 
   if (total <= m_configuration.m_RamAbsAvailable)
   {
      m_RAM_used = total;
      if (std_size && m_RAM_std_size > 0)
      {
         char *buf = m_RAM_std_blocks.back();
         m_RAM_std_blocks.pop_back();
         --m_RAM_std_size;
 
         m_RAM_mutex.UnLock();
 
         return buf;
      }
      else
      {
         m_RAM_mutex.UnLock();
         char *buf;
         if (posix_memalign((void**) &buf, s_block_align, (size_t) size))
         {
            // Report out of mem? Probably should report it at least the first time,
            // then periodically.
            return 0;
         }
         return buf;
      }
   }
   m_RAM_mutex.UnLock();
   return 0;
}

ResMon()

ResourceMonitor & Cache::ResMon ( )

static

Definition at line 135 of file XrdPfc.cc.

{ return  m_instance->RefResMon(); }

Referenced by XrdPfc::ResourceMonitor::perform_purge_check(), ResourceMonitorThread(), and XrdPfc::UnlinkPurgeStateFilesInMap().

Here is the caller graph for this function:

ScheduleFileSync()

void XrdPfc::Cache::ScheduleFileSync ( File * f )

inline

Definition at line 280 of file XrdPfc.hh.

{ schedule_file_sync(f, false, false); }

Stat()

int Cache::Stat ( const char * curl, struct stat & sbuff )

virtual

Returns

<0 - Stat failed, value is -errno. =0 - Stat succeeded, sbuff holds stat information. >0 - Stat could not be done, forward operation to next level.

Reimplemented from XrdOucCache.

Definition at line 1124 of file XrdPfc.cc.

{
   const char *tpfx = "Stat ";
 
   XrdCl::URL url(curl);
   std::string f_name = url.GetPath();
 
   File *file = nullptr;
   {
      XrdSysCondVarHelper lock(&m_active_cond);
      auto it = m_active.find(f_name);
      if (it != m_active.end()) {
         file = it->second;
         // If `file` is nullptr, the file-open is in progress; instead
         // of waiting for the file-open to finish, simply treat it as if
         // the file-open doesn't exist.
         if (file) {
            inc_ref_cnt(file, false, false);
         }
      }
   }
   if (file) {
      int res = file->Fstat(sbuff);
      dec_ref_cnt(file, false);
      TRACE(Debug, tpfx << "from active file " << curl << " -> " << res);
      return res;
   }
 
   int res = m_oss->Stat(f_name.c_str(), &sbuff);
   if (res != XrdOssOK) {
      TRACE(Debug, tpfx << curl << " -> " << res);
      return 1; // res; -- for only-if-cached
   }
   if (S_ISDIR(sbuff.st_mode))
   {
      TRACE(Debug, tpfx << curl << " -> EISDIR");
      return -EISDIR;
   }
 
   long long file_size = DetermineFullFileSize(f_name + Info::s_infoExtension);
   if (file_size < 0) {
      TRACE(Debug, tpfx << curl << " -> " << file_size);
      return 1; // (int) file_size; -- for only-if-cached
   }
   sbuff.st_size = file_size;
   bool is_cached = DecideIfConsideredCached(file_size, sbuff.st_blocks * 512ll);
   if ( ! is_cached)
      sbuff.st_atime = 0;
 
   TRACE(Debug, tpfx << "from disk " << curl << " -> " << res);
 
   return 0;
}

References Debug, DecideIfConsideredCached(), DetermineFullFileSize(), XrdPfc::File::Fstat(), XrdCl::URL::GetPath(), XrdPfc::Info::s_infoExtension, stat, TRACE, and XrdOssOK.

Here is the call graph for this function:

TheOne()

const Cache & Cache::TheOne ( )

static

Definition at line 133 of file XrdPfc.cc.

{ return *m_instance; }

References Cache().

Referenced by XrdPfc::OldStylePurgeDriver(), and XrdPfc::UnlinkPurgeStateFilesInMap().

Here is the call graph for this function:

Here is the caller graph for this function:

Unlink()

int Cache::Unlink ( const char * curl )

virtual

Returns

<0 - Stat failed, value is -errno. =0 - Stat succeeded, sbuff holds stat information.

Reimplemented from XrdOucCache.

Definition at line 1185 of file XrdPfc.cc.

{
   XrdCl::URL url(curl);
   std::string f_name = url.GetPath();
 
   // printf("Unlink url=%s\n\t    fname=%s\n", curl, f_name.c_str());
 
   return UnlinkFile(f_name, false);
}

References XrdCl::URL::GetPath(), and UnlinkFile().

Here is the call graph for this function:

UnlinkFile()

int Cache::UnlinkFile	(	const std::string &	f_name,
		bool	fail_if_open )

Remove cinfo and data files from cache.

Definition at line 1195 of file XrdPfc.cc.

{
   static const char* trc_pfx = "UnlinkFile ";
   ActiveMap_i  it;
   File        *file = 0;
   long long    st_blocks_to_purge = 0;
   {
      XrdSysCondVarHelper lock(&m_active_cond);
 
      it = m_active.find(f_name);
 
      if (it != m_active.end())
      {
         if (fail_if_open)
         {
            TRACE(Info, trc_pfx << f_name << ", file currently open and force not requested - denying request");
            return -EBUSY;
         }
 
         // Null File* in m_active map means an operation is ongoing, probably
         // Attach() with possible File::Open(). Ask for retry.
         if (it->second == 0)
         {
            TRACE(Info, trc_pfx << f_name << ", an operation on this file is ongoing - denying request");
            return -EAGAIN;
         }
 
         file = it->second;
         st_blocks_to_purge = file->initiate_emergency_shutdown();
         it->second = 0;
      }
      else
      {
         it = m_active.insert(std::make_pair(f_name, (File*) 0)).first;
      }
   }
 
   if (file) {
      RemoveWriteQEntriesFor(file);
   } else {
      struct stat f_stat;
      if (m_oss->Stat(f_name.c_str(), &f_stat) == XrdOssOK)
         st_blocks_to_purge = f_stat.st_blocks;
   }
 
   std::string i_name = f_name + Info::s_infoExtension;
 
   // Unlink file & cinfo
   int f_ret = m_oss->Unlink(f_name.c_str());
   int i_ret = m_oss->Unlink(i_name.c_str());
 
   if (st_blocks_to_purge)
      m_res_mon->register_file_purge(f_name, st_blocks_to_purge);
 
   TRACE(Debug, trc_pfx << f_name << ", f_ret=" << f_ret << ", i_ret=" << i_ret);
 
   {
      XrdSysCondVarHelper lock(&m_active_cond);
 
      m_active.erase(it);
   }
 
   return std::min(f_ret, i_ret);
}

References Debug, XrdPfc::File::initiate_emergency_shutdown(), RemoveWriteQEntriesFor(), XrdPfc::Info::s_infoExtension, stat, TRACE, and XrdOssOK.

Referenced by ExecuteCommandUrl(), XrdPfc::File::Sync(), and Unlink().

Here is the call graph for this function:

Here is the caller graph for this function:

VCheck()

static bool XrdPfc::Cache::VCheck ( XrdVersionInfo & urVersion )

inlinestatic

Version check.

Definition at line 235 of file XrdPfc.hh.

{ return true; }

WriteFileSizeXAttr()

void Cache::WriteFileSizeXAttr	(	int	cinfo_fd,
		long long	file_size )

Definition at line 914 of file XrdPfc.cc.

{
   if (m_metaXattr) {
      int res = XrdSysXAttrActive->Set("pfc.fsize", &file_size, sizeof(long long), 0, cinfo_fd, 0);
      if (res != 0) {
         TRACE(Debug, "WriteFileSizeXAttr error setting xattr " << res);
      }
   }
}

References Debug, TRACE, and XrdSysXAttrActive.

WritesSinceLastCall()

long long Cache::WritesSinceLastCall

(

)

Definition at line 324 of file XrdPfc.cc.

{
   // Called from ResourceMonitor for an alternative estimation of disk writes.
   XrdSysCondVarHelper lock(&m_writeQ.condVar);
   long long ret = m_writeQ.writes_between_purges;
   m_writeQ.writes_between_purges = 0;
   return ret;
}

Referenced by XrdPfc::ResourceMonitor::perform_purge_check().

Here is the caller graph for this function:

Member Data Documentation

schedP

XrdScheduler * Cache::schedP = nullptr

static

Definition at line 292 of file XrdPfc.hh.

Referenced by XrdPfc::IO::Detach(), XrdPfc::ResourceMonitor::perform_purge_check(), Prepare(), and XrdOucGetCache().

---

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

• XrdPfc.hh
• XrdPfc.cc
• XrdPfcCommand.cc
• XrdPfcConfiguration.cc