#include <XrdCephOssReadVFile.hh>

Inheritance diagram for XrdCephOssReadVFile:

Collaboration diagram for XrdCephOssReadVFile:

Public Member Functions
	XrdCephOssReadVFile (XrdCephOss cephoss, XrdCephOssFile cephossDF, const std::string &algname)

virtual	~XrdCephOssReadVFile ()

virtual int	Close (long long *retsz=0)

virtual int	Fstat (struct stat *buff)

virtual int	Fsync (void)

virtual int	Ftruncate (unsigned long long)

virtual int	Open (const char *path, int flags, mode_t mode, XrdOucEnv &env)

virtual ssize_t	Read (off_t offset, size_t blen)

virtual ssize_t	Read (void *buff, off_t offset, size_t blen)

virtual int	Read (XrdSfsAio *aoip)

virtual ssize_t	ReadRaw (void *, off_t, size_t)

virtual ssize_t	ReadV (XrdOucIOVec *readV, int rdvcnt)

virtual ssize_t	Write (const void *buff, off_t offset, size_t blen)

virtual int	Write (XrdSfsAio *aiop)

Public Member Functions inherited from XrdCephOssFile
	XrdCephOssFile (XrdCephOss *cephoss)

virtual	~XrdCephOssFile ()

virtual int	getFileDescriptor () const

Public Member Functions inherited from XrdOssDF
	XrdOssDF (const char *tid="", uint16_t dftype=0, int fdnum=-1)

virtual	~XrdOssDF ()

uint16_t	DFType ()

virtual int	Fchmod (mode_t mode)

virtual int	Fctl (int cmd, int alen, const char args, char *resp=0)

virtual void	Flush ()
	Flush filesystem cached pages for this file (used for checksums).

virtual int	Fsync (XrdSfsAio *aiop)

virtual int	getFD ()

virtual off_t	getMmap (void **addr)

virtual const char *	getTID ()

virtual int	isCompressed (char *cxidp=0)

virtual int	Opendir (const char *path, XrdOucEnv &env)

virtual ssize_t	pgRead (void buffer, off_t offset, size_t rdlen, uint32_t csvec, uint64_t opts)

virtual int	pgRead (XrdSfsAio *aioparm, uint64_t opts)

virtual ssize_t	pgWrite (void buffer, off_t offset, size_t wrlen, uint32_t csvec, uint64_t opts)

virtual int	pgWrite (XrdSfsAio *aioparm, uint64_t opts)

virtual int	Readdir (char *buff, int blen)

virtual int	StatRet (struct stat *buff)

virtual ssize_t	WriteV (XrdOucIOVec *writeV, int wrvcnt)

Protected Attributes
std::string	m_algname = "passthrough"

XrdCephOss *	m_cephoss = nullptr

bool	m_extraLogging = true

std::unique_ptr< XrdCephBuffer::IXrdCephReadVAdapter >	m_readVAdapter

std::atomic< long >	m_timer_count {0}
	timer for the reads against ceph

std::atomic< long >	m_timer_longest {0}
	number of reads

std::atomic< long >	m_timer_read_ns {0}

std::atomic< long >	m_timer_size {0}
	number of reads

XrdCephOssFile *	m_xrdOssDF = nullptr

Protected Attributes inherited from XrdCephOssFile
XrdCephOss *	m_cephOss

int	m_fd

Protected Attributes inherited from XrdOssDF
uint16_t	dfType

int	fd

off_t	pgwEOF

short	rsvd

const char *	tident

Additional Inherited Members
Static Public Attributes inherited from XrdOssDF
static const uint16_t	DF_isDir = 0x0001
	Object is for a directory.

static const uint16_t	DF_isFile = 0x0002
	Object is for a file.

static const uint16_t	DF_isProxy = 0x0010
	Object is a proxy object.

static const uint64_t	doCalc = 0x4000000000000000ULL
	pgw: Calculate checksums

static const int	Fctl_ckpObj = 0

static const int	Fctl_utimes = 1

static const uint64_t	Verify = 0x8000000000000000ULL
	all: Verify checksums

Detailed Description

Decorator class XrdCephOssReadVFile designed to wrap XrdCephOssFile Functionality for ReadV access to/from data in Ceph to avoid inefficient small reads / writes from the client side. Initially for monitoring purposes

