core::SimulationEngine Class Reference

Encapsulates the state and logic of the event-driven simulation loop. More...

#include "sim_threading.h"

Public Member Functions
	SimulationEngine (World *world, pool::ThreadPool &pool, std::shared_ptr< ProgressReporter > reporter, std::string output_dir, std::shared_ptr< OutputMetadataCollector > metadata_collector=nullptr, ReceiverOutputSink *output_sink=nullptr, std::function< bool()> cancel_callback=nullptr, bool eager_context_stream_open=false)
	Constructs the simulation engine.
void	run ()
	Starts and runs the main simulation loop until completion.
bool	cancelled () const noexcept
	Returns true after cooperative cancellation has been requested.
void	processStreamingPhysics (RealType t_event)
	Advances the time-stepped inner loop for active streaming systems.
void	processEvent (const Event &event)
	Dispatches a discrete simulation event to its specific handler.
void	handleTxPulsedStart (radar::Transmitter *tx, RealType t_event)
	Handles the start of a pulsed transmission.
void	handleRxPulsedWindowStart (radar::Receiver *rx, RealType t_event)
	Handles the opening of a pulsed receiver's listening window.
void	handleRxPulsedWindowEnd (radar::Receiver *rx, RealType t_event)
	Handles the closing of a pulsed receiver's listening window, triggering finalization.
void	handleTxStreamingStart (const ActiveStreamingSource &source)
	Handles a streaming transmitter turning on.
void	handleTxStreamingEnd (radar::Transmitter *tx)
	Handles a streaming transmitter turning off.
void	handleRxStreamingStart (radar::Receiver *rx)
	Handles a streaming receiver starting to record.
void	handleRxStreamingEnd (radar::Receiver *rx)
	Handles a streaming receiver stopping recording.

Detailed Description

Encapsulates the state and logic of the event-driven simulation loop.

Breaking the simulation loop into this class allows for easily testable, focused functions with low cyclomatic complexity.

Definition at line 109 of file sim_threading.h.

Constructor & Destructor Documentation

SimulationEngine()

core::SimulationEngine::SimulationEngine	(	World *	world,
		pool::ThreadPool &	pool,
		std::shared_ptr< ProgressReporter >	reporter,
		std::string	output_dir,
		std::shared_ptr< OutputMetadataCollector >	metadata_collector = nullptr,
		ReceiverOutputSink *	output_sink = nullptr,
		std::function< bool()>	cancel_callback = nullptr,
		bool	eager_context_stream_open = false
	)

Constructs the simulation engine.

Parameters

world	Pointer to the simulation world containing all entities.
pool	Reference to the thread pool for asynchronous tasks.
reporter	Shared pointer to the thread-safe progress reporter.
output_dir	Output directory for the simulation files.

Definition at line 754 of file sim_threading.cpp.

                                                                             :
        _world(world), _pool(pool), _reporter(std::move(reporter)), _metadata_collector(std::move(metadata_collector)),
        _output_sink(output_sink), _cancel_callback(std::move(cancel_callback)),
        _eager_context_stream_open(eager_context_stream_open), _last_report_time(std::chrono::steady_clock::now()),
        _next_context_heartbeat_time(params::startTime() + 1.0), _output_dir(std::move(output_dir)),
        _internal_stop_time(params::endTime())
    {
        _streaming_tracker_caches.resize(_world->getReceivers().size());
        _if_pulse_tracker_caches.resize(_world->getReceivers().size());
        _fmcw_if_block_buffers.resize(_world->getReceivers().size());
        _fmcw_if_block_start_times.resize(_world->getReceivers().size(), params::startTime());
        _streaming_output_block_buffers.resize(_world->getReceivers().size());
        _streaming_output_processed_buffers.resize(_world->getReceivers().size());
        _streaming_output_block_start_times.resize(_world->getReceivers().size(), params::startTime());
        _streaming_output_block_start_indices.resize(_world->getReceivers().size(), 0);
        _streaming_downsamplers.resize(_world->getReceivers().size());
        _streaming_downsample_base_indices.resize(_world->getReceivers().size(), 0);
        _streaming_downsample_segment_start_times.resize(_world->getReceivers().size(), params::startTime());
        _streaming_output_sample_cursors.resize(_world->getReceivers().size(), 0);
        _streaming_output_stream_ids.resize(_world->getReceivers().size(), 0);
        _streaming_output_stream_open.resize(_world->getReceivers().size(), false);
        _streaming_output_file_metadata.resize(_world->getReceivers().size());
        for (auto& block : _streaming_output_block_buffers)
        {
            block.reserve(streaming_output_block_size);
        }
        for (auto& block : _streaming_output_processed_buffers)
        {
            block.reserve(streaming_output_block_size);
        }
    }

