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) 2014 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);
#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);

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

      // Simulation loop
      int forever = 1;
      while( forever ) // to keep cfront quiet
      {
         // process incoming data
         simulator.drainInput(initializer.deltaTime());

         if ( keybrdTick( ) == 0 )  break;

         // Perform simulation for the current time frame
         simulator.simulate();

         // Publish simulation changes
         simulator.publish();

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

      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;

   // Track reflected objects from other simulations.
   rel = new DtReflectedEntityList( myExConn );
   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);

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

   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;
   mySimulatedVehicles.setParallel(myMode);
   myPublishedObjects.setParallel(myMode);
}

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);
}