|
|
|
#include <TStorageFactoryFile.h>
Inherits ROOT::TFile.
Public Member Functions | |
| ClassDef (TStorageFactoryFile, 0) | |
| virtual Bool_t | ReadBuffer (char *buf, Int_t len) |
| virtual Bool_t | ReadBuffer (char *buf, Long64_t pos, Int_t len) |
| virtual Bool_t | ReadBufferAsync (Long64_t off, Int_t len) |
| virtual Bool_t | ReadBuffers (char *buf, Long64_t *pos, Int_t *len, Int_t nbuf) |
| void | ResetErrno (void) const |
| TStorageFactoryFile (const char *name, Option_t *option, const char *ftitle, Int_t compress, Int_t netopt, Bool_t parallelopen=kFALSE) | |
| TStorageFactoryFile (const char *name, Option_t *option="", const char *ftitle="", Int_t compress=1) | |
| virtual Bool_t | WriteBuffer (const char *buf, Int_t len) |
| ~TStorageFactoryFile (void) | |
Protected Member Functions | |
| virtual Int_t | SysClose (Int_t fd) |
| virtual Int_t | SysOpen (const char *pathname, Int_t flags, UInt_t mode) |
| virtual Long64_t | SysSeek (Int_t fd, Long64_t offset, Int_t whence) |
| virtual Int_t | SysStat (Int_t fd, Long_t *id, Long64_t *size, Long_t *flags, Long_t *modtime) |
| virtual Int_t | SysSync (Int_t fd) |
Private Member Functions | |
| void | Initialize (const char *name, Option_t *option="") |
| Bool_t | ReadBuffersSync (char *buf, Long64_t *pos, Int_t *len, Int_t nbuf) |
| TStorageFactoryFile (void) | |
Private Attributes | |
| Storage * | storage_ |
TFile wrapper around StorageFactory and Storage.
Definition at line 14 of file TStorageFactoryFile.h.
| TStorageFactoryFile::TStorageFactoryFile | ( | const char * | name, |
| Option_t * | option, | ||
| const char * | ftitle, | ||
| Int_t | compress, | ||
| Int_t | netopt, | ||
| Bool_t | parallelopen = kFALSE |
||
| ) |
Definition at line 105 of file TStorageFactoryFile.cc.
References Initialize().
: TFile(path, "NET", ftitle, compress), // Pass "NET" to prevent local access in base class storage_(0) { Initialize(path, option); }
| TStorageFactoryFile::TStorageFactoryFile | ( | const char * | name, |
| Option_t * | option = "", |
||
| const char * | ftitle = "", |
||
| Int_t | compress = 1 |
||
| ) |
Definition at line 117 of file TStorageFactoryFile.cc.
References Initialize().
: TFile(path, "NET", ftitle, compress), // Pass "NET" to prevent local access in base class storage_(0) { Initialize(path, option); }
| TStorageFactoryFile::~TStorageFactoryFile | ( | void | ) |
Definition at line 212 of file TStorageFactoryFile.cc.
References storage_.
{
Close();
delete storage_;
}
| TStorageFactoryFile::TStorageFactoryFile | ( | void | ) | [private] |
Definition at line 94 of file TStorageFactoryFile.cc.
References storageCounter(), and StorageAccount::Stamp::tick().
: storage_(0) { StorageAccount::Stamp stats(storageCounter(s_statsCtor, "construct")); stats.tick(0); }
| TStorageFactoryFile::ClassDef | ( | TStorageFactoryFile | , |
| 0 | |||
| ) |
| void TStorageFactoryFile::Initialize | ( | const char * | name, |
| Option_t * | option = "" |
||
| ) | [private] |
Definition at line 128 of file TStorageFactoryFile.cc.
References edm::Exception::categoryCode(), SurfaceDeformationFactory::create(), alignCSCRings::e, Exception, StorageFactory::get(), edm::Service< T >::isAvailable(), edm::errors::NotFound, IOFlags::OpenCreate, IOFlags::OpenRead, IOFlags::OpenWrite, getHLTPrescaleColumns::path, SiPixelLorentzAngle_cfi::read, Storage::size(), storage_, storageCounter(), StorageAccount::Stamp::tick(), and update.
Referenced by TStorageFactoryFile().
{
StorageAccount::Stamp stats(storageCounter(s_statsCtor, "construct"));
// Enable AsyncReading.
// This was the default for 5.27, but turned off by default for 5.32.
// In our testing, AsyncReading is the fastest mechanism available.
// In 5.32, the AsyncPrefetching mechanism is preferred, but has been a
// performance hit in our "average case" tests.
gEnv->SetValue("TFile.AsyncReading", 1);
// Parse options; at the moment we only accept read!
fOption = option;
fOption.ToUpper();
if (fOption == "NEW")
fOption = "CREATE";
Bool_t create = (fOption == "CREATE");
Bool_t recreate = (fOption == "RECREATE");
Bool_t update = (fOption == "UPDATE");
Bool_t read = (fOption == "READ");
if (!create && !recreate && !update && !read)
{
read = true;
fOption = "READ";
}
if (recreate)
{
if (!gSystem->AccessPathName(path, kFileExists))
gSystem->Unlink(path);
recreate = false;
create = true;
fOption = "CREATE";
}
assert(!recreate);
if (update && gSystem->AccessPathName(path, kFileExists))
{
update = kFALSE;
create = kTRUE;
}
assert(read || update || create);
int openFlags = IOFlags::OpenRead;
if (!read) openFlags |= IOFlags::OpenWrite;
if (create) openFlags |= IOFlags::OpenCreate;
//if (recreate) openFlags |= IOFlags::OpenCreate | IOFlags::OpenTruncate;
// Open storage
if (! (storage_ = StorageFactory::get()->open(path, openFlags)))
{
MakeZombie();
gDirectory = gROOT;
throw cms::Exception("TStorageFactoryFile::TStorageFactoryFile()")
<< "Cannot open file '" << path << "'";
}
// Record the statistics.
try {
edm::Service<edm::storage::StatisticsSenderService> statsService;
if (statsService.isAvailable()) {
statsService->setSize(storage_->size());
}
} catch (edm::Exception e) {
if (e.categoryCode() != edm::errors::NotFound) {
throw;
}
}
fRealName = path;
fD = 0; // sorry, meaningless
fWritable = read ? kFALSE : kTRUE;
Init(create);
stats.tick(0);
}
| Bool_t TStorageFactoryFile::ReadBuffer | ( | char * | buf, |
| Int_t | len | ||
| ) | [virtual] |
Definition at line 232 of file TStorageFactoryFile.cc.
References trackerHits::c, n, s_statsCPrefetch, s_statsCRead, s_statsRead, s_statsXRead, storage_, storageCounter(), StorageAccount::Stamp::tick(), and IOInput::xread().
Referenced by ReadBuffer().
{
// Check that it's valid to access this file.
if (IsZombie())
{
Error("ReadBuffer", "Cannot read from a zombie file");
return kTRUE;
}
if (! IsOpen())
{
Error("ReadBuffer", "Cannot read from a file that is not open");
return kTRUE;
}
// Read specified byte range from the storage. Returns kTRUE in
// case of error. Note that ROOT uses this function recursively
// to fill the cache; we use a flag to make sure our accounting
// is reflected in a comprehensible manner. The "read" counter
// will include both, "readc" indicates how much read from the
// cache, "readu" indicates how much we failed to read from the
// cache (excluding those recursive reads), and "readx" counts
// the amount actually passed to read from the storage object.
StorageAccount::Stamp stats(storageCounter(s_statsRead, "read"));
// If we have a cache, read from there first. This returns 0
// if the block hasn't been prefetched, 1 if it was in cache,
// and 2 if there was an error.
if (TFileCacheRead *c = GetCacheRead())
{
Long64_t here = GetRelOffset();
Bool_t async = c->IsAsyncReading();
StorageAccount::Stamp cstats(async
? storageCounter(s_statsCPrefetch, "readPrefetchToCache")
: storageCounter(s_statsCRead, "readViaCache"));
Int_t st = ReadBufferViaCache(async ? 0 : buf, len);
if (st == 2) {
return kTRUE;
}
if (st == 1) {
if (async) {
cstats.tick(len);
Seek(here);
} else {
cstats.tick(len);
stats.tick(len);
return kFALSE;
}
}
}
// FIXME: Re-enable read-ahead if the data wasn't in cache.
// if (! st) storage_->caching(true, -1, s_readahead);
// A real read
StorageAccount::Stamp xstats(storageCounter(s_statsXRead, "readActual"));
IOSize n = storage_->xread(buf, len);
xstats.tick(n);
stats.tick(n);
return n ? kFALSE : kTRUE;
}
| Bool_t TStorageFactoryFile::ReadBuffer | ( | char * | buf, |
| Long64_t | pos, | ||
| Int_t | len | ||
| ) | [virtual] |
Definition at line 223 of file TStorageFactoryFile.cc.
References ReadBuffer().
{
// This function needs to be optimized to minimize seeks.
// See TFile::ReadBuffer(char *buf, Long64_t pos, Int_t len) in ROOT 5.27.06.
Seek(pos);
return ReadBuffer(buf, len);
}
| Bool_t TStorageFactoryFile::ReadBufferAsync | ( | Long64_t | off, |
| Int_t | len | ||
| ) | [virtual] |
Definition at line 299 of file TStorageFactoryFile.cc.
References StorageFactory::CACHE_HINT_APPLICATION, StorageFactory::CACHE_HINT_STORAGE, StorageFactory::cacheHint(), f, reco::get(), o2o::iov, Storage::prefetch(), PREFETCH_PROBE_LENGTH, s_statsARead, storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
// Check that it's valid to access this file.
if (IsZombie())
{
Error("ReadBufferAsync", "Cannot read from a zombie file");
return kTRUE;
}
if (! IsOpen())
{
Error("ReadBufferAsync", "Cannot read from a file that is not open");
return kTRUE;
}
StorageAccount::Stamp stats(storageCounter(s_statsARead, "readAsync"));
// If asynchronous reading is disabled, bail out now, regardless
// whether the underlying storage supports prefetching. If it is
// forced on, pretend it's on, even if the storage doesn't support
// it, as this turns off the caching in ROOT's side.
StorageFactory *f = StorageFactory::get();
// Verify that we never using async reads in app-only mode
if (f->cacheHint() == StorageFactory::CACHE_HINT_APPLICATION)
return kTRUE;
// Let the I/O method indicate if it can do client-side prefetch.
// If it does, then for example TTreeCache will drop its own cache
// and will use the client-side cache of the actual I/O layer.
// If len is zero ROOT is probing for prefetch support.
if (len) {
// FIXME: Synchronise caching.
// storage_->caching(true, -1, 0);
;
}
IOPosBuffer iov(off, (void *) 0, len ? len : PREFETCH_PROBE_LENGTH);
if (storage_->prefetch(&iov, 1))
{
stats.tick(len);
return kFALSE;
}
// Always ask ROOT to use async reads in storage-only mode,
// regardless of whether the storage system supports it.
if (f->cacheHint() == StorageFactory::CACHE_HINT_STORAGE)
return kFALSE;
// Prefetching not available right now.
return kTRUE;
}
| Bool_t TStorageFactoryFile::ReadBuffers | ( | char * | buf, |
| Long64_t * | pos, | ||
| Int_t * | len, | ||
| Int_t | nbuf | ||
| ) | [virtual] |
Definition at line 420 of file TStorageFactoryFile.cc.
References i, o2o::iov, Storage::prefetch(), ReadBuffersSync(), s_statsARead, storage_, storageCounter(), summarizeEdmComparisonLogfiles::success, StorageAccount::Stamp::tick(), and pileupDistInMC::total.
{
// Check that it's valid to access this file.
if (IsZombie())
{
Error("ReadBuffers", "Cannot read from a zombie file");
return kTRUE;
}
if (! IsOpen())
{
Error("ReadBuffers", "Cannot read from a file that is not open");
return kTRUE;
}
// For synchronous reads, we have special logic to optimize the I/O requests
// from ROOT before handing it to the storage.
if (buf)
{
return ReadBuffersSync(buf, pos, len, nbuf);
}
// For an async read, we assume the storage system is smart enough to do the
// optimization itself.
// Read from underlying storage.
void* const nobuf = 0;
Int_t total = 0;
std::vector<IOPosBuffer> iov;
iov.reserve(nbuf);
for (Int_t i = 0; i < nbuf; ++i)
{
iov.push_back(IOPosBuffer(pos[i], nobuf, len[i]));
total += len[i];
}
// Null buffer means asynchronous reads into I/O system's cache.
bool success;
StorageAccount::Stamp astats(storageCounter(s_statsARead, "readAsync"));
// Synchronise low-level cache with the supposed cache in TFile.
// storage_->caching(true, -1, 0);
success = storage_->prefetch(&iov[0], nbuf);
astats.tick(total);
// If it didn't suceeed, pass down to the base class.
return success ? kFALSE : TFile::ReadBuffers(buf, pos, len, nbuf);
}
| Bool_t TStorageFactoryFile::ReadBuffersSync | ( | char * | buf, |
| Long64_t * | pos, | ||
| Int_t * | len, | ||
| Int_t | nbuf | ||
| ) | [private] |
Most storage systems are not prepared for the onslaught of small reads that ROOT will perform, even if they implement a vectored read interface.
Typically, on the server side, the loop is unrolled and the reads are issued sequentially - giving the OS no hint that you're about to read a very close-by byte in the near future. Normally, OS read-ahead takes care of such situations; because the storage server has so many clients, and ROOT reads look random to the OS, the read-ahead becomes disabled.
Hence, this function will repack the application-layer request into an optimized storage-layer request. The resulting request to the storage layer typically has a slightly larger number of bytes, but far less individual reads.
On average, the server's disks see a smaller number of overall reads, the number of bytes transferred over the network increases modestly (around 10%), and the single application request becomes one-to-two I/O transactions. A clear win for all cases except high-latency WAN.
Definition at line 353 of file TStorageFactoryFile.cc.
References ReadRepacker::bufferUsed(), i, o2o::iov, ReadRepacker::iov(), ReadRepacker::pack(), pos, Storage::readv(), ReadRepacker::realBytesProcessed(), query::result, s_statsXRead, storage_, storageCounter(), StorageAccount::Stamp::tick(), and ReadRepacker::unpack().
Referenced by ReadBuffers().
{
Int_t remaining = nbuf; // Number of read requests left to process.
Int_t pack_count; // Number of read requests processed by this iteration.
IOSize remaining_buffer_size=0;
// Calculate the remaining buffer size for the ROOT-owned buffer by adding
// the size of the various requests.
for (Int_t i=0; i<nbuf; i++) remaining_buffer_size+=len[i];
char *current_buffer = buf;
Long64_t *current_pos = pos;
Int_t *current_len = len;
ReadRepacker repacker;
while (remaining > 0) {
pack_count = repacker.pack(static_cast<long long int *>(current_pos), current_len, remaining, current_buffer, remaining_buffer_size);
int real_bytes_processed = repacker.realBytesProcessed();
IOSize io_buffer_used = repacker.bufferUsed();
// Issue readv, then unpack buffers.
StorageAccount::Stamp xstats(storageCounter(s_statsXRead, "readActual"));
std::vector<IOPosBuffer> &iov = repacker.iov();
IOSize result = storage_->readv(&iov[0], iov.size());
if (result != io_buffer_used) {
return kTRUE;
}
xstats.tick(io_buffer_used);
repacker.unpack(current_buffer);
// Update the location of the unused part of the input buffer.
remaining_buffer_size -= real_bytes_processed;
current_buffer += real_bytes_processed;
current_pos += pack_count;
current_len += pack_count;
remaining -= pack_count;
}
assert(remaining_buffer_size == 0);
return kFALSE;
}
| void TStorageFactoryFile::ResetErrno | ( | void | ) | const |
Definition at line 616 of file TStorageFactoryFile.cc.
{
TSystem::ResetErrno();
}
| Int_t TStorageFactoryFile::SysClose | ( | Int_t | fd | ) | [protected, virtual] |
Definition at line 564 of file TStorageFactoryFile.cc.
References Storage::close(), s_statsClose, storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
StorageAccount::Stamp stats(storageCounter(s_statsClose, "close"));
if (storage_)
{
storage_->close();
delete storage_;
storage_ = 0;
}
stats.tick();
return 0;
}
| Int_t TStorageFactoryFile::SysOpen | ( | const char * | pathname, |
| Int_t | flags, | ||
| UInt_t | mode | ||
| ) | [protected, virtual] |
Definition at line 531 of file TStorageFactoryFile.cc.
References Storage::close(), Exception, StorageFactory::get(), O_NONBLOCK, IOFlags::OpenAppend, IOFlags::OpenCreate, IOFlags::OpenExclusive, IOFlags::OpenNonBlock, IOFlags::OpenRead, IOFlags::OpenTruncate, IOFlags::OpenWrite, s_statsOpen, storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
StorageAccount::Stamp stats(storageCounter(s_statsOpen, "open"));
if (storage_)
{
storage_->close();
delete storage_;
storage_ = 0;
}
int openFlags = IOFlags::OpenRead;
if (flags & O_WRONLY) openFlags = IOFlags::OpenWrite;
else if (flags & O_RDWR) openFlags |= IOFlags::OpenWrite;
if (flags & O_CREAT) openFlags |= IOFlags::OpenCreate;
if (flags & O_APPEND) openFlags |= IOFlags::OpenAppend;
if (flags & O_EXCL) openFlags |= IOFlags::OpenExclusive;
if (flags & O_TRUNC) openFlags |= IOFlags::OpenTruncate;
if (flags & O_NONBLOCK) openFlags |= IOFlags::OpenNonBlock;
if (! (storage_ = StorageFactory::get()->open(pathname, openFlags)))
{
MakeZombie();
gDirectory = gROOT;
throw cms::Exception("TStorageFactoryFile::SysOpen()")
<< "Cannot open file '" << pathname << "'";
}
stats.tick();
return 0;
}
| Long64_t TStorageFactoryFile::SysSeek | ( | Int_t | fd, |
| Long64_t | offset, | ||
| Int_t | whence | ||
| ) | [protected, virtual] |
Definition at line 580 of file TStorageFactoryFile.cc.
References Storage::CURRENT, Storage::END, evf::evtn::offset(), Storage::position(), s_statsSeek, Storage::SET, storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
StorageAccount::Stamp stats(storageCounter(s_statsSeek, "seek"));
Storage::Relative rel = (whence == SEEK_SET ? Storage::SET
: whence == SEEK_CUR ? Storage::CURRENT
: Storage::END);
offset = storage_->position(offset, rel);
stats.tick();
return offset;
}
| Int_t TStorageFactoryFile::SysStat | ( | Int_t | fd, |
| Long_t * | id, | ||
| Long64_t * | size, | ||
| Long_t * | flags, | ||
| Long_t * | modtime | ||
| ) | [protected, virtual] |
Definition at line 602 of file TStorageFactoryFile.cc.
References s_statsStat, Storage::size(), storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
StorageAccount::Stamp stats(storageCounter(s_statsStat, "stat"));
// FIXME: Most of this is unsupported or makes no sense with Storage
*id = ::Hash(fRealName);
*size = storage_->size();
*flags = 0;
*modtime = 0;
stats.tick();
return 0;
}
| Int_t TStorageFactoryFile::SysSync | ( | Int_t | fd | ) | [protected, virtual] |
Definition at line 593 of file TStorageFactoryFile.cc.
References Storage::flush(), s_statsFlush, storage_, storageCounter(), and StorageAccount::Stamp::tick().
{
StorageAccount::Stamp stats(storageCounter(s_statsFlush, "flush"));
storage_->flush();
stats.tick();
return 0;
}
| Bool_t TStorageFactoryFile::WriteBuffer | ( | const char * | buf, |
| Int_t | len | ||
| ) | [virtual] |
Definition at line 468 of file TStorageFactoryFile.cc.
References n, s_statsCWrite, s_statsWrite, s_statsXWrite, storage_, storageCounter(), StorageAccount::Stamp::tick(), and IOOutput::xwrite().
{
// Check that it's valid to access this file.
if (IsZombie())
{
Error("WriteBuffer", "Cannot write to a zombie file");
return kTRUE;
}
if (! IsOpen())
{
Error("WriteBuffer", "Cannot write to a file that is not open");
return kTRUE;
}
if (! fWritable)
{
Error("WriteBuffer", "File is not writable");
return kTRUE;
}
StorageAccount::Stamp stats(storageCounter(s_statsWrite, "write"));
StorageAccount::Stamp cstats(storageCounter(s_statsCWrite, "writeViaCache"));
// Try first writing via a cache, and if that's not possible, directly.
Int_t st;
switch ((st = WriteBufferViaCache(buf, len)))
{
case 0:
// Actual write.
{
StorageAccount::Stamp xstats(storageCounter(s_statsXWrite, "writeActual"));
IOSize n = storage_->xwrite(buf, len);
xstats.tick(n);
stats.tick(n);
// FIXME: What if it's a short write?
return n > 0 ? kFALSE : kTRUE;
}
case 1:
cstats.tick(len);
stats.tick(len);
return kFALSE;
case 2:
default:
Error("WriteBuffer", "Error writing to cache");
return kTRUE;
}
}
Storage* TStorageFactoryFile::storage_ [private] |
Definition at line 56 of file TStorageFactoryFile.h.
Referenced by Initialize(), ReadBuffer(), ReadBufferAsync(), ReadBuffers(), ReadBuffersSync(), SysClose(), SysOpen(), SysSeek(), SysStat(), SysSync(), WriteBuffer(), and ~TStorageFactoryFile().
1.7.3