Definition at line 46 of file XrdCephOssReadVFile.hh.

Constructor & Destructor Documentation

◆ XrdCephOssReadVFile()

XrdCephOssReadVFile::XrdCephOssReadVFile	(	XrdCephOss *	cephoss,
		XrdCephOssFile *	cephossDF,
		const std::string &	algname )

explicit

Definition at line 49 of file XrdCephOssReadVFile.cc.

                                                                                                              :
XrdCephOssFile(cephoss), m_cephoss(cephoss), m_xrdOssDF(cephossDF),m_algname(algname)
{
  if (!m_xrdOssDF) XrdCephEroute.Say("XrdCephOssReadVFile::Null m_xrdOssDF");
 
  if (m_algname == "passthrough") { // #TODO consider to use a factory method. but this is simple enough for now
      m_readVAdapter =  std::unique_ptr<XrdCephBuffer::IXrdCephReadVAdapter>(new XrdCephBuffer::XrdCephReadVNoOp());
  } else if (m_algname == "basic") { 
      m_readVAdapter =  std::unique_ptr<XrdCephBuffer::IXrdCephReadVAdapter>(new XrdCephBuffer::XrdCephReadVBasic());
  } else {
    XrdCephEroute.Say("XrdCephOssReadVFile::ERROR Invalid ReadV algorithm passed; defaulting to passthrough");
    m_algname = "passthrough";
    m_readVAdapter =  std::unique_ptr<XrdCephBuffer::IXrdCephReadVAdapter>(new XrdCephBuffer::XrdCephReadVNoOp());
  }
  LOGCEPH("XrdCephOssReadVFile Algorithm type: " << m_algname);
}

References XrdCephOssFile::XrdCephOssFile(), if(), LOGCEPH, m_algname, m_cephoss, m_readVAdapter, m_xrdOssDF, and XrdCephEroute.

Here is the call graph for this function:

◆ ~XrdCephOssReadVFile()

XrdCephOssReadVFile::~XrdCephOssReadVFile ( )

virtual

Definition at line 66 of file XrdCephOssReadVFile.cc.

                                          {
  if (m_xrdOssDF) {
    delete m_xrdOssDF;
    m_xrdOssDF = nullptr;
  }
 
}

References m_xrdOssDF.

Member Function Documentation

◆ Close()

int XrdCephOssReadVFile::Close ( long long * retsz = 0 )

virtual

Close a directory or file.

Parameters

retsz If not nil, where the size of the file is to be returned.

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 84 of file XrdCephOssReadVFile.cc.

                                               {
  LOGCEPH("XrdCephOssReadVFile::Close: retsz: " << retsz << " Time_ceph_s: " << m_timer_read_ns.load()*1e-9 << " count: " 
      <<  m_timer_count.load()  << " size_B: " << m_timer_size.load()
      << " longest_s:" <<  m_timer_longest.load()*1e-9);
 
  return m_xrdOssDF->Close(retsz);
}

References LOGCEPH, m_timer_count, m_timer_longest, m_timer_read_ns, m_timer_size, and m_xrdOssDF.

◆ Fstat()

int XrdCephOssReadVFile::Fstat ( struct stat * buf )

virtual

Return state information for this file.

Parameters

buf	- Pointer to the structure where info it to be returned.

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 197 of file XrdCephOssReadVFile.cc.

                                                {
  return m_xrdOssDF->Fstat(buff);
}

References m_xrdOssDF, and stat.

◆ Fsync()

int XrdCephOssReadVFile::Fsync ( void )

virtual

Synchronize associated file with media (synchronous).

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 209 of file XrdCephOssReadVFile.cc.

                               {
  return m_xrdOssDF->Fsync();
}

References m_xrdOssDF.

◆ Ftruncate()

int XrdCephOssReadVFile::Ftruncate ( unsigned long long flen )

virtual

Set the size of the associated file.

Parameters

flen	- The new size of the file.

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 213 of file XrdCephOssReadVFile.cc.

                                                         {
  return m_xrdOssDF->Ftruncate(len);
}

References m_xrdOssDF.

◆ Open()

