librheolef-dev 6.7-6 / usr / include / rheolef / diststream.h

# ifndef _RHEOLEF_DISTSTREAM_H
# define _RHEOLEF_DISTSTREAM_H
///
/// This file is part of Rheolef.
///
/// Copyright (C) 2000-2009 Pierre Saramito <Pierre.Saramito@imag.fr>
///
/// Rheolef 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; either version 2 of the License, or
/// (at your option) any later version.
///
/// Rheolef 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 Rheolef; if not, write to the Free Software
/// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
/// 
/// =========================================================================
//
// distributed i/o
//
// author: Pierre.Saramito@imag.fr
//
// date: 13 nov 1998
//
#include "rheolef/distributed.h"
#include "rheolef/dis_macros.h"
#include "rheolef/catchmark.h"
namespace rheolef {

/*Class:odiststream
NAME:  @code{idiststream}, @code{odiststream} - distributed interface for large data streams 
@clindex idiststream
@clindex odiststream
@findex din
@findex dout
@findex derr
@fiindex @file{.gz} gzip
@cindex RHEOPATH environment variable
DESCRIPTION:
  @noindent
  This class provides a parallel and distributed stream interface for large
  data management.
  File decompresion is assumed using
  @code{gzip} and a recursive seach in a directory list
  is provided for input.
  @example
     odiststream foo("NAME", "suffix");
  @end example
  @noindent
  is similar
  @example
     ofstream foo("NAME.suffix").
  @end example
  The main difference is that writting to @code{ofstream} is executed on all processes
  while @code{odiststream} manage nicely the multi-process environment.
  For convenience, the standard streams @code{cin}, @code{cout} and @code{cerr}
  are extended to @code{din}, @code{dout} and @code{derr}.
  @noindent
  Notice that, as @code{orheostream} (@pxref{rheostream class}),
  if @var{NAME} does not end with @file{.suffix}, then
  @file{.suffix} is automatically added.
  By default, compression is performed on the fly with gzip, adding an additional @file{.gz} suffix.
  The @code{flush} action is nicely handled in compression mode:
  @example
     foo.flush();
  @end example
  This feature allows intermediate results to be available during long computations.
  The compression can be deactivated while opening a file by an optional argument:
  @example
     odiststream foo("NAME", "suffix", io::nogz);
  @end example
  An existing compressed file can be reopen in @code{append} mode: new results will
  be appended at the end of an existing file:
  @example
     odiststream foo("NAME", "suffix", io::app);
  @end example

  @noindent
  Conversely,
  @example
 	irheostream foo("NAME","suffix");
  @end example
  @noindent
  is similar to
  @example
	ifstream foo("NAME.suffix").
  @end example
  @noindent
  Also @code{idiststream} manage nicely the multi-process environment.
  However, we look at a search path environment variable 
  @code{RHEOPATH} in order to find @var{NAME} while suffix is assumed.
  Moreover, @code{gzip} compressed files, ending
  with the @file{.gz} suffix is assumed, and decompression is done.


AUTHORS:
    LMC-IMAG, 38041 Grenoble cedex 9, France
   | Pierre.Saramito@imag.fr
DATE:   31 october 1997
METHODS: @rheostream
End:
*/
//<verbatim:
class odiststream {
public:
  typedef std::size_t size_type;

// allocators/deallocators:

  odiststream();
  odiststream (std::string filename, std::string suffix = "",
              io::mode_type mode = io::out, const communicator& comm = communicator());
  odiststream (std::string filename,
              io::mode_type mode, const communicator& comm = communicator());
  odiststream (std::string filename, std::string suffix, const communicator& comm);
  odiststream (std::string filename, const communicator& comm);
  odiststream(std::ostream& os, const communicator& comm = communicator());
  ~odiststream();

// modifiers:

   void open (std::string filename, std::string suffix = "",
	     io::mode_type mode = io::out, const communicator& comm = communicator());
   void open (std::string filename, 
	     io::mode_type mode, const communicator& comm = communicator());
   void open (std::string filename, std::string suffix,
	     const communicator& comm);
   void open (std::string filename, const communicator& comm);
   void flush();
   void close();

// accessors:

   const communicator& comm() const { return _comm; }
   bool good() const;
   operator bool() const { return good(); }
   static size_type io_proc();

// internals:

   std::ostream& os();
   bool nop();

protected:
// data:
   std::ostream* _ptr_os;
   bool          _use_alloc;
   communicator  _comm;
};
//>verbatim:

// ------------------------------------------------------------------
// inlined
// ------------------------------------------------------------------
inline
odiststream::odiststream()
 : _ptr_os(0), _use_alloc(false), _comm()
{
}
inline
odiststream::odiststream (std::string filename, std::string suffix, io::mode_type mode, const communicator& comm)
 : _ptr_os(0), _use_alloc(false), _comm()
{
  open (filename, suffix, mode, comm);
}
inline
odiststream::odiststream (std::string filename, io::mode_type mode, const communicator& comm)
 : _ptr_os(0), _use_alloc(false), _comm()
{
  open (filename, mode, comm);
}
inline
odiststream::odiststream (std::string filename, std::string suffix, const communicator& comm)
 : _ptr_os(0), _use_alloc(false), _comm()
{
  open (filename, suffix, comm);
}
inline
odiststream::odiststream (std::string filename, const communicator& comm)
 : _ptr_os(0), _use_alloc(false), _comm()
{
  open (filename, comm);
}
inline
odiststream::odiststream(std::ostream& os, const communicator& comm)
 : _ptr_os(&os), _use_alloc(false), _comm(comm)
{
}
inline
void
odiststream::open (std::string filename, io::mode_type mode, const communicator& comm)
{
  open (filename, std::string(""), mode, comm);
}
inline
void 
odiststream::open (std::string filename, std::string suffix, const communicator& comm)
{
  open (filename, suffix, io::out, comm);
}
inline
void
odiststream::open (std::string filename, const communicator& comm)
{
  open (filename, std::string(""), io::out, comm);
}
inline
std::ostream&
odiststream::os() {
  check_macro (_ptr_os != 0, "try to use an uninitialized odiststream");
  return *_ptr_os;
}
#ifndef _RHEOLEF_HAVE_MPI
inline bool odiststream::nop() { return false; }
#else
inline bool odiststream::nop() { return size_type(_comm.rank()) != io_proc(); }
#endif //_RHEOLEF_HAVE_MPI

// ------------------------------------------------------------------
// standard i/o:
// ------------------------------------------------------------------
# define define_sequential_odiststream_raw_macro(arg) 	\
    inline 						\
    odiststream&					\
    operator << (odiststream& s, arg) {			\
        if (s.nop()) return s; s.os() << x; return s;	\
    }
# define define_sequential_odiststream_macro(T) 	\
    define_sequential_odiststream_raw_macro(const T& x)

# define define_distributed_odiststream_macro(T) 	\
    inline 						\
    odiststream&					\
    operator << (odiststream& s, const T& x) {		\
        s.os() << x; return s;				\
    }

template <class T>
inline
odiststream&
operator << (odiststream& s, T x)
{
    if (s.nop()) return s;
    s.os() << x;
    return s;
}
define_sequential_odiststream_macro(char)
define_sequential_odiststream_macro(int)
define_sequential_odiststream_macro(unsigned int)
define_sequential_odiststream_macro(long int)
define_sequential_odiststream_macro(long unsigned int)
define_sequential_odiststream_macro(float)
define_sequential_odiststream_macro(double)
define_sequential_odiststream_macro(long double)
define_sequential_odiststream_macro(char*const)
define_sequential_odiststream_macro(std::string)
#ifdef _RHEOLEF_HAVE_FLOAT128
define_sequential_odiststream_macro(float128)
#endif // _RHEOLEF_HAVE_FLOAT128 
#ifdef _RHEOLEF_HAVE_QD
define_sequential_odiststream_macro(dd_real)
define_sequential_odiststream_macro(qd_real)
#endif // _RHEOLEF_HAVE_QD
#ifdef _RHEOLEF_HAVE_DOUBLEDOUBLE
define_sequential_odiststream_macro(doubledouble)
#endif // _RHEOLEF_HAVE_DOUBLEDOUBLE
define_sequential_odiststream_raw_macro(char *x)
define_sequential_odiststream_raw_macro(std::ostream& (*x)(std::ostream&))

//<verbatim:
class idiststream {
public:
  typedef std::size_t size_type;

// allocators/deallocators:
     
  idiststream();
  idiststream (std::istream& is, const communicator& comm = communicator());
  idiststream (std::string filename, std::string suffix = "",
             const communicator& comm = communicator());
  ~idiststream();

// modifiers:

  void open (std::string filename, std::string suffix = "",
             const communicator& comm = communicator());
  void close();

// accessors:

  const communicator& comm() const { return _comm; }
  bool good() const;
  operator bool() const { return good(); }
  static size_type io_proc();

// internals:

  std::istream& is();
  bool nop();
  bool do_load();

protected:
// data:
  std::istream* _ptr_is;
  bool          _use_alloc;
  communicator  _comm;
};
//>verbatim:
// ------------------------------------------------------------------
// inlined
// ------------------------------------------------------------------
inline
idiststream::idiststream()
 : _ptr_is(0), _use_alloc(false), _comm()
{
}
inline
idiststream::idiststream (std::istream& is, const communicator& comm)
 : _ptr_is(&is), _use_alloc(false), _comm(comm)
{
}
inline
idiststream::idiststream (std::string filename, std::string suffix, const communicator& comm)
 : _ptr_is(0), _use_alloc(false), _comm()
{
  open (filename, suffix, comm);
}
inline 
std::istream&
idiststream::is()
{
  check_macro (_ptr_is != 0, "try to use an uninitialized idiststream");
  return *_ptr_is;
}
#ifndef _RHEOLEF_HAVE_MPI
inline bool idiststream::nop()     { return false; }
inline bool idiststream::do_load() { return true; }
#else
inline bool idiststream::nop()     { return size_type(_comm.rank()) != io_proc(); }
inline bool idiststream::do_load() { return size_type(_comm.rank()) == io_proc(); }
#endif //_RHEOLEF_HAVE_MPI

// ------------------------------------------------------------------
// standard i/o:
// ------------------------------------------------------------------
#ifdef _RHEOLEF_HAVE_MPI
# define define_sequential_idiststream_macro(T)		\
inline							\
idiststream&						\
operator >> (idiststream& s, T& x)			\
{							\
  if (s.do_load()) { (s.is()) >> x; }			\
  mpi::broadcast (mpi::communicator(), x, s.io_proc());	\
  return s;						\
}
#else // _RHEOLEF_HAVE_MPI
# define define_sequential_idiststream_macro(T)		\
inline							\
idiststream&						\
operator >> (idiststream& s, T& x)			\
{							\
  (s.is()) >> x; 					\
  return s;						\
}
#endif // _RHEOLEF_HAVE_MPI
# define define_distributed_idiststream_macro(T) 	\
inline							\
idiststream&						\
operator >> (idiststream& s, T& x)			\
{							\
  s.is() >> x; 						\
  return s;						\
}

define_sequential_idiststream_macro(char)
define_sequential_idiststream_macro(int)
define_sequential_idiststream_macro(long int)
define_sequential_idiststream_macro(unsigned int)
define_sequential_idiststream_macro(long unsigned int)
define_sequential_idiststream_macro(float)
define_sequential_idiststream_macro(double)
define_sequential_idiststream_macro(long double)
define_sequential_idiststream_macro(std::string)
#ifdef _RHEOLEF_HAVE_FLOAT128
define_sequential_idiststream_macro(float128)
#endif // _RHEOLEF_HAVE_FLOAT128 
#ifdef _RHEOLEF_HAVE_QD
define_sequential_idiststream_macro(dd_real)
define_sequential_idiststream_macro(qd_real)
#endif // _RHEOLEF_HAVE_QD
#ifdef _RHEOLEF_HAVE_DOUBLEDOUBLE
define_sequential_idiststream_macro(doubledouble)
#endif // _RHEOLEF_HAVE_DOUBLEDOUBLE

// predefined distributed streams
extern idiststream din;
extern odiststream dout;
extern odiststream dlog;
extern odiststream derr;

bool dis_scatch (idiststream& ips, const communicator& comm, std::string ch);

inline
bool dis_scatch (idiststream& ips, std::string ch)
{
  return dis_scatch (ips, ips.comm(), ch);
}
inline
idiststream&
operator>> (idiststream& ids, const catchmark& m)
{
  if (ids.nop()) return ids;
  ids.is() >> setmark(m.mark());
  std::string label = "#" + m.mark();
  if (!scatch(ids.is(),label)) {
    bool verbose  = iorheo::getverbose(ids.is());
    if (verbose) warning_macro ("catchmark: label `"<< label <<"' not found on input");
  }
  return ids;
}
inline
odiststream&
operator<< (odiststream& ods, const catchmark& m)
{
    if (ods.nop()) return ods;
    ods.os() << setmark(m.mark())
             << "#" << m.mark() << std::endl;
    return ods;
}
// system utilities:
int dis_system (const std::string& command, const communicator& comm = communicator());
bool dis_file_exists (const std::string& filename, const communicator& comm = communicator());


} // namespace rheolef
# endif // _RHEOLEF_DISTSTREAM_H