Loading...
Searching...
No Matches
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)
bool | blocksize_str2value (const char *from, const char *str, long long &val, long long min, long long max) const
void | ClearPurgeProtectedSet ()
bool | Config (const char *config_filename, const char *parameters, XrdOucEnv *env)
| 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 () const
File * | GetNextFileToPrefetch ()
XrdOss * | GetOss () const
PurgePin * | GetPurgePin () const
XrdSysTrace * | GetTrace () const
bool | is_prefetch_enabled () const
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 ()
bool | prefetch_str2value (const char *from, const char *str, int &val, int min, int max) const
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 162 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;
}
XrdOucCache::XrdOucCache
XrdOucCache(const char *ctype)
Definition XrdOucCache.hh:700

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 221 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();
}
TRACE
#define TRACE(act, x)
Definition XrdTrace.hh:63
XrdPfc::Block::get_size
int get_size() const
Definition XrdPfcFile.hh:136
XrdPfc::Block::m_offset
long long m_offset
Definition XrdPfcFile.hh:114
XrdPfc::Block::get_file
File * get_file() const
Definition XrdPfcFile.hh:140
XrdPfc::File::GetLocalPath
const std::string & GetLocalPath() const
Definition XrdPfcFile.hh:262

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 (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;
}
obfuscateAuth
std::string obfuscateAuth(const std::string &input)
Definition XrdOucUtils.cc:1534
TRACE_PC
#define TRACE_PC(act, pre_code, x)
Definition XrdPfcTrace.hh:59
Debug
bool Debug
Definition XrdSecgsiProxy.cc:95
Error
@ Error
Definition XrdTlsNotaryUtils.hh:42
XrdOucCacheIO::Path
virtual const char * Path()=0
XrdOucCacheIO::Location
virtual const char * Location(bool refresh=false)
Definition XrdOucCache.hh:161
XrdPfc::Cache::RefConfiguration
const Configuration & RefConfiguration() const
Reference XrdPfc configuration.
Definition XrdPfc.hh:215
XrdPfc::Cache::GetInstance
static Cache & GetInstance()
Singleton access.
Definition XrdPfc.cc:132
XrdPfc::Cache::Decide
bool Decide(XrdOucCacheIO *)
Makes decision if the original XrdOucCacheIO should be cached.
Definition XrdPfc.cc:137
XrdPfc::IOFile::HasFile
bool HasFile() const
Check if File was opened successfully.
Definition XrdPfcIOFile.hh:48

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

Here is the call graph for this function:

## blocksize_str2value()

bool Cache::blocksize_str2value | ( | const char * | from,
|  | const char * | str,
|  | long long & | val,
|  | long long | min,
|  | long long | max ) const

Definition at line 103 of file XrdPfcConfiguration.cc.

{
if (XrdOuca2x::a2sz(m_log, "Error parsing block-size", str, &val, min, max))
return false;
if (val & 0xFFF) {
val &= ~0x0FFF;
val += 0x1000;
m_log.Emsg(from, "blocksize must be a multiple of 4 kB. Rounded up.");
}
return true;
}
XrdOuca2x::a2sz
static int a2sz(XrdSysError &, const char *emsg, const char *item, long long *val, long long minv=-1, long long maxv=-1)
Definition XrdOuca2x.cc:257

References XrdOuca2x::a2sz().

Here is the call graph for this function:

## ClearPurgeProtectedSet()

void Cache::ClearPurgeProtectedSet | ( |  | ) |

Definition at line 685 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,
|  | XrdOucEnv * | env )

Parse configuration file.

Parameters

  config_filename | path to configuration file
  parameters | optional parameters to be passed
  env | optional environment to use for configuration

Returns
  parse status

