Gaudi::NTuple::DiskBuffer::Writer Class Reference

#include </builds/gaudi/Gaudi/GaudiUtils/include/Gaudi/NTuple/DiskBuffer.h>

Collaboration diagram for Gaudi::NTuple::DiskBuffer::Writer:

Public Member Functions
	Writer (std::filesystem::path path, std::string_view header, int zstdLevel=0, std::size_t blockSize=64 *1024)
	Create (truncating) path and write the preamble and header.
	~Writer ()
	Writer (const Writer &)=delete
Writer &	operator= (const Writer &)=delete
void	beginRecord ()
void	append (const void *data, std::size_t bytes)
void	endRecord ()
void	close ()
	Flush and close; throws on failure. Safe to call twice.
std::uint64_t	entries () const
	Records written to the file; records lost to a failed flush do not count.
std::uint64_t	bytes () const
	Payload bytes of those records.
std::uint64_t	storedBytes () const
	Bytes written to the file after the header.
int	zstdLevel () const
std::size_t	blockSize () const
const std::filesystem::path &	path () const

Private Member Functions
void	flush ()
void	write (const void *data, std::size_t bytes)

Private Attributes
std::filesystem::path	m_path
FileHandle	m_file
int	m_level
std::size_t	m_blockSize
std::vector< char >	m_buf
std::size_t	m_recordStart { 0 }
bool	m_inRecord { false }
std::uint64_t	m_entries { 0 }
std::uint64_t	m_bytes { 0 }
std::uint64_t	m_stored { 0 }
std::uint64_t	m_pending { 0 }
std::uint64_t	m_pendingBytes { 0 }

Detailed Description

Definition at line 61 of file DiskBuffer.h.

Constructor & Destructor Documentation

Writer() [1/2]

Writer::Writer	(	std::filesystem::path	path,
		std::string_view	header,
		int	zstdLevel = 0,
		std::size_t	blockSize = 64 * 1024 )

Create (truncating) path and write the preamble and header.

zstdLevel 0 stores the blocks as they are; anything else needs zstdAvailable(). A block is written once blockSize bytes of records have accumulated, so that is the memory held per open file and what zstd sees at a time.

Definition at line 63 of file NTupleDiskBuffer.cpp.

    : m_path( std::move( path ) ), m_level( zstdLevel ), m_blockSize( blockSize ) {
  if ( header.size() > MAX_HEADER ) throw Error( m_path, "header too large" );
  if ( m_level != 0 && !zstdAvailable() ) throw Error( m_path, "zstd compression is not available in this build" );
  if ( m_blockSize == 0 || m_blockSize > MAX_BLOCK ) throw Error( m_path, "invalid block size" );
  m_file.reset( std::fopen( m_path.c_str(), "wb" ) );
  if ( !m_file ) throw Error( m_path, "cannot create buffer file", errno );
  std::setvbuf( m_file.get(), nullptr, _IONBF, 0 ); // m_buf is the buffer
  std::vector<char> preamble( MAGIC, MAGIC + sizeof( MAGIC ) );
  char              words[4 * sizeof( std::uint32_t )];
  put32( words, BYTE_ORDER_MARK );
  put32( words + 4, VERSION );
  put32( words + 8, m_level ? CODEC_ZSTD : CODEC_NONE );
  put32( words + 12, static_cast<std::uint32_t>( header.size() ) );
  preamble.insert( preamble.end(), words, words + sizeof( words ) );
  preamble.insert( preamble.end(), header.begin(), header.end() );
  write( preamble.data(), preamble.size() );
  m_stored = 0; // count from here
  // m_buf grows with the data: reserving m_blockSize up front costs a full block per tuple whether it fills or not
}

~Writer()

Writer::~Writer ( )

default

Writer() [2/2]

Gaudi::NTuple::DiskBuffer::Writer::Writer ( const Writer & )

delete

Member Function Documentation

append()

