00001 #ifndef MPIStreamFilter_h
00002 #define MPIStreamFilter_h
00003
00004 #include "RecordStreamFilter.h"
00005 #include "RecordBuffer.h"
00006 #include <mpi.h>
00007
00008
00009 namespace BIE {
00010
00011 class MPICommunicationSession;
00012
00014 class MPIStreamFilter : public RecordStreamFilter {
00015 public:
00016
00017 MPIStreamFilter(RecordStream * stream, MPI_Comm communicator,
00018 int sessionId, int bufferCapacityInWords=1<<11);
00019 virtual ~MPIStreamFilter();
00020 void compute();
00021 void startNewSession(int newSessionId);
00022 void finishSession();
00023 void flush();
00024 void setHeaderRecordBuffer(RecordBuffer *hdrbuf);
00025 static void *packRecordBuffer(RecordBuffer *rb, int* packedsize);
00026 static int getRecordBufferSizeInBytes(RecordType *rt);
00027
00028 friend class MPICommunicationSession;
00029
00030 private:
00031 typedef void (BIE::MPIStreamFilter::*storeFn)(int fieldindex);
00032
00033 void storeInt(int fieldindex);
00034 void storeReal(int fieldindex);
00035 void storeBool(int fieldindex);
00036
00037 RecordType *_rtype;
00038 int _numFields;
00039 void *_buffer;
00040 void *_bufferIndex;
00041 int _bufferSize;
00042 int _recordSize;
00043 int _capacityInRecords;
00044 int _numRecords;
00045 storeFn *_sfn;
00046 filterIn *input;
00047 filterOut *output;
00048
00049 bool _inSession;
00050
00051 MPICommunicationSession *_mpiSendSession;
00052 static int _defaultBufferSize;
00053
00054 RecordBuffer *_hdrBuffer;
00055 void *_hdr;
00056 };
00057 }
00058 #endif
