#include <XrdFile.h>

Inheritance diagram for XrdFile:

Public Member Functions
virtual void	abort (void)

void	close (void) override

virtual void	create (const char *name, bool exclusive=false, int perms=0666)

virtual void	create (const std::string &name, bool exclusive=false, int perms=0666)

virtual void	open (const char *name, int flags=IOFlags::OpenRead, int perms=0666)

virtual void	open (const std::string &name, int flags=IOFlags::OpenRead, int perms=0666)

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

virtual IOOffset	position (IOOffset offset, Relative whence=SET)=0

virtual IOOffset	position (void) const

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

IOSize	read (IOBuffer into)

IOSize	read (IOBuffer into, IOOffset pos)

IOSize	read (void *into, IOSize n) override

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

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

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

int	read (void)

virtual IOSize	readv (IOBuffer *into, IOSize buffers)

IOSize	readv (IOBuffer *into, IOSize n) override

virtual IOSize	readv (IOPosBuffer *into, IOSize buffers)

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)=0

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

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

IOSize	write (IOBuffer from)

IOSize	write (IOBuffer from, IOOffset pos)

IOSize	write (unsigned char byte)

	XrdFile (const char *name, int flags=IOFlags::OpenRead, int perms=0666)

	XrdFile (const std::string &name, int flags=IOFlags::OpenRead, int perms=0666)

	XrdFile (IOFD fd)

	XrdFile (void)

	~XrdFile (void) override

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

virtual void	flush (void)

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 (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)

virtual IOSize	writev (const IOPosBuffer *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)

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

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	addConnection (cms::Exception &)

std::shared_ptr< XrdCl::File >	getActiveFile ()

Private Attributes
bool	m_close

std::string	m_name

IOOffset	m_offset

std::atomic< unsigned int >	m_op_count

edm::propagate_const< std::shared_ptr< XrdAdaptor::RequestManager > >	m_requestmanager

IOOffset	m_size

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

Detailed Description

Definition at line 17 of file XrdFile.h.

Constructor & Destructor Documentation

◆ XrdFile() [1/4]

XrdFile::XrdFile ( void )

Definition at line 22 of file XrdFile.cc.

22 : m_offset(0), m_size(-1), m_close(false), m_name(), m_op_count(0) {}

◆ XrdFile() [2/4]

XrdFile::XrdFile ( IOFD fd )

◆ XrdFile() [3/4]

XrdFile::XrdFile	(	const char *	name,
		int	flags = `IOFlags::OpenRead`,
		int	perms = `0666`
	)

Definition at line 24 of file XrdFile.cc.

     : m_offset(0), m_size(-1), m_close(false), m_name(), m_op_count(0) {
   open(name, flags, perms);
 }

References HLT_2018_cff::flags, Skims_PA_cff::name, and open().

◆ XrdFile() [4/4]

XrdFile::XrdFile	(	const std::string &	name,
		int	flags = `IOFlags::OpenRead`,
		int	perms = `0666`
	)

Definition at line 29 of file XrdFile.cc.

     : m_offset(0), m_size(-1), m_close(false), m_name(), m_op_count(0) {
   open(name.c_str(), flags, perms);
 }

References HLT_2018_cff::flags, Skims_PA_cff::name, and open().

◆ ~XrdFile()

XrdFile::~XrdFile ( void )

override

Definition at line 34 of file XrdFile.cc.

                       {
   if (m_close)
     edm::LogError("XrdFileError") << "Destructor called on XROOTD file '" << m_name << "' but the file is still open";
 }

References m_close, and m_name.

Member Function Documentation

◆ abort()

void XrdFile::abort ( void )

virtual

Definition at line 168 of file XrdFile.cc.

                         {
   m_requestmanager = nullptr;  // propagate_const<T> has no reset() function
   m_close = false;
   m_offset = 0;
   m_size = -1;
 }

References m_close, m_offset, m_requestmanager, and m_size.

◆ addConnection()

void XrdFile::addConnection ( cms::Exception & ex )

private

