#include <XrdCephOssBufferedFile.hh>

Inheritance diagram for XrdCephOssBufferedFile:

Collaboration diagram for XrdCephOssBufferedFile:

Public Member Functions
	XrdCephOssBufferedFile (XrdCephOss cephoss, XrdCephOssFile cephossDF, size_t buffersize, const std::string &bufferIOmode, size_t maxNumberSimulBuffers)
virtual	~XrdCephOssBufferedFile ()
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 Member Functions
std::unique_ptr< XrdCephBuffer::IXrdCephBufferAlg >	createBuffer ()

Protected Attributes
std::mutex	m_buf_mutex
std::unique_ptr< XrdCephBuffer::IXrdCephBufferAlg >	m_bufferAlg
std::string	m_bufferIOmode
std::map< size_t, std::unique_ptr< XrdCephBuffer::IXrdCephBufferAlg > >	m_bufferReadAlgs
size_t	m_bufsize = 1610241024L
std::atomic< size_t >	m_bytesRead = {0}
std::atomic< size_t >	m_bytesReadAIO = {0}
	number of bytes read or written
std::atomic< size_t >	m_bytesReadV = {0}
	number of bytes read or written
std::atomic< size_t >	m_bytesWrite = {0}
	number of bytes read or written
std::atomic< size_t >	m_bytesWriteAIO = {0}
	number of bytes read or written
XrdCephOss *	m_cephoss = nullptr
	create a new instance of the buffer
int	m_flags = 0
	number of ms to sleep if a retry is requested
int	m_maxBufferRetries {5}
	set the maximum of buffers to open on a single instance (e.g. for simultaneous file reads)
int	m_maxBufferRetrySleepTime_ms
	How many times to retry a ready from a buffer with EBUSY errors.
size_t	m_maxCountReadBuffers {10}
	any data access method on the buffer will use this
std::string	m_path
std::chrono::time_point< std::chrono::system_clock >	m_timestart
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 XrdCephOssBufferedFile designed to wrap XrdCephOssFile Functionality for buffered access to/from data in Ceph to avoid inefficient small reads / writes from the client side

Definition at line 48 of file XrdCephOssBufferedFile.hh.

Constructor & Destructor Documentation

◆ XrdCephOssBufferedFile()

XrdCephOssBufferedFile::XrdCephOssBufferedFile	(	XrdCephOss *	cephoss,
		XrdCephOssFile *	cephossDF,
		size_t	buffersize,
		const std::string &	bufferIOmode,
		size_t	maxNumberSimulBuffers )

Definition at line 58 of file XrdCephOssBufferedFile.cc.

                                                                             :
                  XrdCephOssFile(cephoss), m_cephoss(cephoss), m_xrdOssDF(cephossDF), 
                  m_maxCountReadBuffers(maxNumberSimulBuffers),
                  m_maxBufferRetrySleepTime_ms(1000), 
                  m_bufsize(buffersize),
                  m_bufferIOmode(bufferIOmode)     
{
 
}

References XrdCephOssFile::XrdCephOssFile(), m_bufferIOmode, m_bufsize, m_cephoss, m_maxBufferRetrySleepTime_ms, m_maxCountReadBuffers, and m_xrdOssDF.

Here is the call graph for this function:

◆ ~XrdCephOssBufferedFile()

XrdCephOssBufferedFile::~XrdCephOssBufferedFile ( )

virtual

Definition at line 70 of file XrdCephOssBufferedFile.cc.

                                                {
    // XrdCephEroute.Say("XrdCephOssBufferedFile::Destructor");
 
  // remember to delete the inner XrdCephOssFile object
  if (m_xrdOssDF) {
    delete m_xrdOssDF;
    m_xrdOssDF = nullptr;
  }
 
}

References m_xrdOssDF.

Member Function Documentation

◆ Close()

