stateDataManager.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 2019 MAK Technologies, Inc.
** All rights reserved.
*******************************************************************************/

#pragma once


#include "vrfStateData/vrfStateDataDefines.h"
#include <vrfStateData/stateDataTypes.h>
#include "vrfutil/vrfCallbackQueue.h"
#include "vrfutil/rwUUID.h"
#include <tbb/spin_rw_mutex.h>
#include <tbb/concurrent_queue.h>
#include <vlutil/vlTime.h>
#include "legionDataStore/view.h"
#include "legionDataStore/viewManager.h"
#include "legionDataStore/simulationDatabase.h"
#include "vlpi/entityIdentifier.h"
#include <atomic>

#include <queue>
#include <memory>
#include <vector>
#include <set>
#include <unordered_map>
#include <atomic>

#include <boost/signals2.hpp>
#include <boost/function.hpp>

class Coordinate_System;

namespace Legion
{
   class SimulationDatabase;
   class EventManager;
   class ControlManager;
   struct AttributeDescriptor;
}

class DtVrfParameterDb;
class DtEntityType;

namespace makVrf
{
   struct DtEmitterSystemBeamStateData;
   class DtStateData;
   class DtEmitterSystemStateComponent;
   class DtEntityStateComponent;
   class DtFrameStateInterface;
   class DtSimulatedObjectStateComponent;

   struct DtTrackJamBeamInfo
   {
      const DtEmitterSystemBeamStateData* beamSR;
      const DtEmitterSystemStateComponent* emitterSystemSR;
   };

   typedef std::vector<DtTrackJamBeamInfo> DtTrackJamBeamInfoList;

   class DtStateComponent;
   class DtFrameDataInterface;
   class DtStateDataManager;

   using DtStateDataManagerCreatorFcn = DtStateDataManager* (*)(Legion::SimulationDatabase*, int, unsigned int);

   class DT_DLL_vrfStateData DtStateDataManager
   {
   protected:
      struct StateComponent;

   public:

      enum class FrameStatus
      {
         Normal = 0,
         SkippedBeginFrame,
         SkipAdvanceFrame
      };

      typedef std::set<const DtFrameStateInterface*> CurrentFrameStates;
      using PreAddCallback = boost::function<bool (DtStateData*)>;
      typedef boost::signals2::signal<void (DtStateComponent*)> StateComponentSignal;
      typedef std::vector<DtStateData*> StateData;

      DtStateDataManager(Legion::SimulationDatabase* = nullptr, int threadPoolSize = -1, unsigned int clientId = 0);
      virtual ~DtStateDataManager();

      virtual DtStateData* createStateData(bool isLocal, const DtEntityType&, const DtUUID& uuid = DtUUID(), int publishFlag = 0,
         PreAddCallback callback = 0, const DtEntityIdentifier& entId = DtEntityIdentifier::nullId(), bool addToSimObjectManager = true,
         const DtString& globalId = DtString::nullString());
      virtual void setParameterDatabase(DtVrfParameterDb*);
      DtVrfParameterDb* parameterDatabase() const
      {
         return myParameterDatabase;
      }

      DtVrfCallbackQueue& jobProcessor()
      {
         return myJobProcessor;
      }

      virtual DtStateData* createStateData(Legion::UInt64 legionHandle, PreAddCallback callback = 0);

      virtual void removeAndDeleteAll(const std::set<DtUUID>& except = std::set<DtUUID>());

      virtual void removeStateData(DtStateData*);
      virtual void removeStateData(const DtUUID&);
      virtual void removeStateDataByStateId(makVrf::SimStateID);
      virtual void removeStateDataByUniqueId(unsigned long long);

      virtual void mapStateDataByUniqueId(unsigned long long, DtStateData*);
      virtual void unmapStateDataByUniqueId(unsigned long long);

      virtual void beginFrame(double simTime);
      virtual void advanceFrame();

      virtual DtStateData* advanceFrameFor(unsigned long long uniqueId);

      virtual void registerLegionComponents(Legion::SimulationDatabase*);

      Legion::EventManager* eventManager();

      Legion::SimulationDatabase* database()
      {
         return myDatabase;
      }

