Skip to content

Commit

Permalink
Pass input source as a pointer instead of reference.
Browse files Browse the repository at this point in the history
  • Loading branch information
@alja @abh3
alja authored and abh3 committed Jun 30, 2016
1 parent 99dbf85 commit ac137cd
Show file tree
Hide file tree
Showing 10 changed files with 113 additions and 66 deletions.
42 changes: 37 additions & 5 deletions src/XrdFileCache/XrdFileCache.cc
View file
Expand Up @@ -30,11 +30,9 @@
#include "XrdOuc/XrdOucUtils.hh"

#include "XrdFileCache.hh"
#include "XrdFileCacheInfo.hh"
#include "XrdFileCacheIOEntireFile.hh"
#include "XrdFileCacheIOFileBlock.hh"
//#include "XrdFileCacheConfiguration.cc"
//#include "XrdFileCachePurge.cc"


using namespace XrdFileCache;

Expand Down Expand Up @@ -146,9 +144,9 @@ XrdOucCacheIO2 *Cache::Attach(XrdOucCacheIO2 *io, int Options)
clLog()->Info(XrdCl::AppMsg, "Cache::Attach() %s", io->Path());
IO* cio;
if (Cache::GetInstance().RefConfiguration().m_hdfsmode)
cio = new IOFileBlock(*io, m_stats, *this);
cio = new IOFileBlock(io, m_stats, *this);
else
cio = new IOEntireFile(*io, m_stats, *this);
cio = new IOEntireFile(io, m_stats, *this);

return cio;
}
Expand Down Expand Up @@ -316,6 +314,7 @@ Cache::DeRegisterPrefetchFile(File* file)
}
m_prefetch_condVar.UnLock();
}

//______________________________________________________________________________

File*
Expand All @@ -336,6 +335,39 @@ Cache::GetNextFileToPrefetch()
return f;
}

//______________________________________________________________________________

int
Cache::Prepare(const char *url, int oflags, mode_t mode)
{
return 1;
}

//______________________________________________________________________________

int
Cache::Stat(const char *curl, struct stat &sbuff)
{
XrdCl::URL url(curl);
std::string fname = Cache::GetInstance().RefConfiguration().m_cache_dir + url.GetPath();
fname += ".cinfo";

XrdOucEnv myEnv;
XrdOssDF* df = m_output_fs->newFile(Cache::GetInstance().RefConfiguration().m_username.c_str());
int res = df->Open(fname.c_str(), O_RDONLY, 0600, myEnv);
if (res != 0)
return 1;

df->Fstat(&sbuff);

Info cinfo(0);
cinfo.Read(df);
sbuff.st_size = cinfo.GetSizeInBits();
delete df;
return 0;
}


//______________________________________________________________________________


Expand Down
7 changes: 7 additions & 0 deletions src/XrdFileCache/XrdFileCache.hh
View file
Expand Up @@ -99,6 +99,13 @@ namespace XrdFileCache
// this is an obsolete method
virtual XrdOucCache* Create(XrdOucCache::Parms&, XrdOucCacheIO::aprParms*);

// Virtual function of XrdOucCache2. Used for deferred open.
virtual int Prepare(const char *url, int oflags, mode_t mode);


// virtual function of XrdOucCache2::Stat()
virtual int Stat(const char *url, struct stat &sbuff);

//--------------------------------------------------------------------
//! \brief Makes decision if the original XrdOucCacheIO should be cached.
//!
Expand Down
27 changes: 15 additions & 12 deletions src/XrdFileCache/XrdFileCacheFile.cc
View file
Expand Up @@ -70,7 +70,7 @@ namespace
Cache* cache() { return &Cache::GetInstance(); }
}

File::File(XrdOucCacheIO2 &inputIO, std::string& disk_file_path, long long iOffset, long long iFileSize) :
File::File(XrdOucCacheIO2 *inputIO, std::string& disk_file_path, long long iOffset, long long iFileSize) :
m_input(inputIO),
m_output(NULL),
m_infoFile(NULL),
Expand All @@ -91,9 +91,9 @@ m_prefetchHitCnt(0),
m_prefetchScore(1),
m_prefetchCurrentCnt(0)
{
clLog()->Debug(XrdCl::AppMsg, "File::File() %s", m_input.Path());
clLog()->Debug(XrdCl::AppMsg, "File::File() %s", m_input->Path());
if (!Open()) {
clLog()->Error(XrdCl::AppMsg, "File::File() Open failed %s !!!", m_input.Path());
clLog()->Error(XrdCl::AppMsg, "File::File() Open failed %s !!!", m_input->Path());
}
}