Definition at line 517 of file XrdFile.cc.

                                             {
   if (m_requestmanager.get()) {
     m_requestmanager->addConnections(ex);
   }
 }

References edm::propagate_const< T >::get(), and m_requestmanager.

Referenced by open(), position(), read(), resize(), and write().

◆ close()

void XrdFile::close ( void )

overridevirtual

Reimplemented from Storage.

Definition at line 153 of file XrdFile.cc.

                         {
   if (!m_requestmanager.get()) {
     edm::LogError("XrdFileError") << "XrdFile::close(name='" << m_name << "') called but the file is not open";
     m_close = false;
     return;
   }
  
   m_requestmanager = nullptr;  // propagate_const<T> has no reset() function
  
   m_close = false;
   m_offset = 0;
   m_size = -1;
   edm::LogInfo("XrdFileInfo") << "Closed " << m_name;
 }

References edm::propagate_const< T >::get(), m_close, m_name, m_offset, m_requestmanager, and m_size.

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

◆ create() [1/2]

void XrdFile::create	(	const char *	name,
		bool	exclusive = `false`,
		int	perms = `0666`
	)

virtual

Definition at line 40 of file XrdFile.cc.

                                                                     {
   open(name,
        (IOFlags::OpenCreate | IOFlags::OpenWrite | IOFlags::OpenTruncate | (exclusive ? IOFlags::OpenExclusive : 0)),
        perms);
 }

References generator_cfi::exclusive, Skims_PA_cff::name, open(), IOFlags::OpenCreate, IOFlags::OpenExclusive, IOFlags::OpenTruncate, and IOFlags::OpenWrite.

◆ create() [2/2]

void XrdFile::create	(	const std::string &	name,
		bool	exclusive = `false`,
		int	perms = `0666`
	)

virtual

Definition at line 46 of file XrdFile.cc.

                                                                          {
   open(name.c_str(),
        (IOFlags::OpenCreate | IOFlags::OpenWrite | IOFlags::OpenTruncate | (exclusive ? IOFlags::OpenExclusive : 0)),
        perms);
 }

References generator_cfi::exclusive, Skims_PA_cff::name, open(), IOFlags::OpenCreate, IOFlags::OpenExclusive, IOFlags::OpenTruncate, and IOFlags::OpenWrite.

◆ getActiveFile()

std::shared_ptr< XrdCl::File > XrdFile::getActiveFile ( void )

private

Returns a file handle from one of the active sources. Verifies the file is open and throws an exception as necessary.

Definition at line 505 of file XrdFile.cc.

                                                     {
   if (!m_requestmanager.get()) {
     cms::Exception ex("XrdFileLogicError");
     ex << "Xrd::getActiveFile(name='" << m_name << "') no active request manager";
     ex.addContext("Calling XrdFile::getActiveFile()");
     m_requestmanager->addConnections(ex);
     m_close = false;
     throw ex;
   }
   return m_requestmanager->getActiveFile();
 }

References cms::Exception::addContext(), edm::propagate_const< T >::get(), m_close, m_name, and m_requestmanager.

Referenced by open(), and write().

◆ open() [1/2]

void XrdFile::open	(	const char *	name,
		int	flags = `IOFlags::OpenRead`,
		int	perms = `0666`
	)

virtual

