procFile_metric.cc

#include "artdaq-utilities/Plugins/MetricMacros.hh"
#include "fhiclcpp/ParameterSet.h"
#include "messagefacility/MessageLogger/MessageLogger.h"
#define TRACE_NAME "procFile_metric"
#include "trace.h"

#include <sys/stat.h>           // mkfifo
#include <fcntl.h>              // open
#include <stdlib.h>             // exit
#include <ctime>
#include <string>
#include <boost/thread.hpp>
#include <map>

namespace artdaq
{
    class ProcFileMetric : public MetricPlugin
    {
    private:
        std::string pipe_;
        std::unordered_map<std::string, std::string> value_map_;
        bool stopped_;
        boost::thread thread_;
    public:
        explicit ProcFileMetric(fhicl::ParameterSet const& config, std::string const& app_name) : MetricPlugin(config, app_name)
            , pipe_(pset.get<std::string>("pipe", "/tmp/eventQueueStat"))
            , value_map_()
            , stopped_(true)
        {
            auto names = pset.get<std::vector<std::string>>("names", std::vector<std::string>());

            for (auto name : names)
            {
                value_map_[name] = "";
            }

            int sts = mkfifo(pipe_.c_str(), 0777);
            if (sts != 0) { perror("ProcFileMetric mkfifo"); }
            TLOG(10) << "ProcFileMetric mkfifo()" << pipe_ << " sts=" << sts;
            startMetrics();
        }

        ~ProcFileMetric() {
            TLOG(11) << "~ProcFileMetric" ;
            stopMetrics();
        }

        std::string getLibName() const override { return "procFile"; }

        void sendMetric_(const std::string& name, const std::string& value, const std::string&) override {
            if (value_map_.count(name)) {
                TLOG(12) << "sendMetric_ setting value=" << value;
                value_map_[name] = value;
            }
        }

        void sendMetric_(const std::string& name, const int& value, const std::string& unit) override {
            sendMetric_(name, std::to_string(value), unit);
        }

        void sendMetric_(const std::string& name, const double& value, const std::string& unit) override {
            sendMetric_(name, std::to_string(value), unit);
        }

        void sendMetric_(const std::string& name, const float& value, const std::string& unit) override {
            sendMetric_(name, std::to_string(value), unit);
        }

        void sendMetric_(const std::string& name, const unsigned long int& value, const std::string& unit) override
        {
            sendMetric_(name, std::to_string(value), unit);
        }

        void startMetrics_() override
        {
            if (stopped_)
            {
                // start thread
                stopped_ = false;
                boost::thread::attributes attrs;
                attrs.set_stack_size(4096 * 2000); // 8000 KB
                try {
                    thread_ = boost::thread(attrs, boost::bind(&ProcFileMetric::writePipe, this));
                }
                catch (boost::exception const& e)
                {
                    std::cerr << "Creating ProcFile Metric thread failed! e: " << boost::diagnostic_information(e) << std::endl;
                    exit(4);
                }
            }
        }

        void stopMetrics_() override
        {
            if (!stopped_)
            {
                stopped_ = true;
                // do read on pipe to make sure writePipe is not blocking on open
                TLOG(11) << "stopMetrics_ before open " << pipe_;
                int fd = open(pipe_.c_str(), O_RDONLY | O_NONBLOCK);
                if (fd == -1) { perror("stopMetrics_ open(\"r\")"); exit(1); }
                TLOG(10) << "stopMetrics_ between open and unlink" << pipe_ << " fd=" << fd;
                unlink(pipe_.c_str());
                TLOG(11) << "stopMetrics_ unlinked " << pipe_;
# if 0
                char buf[256];
                read(fd, buf, sizeof(buf));
# endif
                usleep(10000);
                close(fd);
                TLOG(11) << "stopMetrics_ after close " << pipe_;
                if (thread_.joinable()) thread_.join();
            }
        }

        void writePipe()
        {
            while (!stopped_)
            {
                TLOG(11) << "writePipe before open";
                int fd = open(pipe_.c_str(), O_WRONLY);
                std::string str;
                for (auto value : value_map_) {
                    TLOG(10) << "writePipe open fd=" << fd << " name=" << value.first << " value=" << value.second; // can't have args b/c name may have %
                    str += value.first + ": " + value.second + "\n";
                    //snprintf(buf, sizeof(buf), "%s: %lu\n", value.first.c_str(), value.second);
                }
                int sts = write(fd, str.c_str(), str.size());
                TLOG(11) << "writePipe write complete sts=" << sts;
                close(fd);
                TLOG(11) << "writePipe after close -- about to usleep";
                usleep(400000); // must wait to make sure other end closes
            }

        }
    };
} //End namespace artdaq

DEFINE_ARTDAQ_METRIC(artdaq::ProcFileMetric)