NAME¶
idiststream, odiststream - distributed interface for large data
streams
DESCRIPTION¶
This class provides a parallel and distributed stream interface for large data
management. File decompresion is assumed using gzip and a recursive
seach in a directory list is provided for input.
odiststream foo("NAME", "suffix");
is similar
ofstream foo("NAME.suffix").
The main difference is that writting to ofstream is executed on all
processes while odiststream manage nicely the multi-process
environment. For convenience, the standard streams cin, cout and
cerr are extended to din, dout and derr. Notice
that, as orheostream (see rheostream(2)), if NAME does not end
with `.suffix', then `.suffix' is automatically added. By
default, compression is performed on the fly with gzip, adding an additional
`.gz' suffix. The flush action is nicely handled in compression
mode:
foo.flush();
This feature allows intermediate results to be available during long
computations. The compression can be deactivated while opening a file by an
optional argument:
odiststream foo("NAME", "suffix", io::nogz);
An existing compressed file can be reopen in append mode: new results
will be appended at the end of an existing file:
odiststream foo("NAME", "suffix", io::app);
Conversely,
irheostream foo("NAME","suffix");
is similar to
ifstream foo("NAME.suffix").
Also idiststream manage nicely the multi-process environment. However, we
look at a search path environment variable RHEOPATH in order to find
NAME while suffix is assumed. Moreover, gzip compressed files,
ending with the `.gz' suffix is assumed, and decompression is done.
IMPLEMENTATION¶
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;
};
IMPLEMENTATION¶
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;
};
