File_Namespace::FileMgr Class Reference

#include <FileMgr.h>

Inheritance diagram for File_Namespace::FileMgr:

Collaboration diagram for File_Namespace::FileMgr:

Public Member Functions
	FileMgr (const int deviceId, GlobalFileMgr *gfm, const std::pair< const int, const int > fileMgrKey, const size_t num_reader_threads=0, const int epoch=-1, const size_t defaultPageSize=2097152)
	Constructor. More...
	FileMgr (const int deviceId, GlobalFileMgr *gfm, const std::pair< const int, const int > fileMgrKey, const bool initOnly)
	FileMgr (GlobalFileMgr *gfm, const size_t defaultPageSize, std::string basePath)
	~FileMgr () override
	Destructor. More...
AbstractBuffer *	createBuffer (const ChunkKey &key, size_t pageSize=0, const size_t numBytes=0) override
	Creates a chunk with the specified key and page size. More...
bool	isBufferOnDevice (const ChunkKey &key) override
void	deleteBuffer (const ChunkKey &key, const bool purge=true) override
	Deletes the chunk with the specified key. More...
void	deleteBuffersWithPrefix (const ChunkKey &keyPrefix, const bool purge=true) override
AbstractBuffer *	getBuffer (const ChunkKey &key, const size_t numBytes=0) override
	Returns the a pointer to the chunk with the specified key. More...
void	fetchBuffer (const ChunkKey &key, AbstractBuffer *destBuffer, const size_t numBytes) override
AbstractBuffer *	putBuffer (const ChunkKey &key, AbstractBuffer *d, const size_t numBytes=0) override
	Puts the contents of d into the Chunk with the given key. More...
AbstractBuffer *	alloc (const size_t numBytes) override
void	free (AbstractBuffer *buffer) override
Page	requestFreePage (size_t pagesize, const bool isMetadata)
MgrType	getMgrType () override
std::string	getStringMgrType () override
std::string	printSlabs () override
void	clearSlabs () override
size_t	getMaxSize () override
size_t	getInUseSize () override
size_t	getAllocated () override
bool	isAllocationCapped () override
FileInfo *	getFileInfoForFileId (const int fileId)
void	init (const size_t num_reader_threads)
void	init (const std::string dataPathToConvertFrom)
void	copyPage (Page &srcPage, FileMgr *destFileMgr, Page &destPage, const size_t reservedHeaderSize, const size_t numBytes, const size_t offset)
void	requestFreePages (size_t npages, size_t pagesize, std::vector< Page > &pages, const bool isMetadata)
	Obtains free pages – creates new files if necessary – of the requested size. More...
void	getChunkMetadataVec (std::vector< std::pair< ChunkKey, ChunkMetadata >> &chunkMetadataVec) override
void	getChunkMetadataVecForKeyPrefix (std::vector< std::pair< ChunkKey, ChunkMetadata >> &chunkMetadataVec, const ChunkKey &keyPrefix) override
void	checkpoint () override
	Fsyncs data files, writes out epoch and fsyncs that. More...
void	checkpoint (const int db_id, const int tb_id) override
int	epoch ()
	Returns current value of epoch - should be one greater than recorded at last checkpoint. More...
size_t	getNumReaderThreads ()
	Returns number of threads defined by parameter num-reader-threads which should be used during initial load and consequent read of data. More...
FILE *	getFileForFileId (const int fileId)
	Returns FILE pointer associated with requested fileId. More...
size_t	getNumChunks () override
int	getDBVersion () const
	Index for looking up chunks. More...
bool	getDBConvert () const
void	createTopLevelMetadata ()
std::string	getFileMgrBasePath () const
void	closeRemovePhysical ()
void	free_page (std::pair< FileInfo *, int > &&page)
const std::pair< const int, const int >	get_fileMgrKey () const

Public Attributes
ChunkKeyToChunkMap	chunkIndex_

Private Member Functions
FileInfo *	createFile (const size_t pageSize, const size_t numPages)
	Adds a file to the file manager repository. More...
FileInfo *	openExistingFile (const std::string &path, const int fileId, const size_t pageSize, const size_t numPages, std::vector< HeaderInfo > &headerVec)
void	createEpochFile (const std::string &epochFileName)
void	openEpochFile (const std::string &epochFileName)
void	writeAndSyncEpochToDisk ()
void	createDBMetaFile (const std::string &DBMetaFileName)
bool	openDBMetaFile (const std::string &DBMetaFileName)
void	writeAndSyncDBMetaToDisk ()
void	setEpoch (int epoch)
void	processFileFutures (std::vector< std::future< std::vector< HeaderInfo >>> &file_futures, std::vector< HeaderInfo > &headerVec)

Private Attributes
GlobalFileMgr *	gfm_
std::pair< const int, const int >	fileMgrKey_
	Global FileMgr. More...
std::string	fileMgrBasePath_
std::vector< FileInfo * >	files_
PageSizeFileMMap	fileIndex_
	A vector of files accessible via a file identifier. More...
size_t	num_reader_threads_
	Maps page sizes to FileInfo objects. More...
size_t	defaultPageSize_
	number of threads used when loading data More...
unsigned	nextFileId_
int	epoch_
	the index of the next file id More...
FILE *	epochFile_
	the current epoch (time of last checkpoint) More...
int	db_version_
FILE *	DBMetaFile_
std::mutex	getPageMutex_
	pointer to DB level metadata More...
mapd_shared_mutex	chunkIndexMutex_
mapd_shared_mutex	files_rw_mutex_
mapd_shared_mutex	mutex_free_page
std::vector< std::pair< FileInfo *, int > >	free_pages

Friends
class	GlobalFileMgr

Detailed Description

Definition at line 85 of file FileMgr.h.

Constructor & Destructor Documentation

FileMgr() [1/3]

File_Namespace::FileMgr::FileMgr	(	const int	deviceId,
		GlobalFileMgr *	gfm,
		const std::pair< const int, const int >	fileMgrKey,
		const size_t	num_reader_threads = 0,
		const int	epoch = -1,
		const size_t	defaultPageSize = 2097152
	)

Constructor.

Definition at line 67 of file FileMgr.cpp.

References init().

    : AbstractBufferMgr(deviceId)
    , gfm_(gfm)
    , fileMgrKey_(fileMgrKey)
    , defaultPageSize_(defaultPageSize)
    , nextFileId_(0)
    , epoch_(epoch) {
  init(num_reader_threads);
}

Here is the call graph for this function:

FileMgr() [2/3]

File_Namespace::FileMgr::FileMgr	(	const int	deviceId,
		GlobalFileMgr *	gfm,
		const std::pair< const int, const int >	fileMgrKey,
		const bool	initOnly
	)

Definition at line 83 of file FileMgr.cpp.

References epochFile_, fileMgrBasePath_, fileMgrKey_, files_, File_Namespace::GlobalFileMgr::getBasePath(), gfm_, and to_string().

    : AbstractBufferMgr(deviceId)
    , gfm_(gfm)
    , fileMgrKey_(fileMgrKey)
    , defaultPageSize_(0)
    , nextFileId_(0)
    , epoch_(0) {
  const std::string fileMgrDirPrefix("table");
  const std::string FileMgrDirDelim("_");
  fileMgrBasePath_ = (gfm_->getBasePath() + fileMgrDirPrefix + FileMgrDirDelim +
                      std::to_string(fileMgrKey_.first) +                     // db_id
                      FileMgrDirDelim + std::to_string(fileMgrKey_.second));  // tb_id
  epochFile_ = nullptr;
  files_.clear();
}

Here is the call graph for this function:

FileMgr() [3/3]

File_Namespace::FileMgr::FileMgr	(	GlobalFileMgr *	gfm,
		const size_t	defaultPageSize,
		std::string	basePath
	)

Definition at line 102 of file FileMgr.cpp.

References init().

    : AbstractBufferMgr(0)
    , gfm_(gfm)
    , fileMgrKey_(0, 0)
    , fileMgrBasePath_(basePath)
    , defaultPageSize_(defaultPageSize)
    , nextFileId_(0)
    , epoch_(-1) {
  init(basePath);
}

Here is the call graph for this function:

~FileMgr()

File_Namespace::FileMgr::~FileMgr ( )

override

