shmPub.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_STANDALONETOOLS_SHMPUB_H
#define RTCTK_STANDALONETOOLS_SHMPUB_H
 
// arg parsing
#include <boost/program_options.hpp>
 
// cfitsio
#include <cfitsio/fitsio.h>
 
// include the numapp for threading
#include <numapp/mempolicy.hpp>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
// include the ipcq for writer
#include <ipcq/writer.hpp>
 
#include <iostream>
#include <vector>
#include <atomic>
#include <chrono>
#include <ctime>
#include <functional>
#include <numeric>
 
namespace rtctk::standaloneTools {
static bool g_stop = false;
 
void SignalHandler(int signal)
{
    std::cout << std::endl << "Signal to exit received " << std::endl;
    g_stop = true;
}
 
 
template<class TopicType, class WriterType = ipcq::Writer<TopicType>>
class ShmPub {
    public:
        ShmPub(int argc, char* argv[]) : m_help_only(false)
        {
            using namespace boost::program_options;
 
            options_description desc("Allowed options");
            desc.add_options()
            ("help,h", "produce help message")
            ("fits-file,f", value<std::string>(&m_filename)->default_value(""), "fits input file: if not provided the app will generate data")
            ("queue-name,q", value<std::string>(&m_queue_name)->default_value("default_shm_queue"), "shm queue name")
            ("queue-size,s", value<size_t>(&m_queue_size)->default_value(1000), "size of the queue")
            ("sample-delay,d", value<int>(&m_sample_delay)->default_value(10), "inter-sample delay in ms")
            ("numa-node,n", value<int>(&m_numa), "numa node for shm queue")
            ("print-every,p", value<int>(&m_print_every)->default_value(0), "when to print to screen the number of sample written")
            ("gen-frames,g",value<int>(&m_gen_frames)->default_value(100), "Number of frames to generate")
            ("repeat-mode,r",bool_switch(&m_repeat_mode), "Repeat output when all samples are written");
 
            variables_map vm;
            store(command_line_parser(argc, argv).options(desc).run(), vm);
            notify(vm);
 
            if (vm.count("help")) {
                m_help_only = true;
                std::cout << desc << std::endl;
            }else {
                m_help_only = false;
                std::cout << "fits-file: " << m_filename << std::endl;
                std::cout << "queue-name: " << m_queue_name << std::endl;
                std::cout << "queue-size: " << m_queue_size << std::endl;
                std::cout << "sample-delay: " << m_sample_delay << std::endl;
                if(vm.count("numa-node")) {
                    std::cout << "numa-node: " << m_numa << std::endl;
                }
                std::cout << "print-every: " << m_print_every << std::endl;
                std::cout << "gen-frames: " << m_gen_frames << std::endl;
                std::cout << "repeat-mode: " << m_repeat_mode << std::endl;
 
                if(vm.count("numa-node")) {
                    m_writer = std::make_unique<WriterType>(m_queue_name.c_str(),
                                                        m_queue_size,
                                                        numapp::MemPolicy::MakeBindNode(m_numa));
                }else {
                    m_writer = std::make_unique<WriterType>(m_queue_name.c_str(),
                                                        m_queue_size);
                }
            }
        }
 
        virtual ~ShmPub() = default;
 
        int Run()
        {
            if(m_help_only) { return 0; }
 
            int ret_val = 0;
 
            try {
 
                signal(SIGINT,SignalHandler);
 
                std::vector<TopicType> data;
 
                // checks if filename has been indicated if it has loads data by calling the user
                // overloaded function ReadFits if not provided calls the user overloaded function
                // GenData
                if(not m_filename.empty()) {
                    std::cout << "Reading data from FITS file: " << m_filename << std::endl;
                    data = ReadFits(m_filename);
                }else {
                    std::cout << "Generating data" << std::endl;
                    data = GenData(m_gen_frames);
                }
 
                // check to make sure m_data is populated
                if(data.empty()) {
                    throw std::runtime_error("Data vector is not populated so will exit");
                }
 
                // calls main loop
                std::cout << "Writing data to shared memory queue" << std::endl;
                WriteToShm(data);
 
            }catch(std::exception const& e) {
 
                std::cout << e.what() << std::endl;
                ret_val = -1;
            }
 
            // Close queue to signal and give readers time detach from queue
            m_writer->Close();
#ifndef UNIT_TEST
            std::this_thread::sleep_for(std::chrono::seconds(2));
#endif
 
            return ret_val;
        }
 