      Legion::SimulationDatabase* unpublishedDatabase()
      {
         return myUnpublishedDatabase ? myUnpublishedDatabase : myDatabase;
      }

      static void SeedUnpublishedEntityId(short appId, short siteId);
      static DtEntityIdentifier NextUnpublishedEntityId();

      virtual void aerodromeDiscovered(DtStateData*);
      virtual void entityDiscovered(DtStateData*);
      virtual void environmentalDiscovered(DtStateData*);
      virtual void unitDiscovered(DtStateData*);

      virtual void setBlockOnAdvanceFrameJobs(bool);
      bool blockOnAdvanceFrameJobs() const
      {
         return myBlockOnAdvanceFrameJobs;
      }

      virtual bool hasOutstandingAdvanceFrameJobs() const;

      virtual void reserve(unsigned int);

      template <typename T_Class>
      void handleDataUpdated(const std::vector<typename Legion::SetView<T_Class>::InstanceUpdateData>& instanceList)
      {
         typename InstancesUpdatedCallbacks::iterator iter = myInstancesUpdatedCallbacks.find(T_Class::type());

         if (iter != myInstancesUpdatedCallbacks.end())
         {
            typename std::vector<typename Legion::SetView<T_Class>::InstanceUpdateData>::const_iterator instIter = instanceList.begin();
            typename std::vector<typename Legion::SetView<T_Class>::InstanceUpdateData>::const_iterator instEnd = instanceList.end();

            while (instIter != instEnd)
            {
               const T_Class& ent = instIter->instance;
               InstanceUpdatedCallbacks::iterator callIter = iter->second.find(ent.instanceHandle());

               if (callIter != iter->second.end())
               {
                  Callbacks::iterator cbIter = callIter->second.begin();
                  Callbacks::iterator cbEnd = callIter->second.end();

                  while (cbIter != cbEnd)
                  {
                     (cbIter->first)(instIter->attributesChanged);
                     ++cbIter;
                  }
               }

               ++instIter;
            }
         }
      }

      template <typename T_Class>
      unsigned int addInstanceUpdatedCallback(Legion::SimulationDatabase* database, 
         DtInstanceUpdatedCallback callback, Legion::InstanceHandle instanceHandle,
         const typename std::vector<Legion::AttributeIndex>& attributeIndexFilter = std::vector<Legion::AttributeIndex>())
      {
         if (isValid())
         {
            unsigned int newHandle = ++myCallbackHandle;
            tbb::spin_mutex::scoped_lock lock(myUpdateCallbackMutex);

            InstancesUpdatedCallbacks::iterator iter = myInstancesUpdatedCallbacks.find(T_Class::type());

            if (iter == myInstancesUpdatedCallbacks.end())
            {
               myInstancesUpdatedCallbacks[T_Class::type()] = InstanceUpdatedCallbacks();
               iter = myInstancesUpdatedCallbacks.find(T_Class::type());
            }

            iter->second[instanceHandle].push_back(std::make_pair(callback, newHandle));

            if (iter->second.size() == 1)
            {
               view<T_Class>(database)->addInstanceUpdatedCallback(typename Legion::SetView<T_Class>::InstanceUpdateCallback(this, 
                  &DtStateDataManager::handleDataUpdated<T_Class>), attributeIndexFilter);
            }

            return newHandle;
         }

         return 0;
      }

      template <typename T_Class>
      void removeInstanceUpdatedCallback(Legion::SimulationDatabase* database, Legion::InstanceHandle instanceHandle, unsigned int callbackHandle)
      {
         if (isValid() && callbackHandle)
         {
            tbb::spin_mutex::scoped_lock lock(myUpdateCallbackMutex);
            InstancesUpdatedCallbacks::iterator iter = myInstancesUpdatedCallbacks.find(T_Class::type());

            if (iter != myInstancesUpdatedCallbacks.end())
            {
               InstanceUpdatedCallbacks::iterator callIter = iter->second.find(instanceHandle);

               if (callIter != iter->second.end())
               {
                  Callbacks::iterator cbIter = callIter->second.begin();
                  Callbacks::iterator cbEnd = callIter->second.end();

                  while (cbIter != cbEnd)
                  {
                     if (cbIter->second == callbackHandle)
                     {
                        callIter->second.erase(cbIter);
                        break;
                     }

                     ++cbIter;
                  }

                  if (callIter->second.size() == 0)
                  {
                     iter->second.erase(callIter);

                     if (iter->second.size() == 0)
                     {
                        myInstancesUpdatedCallbacks.erase(iter);
                        view<T_Class>(database)->removeInstanceUpdatedCallback(typename Legion::SetView<T_Class>::InstanceUpdateCallback(this, 
                           &DtStateDataManager::handleDataUpdated<T_Class>));
                     }
                  }
               }
            }
         }
      }

