#include <LocalCacheFile.h>

Inheritance diagram for LocalCacheFile:

Public Member Functions
void	close (void) override

void	flush (void) override

	LocalCacheFile (std::unique_ptr< Storage > base, const std::string &tmpdir="")

IOOffset	position (IOOffset offset, Relative whence=SET) override

bool	prefetch (const IOPosBuffer *what, IOSize n) override

IOSize	read (IOBuffer into, IOOffset pos)

IOSize	read (void *into, IOSize n) override

virtual IOSize	read (void *into, IOSize n, IOOffset pos)

IOSize	read (void *into, IOSize n, IOOffset pos) override

IOSize	readv (IOBuffer *into, IOSize n) override

IOSize	readv (IOPosBuffer *into, IOSize n) override

void	resize (IOOffset size) override

IOSize	write (const void *from, IOSize n) override

virtual IOSize	write (const void *from, IOSize n, IOOffset pos)

IOSize	write (const void *from, IOSize n, IOOffset pos) override

IOSize	write (IOBuffer from, IOOffset pos)

IOSize	writev (const IOBuffer *from, IOSize n) override

IOSize	writev (const IOPosBuffer *from, IOSize n) override

	~LocalCacheFile (void) override

Public Member Functions inherited from Storage
virtual bool	eof (void) const

Storage &	operator= (const Storage &)=delete

virtual IOOffset	position (void) const

IOSize	read (IOBuffer into)

IOSize	read (IOBuffer into, IOOffset pos)

virtual IOSize	read (void *into, IOSize n)=0

int	read (void)

virtual IOSize	readv (IOBuffer *into, IOSize buffers)

virtual void	rewind (void)

virtual IOOffset	size (void) const

	Storage (const Storage &)=delete

	Storage (void)

virtual IOSize	write (const void *from, IOSize n)=0

IOSize	write (IOBuffer from)

IOSize	write (IOBuffer from, IOOffset pos)

IOSize	write (unsigned char byte)

virtual IOSize	writev (const IOBuffer *from, IOSize buffers)

	~Storage (void) override

Public Member Functions inherited from IOInput
IOSize	read (IOBuffer into)

int	read (void)

IOSize	xread (IOBuffer into)

IOSize	xread (void *into, IOSize n)

IOSize	xreadv (IOBuffer *into, IOSize buffers)

virtual	~IOInput (void)
	Destruct the stream. A no-op. More...

Public Member Functions inherited from IOOutput
IOSize	write (IOBuffer from)

IOSize	write (unsigned char byte)

IOSize	xwrite (const void *from, IOSize n)

IOSize	xwrite (IOBuffer from)

IOSize	xwritev (const IOBuffer *from, IOSize buffers)

virtual	~IOOutput (void)
	Destruct the stream. A no-op. More...

Private Member Functions
void	cache (IOOffset start, IOOffset end)

Private Attributes
unsigned int	cacheCount_

unsigned int	cacheTotal_

bool	closedFile_

edm::propagate_const< std::unique_ptr< File > >	file_

IOOffset	image_

std::vector< char >	present_

edm::propagate_const< std::unique_ptr< Storage > >	storage_

Additional Inherited Members
Public Types inherited from Storage
enum	Relative { SET, CURRENT, END }

Detailed Description

Proxy class to copy a file locally in large chunks.

Definition at line 12 of file LocalCacheFile.h.

Constructor & Destructor Documentation

◆ LocalCacheFile()

LocalCacheFile::LocalCacheFile	(	std::unique_ptr< Storage >	base,
		const std::string &	tmpdir = `""`
	)

Definition at line 21 of file LocalCacheFile.cc.

     : image_(base->size()),
       file_(),
       storage_(std::move(base)),
       closedFile_(false),
       cacheCount_(0),
       cacheTotal_((image_ + CHUNK_SIZE - 1) / CHUNK_SIZE) {
   present_.resize(cacheTotal_, 0);
  
   std::string pattern(tmpdir);
   if (pattern.empty())
     if (char *p = std::getenv("TMPDIR"))
       pattern = p;
   if (pattern.empty())
     pattern = "/tmp";
   pattern += "/cmssw-shadow-XXXXXX";
  
   std::vector<char> temp(pattern.c_str(), pattern.c_str() + pattern.size() + 1);
   int fd = mkstemp(&temp[0]);
   if (fd == -1) {
     edm::Exception ex(edm::errors::FileOpenError);
     ex << "Cannot create temporary file '" << pattern << "': " << strerror(errno) << " (error " << errno << ")";
     ex.addContext("LocalCacheFile::LocalCacheFile");
   }
  
   unlink(&temp[0]);
   file_ = std::make_unique<File>(fd);
   file_->resize(image_);
 }

