22 #include <sys/statvfs.h>
51 Cache * Cache::m_instance = 0;
85 const char *config_filename,
86 const char *parameters,
90 err.
Say(
"++++++ Proxy file cache initialization started.");
101 if (! instance.
Config(config_filename, parameters))
103 err.
Say(
"Config Proxy file cache initialization failed.");
106 err.
Say(
"------ Proxy file cache initialization completed.");
111 for (
int wti = 0; wti < instance.
RefConfiguration().m_wqueue_threads; ++wti)
127 env->
PutPtr(
"XrdFSCtl_PC*", pfcFSctl);
136 double &frac_du,
double &frac_fu)
150 frac_du = std::min( std::max( frac_du, 0.0), 1.0 );
151 frac_fu = std::min( std::max( frac_fu, 0.0), 1.0 );
158 assert (m_instance == 0);
159 m_instance =
new Cache(logger, env);
169 if (! m_decisionpoints.empty())
172 std::string filename = url.
GetPath();
173 std::vector<Decision*>::const_iterator it;
174 for (it = m_decisionpoints.begin(); it != m_decisionpoints.end(); ++it)
178 if (! d->
Decide(filename, *m_oss))
191 m_log(logger,
"XrdPfc_"),
196 m_prefetch_condVar(0),
197 m_prefetch_enabled(false),
199 m_RAM_write_queue(0),
204 m_stats_n_purge_cond(0),
206 m_last_scan_duration(0),
207 m_last_purge_duration(0),
208 m_spt_state(SPTS_Idle)
216 const char* tpfx =
"Attach() ";
237 TRACE(
Error, tpfx <<
"Failed opening local file, falling back to remote access " << io->
Path());
245 ((loc && loc[0] != 0) ? loc :
"<deferred open>"));
265 m_writeQ.condVar.Lock();
267 m_writeQ.queue.push_back(b);
269 m_writeQ.queue.push_front(b);
271 m_writeQ.condVar.Signal();
272 m_writeQ.condVar.UnLock();
277 std::list<Block*> removed_blocks;
278 long long sum_size = 0;
280 m_writeQ.condVar.Lock();
281 std::list<Block*>::iterator i = m_writeQ.queue.begin();
282 while (i != m_writeQ.queue.end())
284 if ((*i)->m_file == file)
286 TRACE(Dump,
"Remove entries for " << (
void*)(*i) <<
" path " << file->
lPath());
287 std::list<Block*>::iterator j = i++;
288 removed_blocks.push_back(*j);
289 sum_size += (*j)->get_size();
290 m_writeQ.queue.erase(j);
298 m_writeQ.condVar.UnLock();
302 m_RAM_write_queue -= sum_size;
314 m_writeQ.condVar.Lock();
315 while (m_writeQ.size == 0)
317 m_writeQ.condVar.Wait();
323 int n_pushed = std::min(m_writeQ.size, m_configuration.
m_wqueue_blocks);
324 long long sum_size = 0;
326 for (
int bi = 0; bi < n_pushed; ++bi)
328 Block* block = m_writeQ.queue.front();
329 m_writeQ.queue.pop_front();
330 m_writeQ.writes_between_purges += block->
get_size();
333 blks_to_write[bi] = block;
335 TRACE(Dump,
"ProcessWriteTasks for block " << (
void*)(block) <<
" path " << block->
m_file->
lPath());
337 m_writeQ.size -= n_pushed;
339 m_writeQ.condVar.UnLock();
343 m_RAM_write_queue -= sum_size;
346 for (
int bi = 0; bi < n_pushed; ++bi)
348 Block* block = blks_to_write[bi];
359 static const size_t s_block_align = sysconf(_SC_PAGESIZE);
365 long long total = m_RAM_used + size;
370 if (std_size && m_RAM_std_size > 0)
372 char *buf = m_RAM_std_blocks.back();
373 m_RAM_std_blocks.pop_back();
384 if (posix_memalign((
void**) &buf, s_block_align, (
size_t) size))
407 m_RAM_std_blocks.push_back(buf);
419 TRACE(
Debug,
"GetFile " << path <<
", io " << io);
428 it = m_active.find(path);
432 if (it == m_active.end())
434 it = m_active.insert(std::make_pair(path, (
File*) 0)).first;
440 it->second->AddIO(io);
441 inc_ref_cnt(it->second,
false,
true);
448 m_active_cond.
Wait();
457 int res = io->
Fstat(st);
460 TRACE(
Error,
"GetFile, could not get valid stat");
461 }
else if (res > 0) {
463 TRACE(
Error,
"GetFile, stat returned positive value, this should NOT happen here");
465 filesize = st.st_size;
481 inc_ref_cnt(file,
false,
true);
508 dec_ref_cnt(f,
true);
515 class DiskSyncer :
public XrdJob
522 DiskSyncer(
File *f,
bool high_debug,
const char *desc =
"") :
525 m_high_debug(high_debug)
537 class CommandExecutor :
public XrdJob
540 std::string m_command_url;
543 CommandExecutor(
const std::string& command,
const char *desc =
"") :
545 m_command_url(command)
559 void Cache::schedule_file_sync(
File* f,
bool ref_cnt_already_set,
bool high_debug)
561 DiskSyncer* ds =
new DiskSyncer(f, high_debug);
563 if ( ! ref_cnt_already_set) inc_ref_cnt(f,
true, high_debug);
570 dec_ref_cnt(f, high_debug);
573 void Cache::inc_ref_cnt(
File* f,
bool lock,
bool high_debug)
579 if (lock) m_active_cond.
Lock();
581 if (lock) m_active_cond.
UnLock();
586 void Cache::dec_ref_cnt(
File* f,
bool high_debug)
608 <<
" -- deleting File object without further ado");
633 schedule_file_sync(f,
true,
true);
642 TRACE_INT(tlvl,
"dec_ref_cnt " << f->
GetLocalPath() <<
", cnt after sync_check and dec_ref_cnt = " << cnt);
656 int len = snprintf(buf, 4096,
"{\"event\":\"file_close\","
657 "\"lfn\":\"%s\",\"size\":%lld,\"blk_size\":%d,\"n_blks\":%d,\"n_blks_done\":%d,"
658 "\"access_cnt\":%lu,\"attach_t\":%lld,\"detach_t\":%lld,\"remotes\":%s,"
659 "\"b_hit\":%lld,\"b_miss\":%lld,\"b_bypass\":%lld,"
660 "\"b_todisk\":%lld,\"b_prefetch\":%lld,\"n_cks_errs\":%d}",
671 suc = m_gstream->
Insert(buf, len + 1);
675 TRACE(
Error,
"Failed g-stream insertion of file_close record, len=" << len);
688 return m_active.find(path) != m_active.end() ||
689 m_purge_delay_set.find(path) != m_purge_delay_set.end();
701 if ( ! m_prefetch_enabled)
706 m_prefetch_condVar.
Lock();
707 m_prefetchList.push_back(file);
708 m_prefetch_condVar.
Signal();
709 m_prefetch_condVar.
UnLock();
717 if ( ! m_prefetch_enabled)
722 m_prefetch_condVar.
Lock();
723 for (PrefetchList::iterator it = m_prefetchList.begin(); it != m_prefetchList.end(); ++it)
727 m_prefetchList.erase(it);
731 m_prefetch_condVar.
UnLock();
737 m_prefetch_condVar.
Lock();
738 while (m_prefetchList.empty())
740 m_prefetch_condVar.
Wait();
745 size_t l = m_prefetchList.size();
746 int idx = rand() % l;
747 File* f = m_prefetchList[idx];
749 m_prefetch_condVar.
UnLock();
761 bool doPrefetch = (m_RAM_used < limit_RAM);
799 static const mode_t groupReadable = S_IRUSR | S_IWUSR | S_IRGRP;
800 static const mode_t worldReadable = S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH;
801 static const char *lfpReason[] = {
"ForAccess",
"ForInfo",
"ForPath" };
803 TRACE(
Debug,
"LocalFilePath '" << curl <<
"', why=" << lfpReason[why]);
805 if (buff && blen > 0) buff[0] = 0;
808 std::string f_name = url.
GetPath();
813 int ret = m_oss->
Lfn2Pfn(f_name.c_str(), buff, blen);
814 TRACE(
Info,
"LocalFilePath '" << curl <<
"', why=" << lfpReason[why] <<
" -> " << ret);
820 m_purge_delay_set.insert(f_name);
823 struct stat sbuff, sbuff2;
827 if (S_ISDIR(sbuff.st_mode))
829 TRACE(
Info,
"LocalFilePath '" << curl <<
"', why=" << lfpReason[why] <<
" -> EISDIR");
834 bool read_ok =
false;
835 bool is_complete =
false;
847 m_active_cond.
Lock();
849 bool is_active = m_active.find(f_name) != m_active.end();
851 if (is_active) m_active_cond.
UnLock();
855 int res = infoFile->
Open(i_name.c_str(), O_RDWR, 0600, myEnv);
858 Info info(m_trace, 0);
859 if (info.
Read(infoFile, i_name.c_str()))
866 if ( ! is_active && is_complete && why ==
ForAccess)
869 info.
Write(infoFile, i_name.c_str());
876 if ( ! is_active) m_active_cond.
UnLock();
880 if ((is_complete || why ==
ForInfo) && buff != 0)
882 int res2 = m_oss->
Lfn2Pfn(f_name.c_str(), buff, blen);
889 {mode_t mode = (forall ? worldReadable : groupReadable);
890 if (((sbuff.st_mode & worldReadable) != mode)
892 {is_complete =
false;
898 TRACE(
Info,
"LocalFilePath '" << curl <<
"', why=" << lfpReason[why] <<
899 (is_complete ?
" -> FILE_COMPLETE_IN_CACHE" :
" -> EREMOTE"));
901 return is_complete ? 0 : -EREMOTE;
906 TRACE(
Info,
"LocalFilePath '" << curl <<
"', why=" << lfpReason[why] <<
" -> ENOENT");
918 TRACE(
Debug,
"WriteFileSizeXAttr error setting xattr " << res);
932 m_oss->
Lfn2Pfn(cinfo_fname.c_str(), pfn, 4096);
933 long long fsize = -1ll;
935 if (res ==
sizeof(
long long))
941 TRACE(
Debug,
"DetermineFullFileSize error getting xattr " << res);
947 int res = infoFile->
Open(cinfo_fname.c_str(), O_RDONLY, 0600, env);
950 Info info(m_trace, 0);
951 if ( ! info.
Read(infoFile, cinfo_fname.c_str()))
963 if (file_size == 0 || bytes_on_disk >= file_size)
966 double frac_on_disk = (double) bytes_on_disk / file_size;
996 static const char* tpfx =
"ConsiderCached ";
1001 std::string f_name = url.
GetPath();
1003 File *file =
nullptr;
1006 auto it = m_active.find(f_name);
1007 if (it != m_active.end()) {
1009 inc_ref_cnt(file,
false,
false);
1014 int res = file->
Fstat(sbuff);
1015 dec_ref_cnt(file,
false);
1019 return sbuff.st_atime > 0 ? 0 : -EREMOTE;
1023 int res = m_oss->
Stat(f_name.c_str(), &sbuff);
1028 if (S_ISDIR(sbuff.st_mode))
1035 if (file_size < 0) {
1036 TRACE(
Debug, tpfx << curl <<
" -> " << file_size);
1037 return (
int) file_size;
1041 return is_cached ? 0 : -EREMOTE;
1057 std::string f_name = url.
GetPath();
1061 if (oflags & (O_WRONLY | O_RDWR | O_APPEND | O_CREAT))
1063 TRACE(
Warning,
"Prepare write access requested on file " << f_name <<
". Denying access.");
1072 CommandExecutor *ce =
new CommandExecutor(f_name,
"CommandExecutor");
1082 m_purge_delay_set.insert(f_name);
1088 TRACE(Dump,
"Prepare defer open " << f_name);
1107 const char *tpfx =
"Stat ";
1110 std::string f_name = url.
GetPath();
1112 File *file =
nullptr;
1115 auto it = m_active.find(f_name);
1116 if (it != m_active.end()) {
1118 inc_ref_cnt(file,
false,
false);
1122 int res = file->
Fstat(sbuff);
1123 dec_ref_cnt(file,
false);
1124 TRACE(
Debug, tpfx <<
"from active file " << curl <<
" -> " << res);
1128 int res = m_oss->
Stat(f_name.c_str(), &sbuff);
1133 if (S_ISDIR(sbuff.st_mode))
1140 if (file_size < 0) {
1141 TRACE(
Debug, tpfx << curl <<
" -> " << file_size);
1144 sbuff.st_size = file_size;
1149 TRACE(
Debug, tpfx <<
"from disk " << curl <<
" -> " << res);
1164 std::string f_name = url.
GetPath();
1178 it = m_active.find(f_name);
1180 if (it != m_active.end())
1184 TRACE(
Info,
"UnlinkCommon " << f_name <<
", file currently open and force not requested - denying request");
1190 if (it->second == 0)
1192 TRACE(
Info,
"UnlinkCommon " << f_name <<
", an operation on this file is ongoing - denying request");
1202 it = m_active.insert(std::make_pair(f_name, (
File*) 0)).first;
1214 int f_ret = m_oss->
Unlink(f_name.c_str());
1215 int i_ret = m_oss->
Unlink(i_name.c_str());
1217 TRACE(
Debug,
"UnlinkCommon " << f_name <<
", f_ret=" << f_ret <<
", i_ret=" << i_ret);
1225 return std::min(f_ret, i_ret);
int DoIt(int argpnt, int argc, char **argv, bool singleshot)
std::string obfuscateAuth(const std::string &input)
#define TRACE_PC(act, pre_code, x)
#define TRACE_INT(act, x)
XrdOucCache * XrdOucGetCache(XrdSysLogger *logger, const char *config_filename, const char *parameters, XrdOucEnv *env)
void * PrefetchThread(void *)
void * PurgeThread(void *)
XrdSysXAttr * XrdSysXAttrActive
void * ResourceMonitorHeartBeatThread(void *)
void * ProcessWriteTaskThread(void *)
int stat(const char *path, struct stat *buf)
const std::string & GetPath() const
Get the path.
virtual int Close(long long *retsz=0)=0
virtual int Open(const char *path, int Oflag, mode_t Mode, XrdOucEnv &env)
virtual int Chmod(const char *path, mode_t mode, XrdOucEnv *envP=0)=0
virtual int Lfn2Pfn(const char *Path, char *buff, int blen)
virtual XrdOssDF * newFile(const char *tident)=0
virtual int Stat(const char *path, struct stat *buff, int opts=0, XrdOucEnv *envP=0)=0
virtual int Unlink(const char *path, int Opts=0, XrdOucEnv *envP=0)=0
virtual const char * Path()=0
virtual int Fstat(struct stat &sbuff)
virtual const char * Location(bool refresh=false)
void * GetPtr(const char *varname)
void PutPtr(const char *varname, void *value)
Attaches/creates and detaches/deletes cache-io objects for disk based cache.
long long DetermineFullFileSize(const std::string &cinfo_fname)
void FileSyncDone(File *, bool high_debug)
File * GetFile(const std::string &, IO *, long long off=0, long long filesize=0)
static const Configuration & Conf()
bool Config(const char *config_filename, const char *parameters)
Parse configuration file.
virtual int LocalFilePath(const char *url, char *buff=0, int blen=0, LFP_Reason why=ForAccess, bool forall=false)
virtual int Stat(const char *url, struct stat &sbuff)
const Configuration & RefConfiguration() const
Reference XrdPfc configuration.
void Purge()
Thread function invoked to scan and purge files from disk when needed.
void ReleaseRAM(char *buf, long long size)
virtual int ConsiderCached(const char *url)
static Cache & GetInstance()
Singleton access.
void ResourceMonitorHeartBeat()
Thread function checking resource usage periodically.
void DeRegisterPrefetchFile(File *)
void ExecuteCommandUrl(const std::string &command_url)
void RegisterPrefetchFile(File *)
void WriteFileSizeXAttr(int cinfo_fd, long long file_size)
void ReleaseFile(File *, IO *)
void AddWriteTask(Block *b, bool from_read)
Add downloaded block in write queue.
Cache(XrdSysLogger *logger, XrdOucEnv *env)
Constructor.
bool Decide(XrdOucCacheIO *)
Makes decision if the original XrdOucCacheIO should be cached.
int UnlinkFile(const std::string &f_name, bool fail_if_open)
Remove cinfo and data files from cache.
virtual XrdOucCacheIO * Attach(XrdOucCacheIO *ioP, int opts=0)=0
Obtain a new IO object that fronts existing XrdOucCacheIO.
static XrdScheduler * schedP
bool IsFileActiveOrPurgeProtected(const std::string &)
File * GetNextFileToPrefetch()
void ProcessWriteTasks()
Separate task which writes blocks from ram to disk.
virtual int Unlink(const char *url)
void RemoveWriteQEntriesFor(File *f)
Remove blocks from write queue which belong to given prefetch. This method is used at the time of Fil...
static const Cache & TheOne()
char * RequestRAM(long long size)
virtual int Prepare(const char *url, int oflags, mode_t mode)
bool DecideIfConsideredCached(long long file_size, long long bytes_on_disk)
static Cache & CreateInstance(XrdSysLogger *logger, XrdOucEnv *env)
Singleton creation.
Base class for selecting which files should be cached.
virtual bool Decide(const std::string &, XrdOss &) const =0
bool FinalizeSyncBeforeExit()
Returns true if any of blocks need sync. Called from Cache::dec_ref_cnt on zero ref cnt.
const char * lPath() const
Log path.
void WriteBlockToDisk(Block *b)
static File * FileOpen(const std::string &path, long long offset, long long fileSize)
Static constructor that also does Open. Returns null ptr if Open fails.
std::string GetRemoteLocations() const
const Info::AStat * GetLastAccessStats() const
size_t GetAccessCnt() const
int Fstat(struct stat &sbuff)
long long GetPrefetchedBytes() const
int GetNDownloadedBlocks() const
void BlocksRemovedFromWriteQ(std::list< Block * > &)
Handle removal of a set of blocks from Cache's write queue.
void initiate_emergency_shutdown()
void Sync()
Sync file cache inf o and output data with disk.
const Stats & RefStats() const
std::string & GetLocalPath()
Stats DeltaStatsFromLastCall()
bool is_in_emergency_shutdown()
Downloads original file into multiple files, chunked into blocks. Only blocks that are asked for are ...
Downloads original file into a single file on local disk. Handles read requests as they come along.
bool HasFile() const
Check if File was opened successfully.
Base cache-io class that implements some XrdOucCacheIO abstract methods.
Status of cached file. Can be read from and written into a binary file.
static const char * s_infoExtension
void WriteIOStatSingle(long long bytes_disk)
Write single open/close time for given bytes read from disk.
bool Write(XrdOssDF *fp, const char *dname, const char *fname=0)
bool IsComplete() const
Get complete status.
long long GetFileSize() const
Get file size.
bool Read(XrdOssDF *fp, const char *dname, const char *fname=0)
Read content of cinfo file into this object.
Statistics of cache utilisation by a File object.
int m_NCksumErrors
number of checksum errors while getting data from remote
long long m_BytesWritten
number of bytes written to disk
void Schedule(XrdJob *jp)
void Say(const char *text1, const char *text2=0, const char *txt3=0, const char *text4=0, const char *text5=0, const char *txt6=0)
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)
static void Wait(int milliseconds)
virtual int Get(const char *Aname, void *Aval, int Avsz, const char *Path, int fd=-1)=0
virtual int Set(const char *Aname, const void *Aval, int Avsz, const char *Path, int fd=-1, int isNew=0)=0
bool Insert(const char *data, int dlen)
Contains parameters configurable from the xrootd config file.
long long m_RamAbsAvailable
available from configuration
long long m_fileUsageMax
cache purge - files usage maximum
long long m_fileUsageBaseline
cache purge - files usage baseline
bool m_allow_xrdpfc_command
flag for enabling access to /xrdpfc-command/ functionality.
long long m_diskUsageHWM
cache purge - disk usage high water mark
int m_prefetch_max_blocks
maximum number of blocks to prefetch per file
void calculate_fractional_usages(long long du, long long fu, double &frac_du, double &frac_fu)
long long m_diskUsageLWM
cache purge - disk usage low water mark
int m_RamKeepStdBlocks
number of standard-sized blocks kept after release
long long m_bufferSize
prefetch buffer size, default 1MB
int m_wqueue_blocks
maximum number of blocks written per write-queue loop
std::string m_username
username passed to oss plugin
double m_onlyIfCachedMinFrac
minimum fraction of downloaded file, used by only-if-cached CGI option
long long m_onlyIfCachedMinSize
minumum size of downloaded file, used by only-if-cached CGI option
long long BytesHit
read from cache
long long BytesBypassed
read from remote and dropped
time_t DetachTime
close time
long long BytesMissed
read from remote and cached
time_t AttachTime
open time