Multithreading publishers

An example that creates multiple F18 entities and demonstrates how to parallelize the simulation.

It demonstrates both DtPublisherContainer and the more generic DtParallelExecutionContainer to tick every F18 entity in parallel.

/****************************************************************************
 * Copyright (c) 2017 VT MAK
 * All rights reserved.
 ****************************************************************************/


#include <vl/exerciseConn.h>
#include <vl/entityPublisher.h>
#include <vl/reflectedEntityList.h>
#include "multiVehicleFed.h"
#include "multiF18Init.h"
#include <vlutil/vlProcessControl.h>
#include <matrix/geodeticCoord.h>
#include <vlutil/vlRtiMismatchException.h>
#include <vlutil/vlMiniDumper.h>

#include <stdlib.h>
#include <iostream>

#ifndef _WIN32
#include <sys/socket.h>
#endif

// *****************************************************************
// *************************** GLOBALS *****************************
// *****************************************************************
DtTime timeOutInterval        = 12.0;              // seconds
DtTime timeThreshold          =  5.0;              // seconds
double translationThreshold   = 1.0;               // meters
double rotationThreshold      = DtDeg2Rad( 3.0 );  // radians

// Static functions
static int keybrdTick( );

#if VXWORKS
#define main realMain
DT_VXWORKS_APP(f18)
#endif

// *****************************************************************
// ***************************  MAIN *******************************
// *****************************************************************
int main( int argc, char* argv[] )
{
   // Used for error handling
   DtINIT_MINIDUMPER( "Parallel-F18" );

   try
   {
      // get the PID so we can run multiple F18's
      int pid = DtGetPid();
      DtString pidStr( pid );

      // get the Application name
      DtString appName = "VR-Link F18 " + pidStr;
      DtF18Initializer initializer( argc, argv, appName );

#if DtDIS
      // Set the default DIS version
      initializer.setDisVersionToSend( 7 );
#else
      // Set the default FED File, executable, and version
      initializer.setFedFileName( DtDefaultRpr2Fom );
      initializer.setExecName( "MAK-RPR-2.0" );
      initializer.setRprFomVersion( 2.0 );
      initializer.setRprFomRevision( 1 );
#endif

      initializer.setApplicationNumber( pid );
      DtString defaultF18Name( initializer.f18Name() );
      defaultF18Name += " " + pidStr;
      initializer.setF18Name( defaultF18Name );
#ifdef DtHLA
      initializer.setFederateType( defaultF18Name );
#endif
      initializer.parseCmdLine();

      // Create an instance of our simulator
      DtMultiVehicleFed simulator;

      // Set up default static values
      DtEntityPublisher::setDfltRotationThreshold( rotationThreshold );
      DtEntityPublisher::setDfltTranslationThreshold( translationThreshold );

      // This is the only DIS specific code in this file.
#if DtDIS
      DtEntityPublisher::setDfltTimeThreshold( timeThreshold );

      // MUST use async I/O in parallel mode for thread saftey
      // Sync I/O in DIS not yet thread safe
      initializer.setUseAsynchIO(true);
#endif

      // Connect to the exercise or RTI
      simulator.initializeExConn(DtTime(timeThreshold), &initializer);

      DtClock* clk = simulator.exerciseConn()->clock();

      // initialize the fleet
      simulator.initializeFleet(0, initializer.count(), &initializer);

      DtTime before;
      DtTime drainDuration;
      DtTime simDuration;
      DtTime pubDuration;
      DtTime frameDuration;

      // Simulation loop
      int forever = 1;
      DtTime dt = initializer.deltaTime();
      DtVehicleSim::setSimDeltaTime(dt);
      DtTime idle = dt;
      while( forever ) // to keep cfront quiet
      {
         // process incoming data
         before = clk->absRealTime();
         simulator.drainInput(dt);
         drainDuration = clk->absRealTime() - before;

         if ( keybrdTick( ) == 0 )  break;

         DtVehicleSim::setSimDeltaTime(dt);

         // Perform simulation for the current time frame
         before = clk->absRealTime();
         simulator.simulate();
         simDuration = clk->absRealTime() - before;

         // Publish simulation changes
         before = clk->absRealTime();
         simulator.publish();
         pubDuration = clk->absRealTime() - before;

         frameDuration = drainDuration + simDuration + pubDuration;
         if (frameDuration < dt)
         {
            idle = dt - frameDuration;
         }
         else
         {
            idle = 0.0;
         }

         std::cout << "total=" << frameDuration << '\t';
         std::cout << "idle="  << idle << '\t';
         std::cout << "drain=" << drainDuration << '\t';
         std::cout << "sim="   << simDuration << '\t';
         std::cout << "pub="   << pubDuration << '\t';
         std::cout << std::endl;

         // wait for real time to catch up to simulated time
         simulator.sleep(idle);
      }

      simulator.dismantleFleet();   // Dismantle the fleet of vehicles
      simulator.dismantleExConn();  // Dismantle exercise connection
   }
   DtCATCH_AND_WARN(std::cout);

   return 0;
}