References cms::Exception::addContext(), cacheTotal_, ztee::fd, file_, edm::errors::FileOpenError, image_, AlCaHLTBitMon_ParallelJobs::p, topSingleLeptonDQM_PU_cfi::pattern, present_, AlCaHLTBitMon_QueryRunRegistry::string, and groupFilesInBlocks::temp.

◆ ~LocalCacheFile()

LocalCacheFile::~LocalCacheFile ( void )

override

Definition at line 51 of file LocalCacheFile.cc.

51 {}

Member Function Documentation

◆ cache()

void LocalCacheFile::cache	(	IOOffset	start,
		IOOffset	end
	)

private

Definition at line 53 of file LocalCacheFile.cc.

                                                        {
   start = (start / CHUNK_SIZE) * CHUNK_SIZE;
   end = std::min(end, image_);
  
   IOSize nread = 0;
   IOSize index = start / CHUNK_SIZE;
  
   while (start < end) {
     IOSize len = std::min(image_ - start, CHUNK_SIZE);
     if (!present_[index]) {
       void *window = mmap(nullptr, len, PROT_READ | PROT_WRITE, MAP_SHARED, file_->fd(), start);
       if (window == MAP_FAILED) {
         edm::Exception ex(edm::errors::FileReadError);
         ex << "Unable to map a window of local cache file: " << strerror(errno) << " (error " << errno << ")";
         ex.addContext("LocalCacheFile::cache()");
         throw ex;
       }
  
       try {
         nread = storage_->read(window, len, start);
       } catch (cms::Exception &e) {
         munmap(window, len);
         std::ostringstream ost;
         ost << "Unable to cache " << len << " byte file segment at " << start << ": ";
         edm::Exception ex(edm::errors::FileReadError, ost.str(), e);
         ex.addContext("LocalCacheFile::cache()");
         throw ex;
       }
  
       munmap(window, len);
  
       if (nread != len) {
         edm::Exception ex(edm::errors::FileReadError);
         ex << "Unable to cache " << len << " byte file segment at " << start << ": got only " << nread << " bytes back";
         ex.addContext("LocalCacheFile::cache()");
         throw ex;
       }
  
       present_[index] = 1;
       ++cacheCount_;
       if (cacheCount_ == cacheTotal_) {
         storage_->close();
         closedFile_ = true;
       }
     }
  
     start += len;
     ++index;
   }
 }

References cms::Exception::addContext(), cacheCount_, cacheTotal_, CHUNK_SIZE, closedFile_, MillePedeFileConverter_cfg::e, mps_fire::end, file_, edm::errors::FileReadError, image_, min(), present_, command_line::start, storage_, and svgfig::window().

Referenced by prefetch(), read(), and readv().

◆ close()

void LocalCacheFile::close ( void )

overridevirtual

Reimplemented from Storage.

Definition at line 162 of file LocalCacheFile.cc.

                                {
   if (!closedFile_) {
     storage_->close();
   }
   file_->close();
 }

References closedFile_, file_, and storage_.

Referenced by esMonitoring.AsyncLineReaderMixin::handle_close(), and esMonitoring.FDJsonServer::handle_close().

◆ flush()

void LocalCacheFile::flush ( void )

overridevirtual

Reimplemented from Storage.

Definition at line 160 of file LocalCacheFile.cc.

160 { nowrite("flush"); }

References nowrite().

◆ position()

IOOffset LocalCacheFile::position	(	IOOffset	offset,
		Relative	whence = `SET`
	)

overridevirtual

Implements Storage.

Definition at line 156 of file LocalCacheFile.cc.

156 { return file_->position(offset, whence); }

References file_, and hltrates_dqm_sourceclient-live_cfg::offset.

◆ prefetch()

bool LocalCacheFile::prefetch	(	const IOPosBuffer *	what,
		IOSize	n
	)

overridevirtual

Reimplemented from Storage.

Definition at line 169 of file LocalCacheFile.cc.

                                                                {
   for (IOSize i = 0; i < n; ++i) {
     IOOffset start = what[i].offset();
     IOOffset end = start + what[i].size();
     cache(start, end);
   }
  
   return file_->prefetch(what, n);
 }

References cache(), mps_fire::end, file_, mps_fire::i, dqmiodumpmetadata::n, IOPosBuffer::offset(), and IOPosBuffer::size().

◆ read() [1/4]

IOSize Storage::read

Definition at line 10 of file Storage.cc.

10 { return read(into.data(), into.size(), pos); }

◆ read() [2/4]

IOSize LocalCacheFile::read	(	void *	into,
		IOSize	n
	)

overridevirtual

Read into into at most n number of bytes.