        virtual std::vector<TopicType> ReadFits(std::string filename) = 0;
 
        virtual std::vector<TopicType> GenData(int num_frames) = 0;
 
    private:
        void WriteToShm(std::vector<TopicType>& data)
        {
            using namespace std::chrono;
 
            size_t n_written = 0;
            auto t_sent = system_clock::now();
            auto t_last = t_sent;
            do {
                for(auto& sample : data) {
                    if(g_stop) { return; }
                    sample.sample_id = n_written;
                    t_sent = system_clock::now();
                    std::error_code err = m_writer->Write(sample, ipcq::Notify::All);
                    if(err) {
                        throw std::runtime_error("Error writing to shm: " + err.message());
                    }
 
                    n_written++;
                    if(n_written && m_print_every && (n_written % m_print_every == 0)) {
                        auto dur = duration_cast<milliseconds>(t_sent - t_last).count();
                        std::cout << "Samples written:  " << n_written << std::endl;
                        std::cout << "Total time to write " << m_print_every << " : " << dur << " ms" << std::endl;
                        std::cout << "Average frame time: " << (float)dur/m_print_every << " ms" << std::endl;
                        t_last = t_sent;
                    }
 
                    std::this_thread::sleep_until(t_sent + milliseconds(m_sample_delay));
                }
 
            }while(m_repeat_mode);
        }
 
        std::string m_queue_name;   //< Queue name to be used by the writer
        size_t m_queue_size;        //< number of position in shm queue
        std::string m_filename;     //< path to fits file being read
        int m_sample_delay;         //< delay between samples being writter (ms)
        int m_numa;                 //< which numa node to provide writer
 
        int m_print_every;          //< print status every N samples
        int m_gen_frames;           //< if generation data how many sample to be generated
        bool m_repeat_mode;         //< data will loop forever with an ever increasing sample_id
        bool m_help_only;            //< if help only will not enter writing loop
 
        std::unique_ptr<WriterType> m_writer; //< the ipcq writer
};
 
    template<class T>
    std::vector<T> read_col_from_fits(fitsfile* fptr, char* name, long nrows, bool output=false)
    {
        int status = 0;
        int col, typecode, anynul;
        long repeat, width;
        float nulval;
 
        fits_get_colnum(fptr, CASESEN, name, &col, &status);
        if (status) {
            fits_report_error(stderr, status);
            throw std::runtime_error("Error getting column: " + std::string(name));
        }
 
        fits_get_coltype(fptr, col, &typecode, &repeat, &width, &status);
        if (status) {
            fits_report_error(stderr, status);
            throw std::runtime_error("Error getting coltype of:" + std::string(name));
        }
 
        if(output) {
            std::cout << "name: " << name << std::endl;
            std::cout << "col: " << col << std::endl;
            std::cout << "typecode: " << typecode << std::endl;
            std::cout << "repeat: " << repeat << std::endl;
            std::cout << "width: " << width << std::endl;
        }
 
        // load in required data
        std::vector<T> data;
        data.resize(repeat*nrows); // we are assuming the vector to be matrix with row major.
        T* d = data.data();
        fits_read_col(fptr, typecode, col, 1, 1, repeat*nrows, &nulval, d, &anynul, &status);
        if (status) {
            fits_report_error(stderr, status);
            throw std::runtime_error("Error reading column: " + std::string(name));
        }
        return data;
    }
 
}
#endif