int XrdCephOssBufferedFile::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 101 of file XrdCephOssBufferedFile.cc.

                                                  {
  // if data is still in the buffer and we are writing, make sure to write it
  if (m_bufferAlg && ((m_flags & O_ACCMODE) != O_RDONLY)) {
    ssize_t rc = m_bufferAlg->flushWriteCache();
    if (rc < 0) {
        LOGCEPH( "XrdCephOssBufferedFile::Close: flush Error fd: " << m_fd << " rc:" << rc );
        // still try to close the file
        ssize_t rc2 = m_xrdOssDF->Close(retsz);
        if (rc2 < 0) {
          LOGCEPH( "XrdCephOssBufferedFile::Close: Close error after flush Error fd: " << m_fd << " rc:" << rc2 );
        }
        return rc; // return the original flush error
    } else {
      LOGCEPH( "XrdCephOssBufferedFile::Close: Flushed data on close fd: " << m_fd << " rc:" << rc );
    }
  } // check for write
  const std::chrono::time_point<std::chrono::system_clock> now =
         std::chrono::system_clock::now();
  const std::time_t t_s = std::chrono::system_clock::to_time_t(m_timestart);
  const std::time_t t_c = std::chrono::system_clock::to_time_t(now);
 
  auto t_dur = std::chrono::duration_cast<std::chrono::milliseconds>(now - m_timestart).count();
 
  LOGCEPH("XrdCephOssBufferedFile::Summary: {\"fd\":" << m_fd << ", \"Elapsed_time_ms\":" << t_dur 
          << ", \"path\":\"" << m_path  
          << "\", read_B:"   << m_bytesRead.load() 
          << ", readV_B:"     << m_bytesReadV.load() 
          << ", readAIO_B:"   << m_bytesReadAIO.load() 
          << ", writeB:"     << m_bytesWrite.load()
          << ", writeAIO_B:" << m_bytesWriteAIO.load()
          << ", startTime:\"" << std::put_time(std::localtime(&t_s), "%F %T") << "\", endTime:\"" 
          << std::put_time(std::localtime(&t_c), "%F %T") << "\""
          << ", nBuffersRead:" << m_bufferReadAlgs.size()
          << "}"); 
 
  return m_xrdOssDF->Close(retsz);
}

References LOGCEPH, m_bufferAlg, m_bufferReadAlgs, m_bytesRead, m_bytesReadAIO, m_bytesReadV, m_bytesWrite, m_bytesWriteAIO, XrdCephOssFile::m_fd, m_flags, m_path, m_timestart, and m_xrdOssDF.

◆ createBuffer()

std::unique_ptr< XrdCephBuffer::IXrdCephBufferAlg > XrdCephOssBufferedFile::createBuffer ( )

protected

Definition at line 313 of file XrdCephOssBufferedFile.cc.

                                                                                   {
    std::unique_ptr<IXrdCephBufferAlg> bufferAlg;
 
    size_t bufferSize {m_bufsize};  // create buffer of default size
    if (m_bufferReadAlgs.size() >= m_maxCountReadBuffers) {
      BUFLOG("XrdCephOssBufferedFile: buffer reached max number of simul-buffers for this file: creating only 1MiB buffer" );
      bufferSize = 1048576;
    } else {
      BUFLOG("XrdCephOssBufferedFile: buffer: got " << m_bufferReadAlgs.size() << " buffers already");
    }
 
    try {
      std::unique_ptr<IXrdCephBufferData> cephbuffer = std::unique_ptr<IXrdCephBufferData>(new XrdCephBufferDataSimple(bufferSize));
      std::unique_ptr<ICephIOAdapter>     cephio;
      if (m_bufferIOmode == "aio") {
          cephio = std::unique_ptr<ICephIOAdapter>(new CephIOAdapterAIORaw(cephbuffer.get(),m_fd));
      } else if (m_bufferIOmode == "io") {
          cephio = std::unique_ptr<ICephIOAdapter>(new CephIOAdapterRaw(cephbuffer.get(),m_fd,
                                                  !m_cephoss->m_useDefaultPreadAlg));
      } else {
            BUFLOG("XrdCephOssBufferedFile: buffer mode needs to be one of aio|io " );
            m_xrdOssDF->Close();
            return bufferAlg; // invalid instance;
      }
 
      LOGCEPH( "XrdCephOssBufferedFile::Open: fd: " << m_fd <<  " Buffer created: " << cephbuffer->capacity() );
      bufferAlg = std::unique_ptr<IXrdCephBufferAlg>(new XrdCephBufferAlgSimple(std::move(cephbuffer),std::move(cephio),m_fd) );
    } catch (const std::bad_alloc &e) {
      BUFLOG("XrdCephOssBufferedFile: Bad memory allocation in buffer: " << e.what() );
    }
 
    return bufferAlg;
  }