int XrdCephOssReadVFile::Open	(	const char *	path,
		int	Oflag,
		mode_t	Mode,
		XrdOucEnv &	env )

virtual

Open a file.

Parameters

path	- Pointer to the path of the file to be opened.
Oflag	- Standard open flags.
Mode	- File open mode (ignored unless creating a file).
env	- Reference to environmental information.

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 74 of file XrdCephOssReadVFile.cc.

                                                                                      {
  int rc = m_xrdOssDF->Open(path, flags, mode, env);
  if (rc < 0) {
    return rc;
  }
  m_fd = m_xrdOssDF->getFileDescriptor();
  LOGCEPH("XrdCephOssReadVFile::Open: fd: " << m_fd << " " << path );
  return rc;
}

References LOGCEPH, XrdCephOssFile::m_fd, and m_xrdOssDF.

◆ Read() [1/3]

ssize_t XrdCephOssReadVFile::Read	(	off_t	offset,
		size_t	size )

virtual

Preread file blocks into the file system cache.

Parameters

offset	- The offset where the read is to start.
size	- The number of bytes to pre-read.

Returns: 0 upon success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 181 of file XrdCephOssReadVFile.cc.

                                                           {
  return m_xrdOssDF->Read(offset,blen);
}

References m_xrdOssDF.

◆ Read() [2/3]

ssize_t XrdCephOssReadVFile::Read	(	void *	buffer,
		off_t	offset,
		size_t	size )

virtual

Read file bytes into a buffer.

Parameters

buffer	- pointer to buffer where the bytes are to be placed.
offset	- The offset where the read is to start.
size	- The number of bytes to read.