Destructor.

Definition at line 113 of file FileMgr.cpp.

References chunkIndex_, and files_.

                  {
  // checkpoint();
  // free memory used by FileInfo objects
  for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
    delete chunkIt->second;
  }
  for (auto file_info : files_) {
    delete file_info;
  }
}

Member Function Documentation

alloc()

AbstractBuffer * File_Namespace::FileMgr::alloc ( const size_t  numBytes = 0 )

override

Definition at line 790 of file FileMgr.cpp.

References logger::FATAL, and LOG.

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

checkpoint() [1/2]

void File_Namespace::FileMgr::checkpoint ( )

override

Fsyncs data files, writes out epoch and fsyncs that.

Definition at line 577 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, epoch_, logger::FATAL, files_, files_rw_mutex_, free_page(), free_pages, LOG, mutex_free_page, VLOG, and writeAndSyncEpochToDisk().

Referenced by File_Namespace::GlobalFileMgr::checkpoint().

                         {
  VLOG(2) << "Checkpointing epoch: " << epoch_;
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
  for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
    /*
    for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
        std::cout << *vecIt << ",";
    }
    cout << "Is dirty: " << chunkIt->second->isDirty_ << endl;
    */
    if (chunkIt->second->is_dirty_) {
      chunkIt->second->writeMetadata(epoch_);
      chunkIt->second->clearDirtyBits();
    }
  }
  chunkIndexWriteLock.unlock();

  mapd_shared_lock<mapd_shared_mutex> read_lock(files_rw_mutex_);
  for (auto fileIt = files_.begin(); fileIt != files_.end(); ++fileIt) {
    int status = (*fileIt)->syncToDisk();
    if (status != 0) {
      LOG(FATAL) << "Could not sync file to disk";
    }
  }

  writeAndSyncEpochToDisk();

  mapd_unique_lock<mapd_shared_mutex> freePagesWriteLock(mutex_free_page);
  for (auto& free_page : free_pages) {
    free_page.first->freePageDeferred(free_page.second);
  }
  free_pages.clear();
}

Here is the call graph for this function:

Here is the caller graph for this function:

checkpoint() [2/2]

void File_Namespace::FileMgr::checkpoint ( const int  db_id, const int  tb_id  )

inlineoverride

Definition at line 197 of file FileMgr.h.

References logger::FATAL, and LOG.

                                                             {
    LOG(FATAL) << "Operation not supported, api checkpoint() should be used instead";
  }

clearSlabs()

void File_Namespace::FileMgr::clearSlabs ( )

inlineoverride

Definition at line 148 of file FileMgr.h.

                                    { /* noop */
  }

closeRemovePhysical()

void File_Namespace::FileMgr::closeRemovePhysical

(

)

Definition at line 448 of file FileMgr.cpp.

References File_Namespace::close(), epochFile_, files_, getFileMgrBasePath(), and File_Namespace::renameForDelete().

Referenced by File_Namespace::GlobalFileMgr::removeTableRelatedDS().

                                  {
  for (auto file_info : files_) {
    if (file_info->f) {
      close(file_info->f);
      file_info->f = nullptr;
    }
  }

  if (epochFile_) {
    close(epochFile_);
    epochFile_ = nullptr;
  }

  /* rename for later deletion the directory containing table related data */
  File_Namespace::renameForDelete(getFileMgrBasePath());
}

Here is the call graph for this function:

Here is the caller graph for this function:

copyPage()

void File_Namespace::FileMgr::copyPage	(	Page &	srcPage,
		FileMgr *	destFileMgr,
		Page &	destPage,
		const size_t	reservedHeaderSize,
		const size_t	numBytes,
		const size_t	offset
	)

Definition at line 465 of file FileMgr.cpp.

References CHECK, defaultPageSize_, File_Namespace::Page::fileId, getFileInfoForFileId(), File_Namespace::Page::pageNum, File_Namespace::FileInfo::read(), and File_Namespace::FileInfo::write().

Referenced by init().

                                            {
  CHECK(offset + numBytes <= defaultPageSize_);
  FileInfo* srcFileInfo = getFileInfoForFileId(srcPage.fileId);
  FileInfo* destFileInfo = destFileMgr->getFileInfoForFileId(destPage.fileId);
  int8_t* buffer = new int8_t[numBytes];

  size_t bytesRead = srcFileInfo->read(
      srcPage.pageNum * defaultPageSize_ + offset + reservedHeaderSize, numBytes, buffer);
  CHECK(bytesRead == numBytes);
  size_t bytesWritten = destFileInfo->write(
      destPage.pageNum * defaultPageSize_ + offset + reservedHeaderSize,
      numBytes,
      buffer);
  CHECK(bytesWritten == numBytes);
  delete[] buffer;
}

Here is the call graph for this function:

Here is the caller graph for this function:

createBuffer()

AbstractBuffer * File_Namespace::FileMgr::createBuffer ( const ChunkKey &  key, size_t  pageSize = 0, const size_t  numBytes = 0  )

override

Creates a chunk with the specified key and page size.

Todo:

Make all accesses to chunkIndex_ thread-safe

Definition at line 611 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, defaultPageSize_, logger::FATAL, LOG, and showChunk().

Referenced by putBuffer().

                                                             {
  size_t actualPageSize = pageSize;
  if (actualPageSize == 0) {
    actualPageSize = defaultPageSize_;
  }
  // we will do this lazily and not allocate space for the Chunk (i.e.
  // FileBuffer yet)
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);

  if (chunkIndex_.find(key) != chunkIndex_.end()) {
    LOG(FATAL) << "Chunk already exists for key: " << showChunk(key);
  }
  chunkIndex_[key] = new FileBuffer(this, actualPageSize, key, numBytes);
  chunkIndexWriteLock.unlock();
  return (chunkIndex_[key]);
}

Here is the call graph for this function:

Here is the caller graph for this function:

createDBMetaFile()

void File_Namespace::FileMgr::createDBMetaFile ( const std::string &  DBMetaFileName )

private

Definition at line 536 of file FileMgr.cpp.

References File_Namespace::create(), DBMetaFile_, logger::FATAL, fileMgrBasePath_, getDBVersion(), LOG, and File_Namespace::write().

Referenced by createTopLevelMetadata().

                                                              {
  std::string DBMetaFilePath(fileMgrBasePath_ + "/" + DBMetaFileName);
  if (boost::filesystem::exists(DBMetaFilePath)) {
    LOG(FATAL) << "DB metadata file `" << DBMetaFilePath << "` already exists.";
  }
  DBMetaFile_ = create(DBMetaFilePath, sizeof(int));
  int db_ver = getDBVersion();
  write(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_ver);
  // LOG(INFO) << "DB metadata file has been created.";
}

Here is the call graph for this function:

Here is the caller graph for this function:

createEpochFile()

void File_Namespace::FileMgr::createEpochFile ( const std::string &  epochFileName )

private

Definition at line 487 of file FileMgr.cpp.

References File_Namespace::create(), epoch_, epochFile_, logger::FATAL, fileMgrBasePath_, LOG, and File_Namespace::write().