References BUFLOG, LOGCEPH, m_bufferIOmode, m_bufferReadAlgs, m_bufsize, m_cephoss, XrdCephOssFile::m_fd, m_maxCountReadBuffers, and m_xrdOssDF.

Referenced by Read(), Read(), Write(), and Write().

Here is the caller graph for this function:

◆ Fstat()

int XrdCephOssBufferedFile::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 240 of file XrdCephOssBufferedFile.cc.

                                                   {
  return m_xrdOssDF->Fstat(buff);
}

References m_xrdOssDF, and stat.

◆ Fsync()

int XrdCephOssBufferedFile::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 304 of file XrdCephOssBufferedFile.cc.

                                  {
  return m_xrdOssDF->Fsync();
}

References m_xrdOssDF.

◆ Ftruncate()

int XrdCephOssBufferedFile::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 308 of file XrdCephOssBufferedFile.cc.

                                                            {
  return m_xrdOssDF->Ftruncate(len);
}

References m_xrdOssDF.

◆ Open()

int XrdCephOssBufferedFile::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 82 of file XrdCephOssBufferedFile.cc.

                                                                                         {
 
  int rc = m_xrdOssDF->Open(path, flags, mode, env);
  if (rc < 0) {
    return rc;
  }
  m_fd = m_xrdOssDF->getFileDescriptor();
  BUFLOG("XrdCephOssBufferedFile::Open got fd: " << m_fd << " " << path);
  m_flags = flags; // e.g. for write/read knowledge
  m_path  = path; // good to keep the path for final stats presentation
 
 
  // start the timer
  //m_timestart = std::chrono::steady_clock::now();
  m_timestart = std::chrono::system_clock::now();
  // return the file descriptor 
  return rc;
}

References BUFLOG, XrdCephOssFile::m_fd, m_flags, m_path, m_timestart, and m_xrdOssDF.

◆ Read() [1/3]

ssize_t XrdCephOssBufferedFile::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 147 of file XrdCephOssBufferedFile.cc.

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

References m_xrdOssDF.

◆ Read() [2/3]

ssize_t XrdCephOssBufferedFile::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 151 of file XrdCephOssBufferedFile.cc.

                                                                          {
  size_t thread_id = std::hash<std::thread::id>{}(std::this_thread::get_id());
 
  IXrdCephBufferAlg * buffer{nullptr};
  // check for, and create if needed, a buffer
  {
    // lock in case need to create a new algorithm instance
    const std::lock_guard<std::mutex> lock(m_buf_mutex);
    auto buffer_itr = m_bufferReadAlgs.find(thread_id);
    if (buffer_itr == m_bufferReadAlgs.end()) {
      // only create a buffer, if we haven't hit the max buffers yet
      auto buffer_ptr = createBuffer();
      if (buffer_ptr) {
        buffer = buffer_ptr.get();
        m_bufferReadAlgs[thread_id] = std::move(buffer_ptr);
      } else {
        // if we can't create a buffer, we just have to pass through the read ... 
        ssize_t rc = m_xrdOssDF->Read(buff, offset, blen);
        if (rc >= 0) {
          LOGCEPH( "XrdCephOssBufferedFile::Read buffers and read failed with rc: " << rc );
        }
        return rc;
      }
    } else {
      buffer = buffer_itr->second.get();
    }
  } // scope of lock
  
  int retry_counter{m_maxBufferRetries};
  ssize_t rc {0};
  while (retry_counter > 0) {
    rc = buffer->read(buff, offset, blen);
    if (rc != -EBUSY) break; // either worked, or is a real non busy error
    LOGCEPH( "XrdCephOssBufferedFile::Read Received EBUSY for fd: " << m_fd << " on try: " << (m_maxBufferRetries-retry_counter) << ". Sleeping .. "
              << " rc:" << rc  << " off:" << offset << " len:" << blen);
    std::this_thread::sleep_for(m_maxBufferRetrySleepTime_ms * 1ms);
    --retry_counter;
  }
  if (retry_counter == 0) {
    // reach maximum attempts for ebusy retry; fail the job
    LOGCEPH( "XrdCephOssBufferedFile::Read Max attempts for fd: " << m_fd << " on try: " << (m_maxBufferRetries-retry_counter) << ". Terminating with -EIO: "
              << " rc:" << rc  << " off:" << offset << " len:" << blen );
    // set a permanent error code:
    rc = -EIO;    
  }
  if (rc >=0) {
    m_bytesRead.fetch_add(rc);
  } else {
    LOGCEPH( "XrdCephOssBufferedFile::Read: Read error  fd: " << m_fd << " rc:" << rc  << " off:" << offset << " len:" << blen);
  }
  // LOGCEPH( "XrdCephOssBufferedFile::Read: Read good  fd: " << m_fd << " rc:" << rc  << " off:" << offset << " len:" << blen);
  return rc;
}