Definition at line 56 of file XrdFile.cc.

                                                              {
   // Actual open
   if ((name == nullptr) || (*name == 0)) {
     edm::Exception ex(edm::errors::FileOpenError);
     ex << "Cannot open a file without a name";
     ex.addContext("Calling XrdFile::open()");
     throw ex;
   }
   if ((flags & (IOFlags::OpenRead | IOFlags::OpenWrite)) == 0) {
     edm::Exception ex(edm::errors::FileOpenError);
     ex << "Must open file '" << name << "' at least for read or write";
     ex.addContext("Calling XrdFile::open()");
     throw ex;
   }
  
   // Translate our flags to system flags
   XrdCl::OpenFlags::Flags openflags = XrdCl::OpenFlags::None;
  
   if (flags & IOFlags::OpenWrite)
     openflags |= XrdCl::OpenFlags::Update;
   else if (flags & IOFlags::OpenRead)
     openflags |= XrdCl::OpenFlags::Read;
  
   if (flags & IOFlags::OpenAppend) {
     edm::Exception ex(edm::errors::FileOpenError);
     ex << "Opening file '" << name << "' in append mode not supported";
     ex.addContext("Calling XrdFile::open()");
     throw ex;
   }
  
   if (flags & IOFlags::OpenCreate) {
     if (!(flags & IOFlags::OpenExclusive))
       openflags |= XrdCl::OpenFlags::Delete;
     openflags |= XrdCl::OpenFlags::New;
     openflags |= XrdCl::OpenFlags::MakePath;
   }
  
   if ((flags & IOFlags::OpenTruncate) && (flags & IOFlags::OpenWrite))
     openflags |= XrdCl::OpenFlags::Delete;
  
   // Translate mode flags
   XrdCl::Access::Mode modeflags = XrdCl::Access::None;
   modeflags |= (perms & S_IRUSR) ? XrdCl::Access::UR : XrdCl::Access::None;
   modeflags |= (perms & S_IWUSR) ? XrdCl::Access::UW : XrdCl::Access::None;
   modeflags |= (perms & S_IXUSR) ? XrdCl::Access::UX : XrdCl::Access::None;
   modeflags |= (perms & S_IRGRP) ? XrdCl::Access::GR : XrdCl::Access::None;
   modeflags |= (perms & S_IWGRP) ? XrdCl::Access::GW : XrdCl::Access::None;
   modeflags |= (perms & S_IXGRP) ? XrdCl::Access::GX : XrdCl::Access::None;
   modeflags |= (perms & S_IROTH) ? XrdCl::Access::GR : XrdCl::Access::None;
   modeflags |= (perms & S_IWOTH) ? XrdCl::Access::GW : XrdCl::Access::None;
   modeflags |= (perms & S_IXOTH) ? XrdCl::Access::GX : XrdCl::Access::None;
  
   m_requestmanager = RequestManager::getInstance(name, openflags, modeflags);
   m_name = name;
  
   // Stat the file so we can keep track of the offset better.
   auto file = getActiveFile();
   XrdCl::XRootDStatus status;
   XrdCl::StatInfo *statInfo = nullptr;
   if (!(status = file->Stat(false, statInfo)).IsOK()) {
     edm::Exception ex(edm::errors::FileOpenError);
     ex << "XrdCl::File::Stat(name='" << name << ") => error '" << status.ToStr() << "' (errno=" << status.errNo
        << ", code=" << status.code << ")";
     ex.addContext("Calling XrdFile::open()");
     addConnection(ex);
     throw ex;
   }
   assert(statInfo);
   m_size = statInfo->GetSize();
   delete (statInfo);
  
   m_offset = 0;
   m_close = true;
  
   // Send the monitoring info, if available.
   // Note: std::getenv is not reentrant.
   // Commenting out until this is available in the new client.
   /*
   char * crabJobId = std::getenv(kCrabJobIdEnv);
   if (crabJobId) {
     kXR_unt32 dictId;
     m_file->SendMonitoringInfo(crabJobId, &dictId);
     edm::LogInfo("XrdFileInfo") << "Set monitoring ID to " << crabJobId << " with resulting dictId " << dictId << ".";
   }
 */
  
   edm::LogInfo("XrdFileInfo") << "Opened " << m_name;
  
   std::vector<std::string> sources;
   m_requestmanager->getActiveSourceNames(sources);
   std::stringstream ss;
   ss << "Active sources: ";
   for (auto const &it : sources)
     ss << it << ", ";
   edm::LogInfo("XrdFileInfo") << ss.str();
 }

References addConnection(), cms::Exception::addContext(), cms::cuda::assert(), FrontierConditions_GlobalTag_cff::file, edm::errors::FileOpenError, HLT_2018_cff::flags, getActiveFile(), m_close, m_name, m_offset, m_requestmanager, m_size, Skims_PA_cff::name, None, IOFlags::OpenAppend, IOFlags::OpenCreate, IOFlags::OpenExclusive, IOFlags::OpenRead, IOFlags::OpenTruncate, IOFlags::OpenWrite, CalibrationSummaryClient_cfi::sources, contentValuesCheck::ss, and mps_update::status.

