Loading...
Searching...
No Matches
XrdPfcIOFile.cc
Go to the documentation of this file.
//----------------------------------------------------------------------------------
// Copyright (c) 2014 by Board of Trustees of the Leland Stanford, Jr., University
// Author: Alja Mrak-Tadel, Matevz Tadel, Brian Bockelman
//----------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//----------------------------------------------------------------------------------
#include "XrdPfcIOFile.hh"
#include "XrdPfcStats.hh"
#include "XrdPfcTrace.hh"
#include "XrdOss/XrdOss.hh"
#include "XrdSfs/XrdSfsInterface.hh"
#include "XrdSys/XrdSysError.hh"
#include "XrdOuc/XrdOucEnv.hh"
#include "XrdOuc/XrdOucPgrwUtils.hh"
#include <cstdio>
#include <fcntl.h>
using namespace XrdPfc;
//______________________________________________________________________________
IOFile::IOFile(XrdOucCacheIO *io, Cache & cache) :
IO(io, cache),
m_file(0)
{
m_file = Cache::GetInstance().GetFile(GetFilename(), this);
}
//______________________________________________________________________________
IOFile::~IOFile()
{
// called from Detach() if no sync is needed or
// from Cache's sync thread
TRACEIO(Debug, "~IOFile() " << this);
}
//______________________________________________________________________________
int IOFile::Fstat(struct stat &sbuff)
{
// This is only called during Cache::Attach / Cache::GetFile() for file creation.
// Should really be separate name but one needs to change virtual interface
// so initialStat() becomes virtual in the base-class.
// Also, IOFileBlock should be ditched.
if ( ! m_file) {
return initialStat(sbuff);
}
return m_file->Fstat(sbuff);
}
//______________________________________________________________________________
long long IOFile::FSize()
{
return m_file->GetFileSize();
}
//______________________________________________________________________________
int IOFile::initialStat(struct stat &sbuff)
{
// Get stat to determine file-size.
// Called indirectly from the constructor, via Cache::GetFile().
// Either read it from cinfo file or obtain it from the remote IO.
static const char* trace_pfx = "initialStat ";
std::string fname = GetFilename();
if (m_cache.GetOss()->Stat(fname.c_str(), &sbuff) == XrdOssOK)
{
long long file_size = m_cache.DetermineFullFileSize(fname + Info::s_infoExtension);
if (file_size >= 0)
{
sbuff.st_size = file_size;
TRACEIO(Info, trace_pfx << "successfully read size " << sbuff.st_size << " from info file");
return 0;
}
TRACEIO(Error, trace_pfx << "failed reading from info file " << XrdSysE2T(-file_size));
}
int res = GetInput()->Fstat(sbuff);
TRACEIO(Debug, trace_pfx << "stat from client res = " << res << ", size = " << sbuff.st_size);
return res;
}
//______________________________________________________________________________
void IOFile::Update(XrdOucCacheIO &iocp)
{
IO::Update(iocp);
m_file->ioUpdated(this);
}
//______________________________________________________________________________
bool IOFile::ioActive()
{
RefreshLocation();
return m_file->ioActive(this);
}
//______________________________________________________________________________
void IOFile::DetachFinalize()
{
// Effectively a destructor.
TRACE(Debug, "DetachFinalize() " << this);
m_file->RequestSyncOfDetachStats();
Cache::GetInstance().ReleaseFile(m_file, this);
if (( ! m_error_counts.empty() || m_incomplete_count > 0) && XRD_TRACE What >= TRACE_Error) {
char info[1024];
int pos = 0, cap = 1024;
bool truncated = false;
for (auto [err, cnt] : m_error_counts) {
int len = snprintf(&info[pos], cap, " ( %d : %d)", err, cnt);
if (len >= cap) {
truncated = true;
break;
}
pos += len;
cap -= len;
}
TRACE(Error, "DetachFinalize() " << this << " n_incomplete_reads=" << m_incomplete_count
<< ", block (error : count) report:" << info << (truncated ? " - truncated" : ""));
}
delete this;
}
//==============================================================================
// Read and pgRead - sync / async and helpers
//==============================================================================
//______________________________________________________________________________
int IOFile::Read(char *buff, long long off, int size)
{
++m_active_read_reqs;
auto *rh = new ReadReqRHCond(ObtainReadSid(), nullptr);
TRACEIO(Dump, "Read() sync " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " off: " << off << " size: " << size);
rh->m_cond.Lock();
int retval = ReadBegin(buff, off, size, rh);
if (retval == -EWOULDBLOCK)
{
rh->m_cond.Wait();
retval = rh->m_retval;
}
rh->m_cond.UnLock();
return ReadEnd(retval, rh);
}
//______________________________________________________________________________
void IOFile::Read(XrdOucCacheIOCB &iocb, char *buff, long long off, int size)
{
struct ZHandler : ReadReqRH
{ using ReadReqRH::ReadReqRH;
IOFile *m_io = nullptr;
void Done(int result) override {
m_io->ReadEnd(result, this);
}
};
++m_active_read_reqs;
auto *rh = new ZHandler(ObtainReadSid(), &iocb);
rh->m_io = this;
TRACEIO(Dump, "Read() async " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " off: " << off << " size: " << size);
int retval = ReadBegin(buff, off, size, rh);
if (retval != -EWOULDBLOCK)
{
rh->Done(retval);
}
}
//______________________________________________________________________________
void IOFile::pgRead(XrdOucCacheIOCB &iocb, char *buff, long long off, int size,
std::vector<uint32_t> &csvec, uint64_t opts, int *csfix)
{
struct ZHandler : ReadReqRH
{ using ReadReqRH::ReadReqRH;
IOFile *m_io = nullptr;
std::function<void (int)> m_lambda {0};
void Done(int result) override {
if (m_lambda) m_lambda(result);
m_io->ReadEnd(result, this);
}
};
++m_active_read_reqs;
auto *rh = new ZHandler(ObtainReadSid(), &iocb);
rh->m_io = this;
TRACEIO(Dump, "pgRead() async " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " off: " << off << " size: " << size);
if (opts & XrdOucCacheIO::forceCS)
rh->m_lambda = [=, &csvec](int result) {
if (result > 0)
XrdOucPgrwUtils::csCalc((const char *)buff, (ssize_t)off, (size_t)result, csvec);
};
int retval = ReadBegin(buff, off, size, rh);
if (retval != -EWOULDBLOCK)
{
rh->Done(retval);
}
}
//______________________________________________________________________________
int IOFile::ReadBegin(char *buff, long long off, int size, ReadReqRH *rh)
{
// protect from reads over the file size
if (off >= FSize()) {
size = 0;
return 0;
}
if (off < 0) {
return -EINVAL;
}
if (off + size > FSize()) {
size = FSize() - off;
}
rh->m_expected_size = size;
return m_file->Read(this, buff, off, size, rh);
}
//______________________________________________________________________________
int IOFile::ReadEnd(int retval, ReadReqRH *rh)
{
TRACEIO(Dump, "ReadEnd() " << (rh->m_iocb ? "a" : "") << "sync " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " retval: " << retval << " expected_size: " << rh->m_expected_size);
if (retval < 0) {
TRACEIO(Debug, "ReadEnd() error in File::Read(), exit status=" << retval << ", error=" << XrdSysE2T(-retval) << " sid: " << Xrd::hex1 << rh->m_seq_id);
} else if (retval < rh->m_expected_size) {
TRACEIO(Debug, "ReadEnd() bytes missed " << rh->m_expected_size - retval << " sid: " << Xrd::hex1 << rh->m_seq_id);
}
if (rh->m_iocb)
rh->m_iocb->Done(retval);
delete rh;
--m_active_read_reqs;
return retval;
}
//==============================================================================
// ReadV
//==============================================================================
//______________________________________________________________________________
int IOFile::ReadV(const XrdOucIOVec *readV, int n)
{
++m_active_read_reqs;
auto *rh = new ReadReqRHCond(ObtainReadSid(), nullptr);
TRACEIO(Dump, "ReadV() sync " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " n_chunks: " << n);
rh->m_cond.Lock();
int retval = ReadVBegin(readV, n, rh);
if (retval == -EWOULDBLOCK)
{
rh->m_cond.Wait();
retval = rh->m_retval;
}
rh->m_cond.UnLock();
return ReadVEnd(retval, rh);
}
//______________________________________________________________________________
void IOFile::ReadV(XrdOucCacheIOCB &iocb, const XrdOucIOVec *readV, int n)
{
struct ZHandler : ReadReqRH
{ using ReadReqRH::ReadReqRH;
IOFile *m_io = nullptr;
void Done(int result) override { m_io-> ReadVEnd(result, this); }
};
++m_active_read_reqs;
auto *rh = new ZHandler(ObtainReadSid(), &iocb);
rh->m_io = this;
TRACEIO(Dump, "ReadV() async " << this << " sid: " << Xrd::hex1 << rh->m_seq_id << " n_chunks: " << n);
int retval = ReadVBegin(readV, n, rh);
if (retval != -EWOULDBLOCK)
{
rh->Done(retval);
}
}
//______________________________________________________________________________
int IOFile::ReadVBegin(const XrdOucIOVec *readV, int n, ReadReqRH *rh)
{
long long file_size = FSize();
for (int i = 0; i < n; ++i)
{
const XrdOucIOVec &vr = readV[i];
if (vr.offset < 0 || vr.offset >= file_size ||
vr.offset + vr.size > file_size)
{
return -EINVAL;
}
rh->m_expected_size += vr.size;
}
rh->m_n_chunks = n;
return m_file->ReadV(this, readV, n, rh);
}
//______________________________________________________________________________
int IOFile::ReadVEnd(int retval, ReadReqRH *rh)
{
TRACEIO(Dump, "ReadVEnd() " << (rh->m_iocb ? "a" : "") << "sync " << this << " sid: " << Xrd::hex1 << rh->m_seq_id <<
" retval: " << retval << " n_chunks: " << rh->m_n_chunks << " expected_size: " << rh->m_expected_size);
if (retval < 0) {
TRACEIO(Debug, "ReadVEnd() error in File::ReadV(), exit status=" << retval << ", error=" << XrdSysE2T(-retval));
} else if (retval < rh->m_expected_size) {
TRACEIO(Debug, "ReadVEnd() bytes missed " << rh->m_expected_size - retval);
}
if (rh->m_iocb)
rh->m_iocb->Done(retval);
delete rh;
--m_active_read_reqs;
return retval;
}
XrdOss.hh
XrdOssOK
#define XrdOssOK
Definition XrdOss.hh:50
XrdOucEnv.hh
XrdOucPgrwUtils.hh
XrdPfcIOFile.hh
XrdPfcStats.hh
XrdPfcTrace.hh
TRACE_Error
#define TRACE_Error
Definition XrdPfcTrace.hh:7
TRACEIO
#define TRACEIO(act, x)
Definition XrdPfcTrace.hh:63
stat
#define stat(a, b)
Definition XrdPosix.hh:101
XRD_TRACE
#define XRD_TRACE
Definition XrdScheduler.cc:48
Debug
bool Debug
Definition XrdSecgsiProxy.cc:95
opts
struct myOpts opts
XrdSfsInterface.hh
XrdSysE2T
const char * XrdSysE2T(int errcode)
Definition XrdSysE2T.cc:104
XrdSysError.hh
Error
@ Error
Definition XrdTlsNotaryUtils.hh:42
TRACE
#define TRACE(act, x)
Definition XrdTrace.hh:63
XrdOss::Stat
virtual int Stat(const char *path, struct stat *buff, int opts=0, XrdOucEnv *envP=0)=0
XrdOucCacheIOCB
Definition XrdOucCache.hh:53
XrdOucCacheIOCB::Done
virtual void Done(int result)=0
XrdOucCacheIO::Fstat
virtual int Fstat(struct stat &sbuff)
Definition XrdOucCache.hh:148
XrdOucCacheIO::forceCS
static const uint64_t forceCS
Definition XrdOucCache.hh:188
XrdOucCacheIO::XrdOucCacheIO
XrdOucCacheIO()
Construct and Destructor.
Definition XrdOucCache.hh:480
XrdOucPgrwUtils::csCalc
static void csCalc(const char *data, off_t offs, size_t count, uint32_t *csval)
Definition XrdOucPgrwUtils.cc:54
XrdPfc::Cache
Attaches/creates and detaches/deletes cache-io objects for disk based cache.
Definition XrdPfc.hh:163
XrdPfc::Cache::DetermineFullFileSize
long long DetermineFullFileSize(const std::string &cinfo_fname)
Definition XrdPfc.cc:926
XrdPfc::Cache::GetFile
File * GetFile(const std::string &, IO *, long long off=0, long long filesize=0)
Definition XrdPfc.cc:389
XrdPfc::Cache::GetInstance
static Cache & GetInstance()
Singleton access.
Definition XrdPfc.cc:132
XrdPfc::Cache::ReleaseFile
void ReleaseFile(File *, IO *)
Definition XrdPfc.cc:471
XrdPfc::Cache::GetOss
XrdOss * GetOss() const
Definition XrdPfc.hh:280
XrdPfc::IOFile::Update
void Update(XrdOucCacheIO &iocp) override
Definition XrdPfcIOFile.cc:100
XrdPfc::IOFile::DetachFinalize
void DetachFinalize() override
Abstract virtual method of XrdPfc::IO Called to destruct the IO object after it is no longer used.
Definition XrdPfcIOFile.cc:114
XrdPfc::IOFile::~IOFile
~IOFile()
Definition XrdPfcIOFile.cc:44
XrdPfc::IOFile::Read
int Read(char *buff, long long off, int size) override
Pass Read request to the corresponding File object.
Definition XrdPfcIOFile.cc:149
XrdPfc::IOFile::Fstat
int Fstat(struct stat &sbuff) override
Definition XrdPfcIOFile.cc:52
XrdPfc::IOFile::ioActive
bool ioActive() override
Abstract virtual method of XrdPfc::IO Called to check if destruction needs to be done in a separate t...
Definition XrdPfcIOFile.cc:107
XrdPfc::IOFile::pgRead
void pgRead(XrdOucCacheIOCB &iocb, char *buff, long long off, int size, std::vector< uint32_t > &csvec, uint64_t opts=0, int *csfix=0) override
Definition XrdPfcIOFile.cc:196
XrdPfc::IOFile::ReadV
int ReadV(const XrdOucIOVec *readV, int n) override
Pass ReadV request to the corresponding File object.
Definition XrdPfcIOFile.cc:275
XrdPfc::IOFile::FSize
long long FSize() override
Definition XrdPfcIOFile.cc:66
XrdPfc::IOFile::IOFile
IOFile(XrdOucCacheIO *io, Cache &cache)
Definition XrdPfcIOFile.cc:36
XrdPfc::IO::IO
IO(XrdOucCacheIO *io, Cache &cache)
Definition XrdPfcIO.cc:7
XrdPfc::IO::m_incomplete_count
int m_incomplete_count
Definition XrdPfcIO.hh:88
XrdPfc::IO::m_error_counts
std::map< int, int > m_error_counts
Definition XrdPfcIO.hh:89
XrdPfc::IO::GetInput
XrdOucCacheIO * GetInput()
Definition XrdPfcIO.cc:31
XrdPfc::IO::m_cache
Cache & m_cache
reference to Cache object
Definition XrdPfcIO.hh:50
XrdPfc::IO::m_active_read_reqs
RAtomic_int m_active_read_reqs
number of active read requests
Definition XrdPfcIO.hh:70
XrdPfc::IO::RefreshLocation
const char * RefreshLocation()
Definition XrdPfcIO.hh:55
XrdPfc::IO::Update
void Update(XrdOucCacheIO &iocp) override
Definition XrdPfcIO.cc:17
XrdPfc::IO::ObtainReadSid
unsigned short ObtainReadSid()
Definition XrdPfcIO.hh:57
XrdPfc::IO::GetFilename
std::string GetFilename()
Definition XrdPfcIO.cc:36
XrdPfc::Info
Status of cached file. Can be read from and written into a binary file.
Definition XrdPfcInfo.hh:41
XrdPfc::Info::s_infoExtension
static const char * s_infoExtension
Definition XrdPfcInfo.hh:309
XrdPfc
Definition XrdPfc.hh:41
Xrd::hex1
@ hex1
Definition XrdSysTrace.hh:42
XrdOucIOVec
Definition XrdOucIOVec.hh:41
XrdOucIOVec::size
int size
Definition XrdOucIOVec.hh:43
XrdOucIOVec::offset
long long offset
Definition XrdOucIOVec.hh:42
XrdPfc::IO::ReadReqRHCond
Definition XrdPfcIO.hh:60
XrdPfc::ReadReqRH
Definition XrdPfcFile.hh:50
XrdPfc::ReadReqRH::m_iocb
XrdOucCacheIOCB * m_iocb
Definition XrdPfcFile.hh:54
XrdPfc::ReadReqRH::m_seq_id
unsigned short m_seq_id
Definition XrdPfcFile.hh:53
XrdPfc::ReadReqRH::ReadReqRH
ReadReqRH(unsigned short sid, XrdOucCacheIOCB *iocb)
Definition XrdPfcFile.hh:56
XrdPfc::ReadReqRH::m_expected_size
int m_expected_size
Definition XrdPfcFile.hh:51
XrdPfc::ReadReqRH::m_n_chunks
int m_n_chunks
Definition XrdPfcFile.hh:52