int keybrdTick( )
{
   char* keyPtr = DtPollBlockingInputLine();
   if ( !keyPtr )  return 1;  // no input

   switch( *keyPtr )
   {
   case 'q':
   case 'Q':
      return 0;
   default:
      std::cout <<  "Unrecognized command key " <<  *keyPtr  << std::endl;
      break;
   }
   return 1;
}

/*******************************************************************************
 ** Copyright (c) 2013 VT MAK
 ** All rights reserved.
 *******************************************************************************/

#include "multiVehicleFed.h"

#include "vehicleSim.h"
#include "multiF18Init.h"

#include <vl/entityStateRepository.h>
#include <vlutil/vlProcessControl.h>
#include <vl/fom.h>
#include <vl/hlaObjectFactory.h>

#include <stdio.h>
#include <iostream>

DtMultiVehicleFed::DtMultiVehicleFed()
{
   setExecutionMode(DtMultiVehicleFed::PARALLEL); // Default to parallel execution
}

DtMultiVehicleFed::~DtMultiVehicleFed(void)
{
}

void DtMultiVehicleFed::initializeExConn( DtTime theTimeOutInterval, DtF18Initializer *initializer)
{
      // Init the exercise Connection 
      myExConn = NULL;
      DtExerciseConn::InitializationStatus status = DtExerciseConn::DtINIT_SUCCESS;
      std::cout << "Opening connection..." << std::endl;

      myExConn = new DtExerciseConn(*initializer, &status);
      if( status != DtExerciseConn::DtINIT_SUCCESS )
      {
         // failed out 
         std::cerr << "Failed to open connection error: " << status << "." << std::endl;
         return;
      }

#if DtHLA
      myExConn->ignoreAdvisories();

      // Replace HLA object creation with instrumented version
      DtFom* fom = myExConn->fom();
      DtHlaObjectFactory* objFactory = myExConn->objectFactory();
      DtObjClassDesc* classDesc;
#if DtHLA_1516
      for (std::list<DtObjClassDesc*>::const_iterator item = fom->objClassList().begin();
         item != fom->objClassList().end();
         ++item)
      {
         classDesc = const_cast<DtObjClassDesc*>(*item);
#else

      for (classDesc = fom->firstObjClass();
         classDesc;
         classDesc = fom->nextObjClass(classDesc))
      {
#endif
      }
#endif

   mySimTime = 0.0; 
   myClock = myExConn->clock();

   std::cout << "Connection open, status=" << status << "." << std::endl;

   if (initializer->execMode() == 2)
   {
      setExecutionMode(SERIAL);
      std::cout << "Serial execution." << std::endl;
   }
   else if(initializer->execMode() == 1)
   {
      setExecutionMode(PARALLEL);
      std::cout << "Parallel execution." << std::endl;
   }
   else
   {
      setExecutionMode(PARALLEL);
      std::cout << "Unknown execution mode, default to parallel execution." << std::endl;
   }

   // Track reflected objects from other simulations.
#if DtDIS
   rel = new DtReflectedEntityList( myExConn, false, false, 10007 );
#else
   rel = new DtReflectedEntityList(myExConn);
#endif
   rel->setTimeoutInterval(theTimeOutInterval);
   // Register a callback when any detonation happens.
   DtDetonationInteraction::addCallback( myExConn, theDetonationCb, this );
}

// Initialize a single fleet vehicle
DtVehicleSim *DtMultiVehicleFed::vehicleFactory(int num, int appNumOffset, 
   const DtF18Initializer *initializer, DtObjectPublisherContainer& pubContainer)
{
   DtEntityIdentifier entityId = DtEntityIdentifier( initializer->siteId(), 
      initializer->applicationNumber() + appNumOffset, initializer->entityNumber() + num );

   DtEntityType type( initializer->entityTypeStr() );
   DtString name = initializer->f18Name();
   if (num > 0)
   {
      name += DtString(num);
   }

   DtVehicleSim* myVehicle = new DtVehicleSim(
      DtDeg2Rad( initializer->refLatitude() ), DtDeg2Rad( initializer->refLongitude() ),
      myExConn, rel, entityId, type, name, initializer->f18Markings(),
      DtDeadReckonTypes( initializer->deadReckonAlgorithm() ), initializer->haveArtParts(),
      pubContainer, initializer->timeSpacing());

   myVehicle->init( initializer->initialPosition(), initializer->initialSpeed(),
      initializer->initialHeading(), initializer->initialTurnRadius(),
      initializer->initialPitch(), initializer->autoRoll(), initializer->initialRoll(), num );

   return myVehicle;
}

void DtMultiVehicleFed::initializeFleet(int appNumOffset, int theNumVehicles, const DtF18Initializer *initializer)
{
   for (int nVehicle=0; nVehicle<theNumVehicles; ++nVehicle) {

      // Create a vehicle
      DtVehicleSim *vehicle = vehicleFactory(nVehicle, appNumOffset, initializer, myPublishedObjects);

      // Add it to the simulation collection
      mySimulatedVehicles.add(vehicle);
   }
}

// Process inputs
void DtMultiVehicleFed::drainInput(double tMax)
{
   // Update the simulation time.
   myClock->setSimTime( mySimTime );

   myExConn->drainInput(tMax);
}

// Simulate all vehicles
void DtMultiVehicleFed::simulate()
{
   // Call ::simulate method for mySimulatedVehicles.
   mySimulatedVehicles.executeMethod(&DtVehicleSim::simulate);
}

// Publish all vehicle outputs
void DtMultiVehicleFed::publish()
{
   myPublishedObjects.tick();
}

void DtMultiVehicleFed::sleep(double deltaTime)
{
   mySimTime += deltaTime; // Adjust simulation time.

   // Wait until real time has caught up to simulated time.
   DtTime sleeptime = mySimTime  - myClock->elapsedRealTime();
   DtSleep( sleeptime );
}

void DtMultiVehicleFed::dismantleFleet()
{
   mySimulatedVehicles.deleteAndClear();

   // Each vehicle's destructor will delete its published objects
   // So we don't need to delete them here.
   myPublishedObjects.clear();
}

void DtMultiVehicleFed::dismantleExConn()
{
   DtDetonationInteraction::removeCallback( myExConn, theDetonationCb, this );
   DtDELETE( rel );
   DtDELETE( myExConn );
}

void DtMultiVehicleFed::setExecutionMode(mode theMode) 
{
   myMode = theMode;
   if (theMode == PARALLEL)
   {
      mySimulatedVehicles.setParallel(true);
      myPublishedObjects.setParallel(true);
   }
   else
   {
      mySimulatedVehicles.setParallel(false);
      myPublishedObjects.setParallel(false);
   }
}

DtMultiVehicleFed::mode DtMultiVehicleFed::executionMode() const
{
   return myMode;
}

// A detonation interaction was received
void DtMultiVehicleFed::theDetonationCb(
   DtDetonationInteraction *theDetonation, void *theUserData)
{
   // Cast is safe because we installed this callback.
   ((DtMultiVehicleFed *) theUserData)->processDetonation(theDetonation);
}

// Process detonation over all vehicles in parallel
void DtMultiVehicleFed::processDetonation(const DtDetonationInteraction *theDetonation)
{
   // Since we can't pass the detonation as an argument to the vehicle,
   // we'll just use a static instead.
   DtVehicleSim::setDetonation(theDetonation);

   // All vehicles process their reaction to the detonation in parallel.
   mySimulatedVehicles.executeMethod(&DtVehicleSim::reactToDetonation);
}