Referenced by create(), open(), and XrdFile().

◆ open() [2/2]

void XrdFile::open	(	const std::string &	name,
		int	flags = `IOFlags::OpenRead`,
		int	perms = `0666`
	)

virtual

Definition at line 52 of file XrdFile.cc.

                                                                   {
   open(name.c_str(), flags, perms);
 }

References HLT_2018_cff::flags, Skims_PA_cff::name, and open().

◆ position() [1/3]

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

overridevirtual

Implements Storage.

Definition at line 457 of file XrdFile.cc.

                                                              {
   if (!m_requestmanager.get()) {
     cms::Exception ex("FilePositionError");
     ex << "XrdFile::position() called on a closed file";
     ex.addContext("Calling XrdFile::position()");
     addConnection(ex);
     throw ex;
   }
   switch (whence) {
     case SET:
       m_offset = offset;
       break;
  
     case CURRENT:
       m_offset += offset;
       break;
  
     // TODO: None of this works with concurrent writers to the file.
     case END:
       assert(m_size != -1);
       m_offset = m_size + offset;
       break;
  
     default:
       cms::Exception ex("FilePositionError");
       ex << "XrdFile::position() called with incorrect 'whence' parameter";
       ex.addContext("Calling XrdFile::position()");
       addConnection(ex);
       throw ex;
   }
  
   if (m_offset < 0)
     m_offset = 0;
   assert(m_size != -1);
   if (m_offset > m_size)
     m_size = m_offset;
  
   return m_offset;
 }

References addConnection(), cms::Exception::addContext(), cms::cuda::assert(), Storage::CURRENT, Storage::END, edm::propagate_const< T >::get(), m_offset, m_requestmanager, m_size, hltrates_dqm_sourceclient-live_cfg::offset, and Storage::SET.

◆ position() [2/3]

virtual IOOffset Storage::position

◆ position() [3/3]

IOOffset Storage::position

Definition at line 72 of file Storage.cc.

                                      {
   Storage *self = const_cast<Storage *>(this);
   return self->position(0, CURRENT);
 }

◆ prefetch()

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

overridevirtual

Reimplemented from Storage.

Definition at line 448 of file XrdFile.cc.

                                                         {
   // The new Xrootd client does not contain any internal buffers.
   // Hence, prefetching is disabled completely.
   return false;
 }

◆ read() [1/7]

IOSize IOInput::read

Read from the input stream into the buffer starting at into and of size n.

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 simply forwards the call to read(void *, IOSize) method.

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).

Definition at line 80 of file IOInput.cc.

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

◆ read() [2/7]

IOSize Storage::read

Definition at line 10 of file Storage.cc.

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

◆ read() [3/7]

IOSize XrdFile::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 176 of file XrdFile.cc.

                                          {
   if (n > 0x7fffffff) {
     edm::Exception ex(edm::errors::FileReadError);
     ex << "XrdFile::read(name='" << m_name << "', n=" << n << ") too many bytes, limit is 0x7fffffff";
     ex.addContext("Calling XrdFile::read()");
     addConnection(ex);
     throw ex;
   }
  
   uint32_t bytesRead = m_requestmanager->handle(into, n, m_offset).get();
   m_offset += bytesRead;
   return bytesRead;
 }

References addConnection(), cms::Exception::addContext(), edm::errors::FileReadError, edm::propagate_const< T >::get(), m_name, m_offset, m_requestmanager, and dqmiodumpmetadata::n.

◆ read() [4/7]

IOSize IOInput::read

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).

◆ read() [5/7]

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() [6/7]

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

overridevirtual

Reimplemented from Storage.