References createBuffer(), LOGCEPH, m_buf_mutex, m_bufferReadAlgs, m_bytesRead, XrdCephOssFile::m_fd, m_maxBufferRetries, m_maxBufferRetrySleepTime_ms, m_xrdOssDF, and XrdCephBuffer::IXrdCephBufferAlg::read().

Here is the call graph for this function:

◆ Read() [3/3]

int XrdCephOssBufferedFile::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 205 of file XrdCephOssBufferedFile.cc.

                                                {
  size_t thread_id = std::hash<std::thread::id>{}(std::this_thread::get_id());
  IXrdCephBufferAlg * buffer{nullptr};
  // check for, and create if needed, a buffer
  {
    // lock in case need to create a new algorithm instance
    const std::lock_guard<std::mutex> lock(m_buf_mutex);
    auto buffer_itr = m_bufferReadAlgs.find(thread_id);
    if (buffer_itr == m_bufferReadAlgs.end()) {
      m_bufferReadAlgs[thread_id] = createBuffer();
      buffer = m_bufferReadAlgs.find(thread_id)->second.get();
    } else {
      buffer = buffer_itr->second.get();
    }
  }
 
  // LOGCEPH("XrdCephOssBufferedFile::AIOREAD: fd: " << m_xrdOssDF->getFileDescriptor() << "  "  << time(nullptr) << " : " 
  //         << aiop->sfsAio.aio_offset << " " 
  //         << aiop->sfsAio.aio_nbytes << " " << aiop->sfsAio.aio_reqprio << " "
  //         << aiop->sfsAio.aio_fildes );
  ssize_t rc = buffer->read_aio(aiop);
  if (rc > 0) {
    m_bytesReadAIO.fetch_add(rc);
  }  else {
    LOGCEPH( "XrdCephOssBufferedFile::Read: ReadAIO error  fd: " << m_fd << " rc:" << rc  
            << " off:" << aiop->sfsAio.aio_offset << " len:" << aiop->sfsAio.aio_nbytes );
  }
  return rc;
}

References aiocb::aio_nbytes, aiocb::aio_offset, createBuffer(), LOGCEPH, m_buf_mutex, m_bufferReadAlgs, m_bytesReadAIO, XrdCephOssFile::m_fd, XrdCephBuffer::IXrdCephBufferAlg::read_aio(), and XrdSfsAio::sfsAio.

Here is the call graph for this function:

◆ ReadRaw()

ssize_t XrdCephOssBufferedFile::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 235 of file XrdCephOssBufferedFile.cc.

                                                                             {
  // #TODO; ReadRaw should bypass the buffer ?
  return m_xrdOssDF->ReadRaw(buff, offset, blen);
}

References m_xrdOssDF.

◆ ReadV()

ssize_t XrdCephOssBufferedFile::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 140 of file XrdCephOssBufferedFile.cc.

                                                                  {
  // don't touch readV in the buffering method
  ssize_t rc = m_xrdOssDF->ReadV(readV,rnum);
  if (rc > 0) m_bytesReadV.fetch_add(rc);
  return rc;
}

References m_bytesReadV, and m_xrdOssDF.

◆ Write() [1/2]

