shmPub.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_STANDALONETOOLS_SHMPUB_HPP
#define RTCTK_STANDALONETOOLS_SHMPUB_HPP
 
// 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 <atomic>
#include <chrono>
#include <ctime>
#include <functional>
#include <iostream>
#include <numeric>
#include <vector>
 
namespace rtctk::standaloneTools {
 
static bool g_stop = false;
 
void SignalHandler(int signal) {
    std::cout << "\nSignal to exit received\n";
    g_stop = true;
}
 
template <class TopicType, class WriterType = ipcq::Writer<TopicType>>
class ShmPub {
public:
    ShmPub(int argc, char* argv[])
        : m_queue_size(0)
        , m_sample_delay(0)
        , m_numa(0)
        , m_print_every(0)
        , m_gen_frames(0)
        , m_sample_id_increment(1)
        , m_repeat_mode(false)
        , m_help_only(false) {
        using namespace boost::program_options;
 
        try {
            options_description desc("Allowed options");
            // clang-format off
            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")
                ("sample-id-increment,i",
                    value<unsigned>(&m_sample_id_increment)->default_value(1),
                    "sample_id increment")
                ("repeat-mode,r",
                    bool_switch(&m_repeat_mode),
                    "Repeat output when all samples are written");
            // clang-format on
 
            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 << "\n";
            } else {
                m_help_only = false;
                std::cout << "fits-file: " << m_filename << "\n";
                std::cout << "queue-name: " << m_queue_name << "\n";
                std::cout << "queue-size: " << m_queue_size << "\n";
                std::cout << "sample-delay: " << m_sample_delay << "\n";
                if (vm.count("numa-node")) {
                    std::cout << "numa-node: " << m_numa << "\n";
                }
                std::cout << "print-every: " << m_print_every << "\n";
                std::cout << "gen-frames: " << m_gen_frames << "\n";
                std::cout << "sample-id-increment: " << m_sample_id_increment << "\n";
                std::cout << "repeat-mode: " << m_repeat_mode << "\n";
 
                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);
                }
            }
        } catch (const std::exception& e) {
            std::cerr << "Exception:" << e.what() << "\n";
        }
    }
 
    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 << "\n";
                data = ReadFits(m_filename);
            } else {
                std::cout << "Generating data\n";
                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\n";
            WriteToShm(data);
 
        } catch (const std::exception& e) {
            std::cout << e.what() << "\n";
            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;
 
    virtual void AdjustSample(TopicType& sample) const {};
 
protected:
    std::string GetQueueName() const {
        return m_queue_name;
    }
 
    size_t GetQueueSize() const {
        return m_queue_size;
    }
 
    int GetSampleDelay() const {
        return m_sample_delay;
    }
 
    int GetNuma() const {
        return m_numa;
    }
 
    bool GetRepeatMode() const {
        return m_repeat_mode;
    }
 
private:
    void WriteToShm(std::vector<TopicType>& data) {
        using namespace std::chrono;
 
        size_t n_written = 0;
        auto t_sent = steady_clock::now();
        auto t_last = t_sent;
        do {
            for (auto& sample : data) {
                if (g_stop) {
                    return;
                }
                AdjustSample(sample);
                if (m_repeat_mode) {
                    sample.sample_id = n_written * m_sample_id_increment;
                }
                t_sent = steady_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 << "\n";
                    std::cout << "Total time to write " << m_print_every << " : " << dur << " ms\n";
                    std::cout << "Average frame time: " << static_cast<float>(dur) / m_print_every
                              << " ms\n";
                    t_last = t_sent;
                }
                while (duration_cast<milliseconds>(steady_clock::now() - t_sent).count() <
                       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
    unsigned m_sample_id_increment;  //< sample id increment
    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>
ReadColumnFromFits(fitsfile* fptr, const std::string& name, long nrows, bool output = false) {
    int status = 0;
    int col, typecode, anynul;
    long repeat, width;
    float nulval;
 
    // The const_cast is a workaround for a buggy cfitsio API. The argument is never actually
    // modified and should have been declared const.
    fits_get_colnum(fptr, CASESEN, const_cast<char*>(name.c_str()), &col, &status);
    if (status) {
        fits_report_error(stderr, status);
        throw std::runtime_error("Error getting column: " + 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:" + name);
    }
 
    if (output) {
        std::cout << "name: " << name << "\n";
        std::cout << "col: " << col << "\n";
        std::cout << "typecode: " << typecode << "\n";
        std::cout << "repeat: " << repeat << "\n";
        std::cout << "width: " << width << "\n";
    }
 
    // 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: " + name);
    }
    return data;
}
 
}  // namespace rtctk::standaloneTools
 
#endif  // RTCTK_STANDALONETOOLS_SHMPUB_HPP