File_Namespace::FileBuffer Class Reference

Represents/provides access to contiguous data stored in the file system. More...

#include <FileBuffer.h>

Inheritance diagram for File_Namespace::FileBuffer:

Collaboration diagram for File_Namespace::FileBuffer:

Public Member Functions
	FileBuffer (FileMgr *fm, const size_t pageSize, const ChunkKey &chunkKey, const size_t initialSize=0)
	Constructs a FileBuffer object. More...
	FileBuffer (FileMgr *fm, const size_t pageSize, const ChunkKey &chunkKey, const SQLTypeInfo sqlType, const size_t initialSize=0)
	FileBuffer (FileMgr *fm, const ChunkKey &chunkKey, const std::vector< HeaderInfo >::const_iterator &headerStartIt, const std::vector< HeaderInfo >::const_iterator &headerEndIt)
	~FileBuffer () override
	Destructor. More...
Page	addNewMultiPage (const int32_t epoch)
void	reserve (const size_t numBytes) override
size_t	freeMetadataPages ()
size_t	freeChunkPages ()
size_t	freePages ()
void	freePagesBeforeEpoch (const int32_t targetEpoch)
void	read (int8_t *const dst, const size_t numBytes=0, const size_t offset=0, const MemoryLevel dstMemoryLevel=CPU_LEVEL, const int32_t deviceId=-1) override
void	write (int8_t *src, const size_t numBytes, const size_t offset=0, const MemoryLevel srcMemoryLevel=CPU_LEVEL, const int32_t deviceId=-1) override
	Writes the contents of source (src) into new versions of the affected logical pages. More...
void	append (int8_t *src, const size_t numBytes, const MemoryLevel srcMemoryLevel=CPU_LEVEL, const int32_t deviceId=-1) override
void	copyPage (Page &srcPage, Page &destPage, const size_t numBytes, const size_t offset=0)
Data_Namespace::MemoryLevel	getType () const override
int8_t *	getMemoryPtr () override
	Not implemented for FileMgr – throws a runtime_error. More...
size_t	pageCount () const override
	Returns the number of pages in the FileBuffer. More...
bool	hasDataPages () const
size_t	pageSize () const override
	Returns the size in bytes of each page in the FileBuffer. More...
virtual size_t	pageDataSize () const
	Returns the size in bytes of the data portion of each page in the FileBuffer. More...
virtual size_t	reservedHeaderSize () const
virtual std::vector< MultiPage >	getMultiPage () const
	Returns vector of MultiPages in the FileBuffer. More...
MultiPage	getMetadataPage () const
size_t	reservedSize () const override
	Returns the total number of bytes allocated for the FileBuffer. More...
size_t	numMetadataPages () const
	Returns the total number of used bytes in the FileBuffer. More...
bool	isMissingPages () const
size_t	numChunkPages () const
std::string	dump () const
void	freePage (const Page &page)
Public Member Functions inherited from Data_Namespace::AbstractBuffer
	AbstractBuffer (const int device_id)
	AbstractBuffer (const int device_id, const SQLTypeInfo sql_type)
virtual	~AbstractBuffer ()
virtual void	read (int8_t *const dst, const size_t num_bytes, const size_t offset=0, const MemoryLevel dst_buffer_type=CPU_LEVEL, const int dst_device_id=-1)=0
virtual void	write (int8_t *src, const size_t num_bytes, const size_t offset=0, const MemoryLevel src_buffer_type=CPU_LEVEL, const int src_device_id=-1)=0
virtual void	append (int8_t *src, const size_t num_bytes, const MemoryLevel src_buffer_type=CPU_LEVEL, const int device_id=-1)=0
virtual void	setMemoryPtr (int8_t *new_ptr)
virtual int	pin ()
virtual int	unPin ()
virtual int	getPinCount ()
size_t	size () const
int	getDeviceId () const
bool	isDirty () const
bool	isAppended () const
bool	isUpdated () const
bool	hasEncoder () const
SQLTypeInfo	getSqlType () const
void	setSqlType (const SQLTypeInfo &sql_type)
Encoder *	getEncoder () const
void	setDirty ()
void	setUpdated ()
void	setAppended ()
void	setSize (const size_t size)
void	clearDirtyBits ()
void	initEncoder (const SQLTypeInfo &tmp_sql_type)
void	syncEncoder (const AbstractBuffer *src_buffer)
void	copyTo (AbstractBuffer *destination_buffer, const size_t num_bytes=0)
void	resetToEmpty ()

Static Public Attributes
static constexpr size_t	headerBufferOffset_ = 32

Private Member Functions
void	writeHeader (Page &page, const int32_t pageId, const int32_t epoch, const bool writeMetadata=false)
	Write header writes header at top of page in format. More...
void	writeMetadata (const int32_t epoch)
void	readMetadata (const Page &page)
void	calcHeaderBuffer ()
void	freePage (const Page &page, const bool isRolloff)
void	freePagesBeforeEpochForMultiPage (MultiPage &multiPage, const int32_t targetEpoch, const int32_t currentEpoch)
void	initMetadataAndPageDataSize ()
int32_t	getFileMgrEpoch ()

Private Attributes
FileMgr *	fm_
const size_t	metadataPageSize_
MultiPage	metadataPages_
std::vector< MultiPage >	multiPages_
size_t	pageSize_
size_t	pageDataSize_
size_t	reservedHeaderSize_
ChunkKey	chunkKey_

Friends
class	FileMgr
class	CachingFileMgr

Additional Inherited Members
Protected Attributes inherited from Data_Namespace::AbstractBuffer
std::unique_ptr< Encoder >	encoder_
SQLTypeInfo	sql_type_
size_t	size_
int	device_id_

Detailed Description

Represents/provides access to contiguous data stored in the file system.

