/* 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 <iostream>
#include <string>
//#define NDEBUG
#include <cassert>
#include <algorithm>
using namespace std;

#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp>
#include <boost/version.hpp>
namespace bi = boost::interprocess;

#include "shmkeys.h"
#include "brmshmimpl.h"
#include "brmtypes.h"

#include "installdir.h"

namespace BRM
{
const constexpr uint32_t ShmCreateMaxRetries = 10;
const constexpr unsigned int NapTimer = 500000;

BRMShmImplParent::BRMShmImplParent(unsigned key, off_t size, bool readOnly)
 : fKey(key), fSize(size), fReadOnly(readOnly){};

BRMShmImplParent::~BRMShmImplParent(){};

BRMShmImpl::BRMShmImpl(unsigned key, off_t size, bool readOnly) : BRMShmImplParent(key, size, readOnly)
{
  string keyName = ShmKeys::keyToName(fKey);

  if (fSize == 0)
  {
    unsigned tries = 0;
  again:

    try
    {
      bi::shared_memory_object shm(bi::open_only, keyName.c_str(), bi::read_write);
      off_t curSize = 0;
      shm.get_size(curSize);

      if (curSize == 0)
        throw bi::interprocess_exception("shm size is zero");
    }
    catch (bi::interprocess_exception&)
    {
      if (++tries > 10)
      {
        log("BRMShmImpl::BRMShmImpl(): retrying on size==0");
        throw;
      }

      cerr << "BRMShmImpl::BRMShmImpl(): retrying on size==0" << endl;
      usleep(500 * 1000);
      goto again;
    }
  }

  try
  {
    bi::permissions perms;
    perms.set_unrestricted();
    bi::shared_memory_object shm(bi::create_only, keyName.c_str(), bi::read_write, perms);
    idbassert(fSize > 0);
    shm.truncate(fSize);
    fShmobj.swap(shm);
  }
  catch (bi::interprocess_exception& b)
  {
    if (b.get_error_code() != bi::already_exists_error)
    {
      ostringstream o;
      o << "BRM caught an exception creating a shared memory segment: " << b.what();
      log(o.str());
      throw;
    }
    bi::shared_memory_object* shm = NULL;
    try
    {
      shm = new bi::shared_memory_object(bi::open_only, keyName.c_str(), bi::read_write);
    }
    catch (exception& e)
    {
      ostringstream o;
      o << "BRM caught an exception attaching to a shared memory segment (" << keyName << "): " << b.what();
      log(o.str());
      throw;
    }
    off_t curSize = 0;
    shm->get_size(curSize);
    idbassert(curSize > 0);
    idbassert(curSize >= fSize);
    fShmobj.swap(*shm);
    delete shm;
    fSize = curSize;
  }

  if (fReadOnly)
  {
    bi::mapped_region ro_region(fShmobj, bi::read_only);
    fMapreg.swap(ro_region);
  }
  else
  {
    bi::mapped_region region(fShmobj, bi::read_write);
    fMapreg.swap(region);
  }
}

int BRMShmImpl::grow(unsigned newKey, off_t newSize)
{
  idbassert(newKey != fKey);
  idbassert(newSize >= fSize);

  string oldName = fShmobj.get_name();

  string keyName = ShmKeys::keyToName(newKey);
  bi::permissions perms;
  perms.set_unrestricted();
  bi::shared_memory_object shm(bi::create_only, keyName.c_str(), bi::read_write, perms);
  shm.truncate(newSize);

  bi::mapped_region region(shm, bi::read_write);

  // Copy old data into new region
  memcpy(region.get_address(), fMapreg.get_address(), fSize);
  // clear new region
  // make some versions of gcc happier...
  memset(reinterpret_cast<void*>(reinterpret_cast<ptrdiff_t>(region.get_address()) + fSize), 0,
         newSize - fSize);

  fShmobj.swap(shm);
  fMapreg.swap(region);

  if (!oldName.empty())
    bi::shared_memory_object::remove(oldName.c_str());

  fKey = newKey;
  fSize = newSize;

  if (fReadOnly)
  {
    bi::mapped_region ro_region(fShmobj, bi::read_only);
    fMapreg.swap(ro_region);
  }

  return 0;
}

int BRMShmImpl::clear(unsigned newKey, off_t newSize)
{
  idbassert(newKey != fKey);

  string oldName = fShmobj.get_name();

  string keyName = ShmKeys::keyToName(newKey);
  bi::permissions perms;
  perms.set_unrestricted();
  bi::shared_memory_object shm(bi::create_only, keyName.c_str(), bi::read_write, perms);
  shm.truncate(newSize);

  bi::mapped_region region(shm, bi::read_write);

  // clear new region
  memset(region.get_address(), 0, newSize);

  fShmobj.swap(shm);
  fMapreg.swap(region);

  if (!oldName.empty())
    bi::shared_memory_object::remove(oldName.c_str());

  fKey = newKey;
  fSize = newSize;

  if (fReadOnly)
  {
    bi::mapped_region ro_region(fShmobj, bi::read_only);
    fMapreg.swap(ro_region);
  }

  return 0;
}

void BRMShmImpl::setReadOnly()
{
  if (fReadOnly)
    return;

  bi::mapped_region ro_region(fShmobj, bi::read_only);
  fMapreg.swap(ro_region);

  fReadOnly = true;
}

void BRMShmImpl::swap(BRMShmImpl& rhs)
{
  fShmobj.swap(rhs.fShmobj);
  fMapreg.swap(rhs.fMapreg);
  std::swap(fKey, rhs.fKey);
  std::swap(fSize, rhs.fSize);
  std::swap(fReadOnly, rhs.fReadOnly);
}

void BRMShmImpl::destroy()
{
  string oldName = fShmobj.get_name();

  if (!oldName.empty())
    bi::shared_memory_object::remove(oldName.c_str());
}

BRMManagedShmImpl::BRMManagedShmImpl(unsigned key, off_t size, bool readOnly)
 : BRMShmImplParent(key, size, readOnly)
{
  string keyName = ShmKeys::keyToName(fKey);
  off_t curSize = 0;

  for (uint32_t tries = 0; fSize == 0 && tries <= ShmCreateMaxRetries; ++tries)
  {
    try
    {
      auto* shmSegment = new boost::interprocess::managed_shared_memory(bi::open_only, keyName.c_str());
      curSize = shmSegment->get_size();

      if (curSize == 0)
      {
        delete shmSegment;
        throw bi::interprocess_exception("shared memory segment size is 0.");
      }
      else
      {
        fShmSegment = shmSegment;
        fSize = curSize;
        return;
      }
    }
    catch (bi::interprocess_exception&)
    {
      if (tries == ShmCreateMaxRetries)
      {
        log("BRMManagedShmImpl::BRMManagedShmImpl(): re-creating shared memory segment\
 b/c of its size == 0. Re-throw.");
        throw;
      }

      cerr << "BRMManagedShmImpl::BRMManagedShmImpl(): re-creating shared memory segment\
 b/c of its size == 0"
           << endl;
      usleep(NapTimer);
    }
  }

  try
  {
    bi::permissions perms;
    perms.set_unrestricted();
    fShmSegment = new bi::managed_shared_memory(bi::create_only, keyName.c_str(), fSize,
                                                0,  // use a default address to map the segment
                                                perms);
    // fSize == 0 on any process startup but managed_shared_memory ctor throws
    // so control flow doesn't get here.
    idbassert(fSize > 0);
  }
  catch (bi::interprocess_exception& b)
  {
    if (b.get_error_code() != bi::already_exists_error)
    {
      ostringstream o;
      o << "BRM caught an exception creating a shared memory segment: " << b.what();
      log(o.str());
      throw;
    }
    bi::managed_shared_memory* shmSegment = nullptr;
    try
    {
      if (fReadOnly)
        shmSegment = new bi::managed_shared_memory(bi::open_read_only, keyName.c_str());
      else
        shmSegment = new bi::managed_shared_memory(bi::open_only, keyName.c_str());
    }
    catch (exception& e)
    {
      ostringstream o;
      o << "BRM caught an exception attaching to a shared memory segment (" << keyName << "): " << b.what();
      log(o.str());
      throw;
    }
    off_t curSize = shmSegment->get_size();

    idbassert(curSize > 0);
    idbassert(curSize >= fSize);
    fShmSegment = shmSegment;
    fSize = curSize;
  }
}

int BRMManagedShmImpl::grow(off_t newSize)
{
  auto keyName = ShmKeys::keyToName(fKey);

  if (newSize > fSize)
  {
    const auto incSize = newSize - fSize;
    if (fShmSegment)
    {
      // Call destructor to unmap the segment.
      delete fShmSegment;
      // Grow the segment.
      bi::managed_shared_memory::grow(keyName.c_str(), incSize);
      // Open only with the assumption ::grow() can be called on read-write shmem.
      fShmSegment = new bi::managed_shared_memory(bi::open_only, keyName.c_str());
      // Update size.
      fSize = newSize;
    }
  }

  return 0;
}

// Dummy method that has no references in the code.
int BRMManagedShmImpl::clear(unsigned newKey, off_t newSize)
{
  return 0;
}

// This method calls for all related shmem pointers to be refreshed.
void BRMManagedShmImpl::setReadOnly()
{
  if (fReadOnly)
    return;
  const bool readOnly = true;
  remap(readOnly);
  fReadOnly = true;
}

void BRMManagedShmImpl::swap(BRMManagedShmImpl& rhs)
{
  fShmSegment->swap(*rhs.fShmSegment);
  std::swap(fKey, rhs.fKey);
  std::swap(fSize, rhs.fSize);
  std::swap(fReadOnly, rhs.fReadOnly);
}

// The method was copied from non-managed shmem impl class
// and it has no refences in MCS 6.x code.
void BRMManagedShmImpl::destroy()
{
  string keyName = ShmKeys::keyToName(fKey);
  try
  {
    bi::shared_memory_object::remove(keyName.c_str());
  }
  catch (bi::interprocess_exception& b)
  {
    std::ostringstream o;
    o << "BRMManagedShmImpl::destroy caught an exception removing a managed shared memory segment: "
      << b.what();
    log(o.str());
    throw;
  }
}

void BRMManagedShmImpl::remap(const bool readOnly)
{
  delete fShmSegment;
  fShmSegment = nullptr;
  string keyName = ShmKeys::keyToName(fKey);
  if (readOnly)
    fShmSegment = new bi::managed_shared_memory(bi::open_read_only, keyName.c_str());
  else
    fShmSegment = new bi::managed_shared_memory(bi::open_only, keyName.c_str());
}

BRMManagedShmImplRBTree::BRMManagedShmImplRBTree(unsigned key, off_t size, bool readOnly)
 : BRMShmImplParent(key, size, readOnly)
{
  // FIXME: Take a right size.
  // Actually this means to open a segment at the default code.
  if (fSize == 0)
    fSize = 64000;

  try
  {
    bi::permissions perms;
    perms.set_unrestricted();

    fShmSegment = new boost::interprocess::managed_shared_memory(boost::interprocess::open_or_create,
                                                                 segmentName, fSize, 0, perms);

    fSize = fShmSegment->get_size();
  }
  catch (exception& e)
  {
    std::cout << "Cannot create boost::interprocess::managed_shared_memory object: " << e.what() << std::endl;
  }
}

BRMManagedShmImplRBTree::~BRMManagedShmImplRBTree()
{
  delete fShmSegment;
}

void BRMManagedShmImplRBTree::setReadOnly()
{
  try
  {
    if (fShmSegment)
    {
      delete fShmSegment;
      fShmSegment = new bi::managed_shared_memory(bi::open_read_only, segmentName);
      fReadOnly = true;
    }
  }
  catch (exception& e)
  {
    std::cout << "BRMManagedShmImpl::setReadOnly() error " << e.what() << std::endl;
    throw;
  }
}

int32_t BRMManagedShmImplRBTree::grow(unsigned key, off_t incSize)
{
  try
  {
    if (fShmSegment)
    {
      // Update key for this process.
      fKey = key;
      // Call destructor to unmap the segment.
      delete fShmSegment;
      // Grow the segment.
      bi::managed_shared_memory::grow(segmentName, incSize);
      // Open only.
      fShmSegment = new bi::managed_shared_memory(bi::open_only, segmentName);
      // Update size.
      fSize = fShmSegment->get_size();
    }
  }
  catch (exception& e)
  {
    std::cout << "BRMManagedShmImpl::grow() error " << e.what() << std::endl;
    throw;
  }

  return 0;
}

// Dummy method that has no references in the code.
int BRMManagedShmImplRBTree::clear(unsigned newKey, off_t newSize)
{
  return 0;
}

void BRMManagedShmImplRBTree::reMapSegment()
{
  try
  {
    if (fShmSegment)
    {
      // Call destructor to unmap the segment.
      delete fShmSegment;
      // Map.
      fShmSegment = new bi::managed_shared_memory(bi::open_only, segmentName);
      // Update size.
      fSize = fShmSegment->get_size();
    }
  }
  catch (exception& e)
  {
    std::cout << "BRMManagedShmImpl::remap() error " << e.what() << std::endl;
    throw;
  }
}

void BRMManagedShmImplRBTree::destroy()
{
  try
  {
    bi::shared_memory_object::remove(segmentName);
  }
  catch (exception& e)
  {
    std::cout << "BRMManagedShmImplRBTree::destroy() " << std::endl;
    throw;
  }
}

}  // namespace BRM