ProcessorInternalStateBufferManager.hpp

Go to the documentation of this file.

 
#include "tpglibs/ProcessorInternalStateNameRegistry.hpp"
#include "tpglibs/ProcessorMetricArray.hpp"
 
#include <string>
#include <vector>
#include <memory>
#include <atomic>
#include <immintrin.h>
#include <array>
#include <cstdint>
#include <mm_malloc.h>
 
#ifndef TPGLIBS_PROCESSORINTERNALSTATEBUFFERMANAGER_HPP_
#define TPGLIBS_PROCESSORINTERNALSTATEBUFFERMANAGER_HPP_
 
namespace tpglibs {
    
    template <typename T>
    class ProcessorInternalStateBufferManager {
        public:
        using signal_t = T;
        
        ProcessorInternalStateBufferManager();
        
        ~ProcessorInternalStateBufferManager();
        
        void write_to_active_buffer();
        
        ProcessorMetricArray<signal_t> switch_buffer_and_read();
        
        ProcessorMetricArray<std::array<int16_t, 16>> switch_buffer_and_read_casted();
        
        void configure_from_registry(ProcessorInternalStateNameRegistry<signal_t>* registry);
        
        void clear() {
            for (auto& buf : m_store_buffers) { _mm_free(buf.m_data); }
        }
        
        protected:
        
        void allocate_buffers(size_t buffer_size);
        
        void allocate_cast_buffers(size_t) {}; // Do nothing for the generic template
        
        void switch_active_buffer();
        
        private:
        std::vector<std::shared_ptr<signal_t>> m_internal_state_item_ptrs;
        
        ProcessorMetricArray<signal_t> m_store_buffers[2]{};
        
        ProcessorMetricArray<std::array<int16_t,16>> m_cast_store_buffers[2]{};
        
        std::atomic<ProcessorMetricArray<std::array<int16_t,16>>*> m_cast_active_buffer{ &m_cast_store_buffers[0] };
        
        std::atomic<ProcessorMetricArray<signal_t>*> m_write_buffer = &m_store_buffers[0];
        
        std::atomic<ProcessorMetricArray<signal_t>*> m_read_buffer = &m_store_buffers[0];
        
        std::atomic<uint16_t> m_write_seq{0};
        
        std::atomic<uint16_t> m_last_read_seq{0};
        
        size_t m_buffer_size;
    };
    
    // Template function implementations
    template <typename T>
    ProcessorInternalStateBufferManager<T>::ProcessorInternalStateBufferManager() {
    }
    
    template <typename T>
    ProcessorInternalStateBufferManager<T>::~ProcessorInternalStateBufferManager() {
        clear();
    }
    
    template <typename T>
    void ProcessorInternalStateBufferManager<T>::configure_from_registry(ProcessorInternalStateNameRegistry<signal_t>* registry) {
        // obtain the number of internal state items
        auto num_items = registry->get_number_of_requested_internal_states();
        allocate_buffers(num_items);
        allocate_cast_buffers(num_items);
        // Writer starts with buffer 0, reader starts with buffer 1 (they must be different!)
        m_write_buffer.store(&m_store_buffers[0], std::memory_order_release);
        m_read_buffer.store(&m_store_buffers[1], std::memory_order_release);
        m_cast_active_buffer.store(&m_cast_store_buffers[0], std::memory_order_release);
        // reset sequence counters
        m_write_seq.store(0, std::memory_order_release);
        m_last_read_seq.store(0, std::memory_order_release);
        // obtain the pointers to the internal state items
        m_internal_state_item_ptrs = registry->get_all_requested_internal_state_item_ptrs();
    }
    
    template <typename T>
    void ProcessorInternalStateBufferManager<T>::allocate_buffers(size_t buffer_size) {
        for (auto& buf : m_store_buffers) {
            buf.m_size = buffer_size;
            buf.m_data = static_cast<T*>(
                _mm_malloc(buf.m_size * sizeof(T), alignof(T))
            );
        }
    }
    
