havoqgt::distributed_db Class Reference

#include <distributed_db.hpp>

Classes
struct	header

Public Types
typedef mapped_type::segment_manager	segment_manager_type

Public Member Functions
	distributed_db (db_create, const char *base_fname)
	distributed_db (db_open, const char *base_fname)
	~distributed_db ()
segment_manager_type *	get_segment_manager ()

Private Types
typedef boost::interprocess::basic_managed_mapped_file< char,boost::interprocess::rbtree_best_fit< boost::interprocess::null_mutex_family >,boost::interprocess::iset_index >	mapped_type

Private Member Functions
uint64_t	get_file_size ()
bool	rank_file_exists ()
void	init_rank_filename (const char *base_fname)

Private Attributes
mapped_type *	m_pm
std::string	m_rank_filename

Detailed Description

Definition at line 86 of file distributed_db.hpp.

Member Typedef Documentation

typedef boost::interprocess::basic_managed_mapped_file<char ,boost::interprocess::rbtree_best_fit<boost::interprocess::null_mutex_family> ,boost::interprocess::iset_index> havoqgt::distributed_db::mapped_type

private

Definition at line 98 of file distributed_db.hpp.

typedef mapped_type::segment_manager havoqgt::distributed_db::segment_manager_type

Definition at line 101 of file distributed_db.hpp.

Constructor & Destructor Documentation

havoqgt::distributed_db::distributed_db ( db_create  , const char *  base_fname  )

inline

Definition at line 106 of file distributed_db.hpp.

  {
    int mpi_rank = havoqgt_env()->world_comm().rank();
    int mpi_size = havoqgt_env()->world_comm().size();

    init_rank_filename(base_fname);
    if(rank_file_exists())
    {
      HAVOQGT_ERROR_MSG("File already exists.");
    }

    m_pm = new mapped_type(boost::interprocess::create_only, m_rank_filename.c_str(), get_file_size()); 

    #ifdef HAVE_POSIX_FALLOCATE
    {
      int fd  = open(m_rank_filename.c_str(), O_RDWR);
      if(fd == -1) {
        HAVOQGT_ERROR_MSG("Error opening file.");
      }
      int ret = posix_fallocate(fd,0,get_file_size());
      if(ret != 0)
      {
        HAVOQGT_ERROR_MSG("posix_fallocate failed.");
      }
      close(fd);
    }
    #else
            #warning posix_fallocate not found;  OSX?
    #endif

    //
    // Create header
    header* phead = m_pm->construct<header>(boost::interprocess::unique_instance)();
    if(mpi_rank == 0) {
      phead->uuid = boost::uuids::random_generator()();
    }
    mpi::mpi_bcast(phead->uuid,0,havoqgt_env()->world_comm().comm());
    //std::cout << "Rank = " << mpi_rank << ", UUID = " << phead->uuid << std::endl;
    phead->comm_rank = mpi_rank;
    phead->comm_size = mpi_size;
    phead->clean_close = false;
  }

Here is the call graph for this function:

havoqgt::distributed_db::distributed_db ( db_open  , const char *  base_fname  )

inline

Definition at line 152 of file distributed_db.hpp.

  {
    int mpi_rank = havoqgt_env()->world_comm().rank();
    int mpi_size = havoqgt_env()->world_comm().size();

    init_rank_filename(base_fname);
    if(!rank_file_exists())
    {
      std::stringstream error;
      error << "ERROR: " << __FILE__ << ":" << __LINE__ << ": file not found.";
      throw std::runtime_error(error.str());
    }
     
    m_pm = new mapped_type(boost::interprocess::open_only, m_rank_filename.c_str()); 

    //
    // Check for header
    std::pair<header*, std::size_t> ret = m_pm->find<header>(boost::interprocess::unique_instance);
    if(ret.second == 0) {
      std::stringstream error;
      error << "ERROR: " << __FILE__ << ":" << __LINE__ << ": header now found.";
      throw std::runtime_error(error.str());
    }

    //
    // Check UUID
    boost::uuids::uuid uuid;
    if(mpi_rank == 0) {
      uuid = ret.first->uuid;
    }
    mpi::mpi_bcast(uuid,0,havoqgt_env()->world_comm().comm());
    if(uuid != ret.first->uuid)
    {
      std::stringstream error;
      error << "ERROR: " << __FILE__ << ":" << __LINE__ << ": UUIDs don't match.";
      throw std::runtime_error(error.str());
    }
    //std::cout << "Rank = " << mpi_rank << ", UUID = " << ret.first->uuid << std::endl;
    if(ret.first->comm_rank != mpi_rank ||
       ret.first->comm_size != mpi_size ||
       !ret.first->clean_close) {
      std::stringstream error;
      error << "ERROR: " << __FILE__ << ":" << __LINE__ << ": DB corrupt.";
      throw std::runtime_error(error.str());
    }
  }

Here is the call graph for this function:

havoqgt::distributed_db::~distributed_db ( )

inline

Definition at line 202 of file distributed_db.hpp.

  {
    //
    // Mark clean close
    std::pair<header*, std::size_t> ret = m_pm->find<header>(boost::interprocess::unique_instance);
    if(ret.second == 0) {
      std::stringstream error;
      error << "ERROR: " << __FILE__ << ":" << __LINE__ << ": header now found.";
      throw std::runtime_error(error.str());
    }
    ret.first->clean_close = true;

    delete m_pm;
    m_pm = nullptr;
    bool shrink_ret = mapped_type::shrink_to_fit(m_rank_filename.c_str());
  }

Member Function Documentation

uint64_t havoqgt::distributed_db::get_file_size ( )

inlineprivate

Definition at line 229 of file distributed_db.hpp.

  {
    const char* fsize = getenv("HAVOQGT_DB_SIZE");
    if(fsize == NULL) {
      //return 700ULL * 1024ULL * 1024ULL * 1024ULL / 24;
      //return 799700000000ULL;
      //return 500ULL * 1024ULL * 1024ULL * 1024ULL;
      // causing probs return 799700000000ULL / 24ULL;
      // MAXES OUT CATALST!! return 799595142400 / 24ULL;
      return 512*1024*1024 / havoqgt_env()->node_local_comm().size();
    }
    return boost::lexical_cast<uint64_t>(fsize);
  } 

Here is the call graph for this function:

Here is the caller graph for this function:

segment_manager_type* havoqgt::distributed_db::get_segment_manager ( )

inline

Definition at line 219 of file distributed_db.hpp.

  {
    return m_pm->get_segment_manager();
  }

Here is the caller graph for this function:

void havoqgt::distributed_db::init_rank_filename ( const char *  base_fname )

inlineprivate

Definition at line 256 of file distributed_db.hpp.

  {
    int mpi_rank = havoqgt_env()->world_comm().rank();
    int mpi_size = havoqgt_env()->world_comm().size();
    std::stringstream sstr;
    sstr << base_fname << "_" << mpi_rank << "_of_" << mpi_size;
    m_rank_filename = sstr.str();
  }

Here is the call graph for this function:

Here is the caller graph for this function:

bool havoqgt::distributed_db::rank_file_exists ( )

inlineprivate

Definition at line 247 of file distributed_db.hpp.

  {
    std::ifstream fin(m_rank_filename.c_str());
    return fin.good();
  }

Here is the caller graph for this function:

Member Data Documentation

mapped_type* havoqgt::distributed_db::m_pm

private

Definition at line 267 of file distributed_db.hpp.

std::string havoqgt::distributed_db::m_rank_filename

private

Definition at line 268 of file distributed_db.hpp.

---

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

• include/havoqgt/distributed_db.hpp