Definition at line 190 of file XrdFile.cc.

                                                        {
   if (n > 0x7fffffff) {
     edm::Exception ex(edm::errors::FileReadError);
     ex << "XrdFile::read(name='" << m_name << "', n=" << n << ") exceeds read size limit 0x7fffffff";
     ex.addContext("Calling XrdFile::read()");
     addConnection(ex);
     throw ex;
   }
   if (n == 0) {
     return 0;
   }
  
   // In some cases, the IO layers above us (particularly, if lazy-download is
   // enabled) will emit very large reads.  We break this up into multiple
   // reads in order to avoid hitting timeouts.
   std::future<IOSize> prev_future, cur_future;
   IOSize bytesRead = 0, prev_future_expected = 0, cur_future_expected = 0;
   bool readReturnedShort = false;
  
   // Check the status of a read operation; updates bytesRead and
   // readReturnedShort.
   auto check_read = [&](std::future<IOSize> &future, IOSize expected) {
     if (!future.valid()) {
       return;
     }
     IOSize result = future.get();
     if (readReturnedShort && (result != 0)) {
       edm::Exception ex(edm::errors::FileReadError);
       ex << "XrdFile::read(name='" << m_name << "', n=" << n
          << ") remote server returned non-zero length read after EOF.";
       ex.addContext("Calling XrdFile::read()");
       addConnection(ex);
       throw ex;
     } else if (result != expected) {
       readReturnedShort = true;
     }
     bytesRead += result;
   };
  
   while (n) {
     IOSize chunk = std::min(n, static_cast<IOSize>(XRD_CL_MAX_READ_SIZE));
  
     // Save prior read state; issue new read.
     prev_future = std::move(cur_future);
     prev_future_expected = cur_future_expected;
     cur_future = m_requestmanager->handle(into, chunk, pos);
     cur_future_expected = chunk;
  
     // Wait for the prior read; update bytesRead.
     check_read(prev_future, prev_future_expected);
  
     // Update counters.
     into = static_cast<char *>(into) + chunk;
     n -= chunk;
     pos += chunk;
   }
  
   // Wait for the last read to finish.
   check_read(cur_future, cur_future_expected);
  
   return bytesRead;
 }

References addConnection(), cms::Exception::addContext(), edm::errors::FileReadError, m_name, m_requestmanager, min(), eostools::move(), dqmiodumpmetadata::n, mps_fire::result, and XRD_CL_MAX_READ_SIZE.

◆ read() [7/7]

int IOInput::read

Read the next single byte from the input stream and return it as an unsigned char cast to an int, -1 to indicate end of intput data.

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

The base class implementation simply forwards the call to read(void *, IOSize) method.

Returns: The byte cast from a unsigned char to an int (in range 0...255, inclusive) if one could be read, or -1 to indicate end of input data.

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).

Definition at line 52 of file IOInput.cc.

                       {
   unsigned char byte;
   IOSize n = read(&byte, 1);
   return n == 0 ? -1 : byte;
 }

◆ readv() [1/4]

IOSize IOInput::readv

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.

Definition at line 111 of file IOInput.cc.

                                                     {
   assert(!buffers || into);
  
   // Keep reading as long as possible; ignore errors if we have read
   // something, otherwise pass it on.
   IOSize status;
   IOSize done = 0;
   try {
     for (IOSize i = 0; i < buffers; done += status, ++i)
       if ((status = read(into[i])) == 0)
         break;
   } catch (cms::Exception &) {
     if (!done)
       throw;
   }
  
   return done;
 }

◆ readv() [2/4]

IOSize XrdFile::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 254 of file XrdFile.cc.

                                               {
   std::vector<IOPosBuffer> new_buf;
   new_buf.reserve(n);
   IOOffset off = 0;
   for (IOSize i = 0; i < n; i++) {
     IOSize size = into[i].size();
     new_buf[i] = IOPosBuffer(off, into[i].data(), size);
     off += size;
   }
   return readv(&(new_buf[0]), n);
 }

References data, mps_fire::i, dqmiodumpmetadata::n, IOBuffer::size(), and Storage::size().

◆ readv() [3/4]

IOSize Storage::readv