Expand Down Expand Up @@ -220,7 +220,7 @@ bool File::InitiateClose()

bool File::Open()
{
clLog()->Dump(XrdCl::AppMsg, "File::Open() open file for disk cache %s", m_input.Path());
clLog()->Dump(XrdCl::AppMsg, "File::Open() open file for disk cache %s", m_input->Path());

XrdOss &m_output_fs = *Cache::GetInstance().GetOss();
// Create the data file itself.
Expand All @@ -232,7 +232,7 @@ bool File::Open()
int res = m_output->Open(m_temp_filename.c_str(), O_RDWR, 0600, myEnv);
if (res < 0)
{
clLog()->Error(XrdCl::AppMsg, "File::Open() can't get data-FD for %s %s", m_temp_filename.c_str(), m_input.Path());
clLog()->Error(XrdCl::AppMsg, "File::Open() can't get data-FD for %s %s", m_temp_filename.c_str(), m_input->Path());
delete m_output;
m_output = 0;

Expand All @@ -254,7 +254,7 @@ bool File::Open()
int res = m_infoFile->Open(ifn.c_str(), O_RDWR, 0600, myEnv);
if (res < 0)
{
clLog()->Error(XrdCl::AppMsg, "File::Open() can't get info-FD %s %s", ifn.c_str(), m_input.Path());
clLog()->Error(XrdCl::AppMsg, "File::Open() can't get info-FD %s %s", ifn.c_str(), m_input->Path());
delete m_infoFile;
m_infoFile = 0;
return false;
Expand All @@ -267,14 +267,17 @@ bool File::Open()

if (m_cfi.Read(m_infoFile, Cache::GetInstance().RefConfiguration().m_prefetch) <= 0)
{
m_fileSize = m_input->FSize();
int ss = (m_fileSize - 1)/m_cfi.GetBufferSize() + 1;
clLog()->Info(XrdCl::AppMsg, "Creating new file info with size %lld. Reserve space for %d blocks %s", m_fileSize, ss, m_input.Path());
clLog()->Info(XrdCl::AppMsg, "Creating new file info with size %lld. Reserve space for %d blocks %s", m_fileSize, ss, m_input->Path());
m_cfi.ResizeBits(ss, Cache::GetInstance().RefConfiguration().m_prefetch);
m_cfi.WriteHeader(m_infoFile);
}
else
{
clLog()->Debug(XrdCl::AppMsg, "Info file read from disk: %s", m_input.Path());
m_fileSize = m_cfi.GetSizeInBits();
printf("%lld file size \n", m_fileSize);
clLog()->Debug(XrdCl::AppMsg, "Info file read from disk: %s", m_input->Path());
}


Expand Down Expand Up @@ -339,12 +342,12 @@ Block* File::RequestBlock(int i, bool prefetch)
const int last_block = m_cfi.GetSizeInBits() - 1;

long long off = i * BS;
long long this_bs = (i == last_block) ? m_input.FSize() - off : BS;
long long this_bs = (i == last_block) ? m_input->FSize() - off : BS;

Block *b = new Block(this, off, this_bs, prefetch); // should block be reused to avoid recreation

BlockResponseHandler* oucCB = new BlockResponseHandler(b);
m_input.Read(*oucCB, (char*)b->get_buff(), off, (int)this_bs);
m_input->Read(*oucCB, (char*)b->get_buff(), off, (int)this_bs);

clLog()->Dump(XrdCl::AppMsg, "File::RequestBlock() this = %p, b=%p, this idx=%d pOn=(%d) %s", (void*)this, (void*)b, i, prefetch, lPath());
m_block_map[i] = b;
Expand Down Expand Up @@ -377,7 +380,7 @@ int File::RequestBlocksDirect(DirectResponseHandler *handler, IntList_t& blocks,

overlap(*ii, BS, req_off, req_size, off, blk_off, size);

m_input.Read( *handler, req_buf + off, *ii * BS + blk_off, size);
m_input->Read( *handler, req_buf + off, *ii * BS + blk_off, size);
clLog()->Dump(XrdCl::AppMsg, "RequestBlockDirect success %d %ld %s", *ii, size, lPath());

total += size;
Expand Down Expand Up @@ -684,7 +687,7 @@ void File::WriteBlockToDisk(Block* b)
int retval = 0;
// write block buffer into disk file
long long offset = b->m_offset - m_offset;
long long size = (b->m_offset + m_cfi.GetBufferSize()) > m_input.FSize() ? (m_input.FSize() - b->m_offset) : m_cfi.GetBufferSize();
long long size = (b->m_offset + m_cfi.GetBufferSize()) > m_input->FSize() ? (m_input->FSize() - b->m_offset) : m_cfi.GetBufferSize();
int buffer_remaining = size;
int buffer_offset = 0;
int cnt = 0;
Expand Down
4 changes: 2 additions & 2 deletions src/XrdFileCache/XrdFileCacheFile.hh
View file
Expand Up @@ -97,7 +97,7 @@ namespace XrdFileCache
private:
enum PrefetchState_e { kOn, kHold, kCanceled };

XrdOucCacheIO2 &m_input; //!< original data source
XrdOucCacheIO2 *m_input; //!< original data source
XrdOssDF *m_output; //!< file handle for data file on disk
XrdOssDF *m_infoFile; //!< file handle for data-info file on disk
Info m_cfi; //!< download status of file blocks and access statistics
Expand Down Expand Up @@ -144,7 +144,7 @@ namespace XrdFileCache
//------------------------------------------------------------------------
//! Constructor.
//------------------------------------------------------------------------
File(XrdOucCacheIO2 &io, std::string &path,
File(XrdOucCacheIO2 *io, std::string &path,
long long offset, long long fileSize);

//------------------------------------------------------------------------
Expand Down
11 changes: 6 additions & 5 deletions src/XrdFileCache/XrdFileCacheIO.hh
View file
Expand Up @@ -13,17 +13,17 @@ namespace XrdFileCache
class IO : public XrdOucCacheIO2
{
public:
IO (XrdOucCacheIO2 &io, XrdOucCacheStats &stats, Cache &cache) :
IO (XrdOucCacheIO2 *io, XrdOucCacheStats &stats, Cache &cache) :
m_io(io), m_statsGlobal(stats), m_cache(cache) {}

//! Original data source.
virtual XrdOucCacheIO *Base() { return &m_io; }
virtual XrdOucCacheIO *Base() { return m_io; }

//! Original data source URL.
virtual long long FSize() { return m_io.FSize(); }
virtual long long FSize() { return m_io->FSize(); }

//! Original data source URL.
virtual const char *Path() { return m_io.Path(); }
virtual const char *Path() { return m_io->Path(); }

virtual int Sync() { return 0; }

Expand All @@ -32,11 +32,12 @@ namespace XrdFileCache
virtual int Write(char *Buffer, long long Offset, int Length)
{ errno = ENOTSUP; return -1; }

virtual void Update(XrdOucCacheIO2 &iocp) { m_io = &iocp; }

protected:
XrdCl::Log* clLog() const { return XrdCl::DefaultEnv::GetLog(); }

XrdOucCacheIO2 &m_io; //!< original data source
XrdOucCacheIO2 *m_io; //!< original data source
XrdOucCacheStats &m_statsGlobal; //!< reference to Cache statistics
Cache &m_cache; //!< reference to Cache needed in detach
};
Expand Down
32 changes: 18 additions & 14 deletions src/XrdFileCache/XrdFileCacheIOEntireFile.cc
View file
Expand Up @@ -31,16 +31,16 @@ using namespace XrdFileCache;
//______________________________________________________________________________


IOEntireFile::IOEntireFile(XrdOucCacheIO2 &io, XrdOucCacheStats &stats, Cache & cache)
IOEntireFile::IOEntireFile(XrdOucCacheIO2 *io, XrdOucCacheStats &stats, Cache & cache)
: IO(io, stats, cache),
m_file(0)
{
clLog()->Info(XrdCl::AppMsg, "IO::IO() [%p] %s", this, m_io.Path());
clLog()->Info(XrdCl::AppMsg, "IO::IO() [%p] %s", this, m_io->Path());

XrdCl::URL url(io.Path());
XrdCl::URL url(m_io->Path());
std::string fname = Cache::GetInstance().RefConfiguration().m_cache_dir + url.GetPath();

m_file = new File(io, fname, 0, io.FSize());
m_file = new File(io, fname, 0, 0);
}

IOEntireFile::~IOEntireFile()
Expand All @@ -55,7 +55,7 @@ XrdOucCacheIO *IOEntireFile::Detach()
{
m_statsGlobal.Add(m_file->GetStats());

XrdOucCacheIO * io = &m_io;
XrdOucCacheIO * io = m_io;

delete m_file;
m_file = 0;
Expand All @@ -72,36 +72,40 @@ void IOEntireFile::Read (XrdOucCacheIOCB &iocb, char *buff, long long offs, int

int IOEntireFile::Read (char *buff, long long off, int size)
{
clLog()->Debug(XrdCl::AppMsg, "IOEntireFile::Read() [%p] %lld@%d %s", this, off, size, m_io.Path());


clLog()->Debug(XrdCl::AppMsg, "IOEntireFile::Read() [%p] %lld@%d %s", this, off, size, m_io->Path());

return m_io->Read(buff, off, size);

// protect from reads over the file size
if (off >= m_io.FSize())
return 0;
// if (off >= m_io->FSize())
// return 0;
if (off < 0)
{
errno = EINVAL;
return -1;
}
if (off + size > m_io.FSize())
size = m_io.FSize() - off;
//if (off + size > m_io->FSize())
// size = m_io->FSize() - off;

ssize_t bytes_read = 0;
ssize_t retval = 0;

retval = m_file->Read(buff, off, size);
clLog()->Debug(XrdCl::AppMsg, "IOEntireFile::Read() read from File retval = %d %s", retval, m_io.Path());
clLog()->Debug(XrdCl::AppMsg, "IOEntireFile::Read() read from File retval = %d %s", retval, m_io->Path());
if (retval >= 0)
{
bytes_read += retval;
buff += retval;
size -= retval;

if (size > 0)
clLog()->Warning(XrdCl::AppMsg, "IOEntireFile::Read() missed %d bytes %s", size, m_io.Path());
clLog()->Warning(XrdCl::AppMsg, "IOEntireFile::Read() missed %d bytes %s", size, m_io->Path());
}
else
{
clLog()->Error(XrdCl::AppMsg, "IOEntireFile::Read(), origin bytes read %d %s", retval, m_io.Path());
clLog()->Error(XrdCl::AppMsg, "IOEntireFile::Read(), origin bytes read %d %s", retval, m_io->Path());
}

return (retval < 0) ? retval : bytes_read;
Expand All @@ -113,7 +117,7 @@ int IOEntireFile::Read (char *buff, long long off, int size)
*/
int IOEntireFile::ReadV (const XrdOucIOVec *readV, int n)
{
clLog()->Warning(XrdCl::AppMsg, "IOEntireFile::ReadV(), get %d requests %s", n, m_io.Path());
clLog()->Warning(XrdCl::AppMsg, "IOEntireFile::ReadV(), get %d requests %s", n, m_io->Path());


return m_file->ReadV(readV, n);
Expand Down
2 changes: 1 addition & 1 deletion src/XrdFileCache/XrdFileCacheIOEntireFile.hh
View file
Expand Up @@ -43,7 +43,7 @@ namespace XrdFileCache
//------------------------------------------------------------------------
//! Constructor
//------------------------------------------------------------------------
IOEntireFile(XrdOucCacheIO2 &io, XrdOucCacheStats &stats, Cache &cache);
IOEntireFile(XrdOucCacheIO2 *io, XrdOucCacheStats &stats, Cache &cache);

//------------------------------------------------------------------------
//! Destructor
Expand Down

0 comments on commit ac137cd

Please sign in to comment.