      bool isValid() const
      {
         return (myDatabase != nullptr);
      }

      const StateData& stateData() const
      {
         return myStateData;
      }

      const DtStateData* stateData(makVrf::SimStateID id) const
      {
         return stateDataFor(id);
      }

      DtStateData* stateData(makVrf::SimStateID id)
      {
         return stateDataFor(id);
      }

      double absRealTime() const
      {
         return myClock.absRealTime();
      }

      virtual void prepareForHandoff(unsigned long long, bool becomingLocal);

      virtual void handoffComplete(makVrf::SimStateID);
      virtual bool awaitingHandoff(makVrf::SimStateID) const;

      using StateDataCreatorFcn = DtStateData*(*)(makVrf::SimStateID, Legion::SimulationDatabase*, DtStateDataManager*);

      using RemoteStateDataCreatorFcn = DtStateData*(*)(makVrf::SimStateID, UInt64 legionId, Legion::SimulationDatabase*, DtStateDataManager*);

      static StateDataCreatorFcn setStateDataCreatorFcn(StateDataCreatorFcn);
      static StateDataCreatorFcn stateDataCreatorFcn();

      static RemoteStateDataCreatorFcn setRemoteStateDataCreatorFcn(RemoteStateDataCreatorFcn);
      static RemoteStateDataCreatorFcn remoteStateDataCreatorFcn();

      static DtStateData* defaultStateDataCreator(makVrf::SimStateID, Legion::SimulationDatabase*, DtStateDataManager*);
      static DtStateData* defaultRemoteStateDataCreator(makVrf::SimStateID, UInt64 legionGlobalId, Legion::SimulationDatabase*, DtStateDataManager*);

      virtual DtStateData* createStateData(makVrf::SimStateID, Legion::SimulationDatabase*);
      virtual DtStateData* createStateData(makVrf::SimStateID, UInt64 legionGlobalId);

      virtual void clear(bool forReuse = false);

      Legion::ControlManager* controlManager();

      virtual DtUUID uuidFor(makVrf::SimStateID) const;

      virtual const DtStateData* stateDataFor(const DtUUID&) const;
      virtual DtStateData* stateDataFor(const DtUUID&);

      virtual const DtStateData* stateDataFor(const DtEntityIdentifier&) const;
      virtual DtStateData* stateDataFor(const DtEntityIdentifier&);

      virtual void setUpdateOnlySpecifiedStateDataMode(bool);
      bool updateOnlySpecifiedStateDataMode() const
      {
         return myUpdateOnlySpecifiedStateDataMode;
      }

      FrameStatus frameStatus() const
      {
         return myFrameStatus;
      }

      virtual const DtStateData* stateDataFor(makVrf::SimStateID) const;
      virtual DtStateData* stateDataFor(makVrf::SimStateID);

      virtual const DtStateData* stateDataForUniqueId(UInt64) const;
      virtual DtStateData* stateDataForUniqueId(UInt64);
      virtual UInt64 uniqueIdFrom(const DtUUID&) const;

      virtual void setRunFullyThreaded(bool, int theadCount = 4);
      bool runFullyThreaded() const
      {
         return myRunFullyThreaded;
      }

      virtual void addStateComponent(DtStateComponent*);
      virtual void removeStateComponent(DtStateComponent*, bool deleteOnRemoval = true);

      virtual void resizeContainer(unsigned);

      template <typename T_Component>
      void resizeContainer()
      {
         resizeContainer(T_Component::ComponentType());
      }