Definition at line 24 of file Storage.cc.

                                                  {
   IOOffset here = position();
   IOSize total = 0;
   for (IOSize i = 0; i < n; ++i) {
     try {
       position(into[i].offset());
       total += read(into[i].data(), into[i].size());
     } catch (cms::Exception &) {
       if (!total)
         throw;
       break;
     }
   }
   position(here);
   return total;
 }

◆ readv() [4/4]

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

overridevirtual

Reimplemented from Storage.

Definition at line 270 of file XrdFile.cc.

                                                  {
   // A trivial vector read - unlikely, considering ROOT data format.
   if (UNLIKELY(n == 0)) {
     return 0;
   }
   if (UNLIKELY(n == 1)) {
     return read(into[0].data(), into[0].size(), into[0].offset());
   }
  
   auto cl = std::make_shared<std::vector<IOPosBuffer>>();
  
   // CMSSW may issue large readv's; Xrootd is only able to handle
   // 1024.  Further, the splitting algorithm may slightly increase
   // the number of buffers.
   IOSize adjust = XRD_CL_MAX_SIZE - 2;
   cl->reserve(n > adjust ? adjust : n);
   IOSize idx = 0, last_idx = 0;
   IOSize final_result = 0;
   std::vector<std::pair<std::future<IOSize>, IOSize>> readv_futures;
   while (idx < n) {
     cl->clear();
     IOSize size = 0;
     while (idx < n) {
       unsigned rollback_count = 1;
       IOSize current_size = size;
       IOOffset offset = into[idx].offset();
       IOSize length = into[idx].size();
       size += length;
       char *buffer = static_cast<char *>(into[idx].data());
       while (length > XRD_CL_MAX_CHUNK) {
         IOPosBuffer ci;
         ci.set_size(XRD_CL_MAX_CHUNK);
         length -= XRD_CL_MAX_CHUNK;
         ci.set_offset(offset);
         offset += XRD_CL_MAX_CHUNK;
         ci.set_data(buffer);
         buffer += XRD_CL_MAX_CHUNK;
         cl->emplace_back(ci);
         rollback_count++;
       }
       IOPosBuffer ci;
       ci.set_size(length);
       ci.set_offset(offset);
       ci.set_data(buffer);
       cl->emplace_back(ci);
  
       if (cl->size() > adjust) {
         while (rollback_count--)
           cl->pop_back();
         size = current_size;
         break;
       } else {
         idx++;
       }
     }
     try {
       readv_futures.emplace_back(m_requestmanager->handle(cl), size);
     } catch (edm::Exception &ex) {
       ex.addContext("Calling XrdFile::readv()");
       throw;
     }
  
     // Assure that we have made some progress.
     assert(last_idx < idx);
     last_idx = idx;
   }
   std::chrono::time_point<std::chrono::high_resolution_clock> start, end;
   start = std::chrono::high_resolution_clock::now();
  
   // If there are multiple readv calls, wait until all return until looking
   // at the results of any.  This guarantees that all readv's have finished
   // by time we call .get() for the first time (in case one of the readv's
   // result in an exception).
   //
   // We cannot have outstanding readv's on function exit as the XrdCl may
   // write into the corresponding buffer at the same time as ROOT.
   if (readv_futures.size() > 1) {
     for (auto &readv_result : readv_futures) {
       if (readv_result.first.valid()) {
         readv_result.first.wait();
       }
     }
   }
  
   for (auto &readv_result : readv_futures) {
     IOSize result = 0;
     try {
       const int retry_count = 5;
       for (int retries = 0; retries < retry_count; retries++) {
         try {
           if (readv_result.first.valid()) {
             result = readv_result.first.get();
           }
         } catch (XrootdException &ex) {
           if ((retries != retry_count - 1) && (ex.getCode() == XrdCl::errInvalidResponse)) {
             edm::LogWarning("XrdAdaptorInternal")
                 << "Got an invalid response from Xrootd server; retrying" << std::endl;
             result = m_requestmanager->handle(cl).get();
           } else {
             throw;
           }
         }
         assert(result == readv_result.second);
       }
     } catch (edm::Exception &ex) {
       ex.addContext("Calling XrdFile::readv()");
       throw;
     } catch (std::exception &ex) {
       edm::Exception newex(edm::errors::StdException);
       newex << "A std::exception was thrown when processing an xrootd request: " << ex.what();
       newex.addContext("Calling XrdFile::readv()");
       throw newex;
     }
     final_result += result;
   }
   end = std::chrono::high_resolution_clock::now();
  
   edm::LogVerbatim("XrdAdaptorInternal")
       << "[" << m_op_count.fetch_add(1) << "] Time for readv: "
       << static_cast<int>(std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count())
       << " (sub-readv requests: " << readv_futures.size() << ")" << std::endl;
  
   return final_result;
 }

