multi_input_processor.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "commrat/mailbox/mailbox.hpp"
#include "commrat/module/traits/processor_bases.hpp"
#include <optional>
#include <tuple>
 
namespace commrat {
 
template<typename ModuleType, typename InputTypesTuple, typename OutputData, typename OutputTypesTuple, std::size_t InputCount>
class MultiInputProcessor {
protected:
    template<std::size_t PrimaryIdx>
    auto receive_primary_input() -> MailboxResult<TimsMessage<std::tuple_element_t<PrimaryIdx, InputTypesTuple>>> {
        auto& module = static_cast<ModuleType&>(*this);
        
        if (!module.input_mailboxes_) {
            return MailboxResult<TimsMessage<std::tuple_element_t<PrimaryIdx, InputTypesTuple>>>(MailboxError::NotInitialized);
        }
        
        using PrimaryType = std::tuple_element_t<PrimaryIdx, InputTypesTuple>;
        auto& primary_mailbox = std::get<PrimaryIdx>(*module.input_mailboxes_);
        
        // BLOCKING receive - drives execution rate
        return primary_mailbox.template receive<PrimaryType>();
    }
    
    template<std::size_t PrimaryIdx, typename PrimaryMsgType>
    std::optional<InputTypesTuple> gather_all_inputs(const PrimaryMsgType& primary_msg) {
        auto& module = static_cast<ModuleType&>(*this);
        
        if (!module.input_mailboxes_) {
            return std::nullopt;
        }
        
        // Create tuple to hold all inputs
        InputTypesTuple all_inputs{};
        
        // Place primary input at its index
        std::get<PrimaryIdx>(all_inputs) = primary_msg.payload;
        
        // Phase 6.10: Sync secondary inputs using getData with primary timestamp from header
        // TimsMessage.header.timestamp is the authoritative timestamp
        bool all_synced = sync_secondary_inputs<PrimaryIdx>(primary_msg.header.timestamp, all_inputs);
        
        if (!all_synced) {
            return std::nullopt;
        }
        
        return all_inputs;
    }
    
    template<typename O = OutputData,
             typename = std::enable_if_t<!std::is_void_v<O>>>
    void call_multi_input_process(const InputTypesTuple& inputs, O& output) {
        call_multi_input_process_impl(inputs, output, std::make_index_sequence<InputCount>{});
    }
    
    void call_multi_input_multi_output_process(const InputTypesTuple& inputs, OutputTypesTuple& outputs) {
        call_multi_input_multi_output_process_impl(inputs, outputs, 
                                                    std::make_index_sequence<InputCount>{},
                                                    std::make_index_sequence<std::tuple_size_v<OutputTypesTuple>>{});
    }
    
private:
    template<std::size_t PrimaryIdx>
    bool sync_secondary_inputs(uint64_t primary_timestamp, InputTypesTuple& all_inputs) {
        return sync_secondary_inputs_impl<PrimaryIdx>(primary_timestamp, all_inputs, 
                                                       std::make_index_sequence<InputCount>{});
    }
    
    template<std::size_t PrimaryIdx, std::size_t... Is>
    bool sync_secondary_inputs_impl(uint64_t primary_timestamp, InputTypesTuple& all_inputs,
                                     std::index_sequence<Is...>) {
        // For each input index (except primary), call getData
        bool all_success = true;
        
        // Fold expression: process each secondary input
        ((Is != PrimaryIdx ? 
          (all_success = sync_input_at_index<Is>(primary_timestamp, all_inputs) && all_success) : 
          true), ...);
        
        return all_success;
    }
    
    template<std::size_t Index>
    bool sync_input_at_index(uint64_t primary_timestamp, InputTypesTuple& all_inputs) {
        auto& module = static_cast<ModuleType&>(*this);
        using InputType = std::tuple_element_t<Index, InputTypesTuple>;
        auto& mailbox = std::get<Index>(*module.input_mailboxes_);
        
        // Non-blocking getData with tolerance
        auto result = mailbox.template getData<InputType>(
            primary_timestamp,
            module.config_.sync_tolerance(),
            InterpolationMode::NEAREST
        );
        
        if (!result.has_value()) {
            // Phase 6.10: Mark input as invalid
            module.mark_input_invalid(Index);
            return false;  // getData failed
        }
        
        // Phase 6.10: Populate metadata for this input
        // Index matches position in Inputs<T1, T2, T3, ...>
        // getData succeeded - data is "new" (successfully retrieved from buffer)
        // Note: is_new_data = true means getData returned a value (not nullopt)
        //       is_new_data = false would indicate using fallback/default data
        module.update_input_metadata(Index, result.value(), true);
        
        // Store payload in tuple
        std::get<Index>(all_inputs) = result->payload;
        return true;
    }
    
    template<std::size_t... Is, typename O = OutputData,
             typename = std::enable_if_t<!std::is_void_v<O>>>
    void call_multi_input_process_impl(const InputTypesTuple& inputs, O& output, std::index_sequence<Is...>) {
        auto& module = static_cast<ModuleType&>(*this);
        
        // Unpack tuple and call process(const T1&, const T2&, ..., Output&)
        using Base = MultiInputProcessorBase<InputTypesTuple, OutputData, InputCount>;
        static_cast<Base*>(&module)->process(std::get<Is>(inputs)..., output);
    }
    
    template<std::size_t... InputIs, std::size_t... OutputIs>
    void call_multi_input_multi_output_process_impl(const InputTypesTuple& inputs, OutputTypesTuple& outputs,
                                                      std::index_sequence<InputIs...>,
                                                      std::index_sequence<OutputIs...>) {
        auto& module = static_cast<ModuleType&>(*this);
        
        // Unpack both tuples and call process(const T1&, ..., O1&, O2&, ...)
        using Base = MultiInputProcessorBase<InputTypesTuple, OutputTypesTuple, InputCount>;
        static_cast<Base*>(&module)->process(std::get<InputIs>(inputs)..., std::get<OutputIs>(outputs)...);
    }
};
 
} // namespace commrat