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 <fcntl.h>     // open
#include <stdlib.h>    // exit
#include <sys/stat.h>  // mkfifo
#include <boost/thread.hpp>
#include <ctime>
#include <map>
#include <string>

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)