      template <typename T_Component>
      StateComponentSignal& signalWhenStateComponentAdded()
      {
         return signalWhenStateComponentAdded(T_Component::ComponentType());
      }

      StateComponentSignal& signalWhenStateComponentAdded(unsigned type)
      {
         tbb::spin_rw_mutex::scoped_lock lock(myComponentMutex, true);
         resizeContainer(type);
         return *(myCurrentFrameStates[type].signal_stateComponentAdded);
      }

      template <typename T_Component>
      StateComponentSignal& signalWhenStateComponentRemoved()
      {
         return signalWhenStateComponentRemoved(T_Component::ComponentType());
      }

      StateComponentSignal& signalWhenStateComponentRemoved(unsigned type)
      {
         tbb::spin_rw_mutex::scoped_lock lock(myComponentMutex, true);
         resizeContainer(type);
         return *(myCurrentFrameStates[type].signal_stateComponentRemoved);
      }

      template <typename T_Component>
      const CurrentFrameStates& stateComponents() const
      {
         static CurrentFrameStates theDefaultSc;

         tbb::spin_rw_mutex::scoped_lock lock(myComponentMutex, false);

         if (T_Component::ComponentType() < myCurrentFrameStates.size())
         {
            return myCurrentFrameStates[T_Component::ComponentType()].currentFrameStates;
         }

         return theDefaultSc;
      }

      virtual void removeAll();

      virtual std::set<const DtStateData*> entitiesTracking(const DtUUID&) const;

      virtual std::set<const DtStateData*> entitiesJamming(const DtUUID&) const;

      virtual DtTrackJamBeamInfoList systemsTrackingObject(const DtUUID&) const;

      virtual DtTrackJamBeamInfoList systemsJammingObject(const DtUUID&) const;

      static void setStateDataManagerCreatorFcn(DtStateDataManagerCreatorFcn fcn);
      static DtStateDataManager* createStateDataManager(Legion::SimulationDatabase* = nullptr, int threadPoolSize = -1, unsigned int clientId = 0);

      virtual void setUnpublishedDatabaseUsage(bool);

      virtual void setCoordinateSystem(const Coordinate_System*);
      const Coordinate_System* coordinateSystem() const
      {
         return myCoordinateSystem;
      }
      boost::signals2::signal<void (const Coordinate_System*)> signal_coordinateSystemChanged;

      typedef std::unordered_map<UInt64, DtStateData*> StateDataByUniqueId;
      const StateDataByUniqueId& entityData() const
      {
         return myEntityData;
      }

      const StateDataByUniqueId& unitData() const
      {
         return myUnitData;
      }

      const StateDataByUniqueId& aerodromeData() const
      {
         return myAerodromeData;
      }

      const StateDataByUniqueId& environmentalData() const
      {
         return myEnvironmentalData;
      }

      virtual void setPublishingOnLegionNetwork(bool b);
      bool publishingOnLegionNetwork() const
      {
         return myPublishingOnLegionNetwork;
      }

      template <typename T_Class>
      Legion::SetView<T_Class>* view(Legion::SimulationDatabase* db)
      {
         tbb::spin_rw_mutex::scoped_lock lock(myViewCreateMutex, true);

         DatabaseViews::iterator iter = myDatabaseViews.find(db);

         if (iter == myDatabaseViews.end())
         {
            myDatabaseViews[db] = Views();
            iter = myDatabaseViews.find(db);
         }

         DtString key(T_Class::type());
         Views::iterator viewIter = iter->second.find(key);

         if (viewIter == iter->second.end())
         {
            iter->second[key] = db->viewManager().createSetView<T_Class>(key, (myUnpublishedDatabase ? Legion::ViewScope::Remote : Legion::ViewScope::All));
            viewIter = iter->second.find(key);
         }

         return (Legion::SetView<T_Class>*)(viewIter->second);
      }

      bool advancingFrame() const
      {
         return myAdvancingFrame;
      }

