Multithreading publishers

Table of Contents

• Example Code
  • F18 Code
  • Multiple Vehicle Federation

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.

Example Code

F18 Code

/****************************************************************************
 * Copyright (c) 1992-2025 MAK Technologies, Inc
 * All rights reserved.
 ****************************************************************************/


#include "multiVehicleFed.h"
#include "multiF18Init.h"

#include <vl/exerciseConn.h>
#include <vl/entityPublisher.h>
#include <vl/reflectedEntityList.h>

#include <vlutil/vlProcessControl.h>
#include <vlutil/vlRtiMismatchException.h>
#include <vlutil/vlMiniDumper.h>
#include <vlutil/vlSimpleKeyboard.h>

#include <matrix/geodeticCoord.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( );
DtSimpleKeyboard input;

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

      initializer.parseCmdLine();

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


      // 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);
      DtExerciseConn* exConn = simulator.exerciseConn();
      DtClock* clock = exConn->clock();

#ifdef DtHLA
      // Let other VR-Link federates know that we are initializing
      // Helps control request update calls when creating many entities
      // Requests are delayed until all federates are initialized or executing
      std::cout << "[" << clock->elapsedRealTime() << ", 0] Initializing" << std::endl;
      exConn->setExecutionStatus(vrlstatus_initializing);
      simulator.drainInput(0.1);
#endif

      DtTime frameTime = initializer.deltaTime();

      // initialize the fleet
      DtVehicleSim::setSimDeltaTime(frameTime);
      simulator.initializeFleet(0, initializer.count(), &initializer);
      simulator.drainInput(frameTime);
      simulator.publish();

#ifdef DtHLA
      // Let other VR-Link federates know that we are initialized
      exConn->setExecutionStatus(vrlstatus_initialized);
      simulator.drainInput(0.1);
      std::cout << "[" << clock->elapsedRealTime() << ", 0] Initialized" << std::endl;

#endif

      // Simulation loop
      int forever = 1;
      DtTime dt = frameTime;
      DtTime outputTime = 0.0;
      double outputInterval = initializer.outputInterval();

#ifdef DtHLA
      // Let other VR-Link federates know that we are executing
      std::cout << "[" << clock->elapsedRealTime() << ", 0] Executing" << std::endl;
      exConn->setExecutionStatus(vrlstatus_executing);
      simulator.drainInput(0.1);
#endif

      // Reset elapsed time to 0 to match initial simulation time
      clock->init();

      unsigned long frameCount = 1;
      double totalDrainDuration = 0.0;
      double totalSimDuration = 0.0;
      double totalPubDuration = 0.0;
      double totalIdle = 0.0;
      double totalFrameDuration = 0.0;
      int numberFrameOverflows = 0;
      double totalOverflow = 0.0;
      DtTime lastTime = clock->absRealTime();
      std::cout << std::fixed;
      while( forever ) // to keep cfront quiet
      {
         DtTime idle = 0.0;
         DtTime startTime = clock->absRealTime();
         dt = startTime - lastTime;

         simulator.setSimTime(clock->elapsedRealTime());

         // process incoming data
         DtTime before = clock->absRealTime();
         simulator.drainInput(frameTime);
         DtTime drainDuration = clock->absRealTime() - before;
         totalDrainDuration += drainDuration;

         if ( input.keybrdTick( ) == -1 )  break;

         DtVehicleSim::setSimDeltaTime(dt);

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

         // Publish simulation changes
         before = clock->absRealTime();
         simulator.publish();
         DtTime pubDuration = clock->absRealTime() - before;
         totalPubDuration += pubDuration;

         //frameDuration = drainDuration + simDuration + pubDuration;
         DtTime frameDuration = clock->absRealTime() - startTime;
         totalFrameDuration += frameDuration;
         if (frameDuration < frameTime)
         {
            idle = frameTime - frameDuration;
            totalIdle += idle;
            dt = frameTime;
         }
         else
         {
            dt = frameDuration;
            idle = -1.0;
            numberFrameOverflows++;
            totalOverflow += frameDuration - frameTime;
         }

         if (outputInterval <= 0.0 || outputTime < startTime)
         {
            outputTime = clock->absRealTime() + outputInterval;
            std::cout << "[" << std::setprecision(2) << simulator.simTime() << ", " << frameCount << "] #entities: " << simulator.numberRemoteEntities()
               << " total=" << std::setprecision(6) << frameDuration << "(" << totalFrameDuration / frameCount << ")"
               << " idle=" << idle << "(" << totalIdle/frameCount << ")"
               << " drain=" << drainDuration << "(" << totalDrainDuration / frameCount << ")"
               << " sim=" << simDuration << "(" << totalSimDuration / frameCount << ")"
               << " pub=" << pubDuration << "(" << totalPubDuration / frameCount << ")"
               << " #FO=" << numberFrameOverflows << "(" << (numberFrameOverflows > 0 ? totalOverflow/numberFrameOverflows : 0) << ")"
               << std::endl;
            numberFrameOverflows = 0;
            totalOverflow = 0.0;
         }

         if (idle > 0.0)
         {
            // wait for real time to catch up to simulated time
            DtTime sleepTime = simulator.sleep(idle);
         }

         lastTime = startTime;
         frameCount++;
      }

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

   return 0;
}