Returns: >= 0 The number of bytes that placed in buffer.; < 0 -errno or -osserr upon failure (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 185 of file XrdCephOssReadVFile.cc.

                                                                       {
  return m_xrdOssDF->Read(buff,offset,blen);
}

References m_xrdOssDF.

◆ Read() [3/3]

int XrdCephOssReadVFile::Read ( XrdSfsAio * aiop )

virtual

Read file bytes using asynchronous I/O.

Parameters

aiop	- Pointer to async I/O object controlling the I/O.

Returns: 0 upon if request started success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 189 of file XrdCephOssReadVFile.cc.

                                             {
  return m_xrdOssDF->Read(aiop);
}

References m_xrdOssDF.

◆ ReadRaw()

ssize_t XrdCephOssReadVFile::ReadRaw	(	void *	buffer,
		off_t	offset,
		size_t	size )

virtual

Read uncompressed file bytes into a buffer.

Parameters

buffer	- pointer to buffer where the bytes are to be placed.
offset	- The offset where the read is to start.
size	- The number of bytes to read.

Returns: >= 0 The number of bytes that placed in buffer.; < 0 -errno or -osserr upon failure (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 193 of file XrdCephOssReadVFile.cc.

                                                                          {
  return m_xrdOssDF->ReadRaw(buff, offset, blen);
}

References m_xrdOssDF.

◆ ReadV()

ssize_t XrdCephOssReadVFile::ReadV	(	XrdOucIOVec *	readV,
		int	rdvcnt )

virtual

Read file bytes as directed by the read vector.

Parameters

readV	pointer to the array of read requests.
rdvcnt	the number of elements in readV.

Returns: >=0 The numbe of bytes read.; < 0 -errno or -osserr upon failure (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 93 of file XrdCephOssReadVFile.cc.

                                                               {
  int fd = m_xrdOssDF->getFileDescriptor();
  LOGCEPH("XrdCephOssReadVFile::ReadV: fd: " << fd  << " " << rnum );
 
  std::stringstream msg_extents; 
  msg_extents << "XrdCephOssReadVFile::Extentslist={\"fd\": " << fd << ", \"EXTENTS\":[";
 
  ExtentHolder extents(rnum);
  for (int i = 0; i < rnum; i++) {
    extents.push_back(Extent(readV[i].offset, readV[i].size));
    msg_extents << "[" << readV[i].offset << "," << readV[i].size << "]," ;
  }
  msg_extents << "]}";
  //XrdCephEroute.Say(msg_extents.str().c_str()); msg_extents.clear();
  if (m_extraLogging) {
    // improve this so no wasted calls if logging is disabled
    LOGCEPH(msg_extents.str());
    msg_extents.clear();
  }
 
  LOGCEPH("XrdCephOssReadVFile::Extents: fd: "<< fd << " "  << extents.size() << " " << extents.len() << " " 
      << extents.begin() << " " << extents.end() << " " << extents.bytesContained() 
      << " " << extents.bytesMissing());
 
  // take the input set of extents and return a vector of merged extents (covering the range to read)
  std::vector<ExtentHolder> mappedExtents = m_readVAdapter->convert(extents);
 
 
  // counter is the iterator to the original readV elements, and is incremented for each chunk that's returned
  int nbytes = 0, curCount = 0, counter(0);
  size_t totalBytesRead(0), totalBytesUseful(0);
 
  // extract the largest range of the extents, and create a buffer.
  size_t buffersize{0};
  for (std::vector<ExtentHolder>::const_iterator ehit = mappedExtents.cbegin(); ehit!= mappedExtents.cend(); ++ehit ) {
    buffersize = std::max(buffersize, ehit->len());
  }
  std::vector<char> buffer;
  buffer.reserve(buffersize);
 
 
  //LOGCEPH("mappedExtents: len: " << mappedExtents.size() );
  for (std::vector<ExtentHolder>::const_iterator ehit = mappedExtents.cbegin(); ehit!= mappedExtents.cend(); ++ehit ) {
    off_t off = ehit->begin();
    size_t len = ehit->len();
 
    //LOGCEPH("outerloop: " << off << " " << len << " " << ehit->end() << " " << " " << ehit->size() );
 
    // read the full extent into the buffer
    long timed_read_ns{0};
    {Timer_ns ts(timed_read_ns); 
      curCount = m_xrdOssDF->Read(buffer.data(), off, len);
    } // timer scope 
    ++m_timer_count;
    auto l = m_timer_longest.load(); 
    m_timer_longest.store(std::max(l,timed_read_ns)); // doesn't quite prevent race conditions
    m_timer_read_ns.fetch_add(timed_read_ns);
    m_timer_size.fetch_add(curCount);
 
    // check that the correct amount of data was read. 
    // std:: clog << "buf Read " << curCount << std::endl;
    if (curCount != (ssize_t)len) {
      return (curCount < 0 ? curCount : -ESPIPE);
    }
    totalBytesRead += curCount;
    totalBytesUseful += ehit->bytesContained(); 
 
 
    // now read out into the original readV requests for each of the held inner extents
    const char* data = buffer.data();
    const ExtentContainer& innerExtents = ehit->extents();
    for (ExtentContainer::const_iterator it = innerExtents.cbegin(); it != innerExtents.cend(); ++it) {
      off_t innerBegin = it->begin() - off;
      off_t innerEnd   = it->end()   - off; 
      //LOGCEPH( "innerloop: " << innerBegin << " " << innerEnd << " " << off << " " 
      //          << it->begin() << " " << it-> end() << " " 
      //          << readV[counter].offset << " " << readV[counter].size);
      std::copy(data+innerBegin, data+innerEnd, readV[counter].data );
      nbytes += it->len();
      ++counter; // next element
    } // inner extents
 
  } // outer extents
  LOGCEPH( "readV returning " << nbytes << " bytes: " << "Read:  " <<totalBytesRead << " Useful: " << totalBytesUseful );
  return nbytes;
 
}

References XrdCephBuffer::ExtentHolder::begin(), XrdCephBuffer::ExtentHolder::bytesContained(), XrdCephBuffer::ExtentHolder::bytesMissing(), XrdCephBuffer::ExtentHolder::end(), XrdOssDF::fd, XrdCephBuffer::ExtentHolder::len(), LOGCEPH, m_extraLogging, m_readVAdapter, m_timer_count, m_timer_longest, m_timer_read_ns, m_timer_size, m_xrdOssDF, XrdOucIOVec::offset, XrdCephBuffer::ExtentHolder::push_back(), XrdCephBuffer::ExtentHolder::size(), XrdOucIOVec::size, and ts().

Here is the call graph for this function:

◆ Write() [1/2]

ssize_t XrdCephOssReadVFile::Write	(	const void *	buffer,
		off_t	offset,
		size_t	size )

virtual

Write file bytes from a buffer.

Parameters

buffer	- pointer to buffer where the bytes reside.
offset	- The offset where the write is to start.
size	- The number of bytes to write.

Returns: >= 0 The number of bytes that were written.; < 0 -errno or -osserr upon failure (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 201 of file XrdCephOssReadVFile.cc.

                                                                              {
  return m_xrdOssDF->Write(buff,offset,blen);
}

References m_xrdOssDF.

◆ Write() [2/2]

int XrdCephOssReadVFile::Write ( XrdSfsAio * aiop )

virtual

Write file bytes using asynchronous I/O.

Parameters

aiop	- Pointer to async I/O object controlling the I/O.

Returns: 0 upon if request started success or -errno or -osserr (see XrdOssError.hh).

Reimplemented from XrdCephOssFile.

Definition at line 205 of file XrdCephOssReadVFile.cc.

                                              {
  return m_xrdOssDF->Write(aiop);
}

References m_xrdOssDF.

Member Data Documentation

◆ m_algname

std::string XrdCephOssReadVFile::m_algname = "passthrough"

protected

Definition at line 80 of file XrdCephOssReadVFile.hh.

Referenced by XrdCephOssReadVFile().

◆ m_cephoss

XrdCephOss* XrdCephOssReadVFile::m_cephoss = nullptr

protected

Definition at line 77 of file XrdCephOssReadVFile.hh.

Referenced by XrdCephOssReadVFile().

◆ m_extraLogging

bool XrdCephOssReadVFile::m_extraLogging = true

protected

Definition at line 79 of file XrdCephOssReadVFile.hh.

Referenced by ReadV().

◆ m_readVAdapter

std::unique_ptr<XrdCephBuffer::IXrdCephReadVAdapter> XrdCephOssReadVFile::m_readVAdapter

protected

Definition at line 81 of file XrdCephOssReadVFile.hh.

Referenced by XrdCephOssReadVFile(), and ReadV().

◆ m_timer_count

std::atomic<long> XrdCephOssReadVFile::m_timer_count {0}

protected

timer for the reads against ceph

Definition at line 84 of file XrdCephOssReadVFile.hh.

84{0};

Referenced by Close(), and ReadV().

◆ m_timer_longest

std::atomic<long> XrdCephOssReadVFile::m_timer_longest {0}

protected

number of reads

Definition at line 86 of file XrdCephOssReadVFile.hh.

86{0};

Referenced by Close(), and ReadV().

◆ m_timer_read_ns

std::atomic<long> XrdCephOssReadVFile::m_timer_read_ns {0}

protected

Definition at line 83 of file XrdCephOssReadVFile.hh.

83{0};

Referenced by Close(), and ReadV().

◆ m_timer_size

std::atomic<long> XrdCephOssReadVFile::m_timer_size {0}

protected

number of reads

Definition at line 85 of file XrdCephOssReadVFile.hh.

85{0};

Referenced by Close(), and ReadV().

◆ m_xrdOssDF

XrdCephOssFile* XrdCephOssReadVFile::m_xrdOssDF = nullptr

protected

Definition at line 78 of file XrdCephOssReadVFile.hh.

Referenced by XrdCephOssReadVFile(), ~XrdCephOssReadVFile(), Close(), Fstat(), Fsync(), Ftruncate(), Open(), Read(), Read(), Read(), ReadRaw(), ReadV(), Write(), and Write().

The documentation for this class was generated from the following files:

Public Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ XrdCephOssReadVFile()

◆ ~XrdCephOssReadVFile()

Member Function Documentation

◆ Close()

◆ Fstat()

◆ Fsync()

◆ Ftruncate()

◆ Open()

◆ Read() [1/3]

◆ Read() [2/3]

◆ Read() [3/3]

◆ ReadRaw()

◆ ReadV()

◆ Write() [1/2]

◆ Write() [2/2]

Member Data Documentation

◆ m_algname

◆ m_cephoss

◆ m_extraLogging

◆ m_readVAdapter

◆ m_timer_count

◆ m_timer_longest

◆ m_timer_read_ns

◆ m_timer_size

◆ m_xrdOssDF