librheolef-dev 6.5-1build1 / 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 {
public:
     typedef std::size_t size_type;

// allocators/deallocators:

     odiststream()
       : _ptr_os(0), _use_alloc(false), _comm() {}

     odiststream(std::ostream& os, const communicator& comm = communicator())
       : _ptr_os(&os), _use_alloc(false), _comm(comm) {}

     odiststream (std::string filename, std::string suffix = "",
	     bool use_gzip = true,
             const communicator& comm = communicator())
       : _ptr_os(0), _use_alloc(false), _comm()
       {
         open (filename, suffix, use_gzip, comm);
       }
     ~odiststream();

// modifiers:

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

// accessors:

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

// internal accesses:
public:
     std::ostream& os() {
	check_macro (_ptr_os != 0, "try to use an uninitialized odiststream");
	return *_ptr_os;
     }
#ifndef _RHEOLEF_HAVE_MPI
     bool nop() { return false; }
#else
     bool nop() { return size_type(_comm.rank()) != io_proc(); }
#endif //_RHEOLEF_HAVE_MPI

// data:
protected:
     std::ostream* _ptr_os;
     bool          _use_alloc;
     communicator  _comm;
};
// 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_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&))

class idiststream {
public:
     typedef std::size_t size_type;

// allocators/deallocators:
     
     idiststream()
       : _ptr_is(0), _use_alloc(false), _comm() {}

     idiststream (std::istream& is, const communicator& comm = communicator())
       : _ptr_is(&is), _use_alloc(false), _comm(comm) {}

     idiststream (std::string filename, std::string suffix = "",
             const communicator& comm = communicator())
       : _ptr_is(0), _use_alloc(false), _comm()
       {
         open (filename, suffix, comm);
       }

      ~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();

// internal accesses:
public:
#ifndef _RHEOLEF_HAVE_MPI
     bool nop() { return false; }
#else
     bool nop() { return size_type(_comm.rank()) != io_proc(); }
#endif //_RHEOLEF_HAVE_MPI

     std::istream& is() {
	check_macro (_ptr_is != 0, "try to use an uninitialized idiststream");
	return *_ptr_is;
     }
#ifndef _RHEOLEF_HAVE_MPI
     bool do_load() { return true; }
#else
     bool do_load() { return size_type(_comm.rank()) == io_proc(); }
#endif // _RHEOLEF_HAVE_MPI

// data:
protected:
     std::istream* _ptr_is;
     bool          _use_alloc;
     communicator  _comm;
};

#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_DOUBLEDOUBLE
define_sequential_idiststream_macro(doubledouble)
#endif // _RHEOLEF_HAVE_DOUBLEDOUBLE

// predefined distributed streams
extern idiststream din;
extern odiststream dout;
extern odiststream dclog;
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