void Writer::append	(	const void *	data,
		std::size_t	bytes )

Definition at line 93 of file NTupleDiskBuffer.cpp.

                                                       {
  if ( !m_inRecord ) throw Error( m_path, "append called outside a record" );
  if ( bytes == 0 ) return;
  if ( m_buf.size() - m_recordStart - sizeof( std::uint32_t ) + bytes > MAX_RECORD )
    throw Error( m_path, "record too large" );
  const char* src = static_cast<const char*>( data );
  m_buf.insert( m_buf.end(), src, src + bytes );
}

beginRecord()

void Writer::beginRecord

(

)

Definition at line 86 of file NTupleDiskBuffer.cpp.

                         {
  if ( m_inRecord ) throw Error( m_path, "beginRecord called inside a record" );
  m_recordStart = m_buf.size();
  m_buf.resize( m_buf.size() + sizeof( std::uint32_t ) ); // length, patched by endRecord
  m_inRecord = true;
}

blockSize()

std::size_t Gaudi::NTuple::DiskBuffer::Writer::blockSize ( ) const

inline

Definition at line 86 of file DiskBuffer.h.

{ return m_blockSize; }

bytes()

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::bytes ( ) const

inline

Payload bytes of those records.

Definition at line 82 of file DiskBuffer.h.

{ return m_bytes; }

close()

void Writer::close

(

)

Flush and close; throws on failure. Safe to call twice.

Definition at line 162 of file NTupleDiskBuffer.cpp.

                   {
  if ( !m_file ) return;
  if ( m_inRecord ) throw Error( m_path, "close called inside a record" );
  flush();
  bool ok  = std::fflush( m_file.get() ) == 0;
  int  err = ok ? 0 : errno;
  // the block is the buffer, so a deferred write error (quota, NFS) only shows up here
  if ( std::fclose( m_file.release() ) != 0 && ok ) {
    ok  = false;
    err = errno;
  }
  if ( !ok ) throw Error( m_path, "close failed", err );
}

endRecord()

void Writer::endRecord

(

)

Definition at line 102 of file NTupleDiskBuffer.cpp.

                       {
  if ( !m_inRecord ) throw Error( m_path, "endRecord called outside a record" );
  const std::size_t payload = m_buf.size() - m_recordStart - sizeof( std::uint32_t );
  put32( m_buf.data() + m_recordStart, static_cast<std::uint32_t>( payload ) );
  m_inRecord = false;
  ++m_pending;
  m_pendingBytes += payload;
  if ( m_buf.size() >= m_blockSize ) flush();
}

entries()

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::entries ( ) const

inline

Records written to the file; records lost to a failed flush do not count.

Definition at line 80 of file DiskBuffer.h.

{ return m_entries; }

flush()

void Writer::flush ( )

private

Definition at line 118 of file NTupleDiskBuffer.cpp.

                   {
  if ( !m_file ) throw Error( m_path, "buffer file is closed" );
  if ( m_buf.empty() ) return;
  // dropped on failure too: the data is lost either way, do not let the buffer grow
  auto drop = [this] {
    m_buf.clear();
    m_pending = m_pendingBytes = 0;
    // a single huge record should not pin its memory for the rest of the job
    if ( m_buf.capacity() > 4 * m_blockSize ) m_buf.shrink_to_fit();
  };
  const long          start       = std::ftell( m_file.get() );
  const std::uint64_t storedStart = m_stored;
  try {
    const char*       data   = m_buf.data();
    std::size_t       stored = m_buf.size();
    std::vector<char> out;
#ifdef GAUDI_USE_ZSTD
    if ( m_level != 0 ) {
      out.resize( ZSTD_compressBound( m_buf.size() ) );
      stored = ZSTD_compress( out.data(), out.size(), m_buf.data(), m_buf.size(), m_level );
      if ( ZSTD_isError( stored ) )
        throw Error( m_path, std::string( "zstd compression failed: " ) + ZSTD_getErrorName( stored ) );
      data = out.data();
    }
#endif
    char hdr[BLOCK_HEADER];
    put32( hdr, static_cast<std::uint32_t>( stored ) );
    put32( hdr + 4, static_cast<std::uint32_t>( m_buf.size() ) );
    write( hdr, sizeof( hdr ) );
    write( data, stored );
  } catch ( ... ) {
    // cut the half-written block off, so the blocks before it stay readable
    if ( start >= 0 && ::ftruncate( ::fileno( m_file.get() ), start ) == 0 ) {
      std::fseek( m_file.get(), 0, SEEK_END );
      m_stored = storedStart;
    }
    drop();
    throw;
  }
  m_entries += m_pending;
  m_bytes += m_pendingBytes;
  drop();
}