If this is a blocking stream, the call will block until some data can be read, end of input is reached, or an exception is thrown. For a non-blocking stream the available input is returned. If none is available, an exception is thrown.

Returns: The number of bytes actually read. This is less or equal to the size of the buffer. Zero indicates that the end of the input has been reached: end of file, or remote end closing for a connected channel like a pipe or a socket. Otherwise the value can be less than requested if limited amount of input is currently available for platform or implementation reasons.

Exceptions

In	case of error, a #IOError exception is thrown. This includes the situation where the input stream is in non-blocking mode and no input is currently available (FIXME: make this simpler; clarify which exception).

Implements IOInput.

Definition at line 104 of file LocalCacheFile.cc.

                                                 {
   IOOffset here = file_->position();
   cache(here, here + n);
  
   return file_->read(into, n);
 }

References cache(), file_, and dqmiodumpmetadata::n.

◆ read() [3/4]

IOSize Storage::read

Definition at line 12 of file Storage.cc.

                                                        {
   // FIXME: this is not thread safe!  split into separate interface
   // that a particular storage can choose to support or not?  make
   // sure that throw semantics are correct here!
   // FIXME: use saveposition object in case exceptions are thrown?
   IOOffset here = position();
   position(pos);
   n = read(into, n);
   position(here);
   return n;
 }

◆ read() [4/4]

IOSize LocalCacheFile::read	(	void *	into,
		IOSize	n,
		IOOffset	pos
	)

overridevirtual

Reimplemented from Storage.

Definition at line 111 of file LocalCacheFile.cc.

                                                               {
   cache(pos, pos + n);
   return file_->read(into, n, pos);
 }

References cache(), file_, and dqmiodumpmetadata::n.

◆ readv() [1/2]

IOSize LocalCacheFile::readv	(	IOBuffer *	into,
		IOSize	buffers
	)

overridevirtual

Read from the input stream into multiple scattered buffers. There are buffers to fill in an array starting at into; the memory those buffers occupy does not need to be contiguous. The buffers are filled in the order given, eac buffer is filled fully before the subsequent buffers.

If this is a blocking stream, the call will block until some data can be read, end of input is reached, or an exception is thrown. For a non-blocking stream the available input is returned. If none is available, an exception is thrown.

The base class implementation uses read(void *, IOSize) method, but derived classes may implement a more efficient alternative.

Returns: The number of bytes actually read. This is less or equal to the size of the buffer. Zero indicates that the end of the input has been reached: end of file, or remote end closing for a connected channel like a pipe or a socket. Otherwise the value can be less than requested if limited amount of input is currently available for platform or implementation reasons. Note that the return value indicates the number of bytes read, not the number of buffers; it is the sum total of bytes filled into all the buffers.

Exceptions

In case of error, a #IOError exception is thrown. However if some data has already been read, the error is swallowed and the method returns the data read so far. It is assumed that persistent errors will occur anyway on the next read and sporadic errors like stream becoming unvailable can be ignored. Use xread() if a different policy is desirable.

Reimplemented from IOInput.

Definition at line 116 of file LocalCacheFile.cc.

                                                      {
   IOOffset start = file_->position();
   IOOffset end = start;
   for (IOSize i = 0; i < n; ++i)
     end += into[i].size();
   cache(start, end);
  
   return file_->readv(into, n);
 }

References cache(), mps_fire::end, file_, mps_fire::i, dqmiodumpmetadata::n, Storage::size(), and command_line::start.

◆ readv() [2/2]

IOSize LocalCacheFile::readv	(	IOPosBuffer *	into,
		IOSize	n
	)

overridevirtual

Reimplemented from Storage.

Definition at line 126 of file LocalCacheFile.cc.

                                                         {
   for (IOSize i = 0; i < n; ++i) {
     IOOffset start = into[i].offset();
     IOOffset end = start + into[i].size();
     cache(start, end);
   }
  
   return storage_->readv(into, n);
 }

References cache(), mps_fire::end, mps_fire::i, dqmiodumpmetadata::n, IOPosBuffer::offset(), IOPosBuffer::size(), and storage_.

◆ resize()

void LocalCacheFile::resize ( IOOffset size )

overridevirtual

Implements Storage.

Definition at line 158 of file LocalCacheFile.cc.

158 { nowrite("resize"); }

References nowrite().

◆ write() [1/4]

IOSize LocalCacheFile::write	(	const void *	from,
		IOSize	n
	)

overridevirtual

Write n bytes of data starting at address from.

Returns: The number of bytes written. Normally this will be n, but can be less, even zero, for example if the stream is non-blocking mode and cannot accept input at this time.

Exceptions

