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
void	freeMetadataPages ()
size_t	freeChunkPages ()
void	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...
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...
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...
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 ()

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)

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

Static Private Attributes
static size_t	headerBufferOffset_ = 32

Friends
class	FileMgr

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 56 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 39 of file FileBuffer.cpp.

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

    : AbstractBuffer(fm->getDeviceId())
    , fm_(fm)
    , metadataPages_(METADATA_PAGE_SIZE)
    , pageSize_(pageSize)
    , chunkKey_(chunkKey) {
  // Create a new FileBuffer
  CHECK(fm_);
  calcHeaderBuffer();
  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)
    , metadataPages_(METADATA_PAGE_SIZE)
    , 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 82 of file FileBuffer.cpp.

References calcHeaderBuffer(), CHECK, File_Namespace::MultiPage::current(), logger::FATAL, File_Namespace::Page::fileId, fm_, LOG, metadataPages_, multiPages_, File_Namespace::EpochedPage::page, pageDataSize_, pageSize_, File_Namespace::MultiPage::push(), readMetadata(), reservedHeaderSize_, and show_chunk().

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

  CHECK(fm_);
  calcHeaderBuffer();
  int32_t lastPageId = -1;
  // Page lastMetadataPage;
  for (auto vecIt = headerStartIt; vecIt != headerEndIt; ++vecIt) {
    int32_t 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 (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
          CHECK(metadataPages_.current().page.fileId != -1);  // was initialized
          readMetadata(metadataPages_.current().page);
          pageDataSize_ = pageSize_ - reservedHeaderSize_;
        }
        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
      readMetadata(metadataPages_.current().page);
      pageDataSize_ = pageSize_ - reservedHeaderSize_;
    }
  }
}

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 388 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 472 of file FileBuffer.cpp.

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

Referenced by File_Namespace::FileMgr::putBuffer().

                                                {
  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;
  int32_t epoch = fm_->epoch();
  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:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::calcHeaderBuffer ( )

private

Definition at line 146 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 368 of file FileBuffer.cpp.

References CHECK, 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().

                                               {
  // FILE *srcFile = fm_->files_[srcPage.fileId]->f;
  // FILE *destFile = fm_->files_[destPage.fileId]->f;
  CHECK(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));
  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);
  free(buffer);
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t File_Namespace::FileBuffer::freeChunkPages

(

)

Definition at line 172 of file FileBuffer.cpp.

References freePage(), and multiPages_.

Referenced by foreign_storage::ForeignStorageCache::eraseChunk(), foreign_storage::ForeignStorageCache::evictChunkByIterator(), and freePages().

                                  {
  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:

void File_Namespace::FileBuffer::freeMetadataPages

(

)

Definition at line 161 of file FileBuffer.cpp.

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

Referenced by freePages().

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

private

Definition at line 156 of file FileBuffer.cpp.

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

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

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

Here is the call graph for this function:

Here is the caller graph for this function:

void File_Namespace::FileBuffer::freePages

(

)

Definition at line 186 of file FileBuffer.cpp.

References freeChunkPages(), and freeMetadataPages().

                           {
  freeMetadataPages();
  freeChunkPages();
}

Here is the call graph for this function:

void File_Namespace::FileBuffer::freePagesBeforeEpoch

(

const int32_t

targetEpoch

)

Definition at line 201 of file FileBuffer.cpp.

References CHECK_LE, File_Namespace::FileMgr::epoch(), fm_, freePagesBeforeEpochForMultiPage(), metadataPages_, and multiPages_.

                                                               {
  // 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
  const int32_t currentEpoch = fm_->epoch();
  CHECK_LE(targetEpoch, currentEpoch);
  freePagesBeforeEpochForMultiPage(metadataPages_, targetEpoch, currentEpoch);
  for (auto& multiPage : multiPages_) {
    freePagesBeforeEpochForMultiPage(multiPage, targetEpoch, currentEpoch);
  }
}

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 191 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:

int8_t* File_Namespace::FileBuffer::getMemoryPtr ( )

inlineoverridevirtual

Not implemented for FileMgr – throws a runtime_error.

Implements Data_Namespace::AbstractBuffer.

Definition at line 122 of file FileBuffer.h.

References logger::FATAL, and LOG.

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

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

inlinevirtual

Returns vector of MultiPages in the FileBuffer.

Definition at line 141 of file FileBuffer.h.

Referenced by File_Namespace::FileMgr::init(), and 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 119 of file FileBuffer.h.

References Data_Namespace::DISK_LEVEL.

{ return DISK_LEVEL; }

size_t File_Namespace::FileBuffer::numMetadataPages ( ) const

inline

Returns the total number of used bytes in the FileBuffer.

Definition at line 149 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 128 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 134 of file FileBuffer.h.

Referenced by File_Namespace::FileMgr::init(), and 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 131 of file FileBuffer.h.

Referenced by File_Namespace::FileMgr::init(), 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 265 of file FileBuffer.cpp.

References CHECK, Data_Namespace::CPU_LEVEL, logger::FATAL, fm_, getMultiPage(), File_Namespace::FileMgr::getNumReaderThreads(), generate_TableFunctionsFactory_init::i, 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());

  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 415 of file FileBuffer.cpp.

