DiskBuffer.cpp

Go to the documentation of this file.

// Mantid Repository : https://github.com/mantidproject/mantid
//
// Copyright © 2018 ISIS Rutherford Appleton Laboratory UKRI,
//   NScD Oak Ridge National Laboratory, European Spallation Source,
//   Institut Laue - Langevin & CSNS, Institute of High Energy Physics, CAS
// SPDX - License - Identifier: GPL - 3.0 +
#include "MantidKernel/DiskBuffer.h"
#include "MantidKernel/ISaveable.h"
#include <sstream>
#include <utility>
 
using namespace Mantid::Kernel;
 
namespace Mantid::Kernel {
 
//----------------------------------------------------------------------------------------------
DiskBuffer::DiskBuffer()
    : m_writeBufferSize(50), m_writeBufferUsed(0), m_nObjectsToWrite(0), m_free(), m_free_bySize(m_free.get<1>()),
      m_fileLength(0) {
  m_free.clear();
}
 
//----------------------------------------------------------------------------------------------
DiskBuffer::DiskBuffer(uint64_t m_writeBufferSize)
    : m_writeBufferSize(m_writeBufferSize), m_writeBufferUsed(0), m_nObjectsToWrite(0), m_free(),
      m_free_bySize(m_free.get<1>()), m_fileLength(0) {
  m_free.clear();
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::toWrite(ISaveable *item) {
  if (item == nullptr)
    return;
  //    if (!m_useWriteBuffer) return;
 
  if (item->getBufPostion()) // already in the buffer and probably have changed
                             // its size in memory
  {
    // forget old memory size
    std::unique_lock<std::mutex> uniqueLock(m_mutex);
    m_writeBufferUsed -= item->getBufferSize();
    // add new size
    size_t newMemorySize = item->getDataMemorySize();
    m_writeBufferUsed += newMemorySize;
    uniqueLock.unlock();
    item->setBufferSize(newMemorySize);
  } else {
    std::lock_guard<std::mutex> lock(m_mutex);
    m_toWriteBuffer.push_front(item);
    m_writeBufferUsed += item->setBufferPosition(m_toWriteBuffer.begin());
    m_nObjectsToWrite++;
  }
 
  // Should we now write out the old data?
  if (m_writeBufferUsed > m_writeBufferSize)
    writeOldObjects();
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::objectDeleted(ISaveable *item) {
  if (item == nullptr)
    return;
  // have it ever been in the buffer?
  std::unique_lock<std::mutex> uniqueLock(m_mutex);
  auto opt2it = item->getBufPostion();
  if (opt2it) {
    m_writeBufferUsed -= item->getBufferSize();
    m_toWriteBuffer.erase(*opt2it);
  } else {
    return;
  }
 
  // indicate to the object that it is not stored in memory any more
  item->clearBufferState();
  uniqueLock.unlock();
 
  // Mark the amount of space used on disk as free
  if (item->wasSaved())
    this->freeBlock(item->getFilePosition(), item->getFileSize());
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::writeOldObjects() {
 
  std::lock_guard<std::mutex> _lock(m_mutex);
  // Holder for any objects that you were NOT able to write.
  std::list<ISaveable *> couldNotWrite;
  size_t objectsNotWritten(0);
  size_t memoryNotWritten(0);
 
  // Iterate through the list
  auto it = m_toWriteBuffer.begin();
  auto it_end = m_toWriteBuffer.end();
 
  ISaveable *obj = nullptr;
 
  for (; it != it_end; ++it) {
    obj = *it;
    if (!obj->isBusy()) {
      uint64_t NumObjEvents = obj->getTotalDataSize();
      uint64_t fileIndexStart;
      if (!obj->wasSaved()) {
        fileIndexStart = this->allocate(NumObjEvents);
        // Write to the disk; this will call the object specific save function;
        // Prevent simultaneous file access (e.g. write while loading)
        obj->saveAt(fileIndexStart, NumObjEvents);
      } else {
        uint64_t NumFileEvents = obj->getFileSize();
        if (NumObjEvents != NumFileEvents) {
          // Event list changed size. The MRU can tell us where it best fits
          // now.
          fileIndexStart = this->relocate(obj->getFilePosition(), NumFileEvents, NumObjEvents);
          // Write to the disk; this will call the object specific save
          // function;
          obj->saveAt(fileIndexStart, NumObjEvents);
        } else // despite object size have not been changed, it can be modified
               // other way. In this case, the method which changed the data
               // should set dataChanged ID
        {
          if (obj->isDataChanged()) {
            fileIndexStart = obj->getFilePosition();
            // Write to the disk; this will call the object specific save
            // function;
            obj->saveAt(fileIndexStart, NumObjEvents);
            // this is questionable operation, which adjust file size in case
            // when the file postions were allocated externaly
            if (fileIndexStart + NumObjEvents > m_fileLength)
              m_fileLength = fileIndexStart + NumObjEvents;
          } else // just clean the object up -- it just occupies memory
            obj->clearDataFromMemory();
        }
      }
      // tell the object that it has been removed from the buffer
      obj->clearBufferState();
    } else // object busy
    {
      // The object is busy, can't write. Save it for later
      couldNotWrite.emplace_back(obj);
      // When a prefix or postfix operator is applied to a function argument,
      // the value of the argument is
      // NOT GUARANTEED to be incremented or decremented before it is passed to
      // the function.
      memoryNotWritten += obj->setBufferPosition(--couldNotWrite.end());
      objectsNotWritten++;
    }
  }
 
  // use last object to clear NeXus buffer and actually write data to HDD
  if (obj) {
    // NXS needs to flush the writes to file by closing and re-opening the data
    // block.
    // For speed, it is best to do this only once per write dump, using last
    // object saved
    obj->flushData();
  }
 
  // Exchange with the new map you built out of the not-written blocks.
  m_toWriteBuffer.swap(couldNotWrite);
  m_writeBufferUsed = memoryNotWritten;
  m_nObjectsToWrite = objectsNotWritten;
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::flushCache() {
  // Now write everything out.
  writeOldObjects();
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::freeBlock(uint64_t const pos, uint64_t const size) {
  if (size == 0 || size == std::numeric_limits<uint64_t>::max())
    return;
  std::lock_guard<std::mutex> lock(m_freeMutex);
 
  // Make the block
  FreeBlock newBlock(pos, size);
  // Insert it
  std::pair<freeSpace_t::iterator, bool> p = m_free.insert(newBlock);
 
  // Failed insert? Should not happen since the map is NOT unique
  // Or, if the map has only 1 item then it cannot do any merging. This solves a
  // hanging bug in MacOS. Refs #3652
  if (!p.second || m_free.size() <= 1) {
    return;
  }
 
  // This is where we inserted
  freeSpace_t::iterator it = p.first;
  if (it != m_free.begin()) {
    freeSpace_t::iterator it_before = it;
    --it_before;
    // There is a block before
    FreeBlock block_before = *it_before;
    if (FreeBlock::merge(block_before, newBlock)) {
      // Change the map by replacing the old "before" block with the new merged
      // one
      m_free.replace(it_before, block_before);
      // Remove the block we just inserted
      m_free.erase(it);
      // For cases where the new block was between two blocks.
      newBlock = block_before;
      it = it_before;
    }
  }
  // Get an iterator to the block AFTER this one
  freeSpace_t::iterator it_after = it;
  ++it_after;
  // There is a block after
  if (it_after != m_free.end()) {
    FreeBlock block_after = *it_after;
    if (FreeBlock::merge(newBlock, block_after)) {
      // Change the map by replacing the old "new" block with the new merged one
      m_free.replace(it, newBlock);
      // Remove the block that was after this one
      m_free.erase(it_after);
    }
  }
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::defragFreeBlocks() {
  std::lock_guard<std::mutex> lock(m_freeMutex);
 
  freeSpace_t::iterator it = m_free.begin();
  FreeBlock thisBlock;
  thisBlock = *it;
 
  while (it != m_free.end()) {
    // Get iterator to the block after "it".
    freeSpace_t::iterator it_after = it;
    ++it_after;
 
    if (FreeBlock::merge(thisBlock, *it_after)) {
      // Change the map by replacing the old "before" block with the new merged
      // one
      m_free.replace(it, thisBlock);
      // Remove the block that was merged out
      m_free.erase(it_after);
      // And stay at this iterator to
    } else {
      // Move on to the next block
      ++it;
      thisBlock = *it;
    }
  }
}
 
//---------------------------------------------------------------------------------------------
uint64_t DiskBuffer::allocate(uint64_t const newSize) {
  std::unique_lock<std::mutex> uniqueLock(m_freeMutex);
 
  // Now, find the first available block of sufficient size.
  freeSpace_bySize_t::iterator it;
  bool putAtFileEnd = true;
  if (m_free.size() > 0) {
    // Unless there is nothing in the free space map
    it = m_free_bySize.lower_bound(newSize);
    putAtFileEnd = (it == m_free_bySize.end());
  }
 
  if (putAtFileEnd) {
    // No block found
    // Go to the end of the file.
    uint64_t retVal = m_fileLength;
    // And we assume the file will grow by this much.
    m_fileLength += newSize;
    // Will place the new block at the end of the file
    return retVal;
  } else {
    //      std::cout << "Block found for allocate " << newSize << '\n';
    uint64_t foundPos = it->getFilePosition();
    uint64_t foundSize = it->getSize();
    // Remove the free block you found - it is no longer free
    m_free_bySize.erase(it);
    uniqueLock.unlock();
    // Block was too large - free the bit of space after it.
    if (foundSize > newSize) {
      this->freeBlock(foundPos + newSize, foundSize - newSize);
    }
    return foundPos;
  }
}
 
//---------------------------------------------------------------------------------------------
uint64_t DiskBuffer::relocate(uint64_t const oldPos, uint64_t const oldSize, const uint64_t newSize) {
  // std::cout << "Relocating " << oldPos << ", " << oldSize << ", " << newSize
  // << '\n';
  // First, release the space in the old block.
  this->freeBlock(oldPos, oldSize);
  return this->allocate(newSize);
}
 
//---------------------------------------------------------------------------------------------
void DiskBuffer::getFreeSpaceVector(std::vector<uint64_t> &free) const {
  free.reserve(m_free.size() * 2);
  freeSpace_bySize_t::const_iterator it = m_free_bySize.begin();
  freeSpace_bySize_t::const_iterator it_end = m_free_bySize.end();
  for (; it != it_end; ++it) {
    free.emplace_back(it->getFilePosition());
    free.emplace_back(it->getSize());
  }
}
 
void DiskBuffer::setFreeSpaceVector(std::vector<uint64_t> const &free) {
  m_free.clear();
 
  if (free.size() % 2 != 0)
    throw std::length_error("Free vector size is not a factor of 2.");
 
  for (auto it = free.begin(); it != free.end(); it += 2) {
    auto it_next = std::next(it);
 
    if (*it == 0 && *it_next == 0) {
      continue; // Not really a free space block!
    }
 
    FreeBlock newBlock(*it, *it_next);
    m_free.insert(newBlock);
  }
}
 
std::string DiskBuffer::getMemoryStr() const {
  std::ostringstream mess;
  mess << "Buffer: " << m_writeBufferUsed << " in " << m_nObjectsToWrite << " objects. ";
  return mess.str();
}
 
} // namespace Mantid::Kernel