References cms::Exception::addContext(), cms::cuda::assert(), edmScanValgrind::buffer, GetRecoTauVFromDQM_MC_cff::cl, KineDebug3::count(), data, end, cppFunctionSkipper::exception, edm::propagate_const< T >::get(), XrdAdaptor::XrootdException::getCode(), charmTagsComputerCvsB_cfi::idx, m_op_count, m_requestmanager, dqmiodumpmetadata::n, fileCollector::now, IOPosBuffer::offset(), hltrates_dqm_sourceclient-live_cfg::offset, IOInput::read(), mps_fire::result, IOPosBuffer::set_data(), IOPosBuffer::set_offset(), IOPosBuffer::set_size(), IOPosBuffer::size(), Storage::size(), command_line::start, edm::errors::StdException, UNLIKELY, cms::Exception::what(), XRD_CL_MAX_CHUNK, and XRD_CL_MAX_SIZE.

◆ resize()

void XrdFile::resize ( IOOffset size )

overridevirtual

Implements Storage.

Definition at line 497 of file XrdFile.cc.

                                {
   cms::Exception ex("FileResizeError");
   ex << "XrdFile::resize(name='" << m_name << "') not implemented";
   ex.addContext("Calling XrdFile::resize()");
   addConnection(ex);
   throw ex;
 }

References addConnection(), cms::Exception::addContext(), and m_name.

◆ write() [1/7]

IOSize XrdFile::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 395 of file XrdFile.cc.

                                                 {
   if (n > 0x7fffffff) {
     edm::Exception ex(edm::errors::FileWriteError);
     ex << "XrdFile::write(name='" << m_name << "', n=" << n << ") too many bytes, limit is 0x7fffffff";
     ex.addContext("Calling XrdFile::write()");
     addConnection(ex);
     throw ex;
   }
   auto file = getActiveFile();
  
   XrdCl::XRootDStatus s = file->Write(m_offset, n, from);
   if (!s.IsOK()) {
     edm::Exception ex(edm::errors::FileWriteError);
     ex << "XrdFile::write(name='" << m_name << "', n=" << n << ") failed with error '" << s.ToStr()
        << "' (errno=" << s.errNo << ", code=" << s.code << ")";
     ex.addContext("Calling XrdFile::write()");
     addConnection(ex);
     throw ex;
   }
   m_offset += n;
   assert(m_size != -1);
   if (m_offset > m_size)
     m_size = m_offset;
  
   return n;
 }

References addConnection(), cms::Exception::addContext(), cms::cuda::assert(), FrontierConditions_GlobalTag_cff::file, edm::errors::FileWriteError, getActiveFile(), m_name, m_offset, m_size, dqmiodumpmetadata::n, and alignCSCRings::s.

◆ write() [2/7]

IOSize IOOutput::write

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.

◆ write() [3/7]

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() [4/7]

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

overridevirtual

Reimplemented from Storage.