operator=()

Writer & Gaudi::NTuple::DiskBuffer::Writer::operator= ( const Writer & )

delete

path()

const std::filesystem::path & Gaudi::NTuple::DiskBuffer::Writer::path ( ) const

inline

Definition at line 87 of file DiskBuffer.h.

{ return m_path; }

storedBytes()

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::storedBytes ( ) const

inline

Bytes written to the file after the header.

Definition at line 84 of file DiskBuffer.h.

{ return m_stored; }

write()

void Writer::write ( const void * data, std::size_t bytes )

private

Definition at line 112 of file NTupleDiskBuffer.cpp.

                                                      {
  if ( std::fwrite( data, 1, bytes, m_file.get() ) != bytes ) throw Error( m_path, "write failed", errno );
  m_stored += bytes;
}

zstdLevel()

int Gaudi::NTuple::DiskBuffer::Writer::zstdLevel ( ) const

inline

Definition at line 85 of file DiskBuffer.h.

{ return m_level; }

Member Data Documentation

m_blockSize

std::size_t Gaudi::NTuple::DiskBuffer::Writer::m_blockSize

private

Definition at line 96 of file DiskBuffer.h.

m_buf

std::vector<char> Gaudi::NTuple::DiskBuffer::Writer::m_buf

private

Definition at line 97 of file DiskBuffer.h.

m_bytes

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::m_bytes { 0 }

private

Definition at line 101 of file DiskBuffer.h.

{ 0 };

m_entries

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::m_entries { 0 }

private

Definition at line 100 of file DiskBuffer.h.

{ 0 };

m_file

FileHandle Gaudi::NTuple::DiskBuffer::Writer::m_file

private

Definition at line 94 of file DiskBuffer.h.

m_inRecord

bool Gaudi::NTuple::DiskBuffer::Writer::m_inRecord { false }

private

Definition at line 99 of file DiskBuffer.h.

{ false };

m_level

int Gaudi::NTuple::DiskBuffer::Writer::m_level

private

Definition at line 95 of file DiskBuffer.h.

m_path

std::filesystem::path Gaudi::NTuple::DiskBuffer::Writer::m_path

private

Definition at line 93 of file DiskBuffer.h.

m_pending

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::m_pending { 0 }

private

Definition at line 103 of file DiskBuffer.h.

{ 0 };      // records in m_buf, counted once their block is written

m_pendingBytes

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::m_pendingBytes { 0 }

private

Definition at line 104 of file DiskBuffer.h.

{ 0 }; // and their payload

m_recordStart

std::size_t Gaudi::NTuple::DiskBuffer::Writer::m_recordStart { 0 }

private

Definition at line 98 of file DiskBuffer.h.

{ 0 }; // offset of the current record's length field

m_stored

std::uint64_t Gaudi::NTuple::DiskBuffer::Writer::m_stored { 0 }

private

Definition at line 102 of file DiskBuffer.h.

{ 0 };

---

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

• GaudiUtils/include/Gaudi/NTuple/DiskBuffer.h
• GaudiUtils/src/Lib/NTupleDiskBuffer.cpp