The FileBuffer consists of logical pages, which can map to any identically-sized page in any file of the underlying file system. A page's metadata (file and page number) are stored in MultiPage objects, and each MultiPage includes page metadata for multiple versions of the same page.

Note that a "Chunk" is brought into a FileBuffer by the FileMgr.

Note(s): Forbid Copying Idiom 4.1

Definition at line 57 of file FileBuffer.h.

Constructor & Destructor Documentation

File_Namespace::FileBuffer::FileBuffer	(	FileMgr *	fm,
		const size_t	pageSize,
		const ChunkKey &	chunkKey,
		const size_t	initialSize = 0
	)

Constructs a FileBuffer object.

Definition at line 37 of file FileBuffer.cpp.

References calcHeaderBuffer(), CHECK, CHECK_GT, fm_, pageDataSize_, pageSize_, and reservedHeaderSize_.

    : AbstractBuffer(fm->getDeviceId())
    , fm_(fm)
    , metadataPageSize_(fm_->getMetadataPageSize())
    , metadataPages_(metadataPageSize_)
    , pageSize_(pageSize)
    , chunkKey_(chunkKey) {
  // Create a new FileBuffer
  CHECK(fm_);
  calcHeaderBuffer();
  CHECK_GT(pageSize_, reservedHeaderSize_);
  pageDataSize_ = pageSize_ - reservedHeaderSize_;
  //@todo reintroduce initialSize - need to develop easy way of
  // differentiating these pre-allocated pages from "written-to" pages
  /*
  if (initalSize > 0) {
      // should expand to initialSize bytes
      size_t initialNumPages = (initalSize + pageSize_ -1) / pageSize_;
      int32_t epoch = fm_->epoch();
      for (size_t pageNum = 0; pageNum < initialNumPages; ++pageNum) {
          Page page = addNewMultiPage(epoch);
          writeHeader(page,pageNum,epoch);
      }
  }
  */
}

Here is the call graph for this function:

File_Namespace::FileBuffer::FileBuffer	(	FileMgr *	fm,
		const size_t	pageSize,
		const ChunkKey &	chunkKey,
		const SQLTypeInfo	sqlType,
		const size_t	initialSize = 0
	)

Definition at line 67 of file FileBuffer.cpp.

References calcHeaderBuffer(), CHECK, fm_, pageDataSize_, pageSize_, and reservedHeaderSize_.

    : AbstractBuffer(fm->getDeviceId(), sqlType)
    , fm_(fm)
    , metadataPageSize_(fm->getMetadataPageSize())
    , metadataPages_(metadataPageSize_)
    , pageSize_(pageSize)
    , chunkKey_(chunkKey) {
  CHECK(fm_);
  calcHeaderBuffer();
  pageDataSize_ = pageSize_ - reservedHeaderSize_;
}

Here is the call graph for this function:

File_Namespace::FileBuffer::FileBuffer	(	FileMgr *	fm,
		const ChunkKey &	chunkKey,
		const std::vector< HeaderInfo >::const_iterator &	headerStartIt,
		const std::vector< HeaderInfo >::const_iterator &	headerEndIt
	)

Definition at line 83 of file FileBuffer.cpp.

References calcHeaderBuffer(), CHECK, File_Namespace::FileMgr::failOnReadError(), logger::FATAL, fm_, initMetadataAndPageDataSize(), LOG, metadataPages_, multiPages_, pageSize_, File_Namespace::MultiPage::push(), and show_chunk().

    : AbstractBuffer(fm->getDeviceId())
    , fm_(fm)
    , metadataPageSize_(fm->getMetadataPageSize())
    , metadataPages_(metadataPageSize_)
    , pageSize_(0)
    , chunkKey_(chunkKey) {
  // We are being assigned an existing FileBuffer on disk

  CHECK(fm_);
  calcHeaderBuffer();
  int32_t lastPageId = -1;
  int32_t curPageId = 0;
  for (auto vecIt = headerStartIt; vecIt != headerEndIt; ++vecIt) {
    curPageId = vecIt->pageId;

    // We only want to read last metadata page
    if (curPageId == -1) {  // stats page
      metadataPages_.push(vecIt->page, vecIt->versionEpoch);
    } else {
      if (curPageId != lastPageId) {
        // protect from bad data on disk, and give diagnostics
        if (fm->failOnReadError()) {
          if (curPageId != lastPageId + 1) {
            LOG(FATAL) << "Failure reading DB file " << show_chunk(chunkKey)
                       << " Current page " << curPageId << " last page " << lastPageId
                       << " epoch " << vecIt->versionEpoch;
          }
        }
        if (lastPageId == -1) {  // If we are on first real page
          initMetadataAndPageDataSize();
        }
        MultiPage multiPage(pageSize_);
        multiPages_.push_back(multiPage);
        lastPageId = curPageId;
      }
      multiPages_.back().push(vecIt->page, vecIt->versionEpoch);
    }
  }
  if (curPageId == -1) {  // meaning there was only a metadata page
    initMetadataAndPageDataSize();
  }
}

Here is the call graph for this function:

File_Namespace::FileBuffer::~FileBuffer ( )

override

Destructor.

Definition at line 130 of file FileBuffer.cpp.

                        {
  // need to free pages
  // NOP
}

Member Function Documentation

Page File_Namespace::FileBuffer::addNewMultiPage

(

const int32_t

epoch

)

Definition at line 408 of file FileBuffer.cpp.

References fm_, multiPages_, pageSize_, File_Namespace::MultiPage::push(), and File_Namespace::FileMgr::requestFreePage().