Definition at line 422 of file XrdFile.cc.

                                                               {
   if (n > 0x7fffffff) {
     edm::Exception ex(edm::errors::FileWriteError);
     ex << "XrdFile::write(name='" << m_name << "', n=" << n << ") too many bytes, limit is 0x7fffffff";
     ex.addContext("Calling XrdFile::write()");
     addConnection(ex);
     throw ex;
   }
   auto file = getActiveFile();
  
   XrdCl::XRootDStatus s = file->Write(pos, n, from);
   if (!s.IsOK()) {
     edm::Exception ex(edm::errors::FileWriteError);
     ex << "XrdFile::write(name='" << m_name << "', n=" << n << ") failed with error '" << s.ToStr()
        << "' (errno=" << s.errNo << ", code=" << s.code << ")";
     ex.addContext("Calling XrdFile::write()");
     addConnection(ex);
     throw ex;
   }
   assert(m_size != -1);
   if (static_cast<IOOffset>(pos + n) > m_size)
     m_size = pos + n;
  
   return n;
 }

References addConnection(), cms::Exception::addContext(), cms::cuda::assert(), FrontierConditions_GlobalTag_cff::file, edm::errors::FileWriteError, getActiveFile(), m_name, m_size, dqmiodumpmetadata::n, and alignCSCRings::s.

◆ write() [5/7]

IOSize IOOutput::write

Write to the output stream the buffer from. This method is simply redirected to write(const void *, IOSize).

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 buffer. The value can be less than requested if the stream is unable to accept all the output for platform or implementation reasons.

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.

Definition at line 59 of file IOOutput.cc.

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

◆ write() [6/7]

IOSize Storage::write

Definition at line 42 of file Storage.cc.

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

◆ write() [7/7]

IOSize IOOutput::write

Write a single byte to the output stream. This method is simply redirected to write(const void *, IOSize).

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 written. Normally this will be one, but can be zero if the stream is in non-blocking mode and cannot accept output 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 – zero will be returned.

Definition at line 39 of file IOOutput.cc.

39 { return write(&byte, 1); }

Member Data Documentation

◆ m_close

bool XrdFile::m_close

private

Definition at line 61 of file XrdFile.h.

Referenced by abort(), close(), getActiveFile(), open(), and ~XrdFile().

◆ m_name

std::string XrdFile::m_name

private

Definition at line 62 of file XrdFile.h.

Referenced by close(), getActiveFile(), open(), read(), resize(), write(), and ~XrdFile().

◆ m_offset

IOOffset XrdFile::m_offset

private

Definition at line 59 of file XrdFile.h.

Referenced by abort(), close(), open(), position(), read(), and write().

◆ m_op_count

std::atomic<unsigned int> XrdFile::m_op_count

private

Definition at line 63 of file XrdFile.h.

Referenced by readv().

◆ m_requestmanager

edm::propagate_const<std::shared_ptr<XrdAdaptor::RequestManager> > XrdFile::m_requestmanager

private

Definition at line 58 of file XrdFile.h.

Referenced by abort(), addConnection(), close(), getActiveFile(), open(), position(), read(), and readv().

◆ m_size

IOOffset XrdFile::m_size

private

Definition at line 60 of file XrdFile.h.

Referenced by abort(), close(), open(), position(), and write().

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ XrdFile() [1/4]

◆ XrdFile() [2/4]

◆ XrdFile() [3/4]

◆ XrdFile() [4/4]

◆ ~XrdFile()

Member Function Documentation

◆ abort()

◆ addConnection()

◆ close()

◆ create() [1/2]

◆ create() [2/2]

◆ getActiveFile()

◆ open() [1/2]

◆ open() [2/2]

◆ position() [1/3]

◆ position() [2/3]

◆ position() [3/3]

◆ prefetch()

◆ read() [1/7]

◆ read() [2/7]

◆ read() [3/7]

◆ read() [4/7]

◆ read() [5/7]

◆ read() [6/7]

◆ read() [7/7]

◆ readv() [1/4]

◆ readv() [2/4]

◆ readv() [3/4]

◆ readv() [4/4]

◆ resize()

◆ write() [1/7]

◆ write() [2/7]

◆ write() [3/7]

◆ write() [4/7]

◆ write() [5/7]

◆ write() [6/7]

◆ write() [7/7]

Member Data Documentation

◆ m_close

◆ m_name

◆ m_offset

◆ m_op_count

◆ m_requestmanager

◆ m_size