References CHECK, Data_Namespace::AbstractBuffer::encoder_, generate_TableFunctionsFactory_init::f, File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileForFileId(), Data_Namespace::AbstractBuffer::initEncoder(), METADATA_PAGE_SIZE, METADATA_VERSION, 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 FileBuffer().

                                              {
  FILE* f = fm_->getFileForFileId(page.fileId);
  fseek(f, page.pageNum * METADATA_PAGE_SIZE + 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(), File_Namespace::FileMgr::epoch(), fm_, multiPages_, pageSize_, and writeHeader().

                                              {
  size_t numPagesRequested = (numBytes + pageSize_ - 1) / pageSize_;
  size_t numCurrentPages = multiPages_.size();
  int32_t epoch = fm_->epoch();

  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 138 of file FileBuffer.h.

Referenced by File_Namespace::FileMgr::init(), and 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 144 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 516 of file FileBuffer.cpp.

References addNewMultiPage(), CHECK, copyPage(), Data_Namespace::CPU_LEVEL, File_Namespace::FileMgr::epoch(), logger::FATAL, File_Namespace::Page::fileId, fm_, File_Namespace::FileMgr::getFileInfoForFileId(), LOG, 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().

Referenced by File_Namespace::FileMgr::putBuffer().

                                               {
  if (srcBufferType != CPU_LEVEL) {
    LOG(FATAL) << "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();
  int32_t epoch = fm_->epoch();

  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 &&
          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:

Here is the caller 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 396 of file FileBuffer.cpp.

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

Referenced by append(), File_Namespace::FileMgr::init(), 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 ? METADATA_PAGE_SIZE : 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 441 of file FileBuffer.cpp.

References Data_Namespace::AbstractBuffer::encoder_, generate_TableFunctionsFactory_init::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_PAGE_SIZE, METADATA_VERSION, metadataPages_, 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(METADATA_PAGE_SIZE, true);
  writeHeader(page, -1, epoch, true);
  FILE* f = fm_->getFileForFileId(page.fileId);
  fseek(f, page.pageNum * METADATA_PAGE_SIZE + 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 FileMgr

friend

Definition at line 57 of file FileBuffer.h.

Member Data Documentation

ChunkKey File_Namespace::FileBuffer::chunkKey_

private

Definition at line 180 of file FileBuffer.h.

Referenced by calcHeaderBuffer(), and writeHeader().

FileMgr* File_Namespace::FileBuffer::fm_

private

Definition at line 172 of file FileBuffer.h.

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

size_t File_Namespace::FileBuffer::headerBufferOffset_ = 32

staticprivate

Definition at line 174 of file FileBuffer.h.

Referenced by calcHeaderBuffer().

MultiPage File_Namespace::FileBuffer::metadataPages_

private

Definition at line 175 of file FileBuffer.h.

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

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

private

Definition at line 176 of file FileBuffer.h.

Referenced by addNewMultiPage(), append(), FileBuffer(), freeChunkPages(), freePagesBeforeEpoch(), File_Namespace::FileMgr::init(), read(), reserve(), and write().

size_t File_Namespace::FileBuffer::pageDataSize_

private

Definition at line 178 of file FileBuffer.h.

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

size_t File_Namespace::FileBuffer::pageSize_

private

Definition at line 177 of file FileBuffer.h.

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

size_t File_Namespace::FileBuffer::reservedHeaderSize_

private

Definition at line 179 of file FileBuffer.h.

Referenced by append(), calcHeaderBuffer(), copyPage(), FileBuffer(), 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