References core::World::getReceivers(), max, and params::startTime().

Here is the call graph for this function:

Member Function Documentation

cancelled()

bool core::SimulationEngine::cancelled ( ) const

noexcept

Returns true after cooperative cancellation has been requested.

Definition at line 130 of file sim_threading.h.

{ return _cancelled; }

handleRxPulsedWindowEnd()

void core::SimulationEngine::handleRxPulsedWindowEnd	(	radar::Receiver *	rx,
		RealType	t_event
	)

Handles the closing of a pulsed receiver's listening window, triggering finalization.

Parameters

rx	Pointer to the receiving radar object.
t_event	The timestamp of the window closing event.

Definition at line 1796 of file sim_threading.cpp.

    {
        rx->setActive(false);
        const auto active_streaming_sources =
            collectStreamingSourcesForWindow(t_event - rx->getWindowLength(), t_event);
 
        RenderingJob job{.ideal_start_time = t_event - rx->getWindowLength(),
                         .duration = rx->getWindowLength(),
                         .responses = rx->drainInbox(),
                         .active_streaming_sources = active_streaming_sources};
 
        rx->enqueueFinalizerJob(std::move(job));
 
        const RealType next_theoretical = t_event - rx->getWindowLength() + 1.0 / rx->getWindowPrf();
        if (const auto next_start = rx->getNextWindowTime(next_theoretical);
            next_start && *next_start <= params::endTime())
        {
            _world->getEventQueue().push({*next_start, EventType::RX_PULSED_WINDOW_START, rx});
        }
    }

References params::endTime(), core::World::getEventQueue(), max, and core::RX_PULSED_WINDOW_START.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleRxPulsedWindowStart()

void core::SimulationEngine::handleRxPulsedWindowStart	(	radar::Receiver *	rx,
		RealType	t_event
	)

Handles the opening of a pulsed receiver's listening window.

Parameters

rx	Pointer to the receiving radar object.
t_event	The timestamp of the window opening event.

Definition at line 1790 of file sim_threading.cpp.

    {
        rx->setActive(true);
        _world->getEventQueue().push({t_event + rx->getWindowLength(), EventType::RX_PULSED_WINDOW_END, rx});
    }

References core::World::getEventQueue(), max, and core::RX_PULSED_WINDOW_END.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleRxStreamingEnd()

void core::SimulationEngine::handleRxStreamingEnd

(

radar::Receiver *

rx

)

Handles a streaming receiver stopping recording.

Parameters

rx	Pointer to the receiving radar object.

Definition at line 1852 of file sim_threading.cpp.

    {
        const auto receiver_it = std::ranges::find_if(_world->getReceivers(), [rx](const auto& receiver_ptr)
                                                      { return receiver_ptr.get() == rx; });
        if (receiver_it != _world->getReceivers().end())
        {
            const auto receiver_index = static_cast<std::size_t>(receiver_it - _world->getReceivers().begin());
            flushFmcwIfBlock(receiver_index);
            flushStreamingOutputBlock(receiver_index, true);
        }
        if (rx->hasFmcwIfResamplingSink() && _world->getSimulationState().t_current >= params::endTime() &&
            _world->getSimulationState().t_current < _internal_stop_time && activePastUserEnd(rx))
        {
            return;
        }
        if (rx->hasFmcwIfResamplingSink())
        {
            rx->endFmcwIfResamplingSegment();
        }
        if (_output_sink != nullptr && receiver_it != _world->getReceivers().end())
        {
            const auto receiver_index = static_cast<std::size_t>(receiver_it - _world->getReceivers().begin());
            if (_streaming_output_stream_open[receiver_index])
            {
                _output_sink->closeStream(_streaming_output_stream_ids[receiver_index]);
                _streaming_output_stream_open[receiver_index] = false;
            }
        }
        rx->setActive(false);
    }

References core::ReceiverOutputSink::closeStream(), params::endTime(), core::World::getReceivers(), core::World::getSimulationState(), max, and core::SimulationState::t_current.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleRxStreamingStart()

void core::SimulationEngine::handleRxStreamingStart

(

radar::Receiver *

rx

)

Handles a streaming receiver starting to record.

Parameters

rx	Pointer to the receiving radar object.