Multiple Vehicle Federation

/*******************************************************************************
 ** Copyright (c) 1992-2025 MAK Technologies, Inc
 ** All rights reserved.
 *******************************************************************************/

#include "multiVehicleFed.h"

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

#include <vl/entityStateRepository.h>
#include <vl/fom.h>
#include <vl/hlaObjectFactory.h>
#include <vl/hlaObjectManager.h>

#include <vlutil/vlProcessControl.h>
#include <vlutil/vlStringUtil.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

#ifdef DtHLA
      // Set and collect the status for the VR-Link State objects
      myExConn->setExecutionStatus(vrlstatus_initializing);
      myExConn->drainInput(1.0, 0.5);
#endif

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

   myPublishedObjects.setMaxNumberThreads(initializer->maxNumberWorkerThreads());
   mySimulatedVehicles.setMaxNumberThreads(initializer->maxNumberWorkerThreads());

   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
   myRel = new DtReflectedEntityList( myExConn, false, false, 10007 );
#else
   myRel = new DtReflectedEntityList(myExConn);
#endif
   myRel->setTimeoutInterval(theTimeOutInterval);
   myRel->discoverOnlyWhenEntityTypeKnown(true);

   // 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;
   getVehicleName(num, initializer->f18Name(), name);

   DtVehicleSim* myVehicle = new DtVehicleSim(
      DtDeg2Rad( initializer->refLatitude() ), DtDeg2Rad( initializer->refLongitude() ),
      myExConn, myRel, 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;
}

DtString& DtMultiVehicleFed::getVehicleName(int num, const DtString& baseName, DtString& name)
{
   name = baseName;
   if (num > 0)
   {
      name += "-";
      name += DtString(num);
   }

   return name;
}

void DtMultiVehicleFed::initializeFleet(int appNumOffset, int theNumVehicles, const DtF18Initializer *initializer)
{
#ifdef DtHLA_1516_EVOLVED

   std::set<std::wstring> objectNames;
   DtString baseName = initializer->f18Name();
   for (int nVehicle = 1; nVehicle <= theNumVehicles; ++nVehicle)
   {
      DtString name;
      objectNames.insert(DtToWString(getVehicleName(nVehicle, baseName, name)));
   }

   myExConn->hlaObjectManager()->reserveNames(objectNames, true);

#endif

   for (int nVehicle=1; 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)
{
   myExConn->drainInput(tMax);
}

void DtMultiVehicleFed::setSimTime(double time)
{
   mySimTime = time; // Adjust simulation time.
   myClock->setSimTime( mySimTime );
}

double DtMultiVehicleFed::simTime() const
{
   return mySimTime;
}

int DtMultiVehicleFed::numberRemoteEntities() const
{
   return myRel->count();
}

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

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

double DtMultiVehicleFed::sleep(double deltaTime)
{
   DtTime before = myClock->absRealTime();
   // Wait until real time has caught up to simulated time.
   DtTime sleeptime = mySimTime  - myClock->elapsedRealTime();
   DtSleep( sleeptime );
   return myClock->absRealTime() - before;
}

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