mariadb-columnstore-engine/utils/testbc/blockrequestprocessor.cpp

/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/***************************************************************************
 *
 *   $Id: blockrequestprocessor.cpp 725 2008-09-26 16:26:47Z jrodriguez $
 *
 *   jrodriguez@calpont.com   *
 *                                                                         *
 ***************************************************************************/

#include "blockrequestprocessor.h"
#include "rwlock_local.h"
#include "dbrm.h"
#include <sys/time.h>
#include <pthread.h>
#include <sstream>
#include <string>
#include <list>
#include <boost/date_time/posix_time/posix_time.hpp>
using namespace std;

namespace dbbc
{
BlockRequestProcessor::BlockRequestProcessor(uint32_t numBlcks, int thrCount, int blocksPerRead,
                                             uint32_t deleteBlocks, uint32_t blckSz)
 : fbMgr(numBlcks, blckSz, deleteBlocks), fIOMgr(fbMgr, fBRPRequestQueue, thrCount, blocksPerRead)
{
}

BlockRequestProcessor::~BlockRequestProcessor()
{
}

void BlockRequestProcessor::stop()
{
  fBRPRequestQueue.stop();
  fIOMgr.stop();
}

int BlockRequestProcessor::check(const BRM::InlineLBIDRange& range, const BRM::VER_t ver, uint32_t& lbidCount)
{
  uint64_t maxLbid = range.start;  // highest existent lbid
  uint64_t rangeLen = range.size;
  uint64_t idx;
  uint64_t adjSz;
  struct timespec start_tm;

  if (fTrace)
    clock_gettime(CLOCK_MONOTONIC, &start_tm);

  for (idx = 0; fbMgr.exists(maxLbid, ver) == true && idx < rangeLen; maxLbid++, idx++)
    ;

  if (idx == rangeLen)  // range is already loaded
  {
    uint32_t fbo;
    BRM::OID_t oid;
    fdbrm.lookup(maxLbid, ver, false, oid, fbo);
    fLogFile << oid << " " << maxLbid << " " << fbo << " " << rangeLen << " " << 0 << " " << 0 << " " << 0
             << " " << right << fixed << ((double)(start_tm.tv_sec + (1.e-9 * start_tm.tv_nsec))) << endl;
    return 0;
  }

  adjSz = rangeLen - idx;
  BRM::InlineLBIDRange adjRange;
  adjRange.start = maxLbid;
  adjRange.size = adjSz;
  fileRequest rqstBlk(adjRange, ver);
  check(rqstBlk);
  lbidCount = rqstBlk.BlocksRead();

  if (fTrace)
  {
    uint32_t fbo;
    BRM::OID_t oid;
    fdbrm.lookup(maxLbid, ver, false, oid, fbo);
    fLogFile << oid << " " << maxLbid << " " << fbo << " " << rangeLen << " " << adjSz << " "
             << rqstBlk.BlocksRead() << " " << rqstBlk.BlocksLoaded() << " " << right << fixed
             << ((double)(start_tm.tv_sec + (1.e-9 * start_tm.tv_nsec))) << endl;
  }

  return rqstBlk.BlocksLoaded();
}  // check

int BlockRequestProcessor::check(fileRequest& rqstBlk)
{
  pthread_mutex_lock(&rqstBlk.frMutex());
  rqstBlk.SetPredicate(fileRequest::SENDING);
  sendRequest(rqstBlk);  // start file read request

  while (rqstBlk.frPredicate() < fileRequest::COMPLETE)
    pthread_cond_wait(&rqstBlk.frCond(), &rqstBlk.frMutex());

  pthread_mutex_unlock(&rqstBlk.frMutex());

  return 0;
}

int BlockRequestProcessor::check(BRM::LBID_t lbid, BRM::VER_t ver, bool flg, bool& wasBlockInCache)
{
  if (fbMgr.exists(lbid, ver) == true)
  {
    wasBlockInCache = true;
    return 0;
  }
  else
  {
    wasBlockInCache = false;
    fileRequest rqstBlk(lbid, ver, flg);
    int ret = check(rqstBlk);
    return ret;
  }
}

int BlockRequestProcessor::sendRequest(fileRequest& blk)
{
  int ret = fBRPRequestQueue.push(blk);
  return ret;
}

const int BlockRequestProcessor::read(const BRM::InlineLBIDRange& range, FileBufferList_t& readList,
                                      const BRM::VER_t ver)
{
  int blksLoaded = 0;
  HashObject_t fb = {0, 0, 0};

  for (int idx = 0; (uint64_t)idx < range.size; idx++)
  {
    fb.lbid = range.start + idx;
    fb.ver = ver;
    fb.poolIdx = 0;
    FileBuffer fbRet(-1, -1);
    bool ret = false;  // fbMgr.find(fb, fbRet);

    if (ret)
    {
      blksLoaded++;
      readList.push_back(fbRet);
    }
  }

  return blksLoaded;
}

FileBuffer* BlockRequestProcessor::getBlockPtr(const BRM::LBID_t lbid, const BRM::VER_t ver)
{
  HashObject_t hashObj = {lbid, ver, 0};
  FileBuffer* fb = fbMgr.findPtr(hashObj);
  return fb;
}

const int BlockRequestProcessor::read(const BRM::LBID_t& lbid, const BRM::VER_t& ver, FileBuffer& fb)
{
  HashObject_t hashObj = {lbid, ver, 0};
  bool ret = fbMgr.find(hashObj, fb);

  if (ret == true)
    return 1;
  else
    return 0;
}

const int BlockRequestProcessor::read(const BRM::LBID_t& lbid, const BRM::VER_t& ver, void* bufferPtr)
{
  HashObject_t hashObj = {lbid, ver, 0};
  bool ret = fbMgr.find(hashObj, bufferPtr);

  if (ret == true)
    return 1;
  else
    return 0;
}

const int BlockRequestProcessor::getBlock(const BRM::LBID_t& lbid, const BRM::VER_t& ver, void* bufferPtr,
                                          bool flg, bool& wasCached)
{
  HashObject_t hashObj = {lbid, ver, 0};
  wasCached = fbMgr.find(hashObj, bufferPtr);

  if (wasCached)
    return 1;

  wasCached = false;
  fileRequest rqstBlk(lbid, ver, flg, (uint8_t*)bufferPtr);
  check(rqstBlk);
  return 1;
}

bool BlockRequestProcessor::exists(BRM::LBID_t lbid, BRM::VER_t ver)
{
  HashObject_t ho = {lbid, ver, 0};

  return fbMgr.exists(ho);
}

void BlockRequestProcessor::flushCache()
{
  fbMgr.flushCache();
}
/**
const uint32_t BlockRequestProcessor::resize(const uint32_t s)
{
        int rc = fbMgr.resize(s);
        return rc;
}
**/
ostream& BlockRequestProcessor::formatLRUList(ostream& os) const
{
  return fbMgr.formatLRUList(os);
}

}  // namespace dbbc