Definition at line 1831 of file sim_threading.cpp.

    {
        rx->setActive(true);
        const auto receiver_it = std::ranges::find_if(_world->getReceivers(), [rx](const auto& receiver_ptr)
                                                      { return receiver_ptr.get() == rx; });
        if (receiver_it != _world->getReceivers().end())
        {
            const auto receiver_index = static_cast<std::size_t>(receiver_it - _world->getReceivers().begin());
            _streaming_downsamplers[receiver_index].reset();
            if (_eager_context_stream_open)
            {
                ensureStreamingOutputStreamOpen(receiver_index, _world->getSimulationState().t_current,
                                                streamingOutputSampleRate(receiver_index));
            }
        }
        if (rx->hasFmcwIfResamplingSink())
        {
            rx->beginFmcwIfResamplingSegment(_world->getSimulationState().t_current);
        }
    }

References core::World::getReceivers(), core::World::getSimulationState(), max, and core::SimulationState::t_current.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleTxPulsedStart()

void core::SimulationEngine::handleTxPulsedStart	(	radar::Transmitter *	tx,
		RealType	t_event
	)

Handles the start of a pulsed transmission.

Parameters

tx	Pointer to the transmitting radar object.
t_event	The timestamp of the transmission event.

Definition at line 1766 of file sim_threading.cpp.

    {
        for (const auto& rx_ptr : _world->getReceivers())
        {
            if (!rx_ptr->checkFlag(Receiver::RecvFlag::FLAG_NODIRECT))
            {
                routeResponse(rx_ptr.get(), simulation::calculateResponse(tx, rx_ptr.get(), tx->getSignal(), t_event));
            }
            for (const auto& target_ptr : _world->getTargets())
            {
                routeResponse(
                    rx_ptr.get(),
                    simulation::calculateResponse(tx, rx_ptr.get(), tx->getSignal(), t_event, target_ptr.get()));
            }
        }
 
        const RealType next_theoretical_time = t_event + 1.0 / tx->getPrf();
        if (const auto next_pulse_opt = tx->getNextPulseTime(next_theoretical_time);
            next_pulse_opt && *next_pulse_opt <= params::endTime())
        {
            _world->getEventQueue().push({*next_pulse_opt, EventType::TX_PULSED_START, tx});
        }
    }

References simulation::calculateResponse(), params::endTime(), core::World::getEventQueue(), core::World::getReceivers(), core::World::getTargets(), max, and core::TX_PULSED_START.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleTxStreamingEnd()

void core::SimulationEngine::handleTxStreamingEnd

(

radar::Transmitter *

tx

)

Handles a streaming transmitter turning off.

Parameters

tx	Pointer to the transmitting radar object.

Definition at line 1823 of file sim_threading.cpp.

    {
        (void)tx;
        // A transmitter stop is a transmit-time boundary, not an instantaneous receive-time cutoff.
        // Ended sources are removed only after all future receive-time samples fail the retarded-time gate.
        cleanupInactiveStreamingSources(_world->getSimulationState().t_current);
    }

References core::World::getSimulationState(), max, and core::SimulationState::t_current.

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

handleTxStreamingStart()

void core::SimulationEngine::handleTxStreamingStart

(

const ActiveStreamingSource &

source

)

Handles a streaming transmitter turning on.

Parameters

tx	Pointer to the transmitting radar object.

Definition at line 1817 of file sim_threading.cpp.

    {
        _world->getSimulationState().active_streaming_transmitters.push_back(source);
        appendStreamingTrackerSource();
    }

References core::SimulationState::active_streaming_transmitters, and core::World::getSimulationState().

Referenced by processEvent().

Here is the call graph for this function:

Here is the caller graph for this function:

processEvent()

void core::SimulationEngine::processEvent

(

const Event &

event

)

Dispatches a discrete simulation event to its specific handler.

Parameters

event

The event to process.

Definition at line 1715 of file sim_threading.cpp.

    {
        // NOLINTBEGIN(cppcoreguidelines-pro-type-static-cast-downcast)
        switch (event.type)
        {
        case EventType::TX_PULSED_START:
            handleTxPulsedStart(static_cast<Transmitter*>(event.source_object), event.timestamp);
            break;
        case EventType::RX_PULSED_WINDOW_START:
            handleRxPulsedWindowStart(static_cast<Receiver*>(event.source_object), event.timestamp);
            break;
        case EventType::RX_PULSED_WINDOW_END:
            handleRxPulsedWindowEnd(static_cast<Receiver*>(event.source_object), event.timestamp);
            break;
        case EventType::TX_STREAMING_START:
            if (const auto source = streamingSourceAtEvent(static_cast<Transmitter*>(event.source_object),
                                                           event.timestamp, _internal_stop_time);
                source.has_value())
            {
                handleTxStreamingStart(*source);
            }
            break;
        case EventType::TX_STREAMING_END:
            handleTxStreamingEnd(static_cast<Transmitter*>(event.source_object));
            break;
        case EventType::RX_STREAMING_START:
            handleRxStreamingStart(static_cast<Receiver*>(event.source_object));
            break;
        case EventType::RX_STREAMING_END:
            handleRxStreamingEnd(static_cast<Receiver*>(event.source_object));
            break;
        }
        // NOLINTEND(cppcoreguidelines-pro-type-static-cast-downcast)
    }