   protected:
      virtual void jobsThread();
      virtual std::set<const DtStateData*> entitiesTracking(int beamType, const DtUUID&) const;
      virtual DtTrackJamBeamInfoList systemsTrackingObject(int beamType, const DtUUID&) const;
      virtual void uuidChanged(makVrf::SimStateID, const DtUUID& oldUUID, const DtUUID& newUUID);
      virtual void hostChanged(const DtStateData*, const DtUUID& oldUUID, const DtUUID& newUUID);
      virtual void subordinatesChanged(const DtStateData*, const std::vector<DtUUID>& oldSubordinates, const std::vector<DtUUID>& newSubordinates);
      virtual void completeAttachments(DtStateData*);
      virtual void completeAttachments(const DtUUID&);
      virtual void completeAttachments();
      virtual void slot_stateComponentAboutToBeDeleted(DtStateComponent*);
      virtual void convertedToLocal(DtStateData*);

      virtual void resolveHostChanged(const DtUUID&, const DtUUID& oldUUID, const DtUUID& newUUID,
         std::set<DtEntityStateComponent*>&);

   public:
      mutable boost::signals2::signal<void (const DtStateData*)> signal_stateDataAboutToBeRemoved;
      mutable boost::signals2::signal<void (makVrf::SimStateID, UInt64)> signal_stateDataRemoved;

      boost::signals2::signal<void (Legion::SimulationDatabase*)> signal_addVrfSpecificDataElements;

      boost::signals2::signal<void (const DtStateData*)> signal_initializeStateDataBeforeAdd;
      mutable boost::signals2::signal<void (DtStateData*)> signal_stateDataAdded;
      boost::signals2::signal<void ()> signal_beginFrame;
      boost::signals2::signal<void ()> signal_preAdvanceFrame;
      boost::signals2::signal<void ()> signal_advanceFrame;
      boost::signals2::signal<void (DtStateData*, bool)> signal_prepareForHandoff;

      boost::signals2::signal<void (DtStateData*)> signal_entityAdded;
      boost::signals2::signal<void (const DtStateData*)> signal_entityRemoved;

      boost::signals2::signal<void (DtStateData*)> signal_unitAdded;
      boost::signals2::signal<void (const DtStateData*)> signal_unitRemoved;

      boost::signals2::signal<void (DtStateData*)> signal_environmentalAdded;
      boost::signals2::signal<void (const DtStateData*)> signal_environmentalRemoved;

      boost::signals2::signal<void (DtStateData*)> signal_aerodromeAdded;
      boost::signals2::signal<void (const DtStateData*)> signal_aerodromeRemoved;

      boost::signals2::signal<void (unsigned long long, const DtUUID& oldUUID, const DtUUID& newUUID)> signal_uuidChanged;

      StateComponentSignal signal_stateComponentAdded;
      StateComponentSignal signal_stateComponentRemoved;
      boost::signals2::signal<void ()> signal_deleted;

      boost::signals2::signal<void (DtStateData*)> signal_convertedToLocal;

   protected:
      virtual bool _internalRemoveStateData(DtStateData*sd);

      virtual void addVrfSpecificDataElements(Legion::SimulationDatabase*);

      virtual void resolveHostChanges();
      virtual void removeAllStateComponents();
      virtual void resolveSubordinateChanges();

      virtual void removeStateDataToBeDeleted();

      void initializeDefaultPublicData();

      virtual void componentDeleted(DtStateComponent* c);

   protected:
      StateData myStateData;

      mutable tbb::spin_rw_mutex myMutex;
      
      Legion::SimulationDatabase* myDatabase;
      Legion::SimulationDatabase* myUnpublishedDatabase;

      std::set<unsigned long long> myAwaitingHandoff;

      std::atomic<makVrf::SimStateID> myNextSimStateID;
      tbb::concurrent_queue<makVrf::SimStateID> myFreeStateIDs;

      bool myRunFullyThreaded;

      DtClock myClock;
      DtVrfCallbackQueue myJobProcessor;

      mutable tbb::spin_rw_mutex myComponentMutex;

      struct StateComponent
      {
         StateComponent()
         : signal_stateComponentAdded(new StateComponentSignal)
         , signal_stateComponentRemoved(new StateComponentSignal)
         {};

         StateComponent(const StateComponent& orig)
         : signal_stateComponentAdded(orig.signal_stateComponentAdded)
         , signal_stateComponentRemoved(orig.signal_stateComponentRemoved)
         , currentFrameStates(orig.currentFrameStates)
         {};

