cnamedpipecomm.h Source File¶

Parallel Simulation API: cnamedpipecomm.h Source File
Parallel Simulation API
//=========================================================================
//  CNAMEDPIPECOMM.H - part of
//
//                     OMNeT++/OMNEST
//            Discrete System Simulation in C++
//
//=========================================================================
 
/*--------------------------------------------------------------*
  Copyright (C) 1992-2017 Andras Varga
  Copyright (C) 2006-2017 OpenSim Ltd.
 
  This file is distributed WITHOUT ANY WARRANTY. See the file
  `license' for details on this and other legal matters.
*--------------------------------------------------------------*/
 
#ifndef __OMNETPP_CNAMEDPIPECOMM_H
#define __OMNETPP_CNAMEDPIPECOMM_H
 
 
#include <cstdio>
#include <list>
#include "omnetpp/simutil.h"
#include "omnetpp/opp_string.h"
#include "omnetpp/cparsimcomm.h"
#include "omnetpp/platdep/platmisc.h"  // for <windows.h>
 
// decide platform
#if defined(_WIN32)
#define USE_WINDOWS_PIPES
#endif
 
namespace omnetpp {
 
#ifdef USE_WINDOWS_PIPES
typedef HANDLE PIPE;
#else
typedef int PIPE;
#endif
 
class cMemCommBuffer;
class SIM_API cNamedPipeCommunications : public cParsimCommunications
{
  protected:
    int numPartitions;
    int myProcId;
 
    // pipes
    opp_string prefix;
    PIPE *rpipes = nullptr;
    PIPE *wpipes = nullptr;
    int maxFdPlus1;
    int rrBase = 0;
 
    // reordering buffer needed because of tag filtering support (filtTag)
    struct ReceivedBuffer {int receivedTag; int sourceProcId; cMemCommBuffer *buffer;};
    std::list<ReceivedBuffer> receivedBuffers;
 
  protected:
    // common impl. for receiveBlocking() and receiveNonblocking()
    bool receive(int filtTag, cCommBuffer *buffer, int& receivedTag, int& sourceProcId, bool blocking);
    bool doReceive(cCommBuffer *buffer, int& receivedTag, int& sourceProcId, bool blocking);
 
  public:
    cNamedPipeCommunications();
    virtual ~cNamedPipeCommunications();
    virtual void init(int numPartitions) override;
    virtual void shutdown() override;
    virtual int getNumPartitions() const override;
    virtual int getProcId() const override;
    virtual cCommBuffer *createCommBuffer() override;
    virtual void recycleCommBuffer(cCommBuffer *buffer) override;
    virtual void send(cCommBuffer *buffer, int tag, int destination) override;
    virtual bool receiveBlocking(int filtTag, cCommBuffer *buffer, int& receivedTag, int& sourceProcId) override;
    virtual bool receiveNonblocking(int filtTag, cCommBuffer *buffer,  int& receivedTag, int& sourceProcId) override;
};
 
}  // namespace omnetpp
 
 
#endif
 
 