asyConMgr.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 1998 MaK Technologies, Inc.
** All rights reserved.
*******************************************************************************/


#pragma once

#ifndef VXWORKS

#include "rtiMsConfig.h"
#include "connectMgr.h"
#include <vlutil/vlThreadSafe.h>
#include <vlutil/vlMutex.h>
#include "fedAmbCallbackQueue.h"
#include <vlutil/vlLockfreeQueue.h>

#ifdef _WIN32
#include "windows.h"
#else
#include "pthread.h"
#include <vlutil/vlSelectParamW.h>
#endif

#ifdef DtIFSPEC13DLC
#define RTI rti13
#endif

namespace MAKRti
{
   class DtSelectParamWithWrite;
}
class DtFedAmbCallbackMsg;

class DT_DLL_LRC DtQueuePacket
{

public:

   DtQueuePacket( DtRtiMsg* msg,
                  MAKRti::DtTransportType transport,
                  const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   ~DtQueuePacket();

   DtRtiMsg* msg;
   MAKRti::DtInetAddr addr;
   MAKRti::DtTransportType transport;
};

typedef MAKRti::DtLockfreeQueue<DtQueuePacket*> DtAsycQueueType;


class DT_DLL_LRC DtAsyncConnectionMgr : public DtConnectionMgr
{
protected:

   DtAsyncConnectionMgr( DtRIDParameters* params, bool isFederate = false, 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient> assistantClient = 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient>() );

public:

   virtual ~DtAsyncConnectionMgr();

   virtual int init( DtRIDParameters* params );

   virtual void setFederateMgr( DtFederateMgr* fedMgrPtr );

   virtual void shutdown();

   virtual bool isAsynchronous();

   virtual void disconnectFederation();
   virtual void connectFederation(
#ifdef DtIFSPEC1516
      rti1516::FederateAmbassador* fedAmb,
#elif defined(DtIFSPEC13)
      RTI::FederateAmbassador* fedAmb,
#endif
      DtLogicalTimeFactory* logicalTimeFactoryPtr,
      MAKRti::DtString fedExName, 
      DtFederateHandle fedHandle,
      int fedExHandle );

   virtual void queueMsg( DtRtiMsg* msg, 
                          MAKRti::DtTransportType transport,
                          const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() ); 

   virtual int send( const DtRtiMsg& msg, 
                     MAKRti::DtTransportType transport,
                     const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   virtual int sendAndDelete( DtRtiMsg* msg, 
                              MAKRti::DtTransportType transport,
                              const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   virtual bool tick();

   virtual bool processConnections();

   virtual DtRtiMsg* getMessage();

   virtual DtNetReadStatus handleMsg( DtRtiMsg& msg );

   virtual bool flush();

   virtual void receiveThread();
   virtual void sendThread();

   virtual bool wait( MAKRti::DtTime periodInSeconds, bool unlockWhileWaiting = false );

#ifdef DtIFSPEC1516
   virtual int readFileDescriptor(rti1516::TransportationType transport);
#elif defined(DtIFSPEC1516E)
   virtual int readFileDescriptor(rti1516e::TransportationType transport);
#else
   virtual int readFileDescriptor(RTI::TransportType transport);
#endif

#ifdef _WIN32
   virtual HANDLE waitEvent();
#ifdef DtIFSPEC1516
   virtual WSAEVENT readFileEvent(rti1516::TransportationType transport);
#elif defined(DtIFSPEC1516E)
   virtual WSAEVENT readFileEvent(rti1516e::TransportationType transport);
#else
   virtual WSAEVENT readFileEvent(RTI::TransportType transport);
#endif
#else
   virtual int waitEvent();
#endif

   virtual void signalReceiveEvent();

   virtual void clearReceiveEvent();

   virtual void signalSendEvent();

   virtual void clearSendEvent();

   virtual void yield();

   virtual bool processConnectionsInCallback();

protected:

   virtual bool doReceiveProcessing(unsigned long timeout);

   virtual void doDisconnectFederation();

   virtual bool doWait(MAKRti::DtTime period);

   virtual void processCallbackMsg(const DtFedAmbCallbackMsg& callbackMsg);

   virtual bool sendPackets();

   virtual DtNetReadStatus receivePacket(DtRtiConnection* conn,
      MAKRti::DtTransportType transportType, unsigned int& receivedPacketCount);

   virtual bool receivePackets();

   virtual DtNetReadStatus processMsgQueueCallbacks(DtRtiMsg& msg,
      unsigned int& receivedPacketCount);

   virtual void processPendingMsgQueueCallbacks();

   virtual void processPendingMsgMakeCallbacks();

   virtual void setMessageTracking(DtRtiMsg& msg);
   
   virtual void destroyPackets();

   virtual void destroyRelConn();   

   virtual void destroyAncillaryRelConn(DtRtiConnection* conn);

   virtual void destroyAncillaryRelConns();

   virtual void startAsynchronousReceiveProcessing();
   virtual void startAsynchronousSendProcessing();

   virtual void stopAsynchronousReceiveProcessing();
   virtual void stopAsynchronousSendProcessing();

   inline DtQueuePacket* getPacket();

   virtual bool createBestEffortConn();

   virtual bool queueReceivePacket(DtRtiMsg* rtiMessage, MAKRti::DtTransportType transportType,
      bool& overflowWarning);

   virtual void testIoIndicator(const char* label, unsigned int queueLevel,
      unsigned int& lowWaterLine, unsigned int& highWaterLine, bool& overflowWarning);

public:
 
   static DtConnectionMgr* create( DtRIDParameters* p,  bool isFederate = false, 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient> assistantClient = 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient>() );
   
#ifdef _WIN32
   static unsigned long runReceive( void* usr );
   static unsigned long runSend( void* usr );
#else
   static void* runReceive( void* usr );
   static void* runSend( void* usr );
#endif

protected:

   static void processCallbackMsg(const DtFedAmbCallbackMsg& callbackMsg, void* usr);

protected:
   bool myTimeToQuitSending;
   bool myTimeToQuitReceiving;
   bool mySendThreadActivated;
   bool myReceiveThreadActivated;
   bool myAsynchronous;
   bool myAsynchronousCallbacks;
   bool myAsynchronousProcessMsg;
   bool myReserved;
   bool myMoreToFlush;
   bool myDoDisconnect;
   bool myNonReflectedList[DtInvalidMsgKind + 1];

#ifdef _WIN32
   HANDLE myReceiveThreadHandle;
   HANDLE mySendThreadHandle;
   HANDLE myReceiveWaitEvent;
   HANDLE* myWaitEvents;
   HANDLE mySendWaitEvent;
   unsigned int myNumWaitEvents;
#else
   DtSelectParamWithWrite myWaitEventsParams;
   pthread_t myReceiveThreadHandle;
   pthread_t mySendThreadHandle;
   int mySendEvent[2];
   int myRecvEvent[2];
#endif

   float myIoPeriod;
   float myTickWaitPeriod;
   unsigned int myMaxQueue;
   unsigned int myMaxCount;

   unsigned int mySendPacketCount;
   unsigned int myReadPacketCount;

   DtAsycQueueType* mySendQueue;
   DtAsycQueueType* myReceiveQueue;

   DtCallbackQueue* myCallbackQueue;

   DtThreadSafe<bool> myAsynchronousConnectionIsOk;

   unsigned int myNextReliableConnToRead;

#ifdef _WIN32
   unsigned int myWaitEvent_BestEffortConn;
   unsigned int myWaitEvent_PrimaryReliableConn;
   unsigned int myWaitEvent_AncillaryReliableConnStart;
   unsigned int myNoBestEffort;
#endif

   unsigned int myIoQueueIndicator;
   unsigned int myReceiveQueueLowWaterLine;
   unsigned int myReceiveQueueHighWaterLine;
   bool myReceiveOverflowWarning;
   unsigned int mySendQueueLowWaterLine;
   unsigned int mySendQueueHighWaterLine;
   bool mySendOverflowWarning;

   bool myShouldWaitOnCurrentBundle;
};

#endif