ssize_t XrdCephOssBufferedFile::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 244 of file XrdCephOssBufferedFile.cc.

                                                                                 {
 
  if (!m_bufferAlg) {
    m_bufferAlg = createBuffer();
    if (!m_bufferAlg) {
      LOGCEPH( "XrdCephOssBufferedFile: Error in creating buffered object");
      return -EINVAL;
    }
  }
 
 
  int retry_counter{m_maxBufferRetries};
  ssize_t rc {0};
  while (retry_counter > 0) {
    rc = m_bufferAlg->write(buff, offset, blen);
    if (rc != -EBUSY) break; // either worked, or is a real non busy error
    LOGCEPH( "XrdCephOssBufferedFile::Write Received EBUSY for fd: " << m_fd << " on try: " << (m_maxBufferRetries-retry_counter) << ". Sleeping .. "
              << " rc:" << rc  << " off:" << offset << " len:" << blen);
    std::this_thread::sleep_for(m_maxBufferRetrySleepTime_ms * 1ms);
    --retry_counter;
  }
  if (retry_counter == 0) {
    // reach maximum attempts for ebusy retry; fail the job
    LOGCEPH( "XrdCephOssBufferedFile::Write Max attempts for fd: " << m_fd << " on try: " << (m_maxBufferRetries-retry_counter) << ". Terminating with -EIO: "
              << " rc:" << rc  << " off:" << offset << " len:" << blen );
    // set a permanent error code:
    rc = -EIO;    
  }
  if (rc >=0) {
    m_bytesWrite.fetch_add(rc);
  }  else {
    LOGCEPH( "XrdCephOssBufferedFile::Write: Write error  fd: " << m_fd << " rc:" << rc  << " off:" << offset << " len:" << blen);
  }
  return rc;
}

References createBuffer(), LOGCEPH, m_bufferAlg, m_bytesWrite, XrdCephOssFile::m_fd, m_maxBufferRetries, and m_maxBufferRetrySleepTime_ms.

Here is the call graph for this function:

◆ Write() [2/2]

int XrdCephOssBufferedFile::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 280 of file XrdCephOssBufferedFile.cc.

                                                 {
  if (!m_bufferAlg) {
    m_bufferAlg = createBuffer();
    if (!m_bufferAlg) {
      LOGCEPH( "XrdCephOssBufferedFile: Error in creating buffered object");
      return -EINVAL;
    }
  }
 
  // LOGCEPH("XrdCephOssBufferedFile::AIOWRITE: fd: " << m_xrdOssDF->getFileDescriptor() << "  "   << time(nullptr) << " : " 
  //         << aiop->sfsAio.aio_offset << " " 
  //         << aiop->sfsAio.aio_nbytes << " " << aiop->sfsAio.aio_reqprio << " "
  //         << aiop->sfsAio.aio_fildes << " " );
  ssize_t rc = m_bufferAlg->write_aio(aiop);
  if (rc > 0) {
     m_bytesWriteAIO.fetch_add(rc);
  }   else {
    LOGCEPH( "XrdCephOssBufferedFile::Write: WriteAIO error  fd: " << m_fd << " rc:" << rc  
            << " off:" << aiop->sfsAio.aio_offset << " len:" << aiop->sfsAio.aio_nbytes );
  }
  return rc;
 
}

References aiocb::aio_nbytes, aiocb::aio_offset, createBuffer(), LOGCEPH, m_bufferAlg, m_bytesWriteAIO, XrdCephOssFile::m_fd, and XrdSfsAio::sfsAio.

Here is the call graph for this function:

Member Data Documentation

◆ m_buf_mutex

std::mutex XrdCephOssBufferedFile::m_buf_mutex

protected

Definition at line 76 of file XrdCephOssBufferedFile.hh.

Referenced by Read(), and Read().

◆ m_bufferAlg

std::unique_ptr<XrdCephBuffer::IXrdCephBufferAlg> XrdCephOssBufferedFile::m_bufferAlg

protected

Definition at line 74 of file XrdCephOssBufferedFile.hh.

Referenced by Close(), Write(), and Write().

◆ m_bufferIOmode

std::string XrdCephOssBufferedFile::m_bufferIOmode

protected

Definition at line 85 of file XrdCephOssBufferedFile.hh.

Referenced by XrdCephOssBufferedFile(), and createBuffer().

◆ m_bufferReadAlgs

std::map<size_t, std::unique_ptr<XrdCephBuffer::IXrdCephBufferAlg> > XrdCephOssBufferedFile::m_bufferReadAlgs

protected

Definition at line 75 of file XrdCephOssBufferedFile.hh.

Referenced by Close(), createBuffer(), Read(), and Read().

◆ m_bufsize

size_t XrdCephOssBufferedFile::m_bufsize = 16*1024*1024L

protected

Definition at line 84 of file XrdCephOssBufferedFile.hh.

