mariadb-columnstore-engine/dbcon/joblist/resourcedistributor.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: $
 *
 ******************************************************************************************/
#include <string>
#include <stdexcept>
#include <iostream>
#include <sstream>
#include <sys/time.h>
using namespace std;

#include "jl_logger.h"
#include "resourcedistributor.h"

namespace joblist
{
const unsigned maxSessionsDefault = 100;

uint64_t ResourceDistributor::requestResource(uint32_t sessionID)
{
  uint64_t resource = getSessionResource(sessionID);

  return requestResource(sessionID, resource);
}

uint64_t ResourceDistributor::requestResource(uint32_t sessionID, uint64_t resource)
{
  if (fTraceOn)
    logMessage(logging::LOG_TYPE_DEBUG, LogRDRequest, resource, sessionID);

  boost::mutex::scoped_lock lk(fResourceLock);

  while (fTotalResource < resource)
  {
    if (fTraceOn)
      logMessage(logging::LOG_TYPE_DEBUG, LogRDRequestWait, resource, sessionID);

    fResourceAvailable.wait(lk);

    if (fTraceOn)
      logMessage(logging::LOG_TYPE_DEBUG, LogRDRequest, resource, sessionID);
  }

  fTotalResource -= resource;

  return resource;
}

void ResourceDistributor::returnResource(uint64_t resource)
{
  if (fTraceOn)
    logMessage(logging::LOG_TYPE_DEBUG, LogRDReturn, resource);

  boost::mutex::scoped_lock lk(fResourceLock);
  fTotalResource += resource;

  fResourceAvailable.notify_all();
}

void ResourceDistributor::logMessage(logging::LOG_TYPE logLevel, logging::Message::MessageID mid,
                                     uint64_t value, uint32_t sessionID)
{
  logging::Message::Args args;
  args.add(fJob);
  args.add(fIdentity);
  args.add(fTotalResource);

  if (value)
    args.add(value);

  Logger log;
  log.logMessage(logLevel, mid, args, logging::LoggingID(5, sessionID));
}

void LockedSessionMap::updateAging(uint32_t sessionID)
{
  boost::mutex::scoped_lock lock(fSessionLock);
  SessionList::iterator pos = find(fSessionAgingList.begin(), fSessionAgingList.end(), sessionID);

  if (fSessionAgingList.end() != pos)
    fSessionAgingList.splice(fSessionAgingList.end(), fSessionAgingList,
                             find(fSessionAgingList.begin(), fSessionAgingList.end(), sessionID));
  else
    fSessionAgingList.push_back(sessionID);
}

uint64_t LockedSessionMap::getSessionResource(uint32_t sessionID)
{
  SessionMap::const_iterator it = fSessionMap.find(sessionID);

  if (fSessionMap.end() != it)
  {
    updateAging(sessionID);
    return it->second;
  }

  return fResourceBlock;
}

bool LockedSessionMap::addSession(uint32_t sessionID, uint64_t resource, uint64_t limit)
{
  bool ret = true;

  if (resource > limit)
  {
    resource = limit;
    ret = false;
  }

  boost::mutex::scoped_lock maplock(fMapLock);
  fSessionMap[sessionID] = resource;
  updateAging(sessionID);

  if (fMaxSessions < fSessionMap.size())
  {
    boost::mutex::scoped_lock lock(fSessionLock);
    uint32_t oldsession = fSessionAgingList.front();
    fSessionMap.erase(oldsession);
    fSessionAgingList.erase(fSessionAgingList.begin());
  }

  return ret;
}

void LockedSessionMap::removeSession(uint32_t sessionID)
{
  boost::mutex::scoped_lock maplock(fMapLock);
  fSessionMap.erase(sessionID);
  boost::mutex::scoped_lock listlock(fSessionLock);
  fSessionAgingList.erase(find(fSessionAgingList.begin(), fSessionAgingList.end(), sessionID));
}

ostream& operator<<(ostream& os, const LockedSessionMap& lsm)
{
  os << "Default Resource Block: " << lsm.fResourceBlock
     << "\tMax Number of saved sessions: " << lsm.fMaxSessions << endl;
  os << "Session Map:\tsessionID\tvalue\n";
  LockedSessionMap::SessionMap::const_iterator smIt = lsm.fSessionMap.begin(), smEnd = lsm.fSessionMap.end();

  for (; smIt != smEnd; ++smIt)
    os << "\t\t" << smIt->first << "\t\t" << smIt->second << endl;

  os << "\nAging List:\tsessionID\n\t\t";
  copy(lsm.fSessionAgingList.begin(), lsm.fSessionAgingList.end(), ostream_iterator<uint32_t>(os, "\n\t\t"));
  os << endl;
  return os;
}
}  // namespace joblist