Referenced by append(), reserve(), and write().

                                                    {
  Page page = fm_->requestFreePage(pageSize_, false);
  MultiPage multiPage(pageSize_);
  multiPage.push(page, epoch);
  multiPages_.emplace_back(multiPage);
  return page;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::append ( int8_t *  src, const size_t  numBytes, const MemoryLevel  srcMemoryLevel = CPU_LEVEL, const int32_t  deviceId = -1  )

override

Definition at line 492 of file FileBuffer.cpp.

References addNewMultiPage(), CHECK, File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileInfoForFileId(), getFileMgrEpoch(), multiPages_, pageDataSize_, File_Namespace::Page::pageNum, pageSize_, reservedHeaderSize_, Data_Namespace::AbstractBuffer::setAppended(), Data_Namespace::AbstractBuffer::size_, File_Namespace::FileInfo::write(), and writeHeader().

                                                {
  setAppended();

  size_t startPage = size_ / pageDataSize_;
  size_t startPageOffset = size_ % pageDataSize_;
  size_t numPagesToWrite =
      (numBytes + startPageOffset + pageDataSize_ - 1) / pageDataSize_;
  size_t bytesLeft = numBytes;
  int8_t* curPtr = src;  // a pointer to the current location in dst being written to
  size_t initialNumPages = multiPages_.size();
  size_ = size_ + numBytes;
  auto epoch = getFileMgrEpoch();
  for (size_t pageNum = startPage; pageNum < startPage + numPagesToWrite; ++pageNum) {
    Page page;
    if (pageNum >= initialNumPages) {
      page = addNewMultiPage(epoch);
      writeHeader(page, pageNum, epoch);
    } else {
      // we already have a new page at current
      // epoch for this page - just grab this page
      page = multiPages_[pageNum].current().page;
    }
    CHECK(page.fileId >= 0);  // make sure page was initialized
    FileInfo* fileInfo = fm_->getFileInfoForFileId(page.fileId);
    size_t bytesWritten;
    if (pageNum == startPage) {
      bytesWritten = fileInfo->write(
          page.pageNum * pageSize_ + startPageOffset + reservedHeaderSize_,
          min(pageDataSize_ - startPageOffset, bytesLeft),
          curPtr);
    } else {
      bytesWritten = fileInfo->write(page.pageNum * pageSize_ + reservedHeaderSize_,
                                     min(pageDataSize_, bytesLeft),
                                     curPtr);
    }
    curPtr += bytesWritten;
    bytesLeft -= bytesWritten;
  }
  CHECK(bytesLeft == 0);
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::calcHeaderBuffer ( )

private

Definition at line 145 of file FileBuffer.cpp.

References chunkKey_, headerBufferOffset_, and reservedHeaderSize_.

Referenced by FileBuffer().

                                  {
  // 3 * sizeof(int32_t) is for headerSize, for pageId and versionEpoch
  // sizeof(size_t) is for chunkSize
  reservedHeaderSize_ = (chunkKey_.size() + 3) * sizeof(int32_t);
  size_t headerMod = reservedHeaderSize_ % headerBufferOffset_;
  if (headerMod > 0) {
    reservedHeaderSize_ += headerBufferOffset_ - headerMod;
  }
}

Here is the caller graph for this function:

void File_Namespace::FileBuffer::copyPage	(	Page &	srcPage,
		Page &	destPage,
		const size_t	numBytes,
		const size_t	offset = 0
	)

Definition at line 390 of file FileBuffer.cpp.

References CHECK, CHECK_LE, checked_malloc(), File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileInfoForFileId(), pageDataSize_, File_Namespace::Page::pageNum, pageSize_, File_Namespace::FileInfo::read(), reservedHeaderSize_, and File_Namespace::FileInfo::write().

Referenced by write().

                                               {
  CHECK_LE(offset + numBytes, pageDataSize_);
  FileInfo* srcFileInfo = fm_->getFileInfoForFileId(srcPage.fileId);
  FileInfo* destFileInfo = fm_->getFileInfoForFileId(destPage.fileId);

  int8_t* buffer = reinterpret_cast<int8_t*>(checked_malloc(numBytes));
  ScopeGuard guard = [&buffer] { free(buffer); };
  size_t bytesRead = srcFileInfo->read(
      srcPage.pageNum * pageSize_ + offset + reservedHeaderSize_, numBytes, buffer);
  CHECK(bytesRead == numBytes);
  size_t bytesWritten = destFileInfo->write(
      destPage.pageNum * pageSize_ + offset + reservedHeaderSize_, numBytes, buffer);
  CHECK(bytesWritten == numBytes);
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::string File_Namespace::FileBuffer::dump

(

)

const

Definition at line 631 of file FileBuffer.cpp.

References chunkKey_, Data_Namespace::AbstractBuffer::hasEncoder(), show_chunk(), and Data_Namespace::AbstractBuffer::size_.

                                 {
  std::stringstream ss;
  ss << "chunk_key = " << show_chunk(chunkKey_) << "\n";
  ss << "has_encoder = " << (hasEncoder() ? "true\n" : "false\n");
  ss << "size_ = " << size_ << "\n";
  return ss.str();
}

Here is the call graph for this function:

size_t File_Namespace::FileBuffer::freeChunkPages

(

)

Definition at line 178 of file FileBuffer.cpp.

References freePage(), and multiPages_.

Referenced by freePages(), and freePagesBeforeEpoch().

                                  {
  size_t num_pages_freed = multiPages_.size();
  for (auto multiPageIt = multiPages_.begin(); multiPageIt != multiPages_.end();
       ++multiPageIt) {
    for (auto pageIt = multiPageIt->pageVersions.begin();
         pageIt != multiPageIt->pageVersions.end();
         ++pageIt) {
      freePage(pageIt->page, false /* isRolloff */);
    }
  }
  multiPages_.clear();
  return num_pages_freed;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t File_Namespace::FileBuffer::freeMetadataPages

(

)

Definition at line 165 of file FileBuffer.cpp.

References freePage(), metadataPages_, File_Namespace::MultiPage::pageVersions, and File_Namespace::MultiPage::pop().

Referenced by freePages().

                                     {
  size_t num_pages_freed = metadataPages_.pageVersions.size();
  for (auto metaPageIt = metadataPages_.pageVersions.begin();
       metaPageIt != metadataPages_.pageVersions.end();
       ++metaPageIt) {
    freePage(metaPageIt->page, false /* isRolloff */);
  }
  while (metadataPages_.pageVersions.size() > 0) {
    metadataPages_.pop();
  }
  return num_pages_freed;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::freePage

(

const Page &

page

)

Definition at line 155 of file FileBuffer.cpp.

Referenced by freeChunkPages(), freeMetadataPages(), and freePagesBeforeEpochForMultiPage().

                                          {
  freePage(page, false);
}

Here is the caller graph for this function:

void File_Namespace::FileBuffer::freePage ( const Page &  page, const bool  isRolloff  )

private

Definition at line 159 of file FileBuffer.cpp.

References CHECK, File_Namespace::Page::fileId, fm_, File_Namespace::FileInfo::freePage(), File_Namespace::FileMgr::getFileInfoForFileId(), getFileMgrEpoch(), and File_Namespace::Page::pageNum.

                                                                {
  FileInfo* fileInfo = fm_->getFileInfoForFileId(page.fileId);
  CHECK(fileInfo);
  fileInfo->freePage(page.pageNum, isRolloff, getFileMgrEpoch());
}

Here is the call graph for this function:

size_t File_Namespace::FileBuffer::freePages

(

)

Definition at line 192 of file FileBuffer.cpp.

References freeChunkPages(), and freeMetadataPages().

                             {
  return freeMetadataPages() + freeChunkPages();
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::freePagesBeforeEpoch

(

const int32_t

targetEpoch

)

Definition at line 206 of file FileBuffer.cpp.

References CHECK_LE, chunkKey_, fm_, freeChunkPages(), freePagesBeforeEpochForMultiPage(), getFileMgrEpoch(), metadataPages_, multiPages_, pageSize_, File_Namespace::MultiPage::pageVersions, and Data_Namespace::AbstractBuffer::size_.

                                                               {
  // This method is only safe to be called within a checkpoint, after the sync and epoch
  // increment where a failure at any point32_t in the process would lead to a safe
  // rollback
  auto currentEpoch = getFileMgrEpoch();
  CHECK_LE(targetEpoch, currentEpoch);
  freePagesBeforeEpochForMultiPage(metadataPages_, targetEpoch, currentEpoch);
  for (auto& multiPage : multiPages_) {
    freePagesBeforeEpochForMultiPage(multiPage, targetEpoch, currentEpoch);
  }

  // Check if all buffer pages can be freed
  if (size_ == 0) {
    size_t max_historical_buffer_size{0};
    for (auto& epoch_page : metadataPages_.pageVersions) {
      // Create buffer that is used to get the buffer size at the epoch version
      FileBuffer buffer{fm_, pageSize_, chunkKey_};
      buffer.readMetadata(epoch_page.page);
      max_historical_buffer_size = std::max(max_historical_buffer_size, buffer.size());
    }

    // Free all chunk pages, if none of the old chunk versions has any data
    if (max_historical_buffer_size == 0) {
      freeChunkPages();
    }
  }
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::freePagesBeforeEpochForMultiPage ( MultiPage &  multiPage, const int32_t  targetEpoch, const int32_t  currentEpoch  )

private

Definition at line 196 of file FileBuffer.cpp.

References freePage(), and File_Namespace::MultiPage::freePagesBeforeEpoch().

Referenced by freePagesBeforeEpoch().

                                                                              {
  std::vector<EpochedPage> epochedPagesToFree =
      multiPage.freePagesBeforeEpoch(targetEpoch, currentEpoch);
  for (const auto& epochedPageToFree : epochedPagesToFree) {
    freePage(epochedPageToFree.page, true /* isRolloff */);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

int32_t File_Namespace::FileBuffer::getFileMgrEpoch ( )

private

Definition at line 626 of file FileBuffer.cpp.

References chunkKey_, File_Namespace::FileMgr::epoch(), fm_, and get_table_prefix().

Referenced by append(), freePage(), freePagesBeforeEpoch(), reserve(), and write().

                                    {
  auto [db_id, tb_id] = get_table_prefix(chunkKey_);
  return fm_->epoch(db_id, tb_id);
}

Here is the call graph for this function:

Here is the caller graph for this function:

int8_t* File_Namespace::FileBuffer::getMemoryPtr ( )

inlineoverridevirtual

Not implemented for FileMgr – throws a runtime_error.

Implements Data_Namespace::AbstractBuffer.

Definition at line 124 of file FileBuffer.h.

References logger::FATAL, and LOG.

                                  {
    LOG(FATAL) << "Operation not supported.";
    return nullptr;  // satisfy return-type warning
  }

MultiPage File_Namespace::FileBuffer::getMetadataPage ( ) const

inline

Definition at line 148 of file FileBuffer.h.

{ return metadataPages_; }

virtual std::vector<MultiPage> File_Namespace::FileBuffer::getMultiPage ( ) const

inlinevirtual

Returns vector of MultiPages in the FileBuffer.

Definition at line 147 of file FileBuffer.h.

Referenced by read().

{ return multiPages_; }

Here is the caller graph for this function:

Data_Namespace::MemoryLevel File_Namespace::FileBuffer::getType ( ) const

inlineoverridevirtual

Implements Data_Namespace::AbstractBuffer.

Definition at line 121 of file FileBuffer.h.

References Data_Namespace::DISK_LEVEL.

{ return DISK_LEVEL; }

bool File_Namespace::FileBuffer::hasDataPages ( ) const

inline

Returns whether or not a buffer has data pages. It is possible for a buffer to represent metadata (have a size and encode) but not contain actual data.

Definition at line 134 of file FileBuffer.h.

{ return pageCount() > 0; }

void File_Namespace::FileBuffer::initMetadataAndPageDataSize ( )

private

Definition at line 639 of file FileBuffer.cpp.

References CHECK, File_Namespace::MultiPage::current(), File_Namespace::Page::fileId, metadataPages_, File_Namespace::EpochedPage::page, pageDataSize_, pageSize_, readMetadata(), and reservedHeaderSize_.

Referenced by FileBuffer().

                                             {
  CHECK(metadataPages_.current().page.fileId != -1);  // was initialized
  readMetadata(metadataPages_.current().page);
  pageDataSize_ = pageSize_ - reservedHeaderSize_;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool File_Namespace::FileBuffer::isMissingPages

(

)

const

Definition at line 645 of file FileBuffer.cpp.

References multiPages_, pageDataSize_, and Data_Namespace::AbstractBuffer::size().

                                      {
  // Detect the case where a page is missing by comparing the amount of pages read
  // with the metadata size.
  return ((size() + pageDataSize_ - 1) / pageDataSize_ != multiPages_.size());
}

Here is the call graph for this function:

size_t File_Namespace::FileBuffer::numChunkPages

(

)

const

Definition at line 651 of file FileBuffer.cpp.

References multiPages_.

                                       {
  size_t total_size = 0;
  for (const auto& multi_page : multiPages_) {
    total_size += multi_page.pageVersions.size();
  }
  return total_size;
}

size_t File_Namespace::FileBuffer::numMetadataPages ( ) const

inline

Returns the total number of used bytes in the FileBuffer.

Definition at line 156 of file FileBuffer.h.

{ return metadataPages_.pageVersions.size(); };

size_t File_Namespace::FileBuffer::pageCount ( ) const

inlineoverridevirtual

Returns the number of pages in the FileBuffer.

Implements Data_Namespace::AbstractBuffer.

Definition at line 130 of file FileBuffer.h.

{ return multiPages_.size(); }

virtual size_t File_Namespace::FileBuffer::pageDataSize ( ) const

inlinevirtual

Returns the size in bytes of the data portion of each page in the FileBuffer.

Definition at line 140 of file FileBuffer.h.

Referenced by File_Namespace::readForThread().

{ return pageDataSize_; }

Here is the caller graph for this function:

size_t File_Namespace::FileBuffer::pageSize ( ) const

inlineoverridevirtual

Returns the size in bytes of each page in the FileBuffer.

Implements Data_Namespace::AbstractBuffer.

Definition at line 137 of file FileBuffer.h.

Referenced by File_Namespace::readForThread(), and writeHeader().

{ return pageSize_; }

Here is the caller graph for this function:

void File_Namespace::FileBuffer::read ( int8_t *const  dst, const size_t  numBytes = 0, const size_t  offset = 0, const MemoryLevel  dstMemoryLevel = CPU_LEVEL, const int32_t  deviceId = -1  )

override

Definition at line 286 of file FileBuffer.cpp.

References threading_serial::async(), CHECK, Data_Namespace::CPU_LEVEL, logger::FATAL, fm_, getMultiPage(), File_Namespace::FileMgr::getNumReaderThreads(), LOG, File_Namespace::readThreadDS::multiPages, multiPages_, pageDataSize_, File_Namespace::readForThread(), File_Namespace::readThreadDS::t_bytesLeft, File_Namespace::readThreadDS::t_curPtr, File_Namespace::readThreadDS::t_endPage, File_Namespace::readThreadDS::t_fm, File_Namespace::readThreadDS::t_isFirstPage, File_Namespace::readThreadDS::t_startPage, and File_Namespace::readThreadDS::t_startPageOffset.

                                              {
  if (dstBufferType != CPU_LEVEL) {
    LOG(FATAL) << "Unsupported Buffer type";
  }

  // variable declarations
  size_t startPage = offset / pageDataSize_;
  size_t startPageOffset = offset % pageDataSize_;
  size_t numPagesToRead =
      (numBytes + startPageOffset + pageDataSize_ - 1) / pageDataSize_;
  /*
  if (startPage + numPagesToRead > multiPages_.size()) {
      cout << "Start page: " << startPage << endl;
      cout << "Num pages to read: " << numPagesToRead << endl;
      cout << "Num multipages: " << multiPages_.size() << endl;
      cout << "Offset: " << offset << endl;
      cout << "Num bytes: " << numBytes << endl;
  }
  */

  CHECK(startPage + numPagesToRead <= multiPages_.size())
      << "Requested page out of bounds";

  size_t numPagesPerThread = 0;
  size_t numBytesCurrent = numBytes;  // total number of bytes still to be read
  size_t bytesRead = 0;               // total number of bytes already being read
  size_t bytesLeftForThread = 0;      // number of bytes to be read in the thread
  size_t numExtraPages = 0;  // extra pages to be assigned one per thread as needed
  size_t numThreads = fm_->getNumReaderThreads();
  std::vector<readThreadDS>
      threadDSArr;  // array of threadDS, needed to avoid racing conditions

  if (numPagesToRead > numThreads) {
    numPagesPerThread = numPagesToRead / numThreads;
    numExtraPages = numPagesToRead - (numThreads * numPagesPerThread);
  } else {
    numThreads = numPagesToRead;
    numPagesPerThread = 1;
  }

  /* set threadDS for the first thread */
  readThreadDS threadDS;
  threadDS.t_fm = fm_;
  threadDS.t_startPage = offset / pageDataSize_;
  if (numExtraPages > 0) {
    threadDS.t_endPage = threadDS.t_startPage + numPagesPerThread + 1;
    numExtraPages--;
  } else {
    threadDS.t_endPage = threadDS.t_startPage + numPagesPerThread;
  }
  threadDS.t_curPtr = dst;
  threadDS.t_startPageOffset = offset % pageDataSize_;
  threadDS.t_isFirstPage = true;

  bytesLeftForThread = min(((threadDS.t_endPage - threadDS.t_startPage) * pageDataSize_ -
                            threadDS.t_startPageOffset),
                           numBytesCurrent);
  threadDS.t_bytesLeft = bytesLeftForThread;
  threadDS.multiPages = getMultiPage();

  if (numThreads == 1) {
    bytesRead += readForThread(this, threadDS);
  } else {
    std::vector<std::future<size_t>> threads;

    for (size_t i = 0; i < numThreads; i++) {
      threadDSArr.push_back(threadDS);
      threads.push_back(
          std::async(std::launch::async, readForThread, this, threadDSArr[i]));

      // calculate elements of threadDS
      threadDS.t_fm = fm_;
      threadDS.t_isFirstPage = false;
      threadDS.t_curPtr += bytesLeftForThread;
      threadDS.t_startPage +=
          threadDS.t_endPage -
          threadDS.t_startPage;  // based on # of pages read on previous iteration
      if (numExtraPages > 0) {
        threadDS.t_endPage = threadDS.t_startPage + numPagesPerThread + 1;
        numExtraPages--;
      } else {
        threadDS.t_endPage = threadDS.t_startPage + numPagesPerThread;
      }
      numBytesCurrent -= bytesLeftForThread;
      bytesLeftForThread = min(
          ((threadDS.t_endPage - threadDS.t_startPage) * pageDataSize_), numBytesCurrent);
      threadDS.t_bytesLeft = bytesLeftForThread;
      threadDS.multiPages = getMultiPage();
    }

    for (auto& p : threads) {
      p.wait();
    }
    for (auto& p : threads) {
      bytesRead += p.get();
    }
  }
  CHECK(bytesRead == numBytes);
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::readMetadata ( const Page &  page )

private

Definition at line 435 of file FileBuffer.cpp.

References CHECK, Data_Namespace::AbstractBuffer::encoder_, f(), File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileForFileId(), Data_Namespace::AbstractBuffer::initEncoder(), METADATA_VERSION, metadataPageSize_, NUM_METADATA, File_Namespace::Page::pageNum, pageSize_, reservedHeaderSize_, SQLTypeInfo::set_comp_param(), SQLTypeInfo::set_compression(), SQLTypeInfo::set_dimension(), SQLTypeInfo::set_notnull(), SQLTypeInfo::set_scale(), SQLTypeInfo::set_size(), SQLTypeInfo::set_subtype(), SQLTypeInfo::set_type(), Data_Namespace::AbstractBuffer::size_, Data_Namespace::AbstractBuffer::sql_type_, and setup::version.

Referenced by initMetadataAndPageDataSize().

                                              {
  FILE* f = fm_->getFileForFileId(page.fileId);
  fseek(f, page.pageNum * metadataPageSize_ + reservedHeaderSize_, SEEK_SET);
  fread((int8_t*)&pageSize_, sizeof(size_t), 1, f);
  fread((int8_t*)&size_, sizeof(size_t), 1, f);
  vector<int32_t> typeData(
      NUM_METADATA);  // assumes we will encode hasEncoder, bufferType,
                      // encodingType, encodingBits all as int
  fread((int8_t*)&(typeData[0]), sizeof(int32_t), typeData.size(), f);
  int32_t version = typeData[0];
  CHECK(version == METADATA_VERSION);  // add backward compatibility code here
  bool has_encoder = static_cast<bool>(typeData[1]);
  if (has_encoder) {
    sql_type_.set_type(static_cast<SQLTypes>(typeData[2]));
    sql_type_.set_subtype(static_cast<SQLTypes>(typeData[3]));
    sql_type_.set_dimension(typeData[4]);
    sql_type_.set_scale(typeData[5]);
    sql_type_.set_notnull(static_cast<bool>(typeData[6]));
    sql_type_.set_compression(static_cast<EncodingType>(typeData[7]));
    sql_type_.set_comp_param(typeData[8]);
    sql_type_.set_size(typeData[9]);
    initEncoder(sql_type_);
    encoder_->readMetadata(f);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::reserve ( const size_t  numBytes )

overridevirtual

Implements Data_Namespace::AbstractBuffer.

Definition at line 135 of file FileBuffer.cpp.

References addNewMultiPage(), getFileMgrEpoch(), multiPages_, pageSize_, and writeHeader().

                                              {
  size_t numPagesRequested = (numBytes + pageSize_ - 1) / pageSize_;
  size_t numCurrentPages = multiPages_.size();
  auto epoch = getFileMgrEpoch();
  for (size_t pageNum = numCurrentPages; pageNum < numPagesRequested; ++pageNum) {
    Page page = addNewMultiPage(epoch);
    writeHeader(page, pageNum, epoch);
  }
}

Here is the call graph for this function:

virtual size_t File_Namespace::FileBuffer::reservedHeaderSize ( ) const

inlinevirtual

Returns the size in bytes of the reserved header portion of each page in the FileBuffer.

Definition at line 144 of file FileBuffer.h.

Referenced by File_Namespace::readForThread().

{ return reservedHeaderSize_; }

Here is the caller graph for this function:

size_t File_Namespace::FileBuffer::reservedSize ( ) const

inlineoverridevirtual

Returns the total number of bytes allocated for the FileBuffer.

Implements Data_Namespace::AbstractBuffer.

Definition at line 151 of file FileBuffer.h.

{ return multiPages_.size() * pageSize_; }

void File_Namespace::FileBuffer::write ( int8_t *  src, const size_t  numBytes, const size_t  offset = 0, const MemoryLevel  srcMemoryLevel = CPU_LEVEL, const int32_t  deviceId = -1  )

override

Writes the contents of source (src) into new versions of the affected logical pages.

This method will write the contents of source (src) into new version of the affected logical pages. New pages are only appended if the value of epoch (in FileMgr)

Definition at line 536 of file FileBuffer.cpp.

References addNewMultiPage(), CHECK, copyPage(), Data_Namespace::CPU_LEVEL, File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileInfoForFileId(), getFileMgrEpoch(), multiPages_, pageDataSize_, File_Namespace::Page::pageNum, pageSize_, File_Namespace::FileMgr::requestFreePage(), reservedHeaderSize_, Data_Namespace::AbstractBuffer::setAppended(), Data_Namespace::AbstractBuffer::setDirty(), Data_Namespace::AbstractBuffer::setUpdated(), Data_Namespace::AbstractBuffer::size_, File_Namespace::FileInfo::write(), and writeHeader().

                                               {
  CHECK(srcBufferType == CPU_LEVEL) << "Unsupported Buffer type";

  bool tempIsAppended = false;
  setDirty();
  if (offset < size_) {
    setUpdated();
  }
  if (offset + numBytes > size_) {
    tempIsAppended = true;  // because is_appended_ could have already been true - to
                            // avoid rewriting header
    setAppended();
    size_ = offset + numBytes;
  }

  size_t startPage = offset / pageDataSize_;
  size_t startPageOffset = offset % pageDataSize_;
  size_t numPagesToWrite =
      (numBytes + startPageOffset + pageDataSize_ - 1) / pageDataSize_;
  size_t bytesLeft = numBytes;
  int8_t* curPtr = src;  // a pointer to the current location in dst being written to
  size_t initialNumPages = multiPages_.size();
  auto epoch = getFileMgrEpoch();

  if (startPage >
      initialNumPages) {  // means there is a gap we need to allocate pages for
    for (size_t pageNum = initialNumPages; pageNum < startPage; ++pageNum) {
      Page page = addNewMultiPage(epoch);
      writeHeader(page, pageNum, epoch);
    }
  }
  for (size_t pageNum = startPage; pageNum < startPage + numPagesToWrite; ++pageNum) {
    Page page;
    if (pageNum >= initialNumPages) {
      page = addNewMultiPage(epoch);
      writeHeader(page, pageNum, epoch);
    } else if (multiPages_[pageNum].current().epoch <
               epoch) {  // need to create new page b/c this current one lags epoch and we
                         // can't overwrite it also need to copy if we are on first or
                         // last page
      Page lastPage = multiPages_[pageNum].current().page;
      page = fm_->requestFreePage(pageSize_, false);
      multiPages_[pageNum].push(page, epoch);
      if (pageNum == startPage && startPageOffset > 0) {
        // copyPage takes care of header offset so don't worry
        // about it
        copyPage(lastPage, page, startPageOffset, 0);
      }
      if (pageNum == (startPage + numPagesToWrite - 1) &&
          bytesLeft > 0) {  // bytesLeft should always > 0
        copyPage(lastPage,
                 page,
                 pageDataSize_ - bytesLeft,
                 bytesLeft);  // these would be empty if we're appending but we won't
                              // worry about it right now
      }
      writeHeader(page, pageNum, epoch);
    } else {
      // we already have a new page at current
      // epoch for this page - just grab this page
      page = multiPages_[pageNum].current().page;
    }
    CHECK(page.fileId >= 0);  // make sure page was initialized
    FileInfo* fileInfo = fm_->getFileInfoForFileId(page.fileId);
    size_t bytesWritten;
    if (pageNum == startPage) {
      bytesWritten = fileInfo->write(
          page.pageNum * pageSize_ + startPageOffset + reservedHeaderSize_,
          min(pageDataSize_ - startPageOffset, bytesLeft),
          curPtr);
    } else {
      bytesWritten = fileInfo->write(page.pageNum * pageSize_ + reservedHeaderSize_,
                                     min(pageDataSize_, bytesLeft),
                                     curPtr);
    }
    curPtr += bytesWritten;
    bytesLeft -= bytesWritten;
    if (tempIsAppended && pageNum == startPage + numPagesToWrite - 1) {  // if last page
      //@todo below can lead to undefined - we're overwriting num
      // bytes valid at checkpoint
      writeHeader(page, 0, multiPages_[0].current().epoch, true);
    }
  }
  CHECK(bytesLeft == 0);
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::writeHeader ( Page &  page, const int32_t  pageId, const int32_t  epoch, const bool  writeMetadata = false  )

private

Write header writes header at top of page in format.

Definition at line 416 of file FileBuffer.cpp.

References chunkKey_, gpu_enabled::copy(), File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileInfoForFileId(), report::header, metadataPageSize_, File_Namespace::Page::pageNum, pageSize(), pageSize_, and File_Namespace::FileInfo::write().

Referenced by append(), reserve(), write(), and writeMetadata().

                                                       {
  int32_t intHeaderSize = chunkKey_.size() + 3;  // does not include chunkSize
  vector<int32_t> header(intHeaderSize);
  // in addition to chunkkey we need size of header, pageId, version
  header[0] =
      (intHeaderSize - 1) * sizeof(int32_t);  // don't need to include size of headerSize
                                              // value - sizeof(size_t) is for chunkSize
  std::copy(chunkKey_.begin(), chunkKey_.end(), header.begin() + 1);
  header[intHeaderSize - 2] = pageId;
  header[intHeaderSize - 1] = epoch;
  FileInfo* fileInfo = fm_->getFileInfoForFileId(page.fileId);
  size_t pageSize = writeMetadata ? metadataPageSize_ : pageSize_;
  fileInfo->write(
      page.pageNum * pageSize, (intHeaderSize) * sizeof(int32_t), (int8_t*)&header[0]);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::writeMetadata ( const int32_t  epoch )

private

Definition at line 461 of file FileBuffer.cpp.

References Data_Namespace::AbstractBuffer::encoder_, f(), File_Namespace::Page::fileId, fm_, SQLTypeInfo::get_comp_param(), SQLTypeInfo::get_compression(), SQLTypeInfo::get_dimension(), SQLTypeInfo::get_notnull(), SQLTypeInfo::get_scale(), SQLTypeInfo::get_size(), SQLTypeInfo::get_subtype(), SQLTypeInfo::get_type(), File_Namespace::FileMgr::getFileForFileId(), Data_Namespace::AbstractBuffer::hasEncoder(), METADATA_VERSION, metadataPages_, metadataPageSize_, NUM_METADATA, File_Namespace::Page::pageNum, pageSize_, File_Namespace::MultiPage::push(), File_Namespace::FileMgr::requestFreePage(), reservedHeaderSize_, Data_Namespace::AbstractBuffer::size_, Data_Namespace::AbstractBuffer::sql_type_, and writeHeader().

                                                  {
  // Right now stats page is size_ (in bytes), bufferType, encodingType,
  // encodingDataType, numElements
  Page page = fm_->requestFreePage(metadataPageSize_, true);
  writeHeader(page, -1, epoch, true);
  FILE* f = fm_->getFileForFileId(page.fileId);
  fseek(f, page.pageNum * metadataPageSize_ + reservedHeaderSize_, SEEK_SET);
  fwrite((int8_t*)&pageSize_, sizeof(size_t), 1, f);
  fwrite((int8_t*)&size_, sizeof(size_t), 1, f);
  vector<int32_t> typeData(
      NUM_METADATA);  // assumes we will encode hasEncoder, bufferType,
                      // encodingType, encodingBits all as int32_t
  typeData[0] = METADATA_VERSION;
  typeData[1] = static_cast<int32_t>(hasEncoder());
  if (hasEncoder()) {
    typeData[2] = static_cast<int32_t>(sql_type_.get_type());
    typeData[3] = static_cast<int32_t>(sql_type_.get_subtype());
    typeData[4] = sql_type_.get_dimension();
    typeData[5] = sql_type_.get_scale();
    typeData[6] = static_cast<int32_t>(sql_type_.get_notnull());
    typeData[7] = static_cast<int32_t>(sql_type_.get_compression());
    typeData[8] = sql_type_.get_comp_param();
    typeData[9] = sql_type_.get_size();
  }
  fwrite((int8_t*)&(typeData[0]), sizeof(int32_t), typeData.size(), f);
  if (hasEncoder()) {  // redundant
    encoder_->writeMetadata(f);
  }
  metadataPages_.push(page, epoch);
}

Here is the call graph for this function:

Friends And Related Function Documentation

friend class CachingFileMgr

friend

Definition at line 59 of file FileBuffer.h.

friend class FileMgr

friend

Definition at line 58 of file FileBuffer.h.

Member Data Documentation

ChunkKey File_Namespace::FileBuffer::chunkKey_

private

Definition at line 202 of file FileBuffer.h.

Referenced by calcHeaderBuffer(), dump(), freePagesBeforeEpoch(), getFileMgrEpoch(), and writeHeader().

FileMgr* File_Namespace::FileBuffer::fm_

private

Definition at line 190 of file FileBuffer.h.

Referenced by addNewMultiPage(), append(), copyPage(), FileBuffer(), freePage(), freePagesBeforeEpoch(), getFileMgrEpoch(), read(), readMetadata(), write(), writeHeader(), and writeMetadata().

constexpr size_t File_Namespace::FileBuffer::headerBufferOffset_ = 32

static

Definition at line 165 of file FileBuffer.h.

Referenced by calcHeaderBuffer().

MultiPage File_Namespace::FileBuffer::metadataPages_

private

Definition at line 197 of file FileBuffer.h.

Referenced by FileBuffer(), freeMetadataPages(), freePagesBeforeEpoch(), initMetadataAndPageDataSize(), and writeMetadata().

const size_t File_Namespace::FileBuffer::metadataPageSize_

private

Definition at line 196 of file FileBuffer.h.

Referenced by readMetadata(), writeHeader(), and writeMetadata().

std::vector<MultiPage> File_Namespace::FileBuffer::multiPages_

private

Definition at line 198 of file FileBuffer.h.

Referenced by addNewMultiPage(), append(), FileBuffer(), freeChunkPages(), freePagesBeforeEpoch(), isMissingPages(), numChunkPages(), read(), reserve(), and write().

size_t File_Namespace::FileBuffer::pageDataSize_

private

Definition at line 200 of file FileBuffer.h.

Referenced by append(), copyPage(), FileBuffer(), initMetadataAndPageDataSize(), isMissingPages(), read(), and write().

size_t File_Namespace::FileBuffer::pageSize_

private

Definition at line 199 of file FileBuffer.h.

Referenced by addNewMultiPage(), append(), copyPage(), FileBuffer(), freePagesBeforeEpoch(), initMetadataAndPageDataSize(), readMetadata(), reserve(), write(), writeHeader(), and writeMetadata().

size_t File_Namespace::FileBuffer::reservedHeaderSize_

private

Definition at line 201 of file FileBuffer.h.

Referenced by append(), calcHeaderBuffer(), copyPage(), FileBuffer(), initMetadataAndPageDataSize(), readMetadata(), write(), and writeMetadata().

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/FileMgr/FileBuffer.h
• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/FileMgr/FileBuffer.cpp