PLearn 0.1
|
An implementation of the PStreamBuf interface using MPI communication. More...
#include <MPIPStreamBuf.h>
Public Member Functions | |
MPIPStreamBuf (int peer_rank, streamsize chunksize=1024) | |
virtual | ~MPIPStreamBuf () |
Protected Member Functions | |
virtual streamsize | read_ (char *p, streamsize n) |
reads up to n characters into p You should override this call in subclasses. | |
virtual void | write_ (const char *p, streamsize n) |
writes exactly n characters from p (unbuffered, must flush) | |
Protected Attributes | |
int | peerrank |
rank of MPI peer | |
Private Types | |
typedef PStreamBuf | inherited |
Private Member Functions | |
void | fill_mpibuf (int msglength) |
Private Attributes | |
char * | mpibuf |
streamsize | mpibuf_capacity |
char * | mpibuf_p |
streamsize | mpibuf_len |
bool | reached_eof |
streamsize | max_chunksize |
Friends | |
class | PLearn::Poll |
An implementation of the PStreamBuf interface using MPI communication.
Definition at line 54 of file MPIPStreamBuf.h.
typedef PStreamBuf PLearn::MPIPStreamBuf::inherited [private] |
Reimplemented from PLearn::PStreamBuf.
Definition at line 60 of file MPIPStreamBuf.h.
PLearn::MPIPStreamBuf::MPIPStreamBuf | ( | int | peer_rank, |
streamsize | chunksize = 1024 |
||
) |
Definition at line 52 of file MPIPStreamBuf.cc.
: PStreamBuf(true, true, chunksize, chunksize, default_ungetsize), mpibuf(0), mpibuf_capacity(0), mpibuf_p(0), mpibuf_len(0), reached_eof(false), max_chunksize(chunksize), peerrank(peer_rank) {}
PLearn::MPIPStreamBuf::~MPIPStreamBuf | ( | ) | [virtual] |
void PLearn::MPIPStreamBuf::fill_mpibuf | ( | int | msglength | ) | [private] |
Definition at line 69 of file MPIPStreamBuf.cc.
References PLearn::max(), mpibuf, mpibuf_capacity, mpibuf_len, mpibuf_p, peerrank, and PLearn::PLMPI::tag.
Referenced by read_().
{ if (msglength>int(mpibuf_capacity)) { if(mpibuf) delete[] mpibuf; // inrease capacity mpibuf_capacity = max(msglength,int(mpibuf_capacity)*2); mpibuf = new char[mpibuf_capacity]; } MPI_Status status; mpibuf_p = mpibuf; MPI_Recv(mpibuf_p, msglength, MPI_CHAR, peerrank, PLMPI::tag, MPI_COMM_WORLD, &status); mpibuf_len = msglength; }
MPIPStreamBuf::streamsize PLearn::MPIPStreamBuf::read_ | ( | char * | p, |
streamsize | n | ||
) | [protected, virtual] |
reads up to n characters into p You should override this call in subclasses.
Default version issues a PLERROR
On success, the number of bytes read is returned. Zero indicates end of file. If we are not at end of file, at least one character should be returned (the call must block until at least one char is available). It is not an error if the number returned is smaller than the number of bytes requested; this may happen for example because fewer bytes are actually available right now (maybe because we were close to end-of-file, or because we are reading from a pipe, or from a terminal). If an error occurs, an exception should be thrown.
Reimplemented from PLearn::PStreamBuf.
Definition at line 85 of file MPIPStreamBuf.cc.
References std::copy(), fill_mpibuf(), m, PLearn::min(), mpibuf_len, mpibuf_p, n, peerrank, reached_eof, and PLearn::PLMPI::tag.
{ int msglength = 0; MPI_Status status; streamsize nread = 0; if (mpibuf_len>0) { streamsize m = min(n,mpibuf_len); copy(mpibuf_p, mpibuf_p+m, p); p += m; mpibuf_p += m; mpibuf_len -= m; n -= m; nread += m; } if (!reached_eof && n>0) { // MPI_Probe(peerrank, tag, MPI_COMM_WORLD, &status); MPI_Get_count(&status, MPI_CHAR, &msglength); if (msglength==0) // we reached EOF reached_eof = true; else { if (msglength<=(int)n) { MPI_Recv(p, msglength, MPI_CHAR, peerrank, PLMPI::tag, MPI_COMM_WORLD, &status); nread += msglength; } else // msglength>n { fill_mpibuf(msglength); copy(mpibuf_p, mpibuf_p+n, p); mpibuf_p += n; mpibuf_len -= n; nread += n; } } } return nread; }
void PLearn::MPIPStreamBuf::write_ | ( | const char * | p, |
streamsize | n | ||
) | [protected, virtual] |
writes exactly n characters from p (unbuffered, must flush)
Reimplemented from PLearn::PStreamBuf.
Definition at line 128 of file MPIPStreamBuf.cc.
References m, max_chunksize, PLearn::min(), peerrank, PLERROR, and PLearn::PLMPI::tag.
{ while (n>0) { streamsize m = min(max_chunksize,n); if( MPI_Send((void *)p, m, MPI_CHAR, peerrank, PLMPI::tag, MPI_COMM_WORLD) != MPI_SUCCESS ) PLERROR("MPI_Send failed"); p += m; n -= m; } }
friend class PLearn::Poll [friend] |
Definition at line 56 of file MPIPStreamBuf.h.
Definition at line 66 of file MPIPStreamBuf.h.
Referenced by write_().
char* PLearn::MPIPStreamBuf::mpibuf [private] |
Definition at line 61 of file MPIPStreamBuf.h.
Referenced by fill_mpibuf(), and ~MPIPStreamBuf().
Definition at line 62 of file MPIPStreamBuf.h.
Referenced by fill_mpibuf().
streamsize PLearn::MPIPStreamBuf::mpibuf_len [private] |
Definition at line 64 of file MPIPStreamBuf.h.
Referenced by fill_mpibuf(), and read_().
char* PLearn::MPIPStreamBuf::mpibuf_p [private] |
Definition at line 63 of file MPIPStreamBuf.h.
Referenced by fill_mpibuf(), and read_().
int PLearn::MPIPStreamBuf::peerrank [protected] |
rank of MPI peer
Definition at line 74 of file MPIPStreamBuf.h.
Referenced by fill_mpibuf(), read_(), and write_().
bool PLearn::MPIPStreamBuf::reached_eof [private] |
Definition at line 65 of file MPIPStreamBuf.h.
Referenced by read_().