Definition at line 431 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 emptyEnv;
XrdOucEnv *myEnv = env ? env : &emptyEnv;
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;
Configuration &CFG = m_configuration;
// 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.
auto orig_runmode = myEnv->Get("oss.runmode");
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;
}
if (orig_runmode) myEnv->Put("oss.runmode", orig_runmode);
else myEnv->Put("oss.runmode", "");
// 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;
}
}
}
// 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)
{
// 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 uvk[32];
if (m_configuration.m_cs_UVKeep < 0)
strcpy(uvk, "lru");
else
sprintf(uvk, "%lld", (long long) m_configuration.m_cs_UVKeep);
float ram_gb = (m_configuration.m_RamAbsAvailable) / float(1024*1024*1024);
char urlcgi_blks[64] = "ignore", urlcgi_npref[32] = "ignore";
if (CFG.m_cgi_blocksize_allowed)
snprintf(urlcgi_blks, sizeof(urlcgi_blks), "%lldk %lldk",
CFG.m_cgi_min_bufferSize >> 10, CFG.m_cgi_max_bufferSize >> 10);
if (CFG.m_cgi_prefetch_allowed)
snprintf(urlcgi_npref, sizeof(urlcgi_npref), "%d %d",
CFG.m_cgi_min_prefetch_max_blocks, CFG.m_cgi_max_prefetch_max_blocks);
char buff[8192];
int loff = 0;
loff = snprintf(buff, sizeof(buff), "Config effective %s pfc configuration:\n"
" pfc.cschk %s uvkeep %s\n"
" pfc.blocksize %lldk\n"
" pfc.prefetch %d\n"
" pfc.urlcgi blocksize %s prefetch %s\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 >> 10,
m_configuration.m_prefetch_max_blocks,
urlcgi_blks, urlcgi_npref,
ram_gb,
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: size_of_dirlist %d, size_of_globlist %d)\n",
m_configuration.m_dirStatsInterval, 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);
}
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 = CFG.m_prefetch_max_blocks > 0 || CFG.m_cgi_max_prefetch_max_blocks > 0;
Info::s_maxNumAccess = CFG.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;
}
XrdOucGetCache
XrdOucCache * XrdOucGetCache(XrdSysLogger *logger, const char *config_filename, const char *parameters, XrdOucEnv *env)
Definition XrdPfc.cc:76
open
#define open
Definition XrdPosix.hh:76
XrdOfsConfigPI::Parse
bool Parse(TheLib what)
Definition XrdOfsConfigPI.cc:490
XrdOfsConfigPI::Plugin
bool Plugin(XrdAccAuthorize *&piP)
Get Authorization plugin.
Definition XrdOfsConfigPI.cc:694
XrdOfsConfigPI::New
static XrdOfsConfigPI * New(const char *cfn, XrdOucStream *cfgP, XrdSysError *errP, XrdVersionInfo *verP=0, XrdSfsFileSystem *sfsP=0)
Definition XrdOfsConfigPI.cc:466
XrdOfsConfigPI::Load
bool Load(int what, XrdOucEnv *envP=0)
Definition XrdOfsConfigPI.cc:382
XrdOfsConfigPI::Push
bool Push(TheLib what, const char *plugP, const char *parmP=0)
Definition XrdOfsConfigPI.cc:722
XrdOfsConfigPI::theOssLib
@ theOssLib
Oss plugin.
Definition XrdOfsConfigPI.hh:74
XrdOssVSInfo::Total
long long Total
Definition XrdOssVS.hh:90
XrdOucEnv::Get
char * Get(const char *varname)
Definition XrdOucEnv.hh:69
XrdOucEnv::Export
static int Export(const char *Var, const char *Val)
Definition XrdOucEnv.cc:170
XrdOucEnv::Put
void Put(const char *varname, const char *value)
Definition XrdOucEnv.hh:85
XrdOucUtils::UserName
static int UserName(uid_t uID, char *uName, int uNsz)
Definition XrdOucUtils.cc:1396
XrdOuca2x::a2ll
static int a2ll(XrdSysError &, const char *emsg, const char *item, long long *val, long long minv=-1, long long maxv=-1)
Definition XrdOuca2x.cc:70
XrdPfc::Cache::Config
bool Config(const char *config_filename, const char *parameters, XrdOucEnv *env)
Parse configuration file.
Definition XrdPfcConfiguration.cc:431
XrdPfc::Info::s_maxNumAccess
static size_t s_maxNumAccess
Definition XrdPfcInfo.hh:311
XrdPfc::CSChk_Both
@ CSChk_Both
Definition XrdPfcTypes.hh:27
XrdPfc::CSChk_None
@ CSChk_None
Definition XrdPfcTypes.hh:27
XrdPfc::Configuration::m_cgi_max_bufferSize
long long m_cgi_max_bufferSize
max buffer size allowed in pfc.blocksize
Definition XrdPfc.hh:115
XrdPfc::Configuration::m_accHistorySize
int m_accHistorySize
max number of entries in access history part of cinfo file
Definition XrdPfc.hh:100
XrdPfc::Configuration::m_cgi_min_prefetch_max_blocks
int m_cgi_min_prefetch_max_blocks
min prefetch block count allowed in pfc.prefetch
Definition XrdPfc.hh:116
XrdPfc::Configuration::m_cgi_prefetch_allowed
bool m_cgi_prefetch_allowed
allow cgi setting of prefetch
Definition XrdPfc.hh:119
XrdPfc::Configuration::m_prefetch_max_blocks
int m_prefetch_max_blocks
default maximum number of blocks to prefetch per file
Definition XrdPfc.hh:112
XrdPfc::Configuration::m_cgi_min_bufferSize
long long m_cgi_min_bufferSize
min buffer size allowed in pfc.blocksize
Definition XrdPfc.hh:114
XrdPfc::Configuration::m_cgi_max_prefetch_max_blocks
int m_cgi_max_prefetch_max_blocks
max prefetch block count allowed in pfc.prefetch
Definition XrdPfc.hh:117
XrdPfc::Configuration::m_cgi_blocksize_allowed
bool m_cgi_blocksize_allowed
allow cgi setting of blocksize
Definition XrdPfc.hh:118
XrdPfc::TmpConfiguration::m_diskUsageLWM
std::string m_diskUsageLWM
Definition XrdPfc.hh:141
XrdPfc::TmpConfiguration::m_diskUsageHWM
std::string m_diskUsageHWM
Definition XrdPfc.hh:142
XrdPfc::TmpConfiguration::m_fileUsageBaseline
std::string m_fileUsageBaseline
Definition XrdPfc.hh:143
XrdPfc::TmpConfiguration::m_fileUsageNominal
std::string m_fileUsageNominal
Definition XrdPfc.hh:144
XrdPfc::TmpConfiguration::m_flushRaw
std::string m_flushRaw
Definition XrdPfc.hh:146
XrdPfc::TmpConfiguration::m_fileUsageMax
std::string m_fileUsageMax
Definition XrdPfc.hh:145