In	case of error, an exception is thrown. However if the stream is in non-blocking mode and cannot accept output, it will not throw an exception – the return value will be less than requested.

Implements IOOutput.

Definition at line 136 of file LocalCacheFile.cc.

                                                    {
   nowrite("write");
   return 0;
 }

References nowrite().

◆ write() [2/4]

IOSize Storage::write

Definition at line 44 of file Storage.cc.

                                                               {
   // FIXME: this is not thread safe!  split into separate interface
   // that a particular storage can choose to support or not?  make
   // sure that throw semantics are correct here!
  
   // FIXME: use saveposition object in case exceptions are thrown?
   IOOffset here = position();
   position(pos);
   n = write(from, n);
   position(here);
   return n;
 }

◆ write() [3/4]

IOSize LocalCacheFile::write	(	const void *	from,
		IOSize	n,
		IOOffset	pos
	)

overridevirtual

Reimplemented from Storage.

Definition at line 141 of file LocalCacheFile.cc.

                                                                {
   nowrite("write");
   return 0;
 }

References nowrite().

◆ write() [4/4]

IOSize Storage::write

Definition at line 42 of file Storage.cc.

42 { return write(from.data(), from.size(), pos); }

◆ writev() [1/2]

IOSize LocalCacheFile::writev	(	const IOBuffer *	from,
		IOSize	buffers
	)

overridevirtual

Write to the output stream from multiple buffers. There are buffers to fill in an array starting at from. The buffers are filled in the order given, each buffer fully before the subsequent buffers. The method uses write(const void *, IOSize), but may be implemented more efficiently in derived classes.

Note that derived classes should not normally call this method, as it simply routes the call back to derived class through the other virtual functions. Use this method only at the "outside edge" when transferring calls from one object to another, not in up/down calls in the inheritance tree.

Returns: The number of bytes actually written. This is less or equal to the size of the buffers. The value can be less than requested if the stream is unable to accept all the output for platform or implementation reasons. Note that the return value indicates the number of bytes written, not the number of buffers; it is the sum total of bytes written from all the buffers.

Exceptions

In	case of error, an exception is thrown. However if the stream is in non-blocking mode and cannot accept output, it will not throw an exception – the return value will be less than requested.

Reimplemented from IOOutput.

Definition at line 146 of file LocalCacheFile.cc.

                                                         {
   nowrite("writev");
   return 0;
 }

References nowrite().

◆ writev() [2/2]

IOSize LocalCacheFile::writev	(	const IOPosBuffer *	from,
		IOSize	n
	)

overridevirtual

Reimplemented from Storage.

Definition at line 151 of file LocalCacheFile.cc.

                                                            {
   nowrite("writev");
   return 0;
 }

References nowrite().

Member Data Documentation

◆ cacheCount_

unsigned int LocalCacheFile::cacheCount_

private

Definition at line 43 of file LocalCacheFile.h.

Referenced by cache().

◆ cacheTotal_

unsigned int LocalCacheFile::cacheTotal_

private

Definition at line 44 of file LocalCacheFile.h.

Referenced by cache(), and LocalCacheFile().

◆ closedFile_

bool LocalCacheFile::closedFile_

private

Definition at line 42 of file LocalCacheFile.h.

Referenced by cache(), and close().

◆ file_

edm::propagate_const<std::unique_ptr<File> > LocalCacheFile::file_

private

Definition at line 40 of file LocalCacheFile.h.

Referenced by cache(), close(), LocalCacheFile(), position(), prefetch(), read(), and readv().

◆ image_

IOOffset LocalCacheFile::image_

private

Definition at line 38 of file LocalCacheFile.h.

Referenced by cache(), and LocalCacheFile().

◆ present_

std::vector<char> LocalCacheFile::present_

private

Definition at line 39 of file LocalCacheFile.h.

Referenced by cache(), and LocalCacheFile().

◆ storage_

edm::propagate_const<std::unique_ptr<Storage> > LocalCacheFile::storage_

private

Definition at line 41 of file LocalCacheFile.h.

Referenced by cache(), close(), and readv().

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ LocalCacheFile()

◆ ~LocalCacheFile()

Member Function Documentation

◆ cache()

◆ close()

◆ flush()

◆ position()

◆ prefetch()

◆ read() [1/4]

◆ read() [2/4]

◆ read() [3/4]

◆ read() [4/4]

◆ readv() [1/2]

◆ readv() [2/2]

◆ resize()

◆ write() [1/4]

◆ write() [2/4]

◆ write() [3/4]

◆ write() [4/4]

◆ writev() [1/2]

◆ writev() [2/2]

Member Data Documentation

◆ cacheCount_

◆ cacheTotal_

◆ closedFile_

◆ file_

◆ image_

◆ present_

◆ storage_