XRootD
Loading...
Searching...
No Matches
XrdPfc.cc
Go to the documentation of this file.
1//----------------------------------------------------------------------------------
2// Copyright (c) 2014 by Board of Trustees of the Leland Stanford, Jr., University
3// Author: Alja Mrak-Tadel, Matevz Tadel, Brian Bockelman
4//----------------------------------------------------------------------------------
5// XRootD is free software: you can redistribute it and/or modify
6// it under the terms of the GNU Lesser General Public License as published by
7// the Free Software Foundation, either version 3 of the License, or
8// (at your option) any later version.
9//
10// XRootD is distributed in the hope that it will be useful,
11// but WITHOUT ANY WARRANTY; without even the implied warranty of
12// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13// GNU General Public License for more details.
14//
15// You should have received a copy of the GNU Lesser General Public License
16// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
17//----------------------------------------------------------------------------------
18
19#include <fcntl.h>
20#include <sstream>
21#include <algorithm>
22#include <sys/statvfs.h>
23
24#include "XrdCl/XrdClURL.hh"
25
26#include "XrdOuc/XrdOucEnv.hh"
27#include "XrdOuc/XrdOucUtils.hh"
29
30#include "XrdSys/XrdSysTimer.hh"
31#include "XrdSys/XrdSysTrace.hh"
32#include "XrdSys/XrdSysXAttr.hh"
33
35
36#include "XrdOss/XrdOss.hh"
37
38#include "XrdPfc.hh"
39#include "XrdPfcTrace.hh"
40#include "XrdPfcFSctl.hh"
41#include "XrdPfcInfo.hh"
42#include "XrdPfcIOFile.hh"
43#include "XrdPfcIOFileBlock.hh"
45
47
48using namespace XrdPfc;
49
50Cache *Cache::m_instance = nullptr;
52
53
55{
57 return 0;
58}
59
61{
63 return 0;
64}
65
66void *PrefetchThread(void*)
67{
69 return 0;
70}
71
72//==============================================================================
73
74extern "C"
75{
77 const char *config_filename,
78 const char *parameters,
79 XrdOucEnv *env)
80{
81 XrdSysError err(logger, "");
82 err.Say("++++++ Proxy file cache initialization started.");
83
84 if ( ! env ||
85 ! (XrdPfc::Cache::schedP = (XrdScheduler*) env->GetPtr("XrdScheduler*")))
86 {
88 XrdPfc::Cache::schedP->Start();
89 }
90
91 Cache &instance = Cache::CreateInstance(logger, env);
92
93 if (! instance.Config(config_filename, parameters, env))
94 {
95 err.Say("Config Proxy file cache initialization failed.");
96 return 0;
97 }
98 err.Say("++++++ Proxy file cache initialization completed.");
99
100 {
101 pthread_t tid;
102
103 XrdSysThread::Run(&tid, ResourceMonitorThread, 0, 0, "XrdPfc ResourceMonitor");
104
105 for (int wti = 0; wti < instance.RefConfiguration().m_wqueue_threads; ++wti)
106 {
107 XrdSysThread::Run(&tid, ProcessWriteTaskThread, 0, 0, "XrdPfc WriteTasks ");
108 }
109
110 if (instance.is_prefetch_enabled())
111 {
112 XrdSysThread::Run(&tid, PrefetchThread, 0, 0, "XrdPfc Prefetch ");
113 }
114 }
115
116 XrdPfcFSctl* pfcFSctl = new XrdPfcFSctl(instance, logger);
117 env->PutPtr("XrdFSCtl_PC*", pfcFSctl);
118
119 return &instance;
120}
121}
122
123//==============================================================================
124
126{
127 assert (m_instance == 0);
128 m_instance = new Cache(logger, env);
129 return *m_instance;
130}
131
132 Cache& Cache::GetInstance() { return *m_instance; }
133const Cache& Cache::TheOne() { return *m_instance; }
134const Configuration& Cache::Conf() { return m_instance->RefConfiguration(); }
135 ResourceMonitor& Cache::ResMon() { return m_instance->RefResMon(); }
136
138{
139 if (! m_decisionpoints.empty())
140 {
141 XrdCl::URL url(io->Path());
142 std::string filename = url.GetPath();
143 std::vector<Decision*>::const_iterator it;
144 for (it = m_decisionpoints.begin(); it != m_decisionpoints.end(); ++it)
145 {
146 XrdPfc::Decision *d = *it;
147 if (! d) continue;
148 if (! d->Decide(filename, *m_oss))
149 {
150 return false;
151 }
152 }
153 }
154
155 return true;
156}
157
159 XrdOucCache("pfc"),
160 m_env(env),
161 m_log(logger, "XrdPfc_"),
162 m_trace(new XrdSysTrace("XrdPfc", logger)),
163 m_traceID("Cache"),
164 m_oss(0),
165 m_gstream(0),
166 m_purge_pin(0),
167 m_prefetch_condVar(0),
168 m_prefetch_enabled(false),
169 m_RAM_used(0),
170 m_RAM_write_queue(0),
171 m_RAM_std_size(0),
172 m_isClient(false),
173 m_active_cond(0)
174{
175 // Default log level is Warning.
176 m_trace->What = 2;
177}
178
180{
181 const char* tpfx = "Attach() ";
182
183 if (Cache::GetInstance().Decide(io))
184 {
185 TRACE(Info, tpfx << obfuscateAuth(io->Path()));
186
187 IO *cio;
188
189 if (Cache::GetInstance().RefConfiguration().m_hdfsmode)
190 {
191 cio = new IOFileBlock(io, *this);
192 }
193 else
194 {
195 IOFile *iof = new IOFile(io, *this);
196
197 if ( ! iof->HasFile())
198 {
199 delete iof;
200 // TODO - redirect instead. But this is kind of an awkward place for it.
201 // errno is set during IOFile construction.
202 TRACE(Error, tpfx << "Failed opening local file, falling back to remote access " << io->Path());
203 return io;
204 }
205
206 cio = iof;
207 }
208
209 TRACE_PC(Debug, const char* loc = io->Location(), tpfx << io->Path() << " location: " <<
210 ((loc && loc[0] != 0) ? loc : "<deferred open>"));
211
212 return cio;
213 }
214 else
215 {
216 TRACE(Info, tpfx << "decision decline " << io->Path());
217 }
218 return io;
219}
220
221void Cache::AddWriteTask(Block* b, bool fromRead)
222{
223 TRACE(Dump, "AddWriteTask() offset=" << b->m_offset << ". file " << b->get_file()->GetLocalPath());
224
225 {
226 XrdSysMutexHelper lock(&m_RAM_mutex);
227 m_RAM_write_queue += b->get_size();
228 }
229
230 m_writeQ.condVar.Lock();
231 if (fromRead)
232 m_writeQ.queue.push_back(b);
233 else
234 m_writeQ.queue.push_front(b);
235 m_writeQ.size++;
236 m_writeQ.condVar.Signal();
237 m_writeQ.condVar.UnLock();
238}
239
241{
242 std::list<Block*> removed_blocks;
243 long long sum_size = 0;
244
245 m_writeQ.condVar.Lock();
246 std::list<Block*>::iterator i = m_writeQ.queue.begin();
247 while (i != m_writeQ.queue.end())
248 {
249 if ((*i)->m_file == file)
250 {
251 TRACE(Dump, "Remove entries for " << (void*)(*i) << " path " << file->lPath());
252 std::list<Block*>::iterator j = i++;
253 removed_blocks.push_back(*j);
254 sum_size += (*j)->get_size();
255 m_writeQ.queue.erase(j);
256 --m_writeQ.size;
257 }
258 else
259 {
260 ++i;
261 }
262 }
263 m_writeQ.condVar.UnLock();
264
265 {
266 XrdSysMutexHelper lock(&m_RAM_mutex);
267 m_RAM_write_queue -= sum_size;
268 }
269
270 file->BlocksRemovedFromWriteQ(removed_blocks);
271}
272
274{
275 std::vector<Block*> blks_to_write(m_configuration.m_wqueue_blocks);
276
277 while (true)
278 {
279 m_writeQ.condVar.Lock();
280 while (m_writeQ.size == 0)
281 {
282 m_writeQ.condVar.Wait();
283 }
284
285 // MT -- optimize to pop several blocks if they are available (or swap the list).
286 // This makes sense especially for smallish block sizes.
287
288 int n_pushed = std::min(m_writeQ.size, m_configuration.m_wqueue_blocks);
289 long long sum_size = 0;
290
291 for (int bi = 0; bi < n_pushed; ++bi)
292 {
293 Block* block = m_writeQ.queue.front();
294 m_writeQ.queue.pop_front();
295 m_writeQ.writes_between_purges += block->get_size();
296 sum_size += block->get_size();
297
298 blks_to_write[bi] = block;
299
300 TRACE(Dump, "ProcessWriteTasks for block " << (void*)(block) << " path " << block->m_file->lPath());
301 }
302 m_writeQ.size -= n_pushed;
303
304 m_writeQ.condVar.UnLock();
305
306 {
307 XrdSysMutexHelper lock(&m_RAM_mutex);
308 m_RAM_write_queue -= sum_size;
309 }
310
311 for (int bi = 0; bi < n_pushed; ++bi)
312 {
313 Block* block = blks_to_write[bi];
314
315 block->m_file->WriteBlockToDisk(block);
316 }
317 }
318}
319
321{
322 // Called from ResourceMonitor for an alternative estimation of disk writes.
323 XrdSysCondVarHelper lock(&m_writeQ.condVar);
324 long long ret = m_writeQ.writes_between_purges;
325 m_writeQ.writes_between_purges = 0;
326 return ret;
327}
328
329//==============================================================================
330
331char* Cache::RequestRAM(long long size)
332{
333 static const size_t s_block_align = sysconf(_SC_PAGESIZE);
334
335 bool std_size = (size == m_configuration.m_bufferSize);
336
337 m_RAM_mutex.Lock();
338
339 long long total = m_RAM_used + size;
340
341 if (total <= m_configuration.m_RamAbsAvailable)
342 {
343 m_RAM_used = total;
344 if (std_size && m_RAM_std_size > 0)
345 {
346 char *buf = m_RAM_std_blocks.back();
347 m_RAM_std_blocks.pop_back();
348 --m_RAM_std_size;
349
350 m_RAM_mutex.UnLock();
351
352 return buf;
353 }
354 else
355 {
356 m_RAM_mutex.UnLock();
357 char *buf;
358 if (posix_memalign((void**) &buf, s_block_align, (size_t) size))
359 {
360 // Report out of mem? Probably should report it at least the first time,
361 // then periodically.
362 return 0;
363 }
364 return buf;
365 }
366 }
367 m_RAM_mutex.UnLock();
368 return 0;
369}
370
371void Cache::ReleaseRAM(char* buf, long long size)
372{
373 bool std_size = (size == m_configuration.m_bufferSize);
374 {
375 XrdSysMutexHelper lock(&m_RAM_mutex);
376
377 m_RAM_used -= size;
378
379 if (std_size && m_RAM_std_size < m_configuration.m_RamKeepStdBlocks)
380 {
381 m_RAM_std_blocks.push_back(buf);
382 ++m_RAM_std_size;
383 return;
384 }
385 }
386 free(buf);
387}
388
389File* Cache::GetFile(const std::string& path, IO* io, long long off, long long filesize)
390{
391 // Called from virtual IOFile constructor.
392
393 TRACE(Debug, "GetFile " << path << ", io " << io);
394
395 ActiveMap_i it;
396
397 {
398 XrdSysCondVarHelper lock(&m_active_cond);
399
400 while (true)
401 {
402 it = m_active.find(path);
403
404 // File is not open or being opened. Mark it as being opened and
405 // proceed to opening it outside of while loop.
406 if (it == m_active.end())
407 {
408 it = m_active.insert(std::make_pair(path, (File*) 0)).first;
409 break;
410 }
411
412 if (it->second != 0)
413 {
414 it->second->AddIO(io);
415 inc_ref_cnt(it->second, false, true);
416
417 return it->second;
418 }
419 else
420 {
421 // Wait for some change in m_active, then recheck.
422 m_active_cond.Wait();
423 }
424 }
425 }
426
427 // This is always true, now that IOFileBlock is unsupported.
428 if (filesize == 0)
429 {
430 struct stat st;
431 int res = io->Fstat(st);
432 if (res < 0) {
433 errno = res;
434 filesize = -1;
435 TRACE(Error, "GetFile, could not get valid stat");
436 } else if (res > 0) {
437 errno = ENOTSUP;
438 filesize = -1;
439 TRACE(Error, "GetFile, stat returned positive value, this should NOT happen here");
440 } else {
441 filesize = st.st_size;
442 }
443 }
444
445 File *file = 0;
446
447 if (filesize >= 0)
448 {
449 file = File::FileOpen(path, off, filesize, io->GetInput());
450 }
451
452 {
453 XrdSysCondVarHelper lock(&m_active_cond);
454
455 if (file)
456 {
457 inc_ref_cnt(file, false, true);
458 it->second = file;
459
460 file->AddIO(io);
461 }
462 else
463 {
464 m_active.erase(it);
465 }
466
467 m_active_cond.Broadcast();
468 }
469
470 return file;
471}
472
474{
475 // Called from virtual IO::DetachFinalize.
476
477 TRACE(Debug, "ReleaseFile " << f->GetLocalPath() << ", io " << io);
478
479 {
480 XrdSysCondVarHelper lock(&m_active_cond);
481
482 f->RemoveIO(io);
483 }
484 dec_ref_cnt(f, true);
485}
486
487
488namespace
489{
490
491class DiskSyncer : public XrdJob
492{
493private:
494 File *m_file;
495 bool m_high_debug;
496
497public:
498 DiskSyncer(File *f, bool high_debug, const char *desc = "") :
499 XrdJob(desc),
500 m_file(f),
501 m_high_debug(high_debug)
502 {}
503
504 void DoIt()
505 {
506 m_file->Sync();
507 Cache::GetInstance().FileSyncDone(m_file, m_high_debug);
508 delete this;
509 }
510};
511
512
513class CommandExecutor : public XrdJob
514{
515private:
516 std::string m_command_url;
517
518public:
519 CommandExecutor(const std::string& command, const char *desc = "") :
520 XrdJob(desc),
521 m_command_url(command)
522 {}
523
524 void DoIt()
525 {
526 Cache::GetInstance().ExecuteCommandUrl(m_command_url);
527 delete this;
528 }
529};
530
531}
532
533//==============================================================================
534
535void Cache::schedule_file_sync(File* f, bool ref_cnt_already_set, bool high_debug)
536{
537 DiskSyncer* ds = new DiskSyncer(f, high_debug);
538
539 if ( ! ref_cnt_already_set) inc_ref_cnt(f, true, high_debug);
540
541 schedP->Schedule(ds);
542}
543
544void Cache::FileSyncDone(File* f, bool high_debug)
545{
546 dec_ref_cnt(f, high_debug);
547}
548
549void Cache::inc_ref_cnt(File* f, bool lock, bool high_debug)
550{
551 // called from GetFile() or SheduleFileSync();
552
553 int tlvl = high_debug ? TRACE_Debug : TRACE_Dump;
554
555 if (lock) m_active_cond.Lock();
556 int rc = f->inc_ref_cnt();
557 if (lock) m_active_cond.UnLock();
558
559 TRACE_INT(tlvl, "inc_ref_cnt " << f->GetLocalPath() << ", cnt at exit = " << rc);
560}
561
562void Cache::dec_ref_cnt(File* f, bool high_debug)
563{
564 // NOT under active lock.
565 // Called from ReleaseFile(), DiskSync callback and stat-like functions.
566
567 int tlvl = high_debug ? TRACE_Debug : TRACE_Dump;
568 int cnt;
569
570 bool emergency_close = false;
571 {
572 XrdSysCondVarHelper lock(&m_active_cond);
573
574 cnt = f->get_ref_cnt();
575 TRACE_INT(tlvl, "dec_ref_cnt " << f->GetLocalPath() << ", cnt at entry = " << cnt);
576
578 {
579 // In this case file has been already removed from m_active map and
580 // does not need to be synced.
581
582 if (cnt == 1)
583 {
584 TRACE_INT(tlvl, "dec_ref_cnt " << f->GetLocalPath() << " is in shutdown, ref_cnt = " << cnt
585 << " -- closing and deleting File object without further ado");
586 emergency_close = true;
587 }
588 else
589 {
590 TRACE_INT(tlvl, "dec_ref_cnt " << f->GetLocalPath() << " is in shutdown, ref_cnt = " << cnt
591 << " -- waiting");
592 f->dec_ref_cnt();
593 return;
594 }
595 }
596 if (cnt > 1)
597 {
598 f->dec_ref_cnt();
599 return;
600 }
601 }
602 if (emergency_close)
603 {
604 f->Close();
605 delete f;
606 return;
607 }
608
609 if (cnt == 1)
610 {
611 if (f->FinalizeSyncBeforeExit())
612 {
613 // Note, here we "reuse" the existing reference count for the
614 // final sync.
615
616 TRACE(Debug, "dec_ref_cnt " << f->GetLocalPath() << ", scheduling final sync");
617 schedule_file_sync(f, true, true);
618 return;
619 }
620 }
621
622 bool finished_p = false;
623 ActiveMap_i act_it;
624 {
625 XrdSysCondVarHelper lock(&m_active_cond);
626
627 cnt = f->dec_ref_cnt();
628 TRACE_INT(tlvl, "dec_ref_cnt " << f->GetLocalPath() << ", cnt after sync_check and dec_ref_cnt = " << cnt);
629 if (cnt == 0)
630 {
631 act_it = m_active.find(f->GetLocalPath());
632 act_it->second = 0;
633
634 finished_p = true;
635 }
636 }
637 if (finished_p)
638 {
639 f->Close();
640 {
641 XrdSysCondVarHelper lock(&m_active_cond);
642 m_active.erase(act_it);
643 m_active_cond.Broadcast();
644 }
645
646 if (m_gstream)
647 {
648 const Stats &st = f->RefStats();
649 const Info::AStat *as = f->GetLastAccessStats();
650
651 char buf[4096];
652 int len = snprintf(buf, 4096, "{\"event\":\"file_close\","
653 "\"lfn\":\"%s\",\"size\":%lld,\"blk_size\":%d,\"n_blks\":%d,\"n_blks_done\":%d,"
654 "\"access_cnt\":%lu,\"attach_t\":%lld,\"detach_t\":%lld,\"remotes\":%s,"
655 "\"b_hit\":%lld,\"b_miss\":%lld,\"b_bypass\":%lld,"
656 "\"b_todisk\":%lld,\"b_prefetch\":%lld,\"n_cks_errs\":%d}",
657 f->GetLocalPath().c_str(), f->GetFileSize(), f->GetBlockSize(),
659 (unsigned long) f->GetAccessCnt(), (long long) as->AttachTime, (long long) as->DetachTime,
660 f->GetRemoteLocations().c_str(),
661 as->BytesHit, as->BytesMissed, as->BytesBypassed,
663 );
664 bool suc = false;
665 if (len < 4096)
666 {
667 suc = m_gstream->Insert(buf, len + 1);
668 }
669 if ( ! suc)
670 {
671 TRACE(Error, "Failed g-stream insertion of file_close record, len=" << len);
672 }
673 }
674
675 delete f;
676 }
677}
678
679bool Cache::IsFileActiveOrPurgeProtected(const std::string& path) const
680{
681 XrdSysCondVarHelper lock(&m_active_cond);
682
683 return m_active.find(path) != m_active.end() ||
684 m_purge_delay_set.find(path) != m_purge_delay_set.end();
685}
686
688{
689 XrdSysCondVarHelper lock(&m_active_cond);
690 m_purge_delay_set.clear();
691}
692
693//==============================================================================
694//=== PREFETCH
695//==============================================================================
696
698{
699 // Can be called with other locks held.
700
701 if ( ! m_prefetch_enabled)
702 {
703 return;
704 }
705
706 m_prefetch_condVar.Lock();
707 m_prefetchList.push_back(file);
708 m_prefetch_condVar.Signal();
709 m_prefetch_condVar.UnLock();
710}
711
712
714{
715 // Can be called with other locks held.
716
717 if ( ! m_prefetch_enabled)
718 {
719 return;
720 }
721
722 m_prefetch_condVar.Lock();
723 for (PrefetchList::iterator it = m_prefetchList.begin(); it != m_prefetchList.end(); ++it)
724 {
725 if (*it == file)
726 {
727 m_prefetchList.erase(it);
728 break;
729 }
730 }
731 m_prefetch_condVar.UnLock();
732}
733
734
736{
737 m_prefetch_condVar.Lock();
738 while (m_prefetchList.empty())
739 {
740 m_prefetch_condVar.Wait();
741 }
742
743 // std::sort(m_prefetchList.begin(), m_prefetchList.end(), myobject);
744
745 size_t l = m_prefetchList.size();
746 int idx = rand() % l;
747 File* f = m_prefetchList[idx];
748
749 m_prefetch_condVar.UnLock();
750 return f;
751}
752
753
755{
756 const long long limit_RAM = m_configuration.m_RamAbsAvailable * 7 / 10;
757
758 while (true)
759 {
760 m_RAM_mutex.Lock();
761 bool doPrefetch = (m_RAM_used < limit_RAM);
762 m_RAM_mutex.UnLock();
763
764 if (doPrefetch)
765 {
767 f->Prefetch();
768 }
769 else
770 {
772 }
773 }
774}
775
776
777//==============================================================================
778//=== Virtuals from XrdOucCache
779//==============================================================================
780
781//------------------------------------------------------------------------------
795
796int Cache::LocalFilePath(const char *curl, char *buff, int blen,
797 LFP_Reason why, bool forall)
798{
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" };
802
803 TRACE(Debug, "LocalFilePath '" << curl << "', why=" << lfpReason[why]);
804
805 if (buff && blen > 0) buff[0] = 0;
806
807 XrdCl::URL url(curl);
808 std::string f_name = url.GetPath();
809 std::string i_name = f_name + Info::s_infoExtension;
810
811 if (why == ForPath)
812 {
813 int ret = m_oss->Lfn2Pfn(f_name.c_str(), buff, blen);
814 TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> " << ret);
815 return ret;
816 }
817
818 {
819 XrdSysCondVarHelper lock(&m_active_cond);
820 m_purge_delay_set.insert(f_name);
821 }
822
823 struct stat sbuff, sbuff2;
824 if (m_oss->Stat(f_name.c_str(), &sbuff) == XrdOssOK &&
825 m_oss->Stat(i_name.c_str(), &sbuff2) == XrdOssOK)
826 {
827 if (S_ISDIR(sbuff.st_mode))
828 {
829 TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> EISDIR");
830 return -EISDIR;
831 }
832 else
833 {
834 bool read_ok = false;
835 bool is_complete = false;
836
837 // Lock and check if the file is active. If NOT, keep the lock
838 // and add dummy access after successful reading of info file.
839 // If it IS active, just release the lock, this ongoing access will
840 // assure the file continues to exist.
841
842 // XXXX How can I just loop over the cinfo file when active?
843 // Can I not get is_complete from the existing file?
844 // Do I still want to inject access record?
845 // Oh, it writes only if not active .... still let's try to use existing File.
846
847 m_active_cond.Lock();
848
849 bool is_active = m_active.find(f_name) != m_active.end();
850
851 if (is_active) m_active_cond.UnLock();
852
853 XrdOssDF* infoFile = m_oss->newFile(m_configuration.m_username.c_str());
854 XrdOucEnv myEnv;
855 int res = infoFile->Open(i_name.c_str(), O_RDWR, 0600, myEnv);
856 if (res >= 0)
857 {
858 Info info(m_trace, 0);
859 if (info.Read(infoFile, i_name.c_str()))
860 {
861 read_ok = true;
862
863 is_complete = info.IsComplete();
864
865 // Add full-size access if reason is for access.
866 if ( ! is_active && is_complete && why == ForAccess)
867 {
868 info.WriteIOStatSingle(info.GetFileSize());
869 info.Write(infoFile, i_name.c_str());
870 }
871 }
872 infoFile->Close();
873 }
874 delete infoFile;
875
876 if ( ! is_active) m_active_cond.UnLock();
877
878 if (read_ok)
879 {
880 if ((is_complete || why == ForInfo) && buff != 0)
881 {
882 int res2 = m_oss->Lfn2Pfn(f_name.c_str(), buff, blen);
883 if (res2 < 0)
884 return res2;
885
886 // Normally, files are owned by us but when direct cache access
887 // is wanted and possible, make sure the file is world readable.
888 if (why == ForAccess)
889 {mode_t mode = (forall ? worldReadable : groupReadable);
890 if (((sbuff.st_mode & worldReadable) != mode)
891 && (m_oss->Chmod(f_name.c_str(), mode) != XrdOssOK))
892 {is_complete = false;
893 *buff = 0;
894 }
895 }
896 }
897
898 TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] <<
899 (is_complete ? " -> FILE_COMPLETE_IN_CACHE" : " -> EREMOTE"));
900
901 return is_complete ? 0 : -EREMOTE;
902 }
903 }
904 }
905
906 TRACE(Info, "LocalFilePath '" << curl << "', why=" << lfpReason[why] << " -> ENOENT");
907 return -ENOENT;
908}
909
910//______________________________________________________________________________
911// If supported, write file_size as xattr to cinfo file.
912//------------------------------------------------------------------------------
913void Cache::WriteFileSizeXAttr(int cinfo_fd, long long file_size)
914{
915 if (m_metaXattr) {
916 int res = XrdSysXAttrActive->Set("pfc.fsize", &file_size, sizeof(long long), 0, cinfo_fd, 0);
917 if (res != 0) {
918 TRACE(Debug, "WriteFileSizeXAttr error setting xattr " << res);
919 }
920 }
921}
922
923//______________________________________________________________________________
924// Determine full size of the data file from the corresponding cinfo-file name.
925// Attempts to read xattr first and falls back to reading of the cinfo file.
926// Returns -error on failure.
927//------------------------------------------------------------------------------
928long long Cache::DetermineFullFileSize(const std::string &cinfo_fname)
929{
930 if (m_metaXattr) {
931 char pfn[4096];
932 m_oss->Lfn2Pfn(cinfo_fname.c_str(), pfn, 4096);
933 long long fsize = -1ll;
934 int res = XrdSysXAttrActive->Get("pfc.fsize", &fsize, sizeof(long long), pfn);
935 if (res == sizeof(long long))
936 {
937 return fsize;
938 }
939 else
940 {
941 TRACE(Debug, "DetermineFullFileSize error getting xattr " << res);
942 }
943 }
944
945 XrdOssDF *infoFile = m_oss->newFile(m_configuration.m_username.c_str());
946 XrdOucEnv env;
947 long long ret;
948 int res = infoFile->Open(cinfo_fname.c_str(), O_RDONLY, 0600, env);
949 if (res < 0) {
950 ret = res;
951 } else {
952 Info info(m_trace, 0);
953 if ( ! info.Read(infoFile, cinfo_fname.c_str())) {
954 ret = -EBADF;
955 } else {
956 ret = info.GetFileSize();
957 }
958 infoFile->Close();
959 }
960 delete infoFile;
961 return ret;
962}
963
964//______________________________________________________________________________
965// Calculate if the file is to be considered cached for the purposes of
966// only-if-cached and setting of atime of the Stat() calls.
967// Returns true if the file is to be conidered cached.
968//------------------------------------------------------------------------------
969bool Cache::DecideIfConsideredCached(long long file_size, long long bytes_on_disk)
970{
971 if (file_size == 0 || bytes_on_disk >= file_size)
972 return true;
973
974 double frac_on_disk = (double) bytes_on_disk / file_size;
975
976 if (file_size <= m_configuration.m_onlyIfCachedMinSize)
977 {
978 if (frac_on_disk >= m_configuration.m_onlyIfCachedMinFrac)
979 return true;
980 }
981 else
982 {
983 if (bytes_on_disk >= m_configuration.m_onlyIfCachedMinSize &&
984 frac_on_disk >= m_configuration.m_onlyIfCachedMinFrac)
985 return true;
986 }
987 return false;
988}
989
990//______________________________________________________________________________
991// Check if the file is cached including m_onlyIfCachedMinSize and m_onlyIfCachedMinFrac
992// pfc configuration parameters. The logic of accessing the Info file is the same
993// as in Cache::LocalFilePath.
1001//------------------------------------------------------------------------------
1002int Cache::ConsiderCached(const char *curl)
1003{
1004 static const char* tpfx = "ConsiderCached ";
1005
1006 TRACE(Debug, tpfx << curl);
1007
1008 XrdCl::URL url(curl);
1009 std::string f_name = url.GetPath();
1010
1011 File *file = nullptr;
1012 {
1013 XrdSysCondVarHelper lock(&m_active_cond);
1014 auto it = m_active.find(f_name);
1015 if (it != m_active.end()) {
1016 file = it->second;
1017 // If the file-open is in progress, `file` is a nullptr
1018 // so we cannot increase the reference count. For now,
1019 // simply treat it as if the file open doesn't exist instead
1020 // of trying to wait and see if it succeeds.
1021 if (file) {
1022 inc_ref_cnt(file, false, false);
1023 }
1024 }
1025 }
1026 if (file) {
1027 struct stat sbuff;
1028 int res = file->Fstat(sbuff);
1029 dec_ref_cnt(file, false);
1030 if (res)
1031 return res;
1032 // DecideIfConsideredCached() already called in File::Fstat().
1033 return sbuff.st_atime > 0 ? 0 : -EREMOTE;
1034 }
1035
1036 struct stat sbuff;
1037 int res = m_oss->Stat(f_name.c_str(), &sbuff);
1038 if (res != XrdOssOK) {
1039 TRACE(Debug, tpfx << curl << " -> " << res);
1040 return res;
1041 }
1042 if (S_ISDIR(sbuff.st_mode))
1043 {
1044 TRACE(Debug, tpfx << curl << " -> EISDIR");
1045 return -EISDIR;
1046 }
1047
1048 long long file_size = DetermineFullFileSize(f_name + Info::s_infoExtension);
1049 if (file_size < 0) {
1050 TRACE(Debug, tpfx << curl << " -> " << file_size);
1051 return (int) file_size;
1052 }
1053 bool is_cached = DecideIfConsideredCached(file_size, sbuff.st_blocks * 512ll);
1054
1055 return is_cached ? 0 : -EREMOTE;
1056}
1057
1058//______________________________________________________________________________
1066//------------------------------------------------------------------------------
1067
1068int Cache::Prepare(const char *curl, int oflags, mode_t mode)
1069{
1070 XrdCl::URL url(curl);
1071 std::string f_name = url.GetPath();
1072 std::string i_name = f_name + Info::s_infoExtension;
1073
1074 // Do not allow write access.
1075 if ((oflags & O_ACCMODE) != O_RDONLY)
1076 {
1077 TRACE(Warning, "Prepare write access requested on file " << f_name << ". Denying access.");
1078 return -EROFS;
1079 }
1080
1081 // Intercept xrdpfc_command requests.
1082 if (m_configuration.m_allow_xrdpfc_command && strncmp("/xrdpfc_command/", f_name.c_str(), 16) == 0)
1083 {
1084 // Schedule a job to process command request.
1085 {
1086 CommandExecutor *ce = new CommandExecutor(f_name, "CommandExecutor");
1087
1088 schedP->Schedule(ce);
1089 }
1090
1091 return -EAGAIN;
1092 }
1093
1094 {
1095 XrdSysCondVarHelper lock(&m_active_cond);
1096 m_purge_delay_set.insert(f_name);
1097 }
1098
1099 struct stat sbuff;
1100 if (m_oss->Stat(i_name.c_str(), &sbuff) == XrdOssOK &&
1101 m_oss->Stat(f_name.c_str(), &sbuff) == XrdOssOK)
1102 {
1103 TRACE(Dump, "Prepare defer open " << f_name);
1104 return 1;
1105 }
1106 else
1107 {
1108 return 0;
1109 }
1110}
1111
1112//______________________________________________________________________________
1113// virtual method of XrdOucCache.
1118//------------------------------------------------------------------------------
1119
1120int Cache::Stat(const char *curl, struct stat &sbuff)
1121{
1122 const char *tpfx = "Stat ";
1123
1124 XrdCl::URL url(curl);
1125 std::string f_name = url.GetPath();
1126
1127 File *file = nullptr;
1128 {
1129 XrdSysCondVarHelper lock(&m_active_cond);
1130 auto it = m_active.find(f_name);
1131 if (it != m_active.end()) {
1132 file = it->second;
1133 // If `file` is nullptr, the file-open is in progress; instead
1134 // of waiting for the file-open to finish, simply treat it as if
1135 // the file-open doesn't exist.
1136 if (file) {
1137 inc_ref_cnt(file, false, false);
1138 }
1139 }
1140 }
1141 if (file) {
1142 int res = file->Fstat(sbuff);
1143 dec_ref_cnt(file, false);
1144 TRACE(Debug, tpfx << "from active file " << curl << " -> " << res);
1145 return res;
1146 }
1147
1148 int res = m_oss->Stat(f_name.c_str(), &sbuff);
1149 if (res != XrdOssOK) {
1150 TRACE(Debug, tpfx << curl << " -> " << res);
1151 return 1; // res; -- for only-if-cached
1152 }
1153 if (S_ISDIR(sbuff.st_mode))
1154 {
1155 TRACE(Debug, tpfx << curl << " -> EISDIR");
1156 return -EISDIR;
1157 }
1158
1159 long long file_size = DetermineFullFileSize(f_name + Info::s_infoExtension);
1160 if (file_size < 0) {
1161 TRACE(Debug, tpfx << curl << " -> " << file_size);
1162 return 1; // (int) file_size; -- for only-if-cached
1163 }
1164 sbuff.st_size = file_size;
1165 bool is_cached = DecideIfConsideredCached(file_size, sbuff.st_blocks * 512ll);
1166 if ( ! is_cached)
1167 sbuff.st_atime = 0;
1168
1169 TRACE(Debug, tpfx << "from disk " << curl << " -> " << res);
1170
1171 return 0;
1172}
1173
1174//______________________________________________________________________________
1175// virtual method of XrdOucCache.
1179//------------------------------------------------------------------------------
1180
1181int Cache::Unlink(const char *curl)
1182{
1183 XrdCl::URL url(curl);
1184 std::string f_name = url.GetPath();
1185
1186 // printf("Unlink url=%s\n\t fname=%s\n", curl, f_name.c_str());
1187
1188 return UnlinkFile(f_name, false);
1189}
1190
1191int Cache::UnlinkFile(const std::string& f_name, bool fail_if_open)
1192{
1193 static const char* trc_pfx = "UnlinkFile ";
1194 ActiveMap_i it;
1195 File *file = 0;
1196 long long st_blocks_to_purge = 0;
1197 bool file_existed = false;
1198 {
1199 XrdSysCondVarHelper lock(&m_active_cond);
1200
1201 it = m_active.find(f_name);
1202
1203 if (it != m_active.end())
1204 {
1205 if (fail_if_open)
1206 {
1207 TRACE(Info, trc_pfx << f_name << ", file currently open and force not requested - denying request");
1208 return -EBUSY;
1209 }
1210
1211 // Null File* in m_active map means an operation is ongoing, probably
1212 // Attach() with possible File::Open(). Ask for retry.
1213 if (it->second == 0)
1214 {
1215 TRACE(Info, trc_pfx << f_name << ", an operation on this file is ongoing - denying request");
1216 return -EAGAIN;
1217 }
1218
1219 file = it->second;
1220 st_blocks_to_purge = file->initiate_emergency_shutdown();
1221 it->second = 0;
1222 }
1223 else
1224 {
1225 it = m_active.insert(std::make_pair(f_name, (File*) 0)).first;
1226 }
1227 }
1228
1229 if (file) {
1230 file_existed = true;
1232 } else {
1233 struct stat f_stat;
1234 if (m_oss->Stat(f_name.c_str(), &f_stat) == XrdOssOK) {
1235 file_existed = true;
1236 st_blocks_to_purge = f_stat.st_blocks;
1237 }
1238 }
1239
1240 std::string i_name = f_name + Info::s_infoExtension;
1241
1242 // Unlink file & cinfo
1243 int f_ret = m_oss->Unlink(f_name.c_str());
1244 int i_ret = m_oss->Unlink(i_name.c_str());
1245
1246 // Count a removal for any file that existed, even with zero blocks, and never
1247 // for an absent one.
1248 if (file_existed)
1249 m_res_mon->register_file_purge(f_name, st_blocks_to_purge);
1250
1251 TRACE(Debug, trc_pfx << f_name << ", f_ret=" << f_ret << ", i_ret=" << i_ret);
1252
1253 {
1254 XrdSysCondVarHelper lock(&m_active_cond);
1255 m_active.erase(it);
1256 m_active_cond.Broadcast();
1257 }
1258
1259 return std::min(f_ret, i_ret);
1260}
int DoIt(int argpnt, int argc, char **argv, bool singleshot)
#define TRACE_Debug
#define XrdOssOK
Definition XrdOss.hh:50
std::string obfuscateAuth(const std::string &input)
#define TRACE_Dump
#define TRACE_PC(act, pre_code, x)
#define TRACE_INT(act, x)
void * ProcessWriteTaskThread(void *)
Definition XrdPfc.cc:60
XrdSysXAttr * XrdSysXAttrActive
void * ResourceMonitorThread(void *)
Definition XrdPfc.cc:54
void * PrefetchThread(void *)
Definition XrdPfc.cc:66
XrdOucCache * XrdOucGetCache(XrdSysLogger *logger, const char *config_filename, const char *parameters, XrdOucEnv *env)
Definition XrdPfc.cc:76
#define stat(a, b)
Definition XrdPosix.hh:101
bool Debug
#define TRACE(act, x)
Definition XrdTrace.hh:63
URL representation.
Definition XrdClURL.hh:31
const std::string & GetPath() const
Get the path.
Definition XrdClURL.hh:217
virtual int Close(long long *retsz=0)=0
virtual int Open(const char *path, int Oflag, mode_t Mode, XrdOucEnv &env)
Definition XrdOss.hh:200
virtual int Fstat(struct stat &sbuff)
virtual const char * Path()=0
virtual const char * Location(bool refresh=false)
XrdOucCache(const char *ctype)
void * GetPtr(const char *varname)
Definition XrdOucEnv.cc:263
void PutPtr(const char *varname, void *value)
Definition XrdOucEnv.cc:298
int get_size() const
long long m_offset
File * get_file() const
Attaches/creates and detaches/deletes cache-io objects for disk based cache.
Definition XrdPfc.hh:163
long long DetermineFullFileSize(const std::string &cinfo_fname)
Definition XrdPfc.cc:928
void FileSyncDone(File *, bool high_debug)
Definition XrdPfc.cc:544
File * GetFile(const std::string &, IO *, long long off=0, long long filesize=0)
Definition XrdPfc.cc:389
static const Configuration & Conf()
Definition XrdPfc.cc:134
virtual int LocalFilePath(const char *url, char *buff=0, int blen=0, LFP_Reason why=ForAccess, bool forall=false)
Definition XrdPfc.cc:796
virtual int Stat(const char *url, struct stat &sbuff)
Definition XrdPfc.cc:1120
const Configuration & RefConfiguration() const
Reference XrdPfc configuration.
Definition XrdPfc.hh:215
static ResourceMonitor & ResMon()
Definition XrdPfc.cc:135
bool IsFileActiveOrPurgeProtected(const std::string &) const
Definition XrdPfc.cc:679
void ClearPurgeProtectedSet()
Definition XrdPfc.cc:687
void ReleaseRAM(char *buf, long long size)
Definition XrdPfc.cc:371
virtual int ConsiderCached(const char *url)
Definition XrdPfc.cc:1002
static Cache & GetInstance()
Singleton access.
Definition XrdPfc.cc:132
bool Config(const char *config_filename, const char *parameters, XrdOucEnv *env)
Parse configuration file.
void DeRegisterPrefetchFile(File *)
Definition XrdPfc.cc:713
void ExecuteCommandUrl(const std::string &command_url)
void RegisterPrefetchFile(File *)
Definition XrdPfc.cc:697
void WriteFileSizeXAttr(int cinfo_fd, long long file_size)
Definition XrdPfc.cc:913
void Prefetch()
Definition XrdPfc.cc:754
void ReleaseFile(File *, IO *)
Definition XrdPfc.cc:473
void AddWriteTask(Block *b, bool from_read)
Add downloaded block in write queue.
Definition XrdPfc.cc:221
Cache(XrdSysLogger *logger, XrdOucEnv *env)
Constructor.
Definition XrdPfc.cc:158
bool Decide(XrdOucCacheIO *)
Makes decision if the original XrdOucCacheIO should be cached.
Definition XrdPfc.cc:137
int UnlinkFile(const std::string &f_name, bool fail_if_open)
Remove cinfo and data files from cache.
Definition XrdPfc.cc:1191
static XrdScheduler * schedP
Definition XrdPfc.hh:302
File * GetNextFileToPrefetch()
Definition XrdPfc.cc:735
long long WritesSinceLastCall()
Definition XrdPfc.cc:320
void ProcessWriteTasks()
Separate task which writes blocks from ram to disk.
Definition XrdPfc.cc:273
virtual int Unlink(const char *url)
Definition XrdPfc.cc:1181
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
virtual XrdOucCacheIO * Attach(XrdOucCacheIO *, int Options=0)
Definition XrdPfc.cc:179
static const Cache & TheOne()
Definition XrdPfc.cc:133
char * RequestRAM(long long size)
Definition XrdPfc.cc:331
virtual int Prepare(const char *url, int oflags, mode_t mode)
Definition XrdPfc.cc:1068
bool DecideIfConsideredCached(long long file_size, long long bytes_on_disk)
Definition XrdPfc.cc:969
static Cache & CreateInstance(XrdSysLogger *logger, XrdOucEnv *env)
Singleton creation.
Definition XrdPfc.cc:125
bool is_prefetch_enabled() const
Definition XrdPfc.hh:307
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)
int GetNBlocks() const
std::string GetRemoteLocations() const
size_t GetAccessCnt() const
int Fstat(struct stat &sbuff)
void AddIO(IO *io)
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.
long long GetPrefetchedBytes() const
int GetBlockSize() const
int GetNDownloadedBlocks() const
const Info::AStat * GetLastAccessStats() const
void BlocksRemovedFromWriteQ(std::list< Block * > &)
Handle removal of a set of blocks from Cache's write queue.
int inc_ref_cnt()
const Stats & RefStats() const
void Sync()
Sync file cache inf o and output data with disk.
int dec_ref_cnt()
int get_ref_cnt()
long long initiate_emergency_shutdown()
long long GetFileSize() const
const std::string & GetLocalPath() const
void RemoveIO(IO *io)
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.
Definition XrdPfcIO.hh:16
XrdOucCacheIO * GetInput()
Definition XrdPfcIO.cc:31
Status of cached file. Can be read from and written into a binary file.
Definition XrdPfcInfo.hh:41
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.
int m_NCksumErrors
number of checksum errors while getting data from remote
long long m_BytesWritten
number of bytes written to disk
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)
XrdPosixStats Stats
Contains parameters configurable from the xrootd config file.
Definition XrdPfc.hh:64
long long BytesHit
read from cache
Definition XrdPfcInfo.hh:64
long long BytesBypassed
read from remote and dropped
Definition XrdPfcInfo.hh:66
time_t DetachTime
close time
Definition XrdPfcInfo.hh:59
long long BytesMissed
read from remote and cached
Definition XrdPfcInfo.hh:65
time_t AttachTime
open time
Definition XrdPfcInfo.hh:58