References XrdOuca2x::a2ll(), XrdOuca2x::a2sz(), Config(), XrdPfc::CSChk_Both, XrdPfc::CSChk_None, Error, XrdOucEnv::Export(), XrdOucEnv::Get(), XrdOfsConfigPI::Load(), XrdPfc::Configuration::m_accHistorySize, XrdPfc::Configuration::m_cgi_blocksize_allowed, XrdPfc::Configuration::m_cgi_max_bufferSize, XrdPfc::Configuration::m_cgi_max_prefetch_max_blocks, XrdPfc::Configuration::m_cgi_min_bufferSize, XrdPfc::Configuration::m_cgi_min_prefetch_max_blocks, XrdPfc::Configuration::m_cgi_prefetch_allowed, 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::Configuration::m_prefetch_max_blocks, 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 1000 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;
}
XrdOssOK
#define XrdOssOK
Definition XrdOss.hh:50
stat
#define stat(a, b)
Definition XrdPosix.hh:101
File
XrdOucString File
Definition XrdSecpwdSrvAdmin.cc:159
XrdPfc::Cache::DetermineFullFileSize
long long DetermineFullFileSize(const std::string &cinfo_fname)
Definition XrdPfc.cc:926
XrdPfc::Cache::DecideIfConsideredCached
bool DecideIfConsideredCached(long long file_size, long long bytes_on_disk)
Definition XrdPfc.cc:967
XrdPfc::File::Fstat
int Fstat(struct stat &sbuff)
Definition XrdPfcFile.cc:629
XrdPfc::Info::s_infoExtension
static const char * s_infoExtension
Definition XrdPfcInfo.hh:309

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;
}
XrdPfc::Cache::Cache
Cache(XrdSysLogger *logger, XrdOucEnv *env)
Constructor.
Definition XrdPfc.cc:158

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;
}
XrdPfc::Decision::Decide
virtual bool Decide(const std::string &, XrdOss &) const =0

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 967 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 711 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 926 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;
}
XrdSysXAttrActive
XrdSysXAttr * XrdSysXAttrActive
Definition XrdSysFAttr.cc:61
XrdOssDF::Close
virtual int Close(long long *retsz=0)=0
XrdOssDF::Open
virtual int Open(const char *path, int Oflag, mode_t Mode, XrdOucEnv &env)
Definition XrdOss.hh:200
XrdSysXAttr::Get
virtual int Get(const char *Aname, void *Aval, int Avsz, const char *Path, int fd=-1)=0
TPC::Info
@ Info
Definition XrdHttpTpcTPC.hh:29

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 << "'");
}
}
Stat
struct stat Stat
Definition XrdCks.cc:49
usage
void usage()
Definition XrdClReplay.cc:1072
XRDOSS_mkpath
#define XRDOSS_mkpath
Definition XrdOss.hh:466
ERRNO_AND_ERRSTR
#define ERRNO_AND_ERRSTR(err_code)
Definition XrdPfcTrace.hh:46
Create
bool Create
Definition XrdSecpwdSrvAdmin.cc:176
XrdOssDF::getFD
virtual int getFD()
Definition XrdOss.hh:426
XrdOss::newFile
virtual XrdOssDF * newFile(const char *tident)=0
XrdOss::Stat
virtual int Stat(const char *path, struct stat *buff, int opts=0, XrdOucEnv *envP=0)=0
XrdOuca2x::a2i
static int a2i(XrdSysError &, const char *emsg, const char *item, int *val, int minv=-1, int maxv=-1)
Definition XrdOuca2x.cc:45
XrdPfc::Cache::UnlinkFile
int UnlinkFile(const std::string &f_name, bool fail_if_open)
Remove cinfo and data files from cache.
Definition XrdPfc.cc:1188
XrdPfc::Cache::GetOss
XrdOss * GetOss() const
Definition XrdPfc.hh:280
XrdPfc::Stats::m_StBlocksAdded
long long m_StBlocksAdded
number of 512-byte blocks the file has grown by
Definition XrdPfcStats.hh:43
XrdPfc::Stats::m_BytesWritten
long long m_BytesWritten
number of bytes written to disk
Definition XrdPfcStats.hh:42
XrdPfc::Configuration::m_data_space
std::string m_data_space
oss space for data files
Definition XrdPfc.hh:88
XrdPfc::Configuration::m_bufferSize
long long m_bufferSize
cache block size, default 128 kB
Definition XrdPfc.hh:107
XrdPfc::Configuration::m_meta_space
std::string m_meta_space
oss space for metadata files (cinfo)
Definition XrdPfc.hh:89
XrdPfc::Configuration::m_username
std::string m_username
username passed to oss plugin
Definition XrdPfc.hh:87

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 542 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 389 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, io->GetInput());
}
{
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;
}
XrdOucCacheIO::Fstat
virtual int Fstat(struct stat &sbuff)
Definition XrdOucCache.hh:148
XrdPfc::File::AddIO
void AddIO(IO *io)
Definition XrdPfcFile.cc:349
XrdPfc::File::FileOpen
static File * FileOpen(const std::string &path, long long offset, long long fileSize, XrdOucCacheIO *inputIO)
Static constructor that also does Open. Returns null ptr if Open fails.
Definition XrdPfcFile.cc:141
XrdPfc::IO::GetInput
XrdOucCacheIO * GetInput()
Definition XrdPfcIO.cc:31