References handleRxPulsedWindowEnd(), handleRxPulsedWindowStart(), handleRxStreamingEnd(), handleRxStreamingStart(), handleTxPulsedStart(), handleTxStreamingEnd(), handleTxStreamingStart(), max, core::RX_PULSED_WINDOW_END, core::RX_PULSED_WINDOW_START, core::RX_STREAMING_END, core::RX_STREAMING_START, core::Event::source_object, core::Event::timestamp, core::TX_PULSED_START, core::TX_STREAMING_END, core::TX_STREAMING_START, and core::Event::type.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

processStreamingPhysics()

void core::SimulationEngine::processStreamingPhysics

(

RealType

t_event

)

Advances the time-stepped inner loop for active streaming systems.

Parameters

t_event

The timestamp of the next discrete event to process up to.

Definition at line 998 of file sim_threading.cpp.

    {
        auto& state = _world->getSimulationState();
        auto& t_current = state.t_current;
 
        if (t_event <= t_current)
        {
            return;
        }
 
        const RealType dt_sim = 1.0 / (params::rate() * params::oversampleRatio());
        const auto first_index = streamingSampleIndexAtOrAfter(t_current, dt_sim);
        const auto final_index = streamingSampleIndexAtOrAfter(t_event, dt_sim);
        const auto sample_count = final_index - first_index;
        const auto progress_report_stride = std::max<std::size_t>(1, sample_count / 1000);
 
        ensureCwPhaseNoiseLookup();
 
        while (t_current < t_event && !isCancellationRequested())
        {
            cleanupInactiveStreamingSources(t_current);
 
            const RealType chunk_end = streamingChunkEnd(t_current, t_event);
            if (chunk_end <= t_current)
            {
                break;
            }
 
            const auto start_index = streamingSampleIndexAtOrAfter(t_current, dt_sim);
            const auto end_index = streamingSampleIndexAtOrAfter(chunk_end, dt_sim);
            for (size_t sample_index = start_index; sample_index < end_index; ++sample_index)
            {
                if (shouldStopStreamingChunk(sample_index, start_index))
                {
                    break;
                }
                processStreamingSample(sample_index, first_index, final_index, progress_report_stride, dt_sim);
            }
 
            t_current = chunk_end;
            emitContextHeartbeatsThrough(t_current);
        }
        cleanupInactiveStreamingSources(t_current);
    }

References core::World::getSimulationState(), max, params::oversampleRatio(), and params::rate().

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

run()

void core::SimulationEngine::run

(

)

Starts and runs the main simulation loop until completion.

Definition at line 790 of file sim_threading.cpp.

    {
        if (_reporter)
        {
            _reporter->report("Initializing event-driven simulation...", 0, 100);
        }
 
        initializeFmcwIfResamplers();
 
        logSimulationMemoryProjection(*_world);
 
        initializeFinalizers();
 
        LOG(Level::INFO, "Starting unified event-driven simulation loop.");
        logStreamingSummaries();
 
        auto& event_queue = _world->getEventQueue();
        auto& state = _world->getSimulationState();
        const RealType end_time = _internal_stop_time;
 
        while (!event_queue.empty() && state.t_current <= end_time)
        {
            if (isCancellationRequested())
            {
                break;
            }
            const Event event = event_queue.top();
            event_queue.pop();
 
            processStreamingPhysics(event.timestamp);
            if (isCancellationRequested())
            {
                break;
            }
            flushFmcwIfBlocks();
            flushStreamingOutputBlocks();
 
            state.t_current = event.timestamp;
 
            processEvent(event);
            updateProgress();
        }
 
        const bool has_active_if_overrender =
            std::ranges::any_of(_world->getReceivers(), [](const auto& receiver)
                                { return receiver->isActive() && receiver->hasFmcwIfResamplingSink(); });
        if (!isCancellationRequested() && has_active_if_overrender)
        {
            processStreamingPhysics(end_time);
        }
        flushFmcwIfBlocks();
        flushStreamingOutputBlocks();
 
        shutdown();
    }

References core::World::getEventQueue(), core::World::getReceivers(), core::World::getSimulationState(), LOG, core::logSimulationMemoryProjection(), max, processEvent(), processStreamingPhysics(), and receiver.

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• packages/libfers/src/core/sim_threading.h
• packages/libfers/src/core/sim_threading.cpp