diststream.cc

Go to the documentation of this file.

 
#include "rheolef/diststream.h"
#include "rheolef/environment.h"
#include "rheolef/rheostream.h" // scatch()
using namespace std;
namespace rheolef {
// ----------------------------------------------------------------------------
// global variables
// ----------------------------------------------------------------------------

idiststream din  (cin);
odiststream dout (cout);
odiststream dlog (clog);
odiststream derr (cerr);
 
// ----------------------------------------------------------------------------
// ----------------------------------------------------------------------------
bool
dis_scatch (idiststream& ips, const communicator& comm, std::string ch)
{
  // TODO: comm is in ips.comm()
  typedef std::size_t size_type;
  size_type io_proc = idiststream::io_proc();
  size_type my_proc = comm.rank();
  bool status = false;
  if (my_proc == io_proc) {
    status = scatch(ips.is(),ch);
  }
#ifdef _RHEOLEF_HAVE_MPI
  mpi::broadcast (comm, status, io_proc);
#endif // _RHEOLEF_HAVE_MPI
  return status;
}
// --------------------------------------------------------------
// intput
// --------------------------------------------------------------
idiststream::size_type
idiststream::io_proc() {
#ifndef _RHEOLEF_HAVE_MPI
    return 0;
#else // _RHEOLEF_HAVE_MPI
    boost::optional<int> opt_io_proc = environment::io_rank();
    check_macro (opt_io_proc, "no process can perform i/o");
    int io_proc = opt_io_proc.get();
    if (io_proc == mpi::any_source) {
        io_proc = 0;
    }
    return size_type(io_proc);
#endif // _RHEOLEF_HAVE_MPI
}
odiststream::size_type
odiststream::io_proc() 
{
  return idiststream::io_proc(); 
}

void
idiststream::open (
  std::string filename, 
  std::string suffix,
  const communicator& comm)
{
  close();
  if (_use_alloc && _ptr_is != 0) {
    delete_macro (_ptr_is);
    _use_alloc = false;
    _ptr_is = 0;
  }
  if (size_type(comm.rank()) == idiststream::io_proc()) {
    _ptr_is = new_macro (irheostream(filename, suffix));
  } else {
    _ptr_is = new_macro (irheostream);
  }
  _comm = comm;
  _use_alloc = true;
}
void
idiststream::close ()
{
  if (_use_alloc && _ptr_is != 0) {
    if (size_type(_comm.rank()) == idiststream::io_proc()) {
      irheostream* ptr_irs = (irheostream*)(_ptr_is);
      (*ptr_irs).close();
    }
  }
}
idiststream::~idiststream ()
{
  close();
  if (_use_alloc && _ptr_is != 0) {
    delete_macro (_ptr_is);
    _use_alloc = false;
    _ptr_is = 0;
  }
}
bool
idiststream::good() const
{
    bool status;
    if (size_type(comm().rank()) != idiststream::io_proc()) {
        status = true;
    } else if (_ptr_is == 0) {
        status = false;
    } else {
        status = (*_ptr_is).good();
    }
#ifdef _RHEOLEF_HAVE_MPI
    mpi::broadcast(comm(), status, 0);
#endif // _RHEOLEF_HAVE_MPI
    return status;
}
// --------------------------------------------------------------
// output
// --------------------------------------------------------------
void
odiststream::open (
  std::string filename, 
  std::string suffix,
  io::mode_type mode,
  const communicator& comm)
{
  close();
  if (_use_alloc && _ptr_os != 0) {
    delete_macro (_ptr_os);
    _use_alloc = false;
    _ptr_os = 0;
  }
  if (size_type(comm.rank()) == odiststream::io_proc()) {
    _ptr_os = new_macro (orheostream(filename, suffix, mode));
  } else {
    _ptr_os = new_macro (orheostream);
  }
  _comm = comm;
  _use_alloc = true;
}
void
odiststream::close ()
{
  if (_use_alloc && _ptr_os != 0) {
    if (size_type(_comm.rank()) == odiststream::io_proc()) {
      orheostream* ptr_ors = (orheostream*)(_ptr_os);
      (*ptr_ors).close();
    }
  }
}
void
odiststream::flush()
{
  if (size_type(_comm.rank()) != odiststream::io_proc()) return;
  if (_use_alloc && _ptr_os != 0) {
    orheostream* ptr_ors = (orheostream*)(_ptr_os);
    (*ptr_ors).flush();
  } else if (_ptr_os != 0) {
    (*_ptr_os).flush(); // call usual ostream::flush(), i.e. not gziped output
  }
}
odiststream::~odiststream ()
{
  close();
  if (_use_alloc && _ptr_os != 0) {
    delete_macro (_ptr_os);
    _use_alloc = false;
    _ptr_os = 0;
  }
}
bool
odiststream::good() const
{
    bool status;
    if (size_type(comm().rank()) != idiststream::io_proc()) {
        status = true;
    } else if (_ptr_os == 0) {
        status = false;
    } else {
        status = (*_ptr_os).good();
    }
#ifdef _RHEOLEF_HAVE_MPI
    mpi::broadcast(comm(), status, 0);
#endif // _RHEOLEF_HAVE_MPI
    return status;
}
// -----------------------------------------
// system utilities
// -----------------------------------------
int
dis_system (const std::string& command, const communicator& comm)
{
  typedef odiststream::size_type size_type;
  size_type io_proc = odiststream::io_proc();
  size_type my_proc = comm.rank();
  int status = 0;
#ifdef _RHEOLEF_HAVE_MPI
  mpi::communicator().barrier();
#endif // _RHEOLEF_HAVE_MPI
  if (my_proc == io_proc) {
    status = std::system (command.c_str());
  }
#ifdef _RHEOLEF_HAVE_MPI
  mpi::communicator().barrier();
  mpi::broadcast (mpi::communicator(), status, io_proc); 
#endif // _RHEOLEF_HAVE_MPI
  return status;
}
bool
dis_file_exists (const std::string& filename, const communicator& comm)
{
  typedef odiststream::size_type size_type;
  size_type io_proc = odiststream::io_proc();
  size_type my_proc = comm.rank();
  bool status = false;
  if (my_proc == io_proc) {
    status = file_exists (filename);
  }
#ifdef _RHEOLEF_HAVE_MPI
  mpi::broadcast (mpi::communicator(), status, io_proc); 
#endif // _RHEOLEF_HAVE_MPI
  return status;
}
 
} // namespace rheolef