References XrdPfc::File::AddIO(), Debug, Error, XrdPfc::File::FileOpen(), XrdOucCacheIO::Fstat(), XrdPfc::IO::GetInput(), 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 298 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::ResourceMonitor::perform_purge_check(), XrdPfc::ResourceMonitor::perform_purge_task_cleanup(), PrefetchThread(), 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 | ( |  | ) | const

inline

Definition at line 294 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 733 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 280 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 284 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 | ( |  | ) | const

inline

Definition at line 295 of file XrdPfc.hh.

{ return m_trace; }

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

Here is the caller graph for this function:

## is_prefetch_enabled()

bool XrdPfc::Cache::is_prefetch_enabled | ( |  | ) | const

inline

Definition at line 307 of file XrdPfc.hh.

{ return m_prefetch_enabled; }

Referenced by XrdOucGetCache().

Here is the caller graph for this function:

## IsFileActiveOrPurgeProtected()

bool Cache::IsFileActiveOrPurgeProtected | ( | const std::string & | path | ) | const

Definition at line 677 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 794 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;
}
XrdOucCache::ForAccess
@ ForAccess
Definition XrdOucCache.hh:570
XrdOucCache::ForPath
@ ForPath
Definition XrdOucCache.hh:570
XrdOucCache::ForInfo
@ ForInfo
Definition XrdOucCache.hh:570

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 752 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);
}
}
}
XrdPfc::Cache::GetNextFileToPrefetch
File * GetNextFileToPrefetch()
Definition XrdPfc.cc:733
XrdPfc::File::Prefetch
void Prefetch()
Definition XrdPfcFile.cc:1621
XrdSysTimer::Wait
static void Wait(int milliseconds)
Definition XrdSysTimer.cc:227

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:

