Home Research Papers Posters Manuals Talks Download People Help Internal

conv-com/StreamingStrategy.h

Go to the documentation of this file.

#ifndef STREAMING_STRATEGY
#define STREAMING_STRATEGY
//#include "convcomlibmanager.h"
#include "convcomlibstrategy.h"

#define MAX_STREAMING_MESSAGE_SIZE 2048*2
#define MAX_NUM_STREAMING_MESSAGES 1000
#define DEFAULT_TIMEOUT 10

CpvExtern(int, streaming_handler_id);
extern void StreamingHandlerFn(void *msg);

struct StreamingMessage {
  char header[CmiReservedHeaderSize];
  CmiUInt4 srcPE;
  CmiUInt4 nmsgs;
};

PUPbytes(StreamingMessage)


class StreamingStrategy : public Strategy {
 protected:
    CkQ<MessageHolder *> *streamingMsgBuf;
    int *streamingMsgCount;
    int *bufSize;
    int bufferMax;
    int msgSizeMax;
    int bufSizeMax;
    double PERIOD;
    //CmiBool shortMsgPackingFlag;
    CmiBool idleFlush;

    //int streaming_handler_id; //Getting rid of multiple send

    void flushPE(int destPE);
    
 public:
    StreamingStrategy(int periodMs=DEFAULT_TIMEOUT, 
                      int bufferMax=MAX_NUM_STREAMING_MESSAGES,
                      int msgSizeMax=MAX_STREAMING_MESSAGE_SIZE, 
                      int bufSizeMax=MAX_STREAMING_MESSAGE_SIZE*MAX_NUM_STREAMING_MESSAGES);
    StreamingStrategy(double periodMs=DEFAULT_TIMEOUT, 
                      int bufferMax=MAX_NUM_STREAMING_MESSAGES, 
                      int msgSizeMax=MAX_STREAMING_MESSAGE_SIZE, 
                      int bufSizeMax=MAX_STREAMING_MESSAGE_SIZE*MAX_NUM_STREAMING_MESSAGES);

    StreamingStrategy(CkMigrateMessage *m) : Strategy(m) {}
    
    virtual void insertMessage(MessageHolder *msg);
    virtual void doneInserting();
    
    virtual void handleMessage(void *msg) {
      CmiAbort("[%d] StreamingStrategy::handleMessage should never be called\n");
    }

    //virtual void beginProcessing(int ignored);

    virtual void pup(PUP::er &p);
    //virtual void enableShortArrayMessagePacking()
    //    {shortMsgPackingFlag=CmiTrue;} //Should be used only for array
                                       //messages

    virtual void disableIdleFlush() { idleFlush = CmiFalse;}

    void registerFlush(void);
    void periodicFlush();

    PUPable_decl(StreamingStrategy);
};
#endif

---