    template <typename T>
    void ProcessorInternalStateBufferManager<T>::write_to_active_buffer() {
        // Increment seq to indicate write start (becomes odd)
        m_write_seq.fetch_add(1, std::memory_order_release);
        
        auto write_ptr = m_write_buffer.load(std::memory_order_acquire);
        
        // Write to write buffer
        for (size_t i = 0; i < m_internal_state_item_ptrs.size(); i++) {
            // Check for nullptr before dereferencing (handles invalid state names)
            if (m_internal_state_item_ptrs[i] != nullptr) {
                write_ptr->m_data[i] = *m_internal_state_item_ptrs[i];
            } else {
                // If pointer is null, write a zeroed value
                write_ptr->m_data[i] = T{};
            }
        }
        
        // Increment seq to indicate write complete (becomes even)
        m_write_seq.fetch_add(1, std::memory_order_release);
    }
    
    template <typename T>
    ProcessorMetricArray<typename ProcessorInternalStateBufferManager<T>::signal_t> ProcessorInternalStateBufferManager<T>::switch_buffer_and_read() {
        // Wait until no write is in progress
        uint16_t current_write_seq;
        do {
            current_write_seq = m_write_seq.load(std::memory_order_acquire);
        } while (current_write_seq & 1); // spin if writer is mid-write (odd seq number)
        
        // Check if there's new data since last read
        uint16_t last_read = m_last_read_seq.load(std::memory_order_acquire);
        
        if (current_write_seq != last_read && current_write_seq > 0) {
            // New data available - swap the buffers between reader and writer
            auto current_read = m_read_buffer.load(std::memory_order_acquire);
            auto current_write = m_write_buffer.load(std::memory_order_acquire);
            
            // Swap: reader gets what writer just finished, writer gets what reader was using
            m_read_buffer.store(current_write, std::memory_order_release);
            m_write_buffer.store(current_read, std::memory_order_release);
            
            // Update last read sequence
            m_last_read_seq.store(current_write_seq, std::memory_order_release);
        }
        
        // Return data from read buffer
        auto read_ptr = m_read_buffer.load(std::memory_order_acquire);
        return *read_ptr;
    }
    
    // Template specializations
    // Specialization for __m256i since it has additional cast buffers.
    template<>
    inline void ProcessorInternalStateBufferManager<__m256i>::clear() {
        // free double buffer for temp values
        for (auto& buf : m_store_buffers) { _mm_free(buf.m_data); }
        // free double buffer for casted values
        for (auto& buf : m_cast_store_buffers) { _mm_free(buf.m_data); }
    }
    
    // Specialization for __m256i -> std::array<int16_t, 16> cast
    template <>
    inline ProcessorMetricArray<std::array<int16_t, 16>> ProcessorInternalStateBufferManager<__m256i>::switch_buffer_and_read_casted() {
        // First, get the raw data using switch_buffer_and_read
        auto raw_data = switch_buffer_and_read();
        
        auto* cast_free = m_cast_active_buffer.load(std::memory_order_acquire);
        
        // Cast each __m256i to std::array<int16_t, 16>
        for (size_t i = 0; i < raw_data.m_size; ++i) {
            // Cast and save to cast buffer
            _mm256_store_si256(
                reinterpret_cast<__m256i*>(cast_free->m_data[i].data()),
                raw_data.m_data[i]
            );
        }
        
        // Flip the active cast buffer
        
        auto* next = (cast_free == &m_cast_store_buffers[0]) ? &m_cast_store_buffers[1] : &m_cast_store_buffers[0];
        m_cast_active_buffer.store(next, std::memory_order_release);
        
        return *cast_free;
    }
    
    // Specialization for __m256i
    template<>
    inline void ProcessorInternalStateBufferManager<__m256i>::allocate_cast_buffers(size_t n) {
        for (auto& buf : m_cast_store_buffers) {
            buf.m_size = n;
            buf.m_data = static_cast<std::array<int16_t,16>*>(
                _mm_malloc(n * sizeof(std::array<int16_t,16>), 32)
            );
        }
        m_cast_active_buffer.store(&m_cast_store_buffers[0], std::memory_order_release);
    }
    
    // Specialization for std::array<int16_t, 16> -> std::array<int16_t, 16> cast (trivial)
    template <>
    inline ProcessorMetricArray<std::array<int16_t, 16>> ProcessorInternalStateBufferManager<std::array<int16_t, 16>>::switch_buffer_and_read_casted() {
        // First, get the raw data using switch_buffer_and_read
        auto raw_data = switch_buffer_and_read();
        
        // For std::array<int16_t, 16>, the cast is trivial - just return the data as-is
        return raw_data;
    }
    
} // namespace tpglibs
 
#endif // TPGLIBS_PROCESSORINTERNALSTATEBUFFERMANAGER_HPP_