## prefetch_str2value()

bool Cache::prefetch_str2value | ( | const char * | from,
|  | const char * | str,
|  | int & | val,
|  | int | min,
|  | int | max ) const

Definition at line 118 of file XrdPfcConfiguration.cc.

{
if (XrdOuca2x::a2i(m_log, "Error parsing prefetch block count", str, &val, min, max))
return false;
return true;
}

References XrdOuca2x::a2i().

Here is the call 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 1066 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)
{
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;
}
}
XrdPfc::Cache::schedP
static XrdScheduler * schedP
Definition XrdPfc.hh:302

References XrdCl::URL::GetPath(), 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 273 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);
}
}
}
XrdPfc::Block::m_file
File * m_file
Definition XrdPfcFile.hh:109
XrdPfc::File::lPath
const char * lPath() const
Log path.
Definition XrdPfcFile.cc:1606
XrdPfc::File::WriteBlockToDisk
void WriteBlockToDisk(Block *b)
Definition XrdPfcFile.cc:1134

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 215 of file XrdPfc.hh.

{ return m_configuration; }

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

Here is the caller graph for this function:

## RefResMon()

ResourceMonitor & XrdPfc::Cache::RefResMon | ( |  | ) |

inline

Definition at line 297 of file XrdPfc.hh.

{ return *m_res_mon; }

## RegisterPrefetchFile()

void Cache::RegisterPrefetchFile | ( | File * | file | ) |

Definition at line 695 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 471 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);
}
XrdPfc::File::RemoveIO
void RemoveIO(IO *io)
Definition XrdPfcFile.cc:386

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 371 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 240 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 331 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 290 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 1117 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::File::GetLog(), XrdPfc::File::GetTrace(), 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 1178 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 1188 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);
m_active_cond.Broadcast();
}
return std::min(f_ret, i_ret);
}
XrdPfc::Cache::RemoveWriteQEntriesFor
void RemoveWriteQEntriesFor(File *f)
Remove blocks from write queue which belong to given prefetch. This method is used at the time of Fil...
Definition XrdPfc.cc:240
XrdPfc::File::initiate_emergency_shutdown
long long initiate_emergency_shutdown()
Definition XrdPfcFile.cc:154

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()

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

inline static

Version check.

Definition at line 245 of file XrdPfc.hh.

{ return true; }

## WriteFileSizeXAttr()

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

Definition at line 911 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);
}
}
}
XrdSysXAttr::Set
virtual int Set(const char *Aname, const void *Aval, int Avsz, const char *Path, int fd=-1, int isNew=0)=0

References Debug, TRACE, and XrdSysXAttrActive.

## WritesSinceLastCall()

long long Cache::WritesSinceLastCall | ( |  | ) |

Definition at line 320 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 302 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