Referenced by init().

                                                            {
  std::string epochFilePath(fileMgrBasePath_ + "/" + epochFileName);
  if (boost::filesystem::exists(epochFilePath)) {
    LOG(FATAL) << "Epoch file `" << epochFilePath << "` already exists";
  }
  epochFile_ = create(epochFilePath, sizeof(int));
  // Write out current epoch to file - which if this
  // function is being called should be 0
  write(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
  epoch_++;
}

Here is the call graph for this function:

Here is the caller graph for this function:

createFile()

FileInfo * File_Namespace::FileMgr::createFile ( const size_t  pageSize, const size_t  numPages  )

private

Adds a file to the file manager repository.

This method will create a FileInfo object for the file being added, and it will create the corresponding file on physical disk with the indicated number of pages pre-allocated.

A pointer to the FileInfo object is returned, which itself has a file pointer (FILE*) and a file identifier (int fileId).

Parameters

fileName	The name given to the file in physical storage.
pageSize	The logical page size for the pages in the file.
numPages	The number of logical pages to initially allocate for the file.

Returns

FileInfo* A pointer to the FileInfo object of the added file.

Definition at line 887 of file FileMgr.cpp.

References CHECK, File_Namespace::create(), logger::FATAL, fileIndex_, fileMgrBasePath_, files_, files_rw_mutex_, LOG, and nextFileId_.

Referenced by requestFreePage(), and requestFreePages().

                                                                          {
  // check arguments
  if (pageSize == 0 || numPages == 0) {
    LOG(FATAL) << "File creation failed: pageSize and numPages must be greater than 0.";
  }

  // create the new file
  FILE* f = create(fileMgrBasePath_,
                   nextFileId_,
                   pageSize,
                   numPages);  // TM: not sure if I like naming scheme here - should be in
                               // separate namespace?
  CHECK(f);

  // instantiate a new FileInfo for the newly created file
  int fileId = nextFileId_++;
  FileInfo* fInfo =
      new FileInfo(this, fileId, f, pageSize, numPages, true);  // true means init file
  CHECK(fInfo);

  mapd_unique_lock<mapd_shared_mutex> write_lock(files_rw_mutex_);
  // update file manager data structures
  files_.push_back(fInfo);
  fileIndex_.insert(std::pair<size_t, int>(pageSize, fileId));

  CHECK(files_.back() == fInfo);  // postcondition
  return fInfo;
}

Here is the call graph for this function:

Here is the caller graph for this function:

createTopLevelMetadata()

void File_Namespace::FileMgr::createTopLevelMetadata

(

)

Definition at line 980 of file FileMgr.cpp.

References createDBMetaFile(), DB_META_FILENAME, db_version_, logger::FATAL, getDBVersion(), LOG, and openDBMetaFile().

Referenced by Data_Namespace::DataMgr::createTopLevelMetadata().

                                     {
  if (openDBMetaFile(DB_META_FILENAME)) {
    if (db_version_ > getDBVersion()) {
      LOG(FATAL) << "DB forward compatibility is not supported. Version of OmniSci "
                    "software used is older than the version of DB being read: "
                 << db_version_;
    }
  } else {
    createDBMetaFile(DB_META_FILENAME);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

deleteBuffer()

void File_Namespace::FileMgr::deleteBuffer ( const ChunkKey &  key, const bool  purge = true  )

override

Deletes the chunk with the specified key.

Definition at line 636 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, logger::FATAL, LOG, and showChunk().

                                                                {
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
  auto chunkIt = chunkIndex_.find(key);
  // ensure the Chunk exists
  if (chunkIt == chunkIndex_.end()) {
    LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
  }
  chunkIndexWriteLock.unlock();
  // chunkIt->second->writeMetadata(-1); // writes -1 as epoch - signifies deleted
  if (purge) {
    chunkIt->second->freePages();
  }
  //@todo need a way to represent delete in non purge case
  delete chunkIt->second;
  chunkIndex_.erase(chunkIt);
}

Here is the call graph for this function:

deleteBuffersWithPrefix()

void File_Namespace::FileMgr::deleteBuffersWithPrefix ( const ChunkKey &  keyPrefix, const bool  purge = true  )

override

Definition at line 653 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

Referenced by File_Namespace::GlobalFileMgr::deleteBuffersWithPrefix().

                                                                                 {
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
  auto chunkIt = chunkIndex_.lower_bound(keyPrefix);
  if (chunkIt == chunkIndex_.end()) {
    return;  // should we throw?
  }
  while (chunkIt != chunkIndex_.end() &&
         std::search(chunkIt->first.begin(),
                     chunkIt->first.begin() + keyPrefix.size(),
                     keyPrefix.begin(),
                     keyPrefix.end()) != chunkIt->first.begin() + keyPrefix.size()) {
    /*
    cout << "Freeing pages for chunk ";
    for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
        std::cout << *vecIt << ",";
    }
    cout << endl;
    */
    if (purge) {
      chunkIt->second->freePages();
    }
    //@todo need a way to represent delete in non purge case
    delete chunkIt->second;
    chunkIndex_.erase(chunkIt++);
  }
}

Here is the caller graph for this function:

epoch()

int File_Namespace::FileMgr::epoch ( )

inline

Returns current value of epoch - should be one greater than recorded at last checkpoint.

Definition at line 204 of file FileMgr.h.

Referenced by File_Namespace::FileBuffer::append(), File_Namespace::FileInfo::freePage(), File_Namespace::FileInfo::openExistingFile(), File_Namespace::FileBuffer::reserve(), setEpoch(), and File_Namespace::FileBuffer::write().

{ return epoch_; }

Here is the caller graph for this function:

fetchBuffer()

void File_Namespace::FileMgr::fetchBuffer ( const ChunkKey &  key, AbstractBuffer *  destBuffer, const size_t  numBytes  )

override

Definition at line 689 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, logger::FATAL, Data_Namespace::AbstractBuffer::getDeviceId(), Data_Namespace::AbstractBuffer::getMemoryPtr(), Data_Namespace::AbstractBuffer::getType(), Data_Namespace::AbstractBuffer::isDirty(), Data_Namespace::AbstractBuffer::isUpdated(), LOG, Data_Namespace::AbstractBuffer::read(), Data_Namespace::AbstractBuffer::reserve(), Data_Namespace::AbstractBuffer::setSize(), showChunk(), Data_Namespace::AbstractBuffer::size(), and Data_Namespace::AbstractBuffer::syncEncoder().

                                                 {
  // reads chunk specified by ChunkKey into AbstractBuffer provided by
  // destBuffer
  if (destBuffer->isDirty()) {
    LOG(FATAL)
        << "Aborting attempt to fetch a chunk marked dirty. Chunk inconsistency for key: "
        << showChunk(key);
  }
  mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
  auto chunkIt = chunkIndex_.find(key);
  if (chunkIt == chunkIndex_.end()) {
    LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
  }
  chunkIndexReadLock.unlock();

  AbstractBuffer* chunk = chunkIt->second;
  // ChunkSize is either specified in function call with numBytes or we
  // just look at pageSize * numPages in FileBuffer
  size_t chunkSize = numBytes == 0 ? chunk->size() : numBytes;
  if (numBytes > 0 && numBytes > chunk->size()) {
    LOG(FATAL) << "Chunk retrieved for key `" << showChunk(key) << "` is smaller ("
               << chunk->size() << ") than number of bytes requested (" << numBytes
               << ")";
  }
  destBuffer->reserve(chunkSize);
  // std::cout << "After reserve chunksize: " << chunkSize << std::endl;
  if (chunk->isUpdated()) {
    chunk->read(destBuffer->getMemoryPtr(),
                chunkSize,
                0,
                destBuffer->getType(),
                destBuffer->getDeviceId());
  } else {
    chunk->read(destBuffer->getMemoryPtr() + destBuffer->size(),
                chunkSize - destBuffer->size(),
                destBuffer->size(),
                destBuffer->getType(),
                destBuffer->getDeviceId());
  }
  destBuffer->setSize(chunkSize);
  destBuffer->syncEncoder(chunk);
}

Here is the call graph for this function:

free()

void File_Namespace::FileMgr::free ( AbstractBuffer *  buffer )

override

Definition at line 795 of file FileMgr.cpp.

References logger::FATAL, and LOG.

                                         {
  LOG(FATAL) << "Operation not supported";
}

free_page()

void File_Namespace::FileMgr::free_page

(

std::pair< FileInfo *, int > &&

page

)

Definition at line 997 of file FileMgr.cpp.

References free_pages, and mutex_free_page.

Referenced by checkpoint(), and File_Namespace::FileInfo::freePage().

                                                      {
  std::unique_lock<mapd_shared_mutex> lock(mutex_free_page);
  free_pages.push_back(page);
}

Here is the caller graph for this function:

get_fileMgrKey()

const std::pair<const int, const int> File_Namespace::FileMgr::get_fileMgrKey ( ) const

inline

Definition at line 234 of file FileMgr.h.

Referenced by File_Namespace::FileInfo::openExistingFile().

{ return fileMgrKey_; }

Here is the caller graph for this function:

getAllocated()

size_t File_Namespace::FileMgr::getAllocated ( )

inlineoverride

Definition at line 152 of file FileMgr.h.

{ return 0; }

getBuffer()

AbstractBuffer * File_Namespace::FileMgr::getBuffer ( const ChunkKey &  key, const size_t  numBytes = 0  )

override

Returns the a pointer to the chunk with the specified key.

Definition at line 680 of file FileMgr.cpp.

References chunkIndex_, chunkIndexMutex_, logger::FATAL, LOG, and showChunk().

                                                                             {
  mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
  auto chunkIt = chunkIndex_.find(key);
  if (chunkIt == chunkIndex_.end()) {
    LOG(FATAL) << "Chunk does not exist for key: " << showChunk(key);
  }
  return chunkIt->second;
}

Here is the call graph for this function:

getChunkMetadataVec()

void File_Namespace::FileMgr::getChunkMetadataVec ( std::vector< std::pair< ChunkKey, ChunkMetadata >> &  chunkMetadataVec )

override

Definition at line 929 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                                                   {
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
  chunkMetadataVec.reserve(chunkIndex_.size());
  for (auto chunkIt = chunkIndex_.begin(); chunkIt != chunkIndex_.end(); ++chunkIt) {
    if (chunkIt->second->has_encoder) {
      ChunkMetadata chunkMetadata;
      chunkIt->second->encoder->getMetadata(chunkMetadata);
      chunkMetadataVec.emplace_back(chunkIt->first, chunkMetadata);
    }
  }
}

getChunkMetadataVecForKeyPrefix()

void File_Namespace::FileMgr::getChunkMetadataVecForKeyPrefix ( std::vector< std::pair< ChunkKey, ChunkMetadata >> &  chunkMetadataVec, const ChunkKey &  keyPrefix  )

override

Definition at line 942 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                               {
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(
      chunkIndexMutex_);  // is this guarding the right structure?  it look slike we oly
                          // read here for chunk
  auto chunkIt = chunkIndex_.lower_bound(keyPrefix);
  if (chunkIt == chunkIndex_.end()) {
    return;  // throw?
  }
  while (chunkIt != chunkIndex_.end() &&
         std::search(chunkIt->first.begin(),
                     chunkIt->first.begin() + keyPrefix.size(),
                     keyPrefix.begin(),
                     keyPrefix.end()) != chunkIt->first.begin() + keyPrefix.size()) {
    /*
    for (auto vecIt = chunkIt->first.begin(); vecIt != chunkIt->first.end(); ++vecIt) {
        std::cout << *vecIt << ",";
    }
    cout << endl;
    */
    if (chunkIt->second->has_encoder) {
      ChunkMetadata chunkMetadata;
      chunkIt->second->encoder->getMetadata(chunkMetadata);
      chunkMetadataVec.emplace_back(chunkIt->first, chunkMetadata);
    }
    chunkIt++;
  }
}

getDBConvert()

bool File_Namespace::FileMgr::getDBConvert

(

)

const

Definition at line 976 of file FileMgr.cpp.

References File_Namespace::GlobalFileMgr::getDBConvert(), and gfm_.

                                 {
  return gfm_->getDBConvert();
}

Here is the call graph for this function:

getDBVersion()

int File_Namespace::FileMgr::getDBVersion

(

)

const

Index for looking up chunks.

Definition at line 972 of file FileMgr.cpp.

References File_Namespace::GlobalFileMgr::getDBVersion(), and gfm_.

Referenced by createDBMetaFile(), createTopLevelMetadata(), and writeAndSyncDBMetaToDisk().

                                {
  return gfm_->getDBVersion();
}

Here is the call graph for this function:

Here is the caller graph for this function:

getFileForFileId()

FILE * File_Namespace::FileMgr::getFileForFileId

(

const int

fileId

)

Returns FILE pointer associated with requested fileId.

See also

FileBuffer

Definition at line 916 of file FileMgr.cpp.

References CHECK, and files_.

Referenced by File_Namespace::FileBuffer::readMetadata(), and File_Namespace::FileBuffer::writeMetadata().

                                                {
  CHECK(fileId >= 0 && static_cast<size_t>(fileId) < files_.size());
  return files_[fileId]->f;
}

Here is the caller graph for this function:

getFileInfoForFileId()

FileInfo* File_Namespace::FileMgr::getFileInfoForFileId ( const int  fileId )

inline

Definition at line 155 of file FileMgr.h.

References logger::init().

Referenced by File_Namespace::FileBuffer::append(), File_Namespace::FileBuffer::copyPage(), copyPage(), File_Namespace::FileBuffer::freePages(), File_Namespace::readForThread(), File_Namespace::FileBuffer::write(), and File_Namespace::FileBuffer::writeHeader().

{ return files_[fileId]; }

Here is the call graph for this function:

Here is the caller graph for this function:

getFileMgrBasePath()

std::string File_Namespace::FileMgr::getFileMgrBasePath ( ) const

inline

Definition at line 230 of file FileMgr.h.

Referenced by closeRemovePhysical().

{ return fileMgrBasePath_; }

Here is the caller graph for this function:

getInUseSize()

size_t File_Namespace::FileMgr::getInUseSize ( )

inlineoverride

Definition at line 151 of file FileMgr.h.

{ return 0; }

getMaxSize()

size_t File_Namespace::FileMgr::getMaxSize ( )

inlineoverride

Definition at line 150 of file FileMgr.h.

{ return 0; }

getMgrType()

MgrType File_Namespace::FileMgr::getMgrType ( )

inlineoverride

Definition at line 145 of file FileMgr.h.

{ return FILE_MGR; };

getNumChunks()

size_t File_Namespace::FileMgr::getNumChunks ( )

inlineoverride

Definition at line 221 of file FileMgr.h.

                                        {
    // @todo should be locked - but this is more for testing now
    return chunkIndex_.size();
  }

getNumReaderThreads()

size_t File_Namespace::FileMgr::getNumReaderThreads ( )

inline

Returns number of threads defined by parameter num-reader-threads which should be used during initial load and consequent read of data.

Definition at line 210 of file FileMgr.h.

Referenced by File_Namespace::FileBuffer::read().

{ return num_reader_threads_; }

Here is the caller graph for this function:

getStringMgrType()

std::string File_Namespace::FileMgr::getStringMgrType ( )

inlineoverride

Definition at line 146 of file FileMgr.h.

{ return ToString(FILE_MGR); }

init() [1/2]

void File_Namespace::FileMgr::init

(

const size_t

num_reader_threads

)

Definition at line 124 of file FileMgr.cpp.

References CHECK_EQ, chunkIndex_, createEpochFile(), epoch_, EPOCH_FILENAME, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, fileMgrKey_, File_Namespace::fileSize(), File_Namespace::GlobalFileMgr::getBasePath(), gfm_, File_Namespace::headerCompare(), logger::INFO, LOG, MAPD_FILE_EXT, nextFileId_, num_reader_threads_, openEpochFile(), openExistingFile(), processFileFutures(), timer_start(), timer_stop(), to_string(), and VLOG.

Referenced by FileMgr().

                                                  {
  // if epoch = -1 this means open from epoch file
  const std::string fileMgrDirPrefix("table");
  const std::string FileMgrDirDelim("_");
  fileMgrBasePath_ = (gfm_->getBasePath() + fileMgrDirPrefix + FileMgrDirDelim +
                      std::to_string(fileMgrKey_.first) +                     // db_id
                      FileMgrDirDelim + std::to_string(fileMgrKey_.second));  // tb_id
  boost::filesystem::path path(fileMgrBasePath_);
  if (boost::filesystem::exists(path)) {
    if (!boost::filesystem::is_directory(path)) {
      LOG(FATAL) << "Specified path '" << fileMgrBasePath_
                 << "' for table data is not a directory.";
    }
    if (epoch_ != -1) {  // if opening at previous epoch
      int epochCopy = epoch_;
      openEpochFile(EPOCH_FILENAME);
      epoch_ = epochCopy;
    } else {
      openEpochFile(EPOCH_FILENAME);
    }

    auto clock_begin = timer_start();

    boost::filesystem::directory_iterator
        endItr;  // default construction yields past-the-end
    int maxFileId = -1;
    int fileCount = 0;
    int threadCount = std::thread::hardware_concurrency();
    std::vector<HeaderInfo> headerVec;
    std::vector<std::future<std::vector<HeaderInfo>>> file_futures;
    for (boost::filesystem::directory_iterator fileIt(path); fileIt != endItr; ++fileIt) {
      if (boost::filesystem::is_regular_file(fileIt->status())) {
        // note that boost::filesystem leaves preceding dot on
        // extension - hence MAPD_FILE_EXT is ".mapd"
        std::string extension(fileIt->path().extension().string());

        if (extension == MAPD_FILE_EXT) {
          std::string fileStem(fileIt->path().stem().string());
          // remove trailing dot if any
          if (fileStem.size() > 0 && fileStem.back() == '.') {
            fileStem = fileStem.substr(0, fileStem.size() - 1);
          }
          size_t dotPos = fileStem.find_last_of(".");  // should only be one
          if (dotPos == std::string::npos) {
            LOG(FATAL) << "File `" << fileIt->path()
                       << "` does not carry page size information in the filename.";
          }
          int fileId = boost::lexical_cast<int>(fileStem.substr(0, dotPos));
          if (fileId > maxFileId) {
            maxFileId = fileId;
          }
          size_t pageSize =
              boost::lexical_cast<size_t>(fileStem.substr(dotPos + 1, fileStem.size()));
          std::string filePath(fileIt->path().string());
          size_t fileSize = boost::filesystem::file_size(filePath);
          CHECK_EQ(fileSize % pageSize, size_t(0));  // should be no partial pages
          size_t numPages = fileSize / pageSize;

          VLOG(4) << "File id: " << fileId << " Page size: " << pageSize
                  << " Num pages: " << numPages;

          file_futures.emplace_back(std::async(
              std::launch::async, [filePath, fileId, pageSize, numPages, this] {
                std::vector<HeaderInfo> tempHeaderVec;
                openExistingFile(filePath, fileId, pageSize, numPages, tempHeaderVec);
                return tempHeaderVec;
              }));
          fileCount++;
          if (fileCount % threadCount == 0) {
            processFileFutures(file_futures, headerVec);
          }
        }
      }
    }

    if (file_futures.size() > 0) {
      processFileFutures(file_futures, headerVec);
    }
    int64_t queue_time_ms = timer_stop(clock_begin);

    LOG(INFO) << "Completed Reading table's file metadata, Elapsed time : "
              << queue_time_ms << "ms Epoch: " << epoch_ << " files read: " << fileCount
              << " table location: '" << fileMgrBasePath_ << "'";

    /* Sort headerVec so that all HeaderInfos
     * from a chunk will be grouped together
     * and in order of increasing PageId
     * - Version Epoch */

    std::sort(headerVec.begin(), headerVec.end(), headerCompare);

    /* Goal of next section is to find sequences in the
     * sorted headerVec of the same ChunkId, which we
     * can then initiate a FileBuffer with */

    VLOG(4) << "Number of Headers in Vector: " << headerVec.size();
    if (headerVec.size() > 0) {
      ChunkKey lastChunkKey = headerVec.begin()->chunkKey;
      auto startIt = headerVec.begin();

      for (auto headerIt = headerVec.begin() + 1; headerIt != headerVec.end();
           ++headerIt) {
        // for (auto chunkIt = headerIt->chunkKey.begin(); chunkIt !=
        // headerIt->chunkKey.end(); ++chunkIt) {
        //    std::cout << *chunkIt << " ";
        //}

        if (headerIt->chunkKey != lastChunkKey) {
          chunkIndex_[lastChunkKey] =
              new FileBuffer(this, /*pageSize,*/ lastChunkKey, startIt, headerIt);
          /*
          if (startIt->versionEpoch != -1) {
              cout << "not skipping bc version != -1" << endl;
              // -1 means that chunk was deleted
              // lets not read it in
              chunkIndex_[lastChunkKey] = new FileBuffer
          (this,/lastChunkKey,startIt,headerIt);

          }
          else {
              cout << "Skipping bc version == -1" << endl;
          }
          */
          lastChunkKey = headerIt->chunkKey;
          startIt = headerIt;
        }
      }
      // now need to insert last Chunk
      // size_t pageSize = files_[startIt->page.fileId]->pageSize;
      // cout << "Inserting last chunk" << endl;
      // if (startIt->versionEpoch != -1) {
      chunkIndex_[lastChunkKey] =
          new FileBuffer(this, /*pageSize,*/ lastChunkKey, startIt, headerVec.end());
      //}
    }
    nextFileId_ = maxFileId + 1;
    // std::cout << "next file id: " << nextFileId_ << std::endl;
  } else {
    if (!boost::filesystem::create_directory(path)) {
      LOG(FATAL) << "Could not create data directory: " << path;
    }
    createEpochFile(EPOCH_FILENAME);
  }

  /* define number of reader threads to be used */
  size_t num_hardware_based_threads =
      std::thread::hardware_concurrency();  // # of threads is based on # of cores on the
                                            // host
  if (num_reader_threads == 0) {            // # of threads has not been defined by user
    num_reader_threads_ = num_hardware_based_threads;
  } else {
    if (num_reader_threads > num_hardware_based_threads) {
      num_reader_threads_ = num_hardware_based_threads;
    } else {
      num_reader_threads_ = num_reader_threads;
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

init() [2/2]

void File_Namespace::FileMgr::init

(

const std::string

dataPathToConvertFrom

)

Definition at line 297 of file FileMgr.cpp.

References CHECK, chunkIndex_, copyPage(), epoch_, EPOCH_FILENAME, File_Namespace::MultiPage::epochs, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), File_Namespace::fileSize(), File_Namespace::GlobalFileMgr::getFileMgr(), File_Namespace::FileBuffer::getMultiPage(), gfm_, File_Namespace::headerCompare(), LOG, MAPD_FILE_EXT, File_Namespace::FileBuffer::multiPages_, nextFileId_, openEpochFile(), openExistingFile(), File_Namespace::FileBuffer::pageDataSize(), File_Namespace::FileBuffer::pageSize(), processFileFutures(), requestFreePage(), File_Namespace::FileBuffer::reservedHeaderSize(), Data_Namespace::AbstractBuffer::setDirty(), Data_Namespace::AbstractBuffer::setSize(), File_Namespace::FileBuffer::size(), Data_Namespace::AbstractBuffer::syncEncoder(), and File_Namespace::FileBuffer::writeHeader().

                                                        {
  int converted_data_epoch = 0;
  boost::filesystem::path path(dataPathToConvertFrom);
  if (boost::filesystem::exists(path)) {
    if (!boost::filesystem::is_directory(path)) {
      LOG(FATAL) << "Specified path `" << path << "` is not a directory.";
    }

    if (epoch_ != -1) {  // if opening at previous epoch
      int epochCopy = epoch_;
      openEpochFile(EPOCH_FILENAME);
      epoch_ = epochCopy;
    } else {
      openEpochFile(EPOCH_FILENAME);
    }

    boost::filesystem::directory_iterator
        endItr;  // default construction yields past-the-end
    int maxFileId = -1;
    int fileCount = 0;
    int threadCount = std::thread::hardware_concurrency();
    std::vector<HeaderInfo> headerVec;
    std::vector<std::future<std::vector<HeaderInfo>>> file_futures;
    for (boost::filesystem::directory_iterator fileIt(path); fileIt != endItr; ++fileIt) {
      if (boost::filesystem::is_regular_file(fileIt->status())) {
        // note that boost::filesystem leaves preceding dot on
        // extension - hence MAPD_FILE_EXT is ".mapd"
        std::string extension(fileIt->path().extension().string());

        if (extension == MAPD_FILE_EXT) {
          std::string fileStem(fileIt->path().stem().string());
          // remove trailing dot if any
          if (fileStem.size() > 0 && fileStem.back() == '.') {
            fileStem = fileStem.substr(0, fileStem.size() - 1);
          }
          size_t dotPos = fileStem.find_last_of(".");  // should only be one
          if (dotPos == std::string::npos) {
            LOG(FATAL) << "File `" + fileIt->path().string() +
                              "` does not carry page size information in the filename.";
          }
          int fileId = boost::lexical_cast<int>(fileStem.substr(0, dotPos));
          if (fileId > maxFileId) {
            maxFileId = fileId;
          }
          size_t pageSize =
              boost::lexical_cast<size_t>(fileStem.substr(dotPos + 1, fileStem.size()));
          std::string filePath(fileIt->path().string());
          size_t fileSize = boost::filesystem::file_size(filePath);
          CHECK(fileSize % pageSize == 0);  // should be no partial pages
          size_t numPages = fileSize / pageSize;

          file_futures.emplace_back(std::async(
              std::launch::async, [filePath, fileId, pageSize, numPages, this] {
                std::vector<HeaderInfo> tempHeaderVec;
                openExistingFile(filePath, fileId, pageSize, numPages, tempHeaderVec);
                return tempHeaderVec;
              }));
          fileCount++;
          if (fileCount % threadCount) {
            processFileFutures(file_futures, headerVec);
          }
        }
      }
    }

    if (file_futures.size() > 0) {
      processFileFutures(file_futures, headerVec);
    }

    /* Sort headerVec so that all HeaderInfos
     * from a chunk will be grouped together
     * and in order of increasing PageId
     * - Version Epoch */

    std::sort(headerVec.begin(), headerVec.end(), headerCompare);

    /* Goal of next section is to find sequences in the
     * sorted headerVec of the same ChunkId, which we
     * can then initiate a FileBuffer with */

    if (headerVec.size() > 0) {
      ChunkKey lastChunkKey = headerVec.begin()->chunkKey;
      auto startIt = headerVec.begin();

      for (auto headerIt = headerVec.begin() + 1; headerIt != headerVec.end();
           ++headerIt) {
        if (headerIt->chunkKey != lastChunkKey) {
          FileMgr* c_fm_ = gfm_->getFileMgr(lastChunkKey);
          FileBuffer* srcBuf = new FileBuffer(this, lastChunkKey, startIt, headerIt);
          chunkIndex_[lastChunkKey] = srcBuf;
          FileBuffer* destBuf = new FileBuffer(c_fm_, srcBuf->pageSize(), lastChunkKey);
          c_fm_->chunkIndex_[lastChunkKey] = destBuf;
          destBuf->syncEncoder(srcBuf);
          destBuf->setSize(srcBuf->size());
          destBuf->setDirty();  // this needs to be set to force writing out metadata
                                // files from "checkpoint()" call

          size_t totalNumPages = srcBuf->getMultiPage().size();
          for (size_t pageNum = 0; pageNum < totalNumPages; pageNum++) {
            Page srcPage = srcBuf->getMultiPage()[pageNum].current();
            Page destPage = c_fm_->requestFreePage(
                srcBuf->pageSize(),
                false);  // may modify and use api "FileBuffer::addNewMultiPage" instead
            MultiPage multiPage(srcBuf->pageSize());
            multiPage.epochs.push_back(converted_data_epoch);
            multiPage.pageVersions.push_back(destPage);
            destBuf->multiPages_.push_back(multiPage);
            size_t reservedHeaderSize = srcBuf->reservedHeaderSize();
            copyPage(
                srcPage, c_fm_, destPage, reservedHeaderSize, srcBuf->pageDataSize(), 0);
            destBuf->writeHeader(destPage, pageNum, converted_data_epoch, false);
          }
          lastChunkKey = headerIt->chunkKey;
          startIt = headerIt;
        }
      }

      // now need to insert last Chunk
      FileMgr* c_fm_ = gfm_->getFileMgr(lastChunkKey);
      FileBuffer* srcBuf = new FileBuffer(this, lastChunkKey, startIt, headerVec.end());
      chunkIndex_[lastChunkKey] = srcBuf;
      FileBuffer* destBuf = new FileBuffer(c_fm_, srcBuf->pageSize(), lastChunkKey);
      c_fm_->chunkIndex_[lastChunkKey] = destBuf;
      destBuf->syncEncoder(srcBuf);
      destBuf->setSize(srcBuf->size());
      destBuf->setDirty();  // this needs to be set to write out metadata file from the
                            // "checkpoint()" call

      size_t totalNumPages = srcBuf->getMultiPage().size();
      for (size_t pageNum = 0; pageNum < totalNumPages; pageNum++) {
        Page srcPage = srcBuf->getMultiPage()[pageNum].current();
        Page destPage = c_fm_->requestFreePage(
            srcBuf->pageSize(),
            false);  // may modify and use api "FileBuffer::addNewMultiPage" instead
        MultiPage multiPage(srcBuf->pageSize());
        multiPage.epochs.push_back(converted_data_epoch);
        multiPage.pageVersions.push_back(destPage);
        destBuf->multiPages_.push_back(multiPage);
        size_t reservedHeaderSize = srcBuf->reservedHeaderSize();
        copyPage(srcPage, c_fm_, destPage, reservedHeaderSize, srcBuf->pageDataSize(), 0);
        destBuf->writeHeader(destPage, pageNum, converted_data_epoch, false);
      }
    }
    nextFileId_ = maxFileId + 1;
  } else {
    if (!boost::filesystem::create_directory(path)) {
      LOG(FATAL) << "Specified path does not exist: " << path;
    }
  }
}

Here is the call graph for this function:

isAllocationCapped()

bool File_Namespace::FileMgr::isAllocationCapped ( )

inlineoverride

Definition at line 153 of file FileMgr.h.

{ return false; }

isBufferOnDevice()

bool File_Namespace::FileMgr::isBufferOnDevice ( const ChunkKey &  key )

override

Definition at line 631 of file FileMgr.cpp.

References chunkIndex_, and chunkIndexMutex_.

                                                  {
  mapd_shared_lock<mapd_shared_mutex> chunkIndexReadLock(chunkIndexMutex_);
  return chunkIndex_.find(key) != chunkIndex_.end();
}

openDBMetaFile()

bool File_Namespace::FileMgr::openDBMetaFile ( const std::string &  DBMetaFileName )

private

Definition at line 547 of file FileMgr.cpp.

References db_version_, DBMetaFile_, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, File_Namespace::open(), and File_Namespace::read().

Referenced by createTopLevelMetadata().

                                                            {
  std::string DBMetaFilePath(fileMgrBasePath_ + "/" + DBMetaFileName);

  if (!boost::filesystem::exists(DBMetaFilePath)) {
    // LOG(INFO) << "DB metadata file does not exist, one will be created.";
    return false;
  }
  if (!boost::filesystem::is_regular_file(DBMetaFilePath)) {
    // LOG(INFO) << "DB metadata file is not a regular file, one will be created.";
    return false;
  }
  if (boost::filesystem::file_size(DBMetaFilePath) < 4) {
    // LOG(INFO) << "DB metadata file is not sized properly, one will be created.";
    return false;
  }
  DBMetaFile_ = open(DBMetaFilePath);
  read(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_version_);

  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

openEpochFile()

void File_Namespace::FileMgr::openEpochFile ( const std::string &  epochFileName )

private

Definition at line 499 of file FileMgr.cpp.

References epoch_, epochFile_, logger::FATAL, anonymous_namespace{StringDictionary.cpp}::file_size(), fileMgrBasePath_, LOG, File_Namespace::open(), and File_Namespace::read().

Referenced by init().

                                                          {
  std::string epochFilePath(fileMgrBasePath_ + "/" + epochFileName);
  if (!boost::filesystem::exists(epochFilePath)) {
    LOG(FATAL) << "Epoch file `" << epochFilePath << "` does not exist";
  }
  if (!boost::filesystem::is_regular_file(epochFilePath)) {
    LOG(FATAL) << "Epoch file `" << epochFilePath << "` is not a regular file";
  }
  if (boost::filesystem::file_size(epochFilePath) < 4) {
    LOG(FATAL) << "Epoch file `" << epochFilePath
               << "` is not sized properly (current size: "
               << boost::filesystem::file_size(epochFilePath) << ", expected size: 4)";
  }
  epochFile_ = open(epochFilePath);
  read(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
  // std::cout << "Epoch after open file: " << epoch_ << std::endl;
  epoch_++;  // we are in new epoch from last checkpoint
}

Here is the call graph for this function:

Here is the caller graph for this function:

openExistingFile()

FileInfo * File_Namespace::FileMgr::openExistingFile ( const std::string &  path, const int  fileId, const size_t  pageSize, const size_t  numPages, std::vector< HeaderInfo > &  headerVec  )

private

Definition at line 868 of file FileMgr.cpp.

References epoch_, fileIndex_, files_, files_rw_mutex_, File_Namespace::open(), and File_Namespace::FileInfo::openExistingFile().

Referenced by init().

                                                                      {
  FILE* f = open(path);
  FileInfo* fInfo = new FileInfo(
      this, fileId, f, pageSize, numPages, false);  // false means don't init file

  fInfo->openExistingFile(headerVec, epoch_);
  mapd_unique_lock<mapd_shared_mutex> write_lock(files_rw_mutex_);
  if (fileId >= static_cast<int>(files_.size())) {
    files_.resize(fileId + 1);
  }
  files_[fileId] = fInfo;
  fileIndex_.insert(std::pair<size_t, int>(pageSize, fileId));
  return fInfo;
}

Here is the call graph for this function:

Here is the caller graph for this function:

printSlabs()

std::string File_Namespace::FileMgr::printSlabs ( )

inlineoverride

Definition at line 147 of file FileMgr.h.

{ return "Not Implemented"; }

processFileFutures()

void File_Namespace::FileMgr::processFileFutures ( std::vector< std::future< std::vector< HeaderInfo >>> &  file_futures, std::vector< HeaderInfo > &  headerVec  )

private

Definition at line 283 of file FileMgr.cpp.

Referenced by init().

                                      {
  for (auto& file_future : file_futures) {
    file_future.wait();
  }
  // concatenate the vectors after thread completes
  for (auto& file_future : file_futures) {
    auto tempHeaderVec = file_future.get();
    headerVec.insert(headerVec.end(), tempHeaderVec.begin(), tempHeaderVec.end());
  }
  file_futures.clear();
}

Here is the caller graph for this function:

putBuffer()

AbstractBuffer * File_Namespace::FileMgr::putBuffer ( const ChunkKey &  key, AbstractBuffer *  d, const size_t  numBytes = 0  )

override

Puts the contents of d into the Chunk with the given key.

Parameters

key	- Unique identifier for a Chunk.
d	- An object representing the source data for the Chunk.

Returns

AbstractBuffer*

Definition at line 734 of file FileMgr.cpp.

References Data_Namespace::AbstractBuffer::append(), CHECK_LT, chunkIndex_, chunkIndexMutex_, Data_Namespace::AbstractBuffer::clearDirtyBits(), createBuffer(), defaultPageSize_, logger::FATAL, Data_Namespace::AbstractBuffer::getDeviceId(), Data_Namespace::AbstractBuffer::getMemoryPtr(), Data_Namespace::AbstractBuffer::getType(), Data_Namespace::AbstractBuffer::isAppended(), Data_Namespace::AbstractBuffer::isDirty(), Data_Namespace::AbstractBuffer::isUpdated(), LOG, Data_Namespace::AbstractBuffer::setSize(), showChunk(), Data_Namespace::AbstractBuffer::size(), Data_Namespace::AbstractBuffer::syncEncoder(), and Data_Namespace::AbstractBuffer::write().

                                                          {
  // obtain a pointer to the Chunk
  mapd_unique_lock<mapd_shared_mutex> chunkIndexWriteLock(chunkIndexMutex_);
  auto chunkIt = chunkIndex_.find(key);
  AbstractBuffer* chunk;
  if (chunkIt == chunkIndex_.end()) {
    chunk = createBuffer(key, defaultPageSize_);
  } else {
    chunk = chunkIt->second;
  }
  chunkIndexWriteLock.unlock();
  size_t oldChunkSize = chunk->size();
  // write the buffer's data to the Chunk
  // size_t newChunkSize = numBytes == 0 ? srcBuffer->size() : numBytes;
  size_t newChunkSize = numBytes == 0 ? srcBuffer->size() : numBytes;
  if (chunk->isDirty()) {
    // multiple appends are allowed,
    // but only single update is allowed
    if (srcBuffer->isUpdated() && chunk->isUpdated()) {
      LOG(FATAL) << "Aborting attempt to write a chunk marked dirty. Chunk inconsistency "
                    "for key: "
                 << showChunk(key);
    }
  }
  if (srcBuffer->isUpdated()) {
    // chunk size is not changed when fixed rows are updated or are marked as deleted.
    // but when rows are vacuumed or varlen rows are updated (new rows are appended),
    // chunk size will change. For vacuum, checkpoint should sync size from cpu to disk.
    // For varlen update, it takes another route via fragmenter using disk-level buffer.
    if (0 == numBytes && !chunk->isDirty()) {
      chunk->setSize(newChunkSize);
    }
    //@todo use dirty flags to only flush pages of chunk that need to
    // be flushed
    chunk->write((int8_t*)srcBuffer->getMemoryPtr(),
                 newChunkSize,
                 0,
                 srcBuffer->getType(),
                 srcBuffer->getDeviceId());
  } else if (srcBuffer->isAppended()) {
    CHECK_LT(oldChunkSize, newChunkSize);
    chunk->append((int8_t*)srcBuffer->getMemoryPtr() + oldChunkSize,
                  newChunkSize - oldChunkSize,
                  srcBuffer->getType(),
                  srcBuffer->getDeviceId());
  }
  // chunk->clearDirtyBits(); // Hack: because write and append will set dirty bits
  //@todo commenting out line above will make sure this metadata is set
  // but will trigger error on fetch chunk
  srcBuffer->clearDirtyBits();
  chunk->syncEncoder(srcBuffer);
  return chunk;
}

Here is the call graph for this function:

requestFreePage()

Page File_Namespace::FileMgr::requestFreePage	(	size_t	pagesize,
		const bool	isMetadata
	)

Definition at line 799 of file FileMgr.cpp.

References CHECK, createFile(), File_Namespace::FileInfo::fileId, fileIndex_, files_, File_Namespace::FileInfo::getFreePage(), getPageMutex_, MAX_FILE_N_METADATA_PAGES, and MAX_FILE_N_PAGES.

Referenced by File_Namespace::FileBuffer::addNewMultiPage(), init(), File_Namespace::FileBuffer::write(), and File_Namespace::FileBuffer::writeMetadata().

                                                                    {
  std::lock_guard<std::mutex> lock(getPageMutex_);

  auto candidateFiles = fileIndex_.equal_range(pageSize);
  int pageNum = -1;
  for (auto fileIt = candidateFiles.first; fileIt != candidateFiles.second; ++fileIt) {
    FileInfo* fileInfo = files_[fileIt->second];
    pageNum = fileInfo->getFreePage();
    if (pageNum != -1) {
      return (Page(fileInfo->fileId, pageNum));
    }
  }
  // if here then we need to add a file
  FileInfo* fileInfo;
  if (isMetadata) {
    fileInfo = createFile(pageSize, MAX_FILE_N_METADATA_PAGES);
  } else {
    fileInfo = createFile(pageSize, MAX_FILE_N_PAGES);
  }
  pageNum = fileInfo->getFreePage();
  CHECK(pageNum != -1);
  return (Page(fileInfo->fileId, pageNum));
}

Here is the call graph for this function:

Here is the caller graph for this function:

requestFreePages()

void File_Namespace::FileMgr::requestFreePages	(	size_t	npages,
		size_t	pagesize,
		std::vector< Page > &	pages,
		const bool	isMetadata
	)

Obtains free pages – creates new files if necessary – of the requested size.

Given a page size and number of pages, this method updates the vector "pages" to include free pages of the requested size. These pages are immediately removed from the free list of the affected file(s). If there are not enough pages available among current files, new files are created and their pages are included in the vector.

Parameters

npages	The number of free pages requested
pagesize	The size of each requested page
pages	A vector containing the free pages obtained by this method

Definition at line 823 of file FileMgr.cpp.

References CHECK, createFile(), File_Namespace::FileInfo::fileId, fileIndex_, files_, File_Namespace::FileInfo::getFreePage(), getPageMutex_, MAX_FILE_N_METADATA_PAGES, and MAX_FILE_N_PAGES.

                                                      {
  // not used currently
  // @todo add method to FileInfo to get more than one page
  std::lock_guard<std::mutex> lock(getPageMutex_);
  auto candidateFiles = fileIndex_.equal_range(pageSize);
  size_t numPagesNeeded = numPagesRequested;
  for (auto fileIt = candidateFiles.first; fileIt != candidateFiles.second; ++fileIt) {
    FileInfo* fileInfo = files_[fileIt->second];
    int pageNum;
    do {
      pageNum = fileInfo->getFreePage();
      if (pageNum != -1) {
        pages.emplace_back(fileInfo->fileId, pageNum);
        numPagesNeeded--;
      }
    } while (pageNum != -1 && numPagesNeeded > 0);
    if (numPagesNeeded == 0) {
      break;
    }
  }
  while (numPagesNeeded > 0) {
    FileInfo* fileInfo;
    if (isMetadata) {
      fileInfo = createFile(pageSize, MAX_FILE_N_METADATA_PAGES);
    } else {
      fileInfo = createFile(pageSize, MAX_FILE_N_PAGES);
    }
    int pageNum;
    do {
      pageNum = fileInfo->getFreePage();
      if (pageNum != -1) {
        pages.emplace_back(fileInfo->fileId, pageNum);
        numPagesNeeded--;
      }
    } while (pageNum != -1 && numPagesNeeded > 0);
    if (numPagesNeeded == 0) {
      break;
    }
  }
  CHECK(pages.size() == numPagesRequested);
}

Here is the call graph for this function:

setEpoch()

void File_Namespace::FileMgr::setEpoch ( int  epoch )

private

Definition at line 992 of file FileMgr.cpp.

References epoch(), epoch_, and writeAndSyncEpochToDisk().

Referenced by File_Namespace::GlobalFileMgr::setTableEpoch().

                                {
  epoch_ = epoch;
  writeAndSyncEpochToDisk();
}

Here is the call graph for this function:

Here is the caller graph for this function:

writeAndSyncDBMetaToDisk()

void File_Namespace::FileMgr::writeAndSyncDBMetaToDisk ( )

private

Definition at line 568 of file FileMgr.cpp.

References DBMetaFile_, logger::FATAL, getDBVersion(), LOG, and File_Namespace::write().

                                       {
  int db_ver = getDBVersion();
  write(DBMetaFile_, 0, sizeof(int), (int8_t*)&db_ver);
  int status = fflush(DBMetaFile_);
  if (status != 0) {
    LOG(FATAL) << "Could not sync DB metadata file to disk";
  }
}

Here is the call graph for this function:

writeAndSyncEpochToDisk()

void File_Namespace::FileMgr::writeAndSyncEpochToDisk ( )

private

Definition at line 518 of file FileMgr.cpp.

References epoch_, epochFile_, logger::FATAL, LOG, and File_Namespace::write().

Referenced by checkpoint(), and setEpoch().

                                      {
  write(epochFile_, 0, sizeof(int), (int8_t*)&epoch_);
  int status = fflush(epochFile_);
  // int status = fcntl(fileno(epochFile_),51);
  if (status != 0) {
    LOG(FATAL) << "Could not flush epoch file to disk";
  }
#ifdef __APPLE__
  status = fcntl(fileno(epochFile_), 51);
#else
  status = fsync(fileno(epochFile_));
#endif
  if (status != 0) {
    LOG(FATAL) << "Could not sync epoch file to disk";
  }
  ++epoch_;
}

Here is the call graph for this function:

Here is the caller graph for this function:

Friends And Related Function Documentation

GlobalFileMgr

friend class GlobalFileMgr

friend

Definition at line 86 of file FileMgr.h.

Member Data Documentation

chunkIndex_

ChunkKeyToChunkMap File_Namespace::FileMgr::chunkIndex_

Definition at line 225 of file FileMgr.h.

Referenced by checkpoint(), createBuffer(), deleteBuffer(), deleteBuffersWithPrefix(), fetchBuffer(), getBuffer(), getChunkMetadataVec(), getChunkMetadataVecForKeyPrefix(), init(), isBufferOnDevice(), putBuffer(), File_Namespace::GlobalFileMgr::writeFileMgrData(), and ~FileMgr().

chunkIndexMutex_

mapd_shared_mutex File_Namespace::FileMgr::chunkIndexMutex_

mutableprivate

Definition at line 253 of file FileMgr.h.

Referenced by checkpoint(), createBuffer(), deleteBuffer(), deleteBuffersWithPrefix(), fetchBuffer(), getBuffer(), getChunkMetadataVec(), getChunkMetadataVecForKeyPrefix(), isBufferOnDevice(), and putBuffer().

db_version_

int File_Namespace::FileMgr::db_version_

private

Definition at line 248 of file FileMgr.h.

Referenced by createTopLevelMetadata(), and openDBMetaFile().

DBMetaFile_

FILE* File_Namespace::FileMgr::DBMetaFile_

private

DB version from dbmeta file, should be compatible with GlobalFileMgr::mapd_db_version_

Definition at line 250 of file FileMgr.h.

Referenced by createDBMetaFile(), openDBMetaFile(), and writeAndSyncDBMetaToDisk().

defaultPageSize_

size_t File_Namespace::FileMgr::defaultPageSize_

private

number of threads used when loading data

Definition at line 244 of file FileMgr.h.

Referenced by copyPage(), createBuffer(), and putBuffer().

epoch_

int File_Namespace::FileMgr::epoch_

private

the index of the next file id

Definition at line 246 of file FileMgr.h.

Referenced by checkpoint(), createEpochFile(), File_Namespace::GlobalFileMgr::getTableEpoch(), init(), openEpochFile(), openExistingFile(), setEpoch(), and writeAndSyncEpochToDisk().

epochFile_

FILE* File_Namespace::FileMgr::epochFile_

private

the current epoch (time of last checkpoint)

Definition at line 247 of file FileMgr.h.

Referenced by closeRemovePhysical(), createEpochFile(), FileMgr(), openEpochFile(), and writeAndSyncEpochToDisk().

fileIndex_

PageSizeFileMMap File_Namespace::FileMgr::fileIndex_

private

A vector of files accessible via a file identifier.

Definition at line 242 of file FileMgr.h.

Referenced by createFile(), openExistingFile(), requestFreePage(), and requestFreePages().

fileMgrBasePath_

std::string File_Namespace::FileMgr::fileMgrBasePath_

private

Definition at line 239 of file FileMgr.h.

Referenced by createDBMetaFile(), createEpochFile(), createFile(), FileMgr(), init(), openDBMetaFile(), and openEpochFile().

fileMgrKey_

std::pair<const int, const int> File_Namespace::FileMgr::fileMgrKey_

private

Global FileMgr.

Definition at line 238 of file FileMgr.h.

Referenced by FileMgr(), and init().

files_

std::vector<FileInfo*> File_Namespace::FileMgr::files_

private

The OS file system path containing files related to this FileMgr

Definition at line 241 of file FileMgr.h.

Referenced by checkpoint(), closeRemovePhysical(), createFile(), FileMgr(), getFileForFileId(), openExistingFile(), requestFreePage(), requestFreePages(), and ~FileMgr().

files_rw_mutex_

mapd_shared_mutex File_Namespace::FileMgr::files_rw_mutex_

mutableprivate

Definition at line 254 of file FileMgr.h.

Referenced by checkpoint(), createFile(), and openExistingFile().

free_pages

std::vector<std::pair<FileInfo*, int> > File_Namespace::FileMgr::free_pages

private

Definition at line 257 of file FileMgr.h.

Referenced by checkpoint(), and free_page().

getPageMutex_

std::mutex File_Namespace::FileMgr::getPageMutex_

private

pointer to DB level metadata

Definition at line 252 of file FileMgr.h.

Referenced by requestFreePage(), and requestFreePages().

gfm_

GlobalFileMgr* File_Namespace::FileMgr::gfm_

private

Definition at line 237 of file FileMgr.h.

Referenced by FileMgr(), getDBConvert(), getDBVersion(), and init().

mutex_free_page

mapd_shared_mutex File_Namespace::FileMgr::mutex_free_page

mutableprivate

Definition at line 256 of file FileMgr.h.

Referenced by checkpoint(), and free_page().

nextFileId_

unsigned File_Namespace::FileMgr::nextFileId_

private

Definition at line 245 of file FileMgr.h.

Referenced by createFile(), and init().

num_reader_threads_

size_t File_Namespace::FileMgr::num_reader_threads_

private

Maps page sizes to FileInfo objects.

Definition at line 243 of file FileMgr.h.

Referenced by init().

---

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

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