Referenced by XrdCephOssBufferedFile(), and createBuffer().

◆ m_bytesRead

std::atomic<size_t> XrdCephOssBufferedFile::m_bytesRead = {0}

protected

Definition at line 88 of file XrdCephOssBufferedFile.hh.

88{0};

Referenced by Close(), and Read().

◆ m_bytesReadAIO

std::atomic<size_t> XrdCephOssBufferedFile::m_bytesReadAIO = {0}

protected

number of bytes read or written

Definition at line 90 of file XrdCephOssBufferedFile.hh.

90{0};

Referenced by Close(), and Read().

◆ m_bytesReadV

std::atomic<size_t> XrdCephOssBufferedFile::m_bytesReadV = {0}

protected

number of bytes read or written

Definition at line 89 of file XrdCephOssBufferedFile.hh.

89{0};

Referenced by Close(), and ReadV().

◆ m_bytesWrite

std::atomic<size_t> XrdCephOssBufferedFile::m_bytesWrite = {0}

protected

number of bytes read or written

Definition at line 91 of file XrdCephOssBufferedFile.hh.

91{0};

Referenced by Close(), and Write().

◆ m_bytesWriteAIO

std::atomic<size_t> XrdCephOssBufferedFile::m_bytesWriteAIO = {0}

protected

number of bytes read or written

Definition at line 92 of file XrdCephOssBufferedFile.hh.

92{0};

Referenced by Close(), and Write().

◆ m_cephoss

XrdCephOss* XrdCephOssBufferedFile::m_cephoss = nullptr

protected

create a new instance of the buffer

Definition at line 72 of file XrdCephOssBufferedFile.hh.

Referenced by XrdCephOssBufferedFile(), and createBuffer().

◆ m_flags

int XrdCephOssBufferedFile::m_flags = 0

protected

number of ms to sleep if a retry is requested

Definition at line 83 of file XrdCephOssBufferedFile.hh.

Referenced by Close(), and Open().

◆ m_maxBufferRetries

int XrdCephOssBufferedFile::m_maxBufferRetries {5}

protected

set the maximum of buffers to open on a single instance (e.g. for simultaneous file reads)

Definition at line 80 of file XrdCephOssBufferedFile.hh.

80{5};

Referenced by Read(), and Write().

◆ m_maxBufferRetrySleepTime_ms

int XrdCephOssBufferedFile::m_maxBufferRetrySleepTime_ms

protected

How many times to retry a ready from a buffer with EBUSY errors.

Definition at line 81 of file XrdCephOssBufferedFile.hh.

Referenced by XrdCephOssBufferedFile(), Read(), and Write().

◆ m_maxCountReadBuffers

size_t XrdCephOssBufferedFile::m_maxCountReadBuffers {10}

protected

any data access method on the buffer will use this

Definition at line 77 of file XrdCephOssBufferedFile.hh.

77{10};

Referenced by XrdCephOssBufferedFile(), and createBuffer().

◆ m_path

std::string XrdCephOssBufferedFile::m_path

protected

Definition at line 86 of file XrdCephOssBufferedFile.hh.

Referenced by Close(), and Open().

◆ m_timestart

std::chrono::time_point<std::chrono::system_clock> XrdCephOssBufferedFile::m_timestart

protected

Definition at line 87 of file XrdCephOssBufferedFile.hh.

Referenced by Close(), and Open().

◆ m_xrdOssDF

XrdCephOssFile* XrdCephOssBufferedFile::m_xrdOssDF = nullptr

protected

Definition at line 73 of file XrdCephOssBufferedFile.hh.

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

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

	XRootD

Public Member Functions

Protected Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ XrdCephOssBufferedFile()

◆ ~XrdCephOssBufferedFile()

Member Function Documentation

◆ Close()

◆ createBuffer()

◆ 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_buf_mutex

◆ m_bufferAlg

◆ m_bufferIOmode

◆ m_bufferReadAlgs

◆ m_bufsize

◆ m_bytesRead

◆ m_bytesReadAIO

◆ m_bytesReadV

◆ m_bytesWrite

◆ m_bytesWriteAIO

◆ m_cephoss

◆ m_flags

◆ m_maxBufferRetries

◆ m_maxBufferRetrySleepTime_ms

◆ m_maxCountReadBuffers

◆ m_path

◆ m_timestart

◆ m_xrdOssDF