         StateComponent& operator=(const StateComponent& rhs)
         {
            signal_stateComponentAdded = rhs.signal_stateComponentAdded;
            signal_stateComponentRemoved = rhs.signal_stateComponentRemoved;
            currentFrameStates = rhs.currentFrameStates;

            return *this;
         }

         std::shared_ptr<StateComponentSignal> signal_stateComponentAdded;
         std::shared_ptr<StateComponentSignal> signal_stateComponentRemoved;
         CurrentFrameStates currentFrameStates{};
      };

      typedef std::vector<StateComponent> CurrentFrameStatesList;
      CurrentFrameStatesList myCurrentFrameStates;

      typedef std::unordered_map<DtUUID, std::set<DtStateData*> > StateDataByUUID;
      StateDataByUUID myStateDataByUUID;

      typedef std::unordered_map<std::string, std::set<DtStateData*> > StateDataByEntityIdentifier;
      StateDataByEntityIdentifier myStateDataByEntityIdentifier;

      static DtStateDataManagerCreatorFcn theStateDataManagerCreatorFcn;

      StateDataByUniqueId myStateDataByUniqueId;

      const Coordinate_System* myCoordinateSystem;
      bool myOwnsCoordinateSystem;

      StateDataByUniqueId myEntityData;
      StateDataByUniqueId myUnitData;
      StateDataByUniqueId myAerodromeData;
      StateDataByUniqueId myEnvironmentalData;

      bool myOwnsDatabase;

      unsigned int myReserved;

      tbb::spin_mutex myUpdateCallbackMutex;
      tbb::spin_rw_mutex myViewCreateMutex;
      typedef std::map<DtString, Legion::View*> Views;
      typedef std::map<Legion::SimulationDatabase*, Views> DatabaseViews;
      DatabaseViews myDatabaseViews;

      typedef std::vector<std::pair<DtInstanceUpdatedCallback, unsigned int> > Callbacks;
      typedef std::map<Legion::InstanceHandle, Callbacks> InstanceUpdatedCallbacks;
      typedef std::map<Legion::AttributeSetId, InstanceUpdatedCallbacks> InstancesUpdatedCallbacks;
      InstancesUpdatedCallbacks myInstancesUpdatedCallbacks;

      std::atomic<unsigned int> myCallbackHandle;

      tbb::spin_mutex myHostsChangedThisFrameMutex;
      std::map<DtUUID, std::map<DtUUID, DtUUID> > myHostsChangedThisFrame;
      std::map<DtUUID, std::set<DtUUID> > myWaitingForAttachment;

      tbb::spin_mutex mySubordinatesChangedThisFrameMutex;

      typedef std::pair<std::set<DtUUID>, std::set<DtUUID> > SubordinateChangeSet;
      std::map<DtUUID, SubordinateChangeSet> mySubordinatesChangedThisFrame;
      std::map<DtUUID, std::set<DtUUID> > myPendingSuperiorSet;

      DtVrfParameterDb* myParameterDatabase;

      std::vector<DtStateData*> myStateDataToBeDeleted;
      bool myPublishingOnLegionNetwork;

      struct ComponentRemoval
      {
         bool deleteOnRemoval = false;
         std::shared_ptr<boost::signals2::connection> deleteConnection;
      };

      std::map<DtStateComponent*, ComponentRemoval> myComponentsToRemove;
      std::set<DtStateComponent*> myComponentsDeleted;
      bool myUpdateOnlySpecifiedStateDataMode;

      using Condition = std::condition_variable;
      using ConditionMutex = std::mutex;
      using ConditionLock = std::unique_lock<ConditionMutex>;
      tbb::spin_mutex myAdvancedThisFrameMutex;
      std::set<unsigned long long> myAdvancedThisFrame;
      bool myAdvancingFrame = false;
      bool myBlockOnAdvanceFrameJobs = true;
      std::unique_ptr<std::thread> myJobsThread;
      mutable Condition myStartJobsCondition;
      mutable ConditionMutex myStartJobsConditionMutex;
      std::atomic<bool> myShutdownJobsThreadFlag;

      FrameStatus myFrameStatus = FrameStatus::Normal;
   };
}