CondorStatusUpdater.cc

Go to the documentation of this file.

#include "DataFormats/Provenance/interface/ModuleDescription.h"
#include "DataFormats/Provenance/interface/ParameterSetID.h"
#include "FWCore/Utilities/interface/TimingServiceBase.h"
#include "FWCore/Utilities/interface/CPUServiceBase.h"
#include "FWCore/ServiceRegistry/interface/Service.h"
#include "FWCore/ServiceRegistry/interface/ServiceMaker.h"
#include "FWCore/ServiceRegistry/interface/ProcessContext.h"
#include "FWCore/ServiceRegistry/interface/ActivityRegistry.h"
#include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
#include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
#include "FWCore/ParameterSet/interface/ParameterSet.h"
#include "FWCore/ParameterSet/interface/Registry.h"
#include "Utilities/StorageFactory/interface/StorageAccount.h"
#include "Utilities/XrdAdaptor/src/XrdStatistics.h"

#include <fcntl.h>
#include <unistd.h>
#include <sys/wait.h>
#include <spawn.h>
#include <iostream>
#include <fstream>
#include <sstream>
#include <cmath>
#include <chrono>
#include <sstream>
#include <atomic>
#include <string>
#include <set>

namespace edm {

    namespace service {

        class CondorStatusService
        {

        public:

            explicit CondorStatusService(ParameterSet const& pset, edm::ActivityRegistry& ar);
            ~CondorStatusService() {}
            CondorStatusService(const CondorStatusService&) = delete;
            CondorStatusService& operator=(const CondorStatusService&) = delete;

            static void fillDescriptions(ConfigurationDescriptions &descriptions);

        private:

            bool isChirpSupported();
            template<typename T> bool updateChirp(const std::string &key_suffix, const T &value);
            bool updateChirpQuoted(const std::string &key_suffix, const std::string &value);
            bool updateChirpImpl(std::string const &key, std::string const &value);
            inline void update();
            void firstUpdate();
            void lastUpdate();
            void updateImpl(time_t secsSinceLastUpdate);

            void preSourceConstruction(ModuleDescription const &md, int maxEvents, int maxLumis, int maxSecondsUntilRampdown);
            void eventPost(StreamContext const& iContext);
            void lumiPost(GlobalContext const&);
            void runPost(GlobalContext const&);
            void beginPre(PathsAndConsumesOfModulesBase const&, ProcessContext const& processContext);
            void beginPost();
            void endPost();
            void filePost(std::string const &, bool);

            bool m_debug;
            std::atomic_flag m_shouldUpdate;
            time_t m_beginJob = 0;
            time_t m_updateInterval = m_defaultUpdateInterval;
            float m_emaInterval = m_defaultEmaInterval;
            float m_rate = 0;
            static constexpr float m_defaultEmaInterval = 15*60; // Time in seconds to average EMA over for event rate.
            static constexpr unsigned int m_defaultUpdateInterval = 3*60;
            std::atomic<time_t> m_lastUpdate;
            std::atomic<std::uint_least64_t> m_events;
            std::atomic<std::uint_least64_t> m_lumis;
            std::atomic<std::uint_least64_t> m_runs;
            std::atomic<std::uint_least64_t> m_files;
            std::string m_tag;
            edm::ParameterSetID m_processParameterSetID;

            std::uint_least64_t m_lastEventCount = 0;
        };

    }

}

using namespace edm::service;

const unsigned int CondorStatusService::m_defaultUpdateInterval;
constexpr float CondorStatusService::m_defaultEmaInterval;

CondorStatusService::CondorStatusService(ParameterSet const& pset, edm::ActivityRegistry& ar)
  :
    m_debug(false),
    m_lastUpdate(0),
    m_events(0),
    m_lumis(0),
    m_runs(0),
    m_files(0)
{
    m_shouldUpdate.clear();
    if (pset.exists("debug"))
    {
        m_debug = true;
    }
    if (!isChirpSupported()) {return;}

    firstUpdate();

    ar.watchPostCloseFile(this, &CondorStatusService::filePost);
    ar.watchPostEvent(this, &CondorStatusService::eventPost);
    ar.watchPostGlobalEndLumi(this, &CondorStatusService::lumiPost);
    ar.watchPostGlobalEndRun(this, &CondorStatusService::runPost);
    ar.watchPreBeginJob(this, &CondorStatusService::beginPre);
    ar.watchPostBeginJob(this, &CondorStatusService::beginPost);
    ar.watchPostEndJob(this, &CondorStatusService::endPost);

    if (pset.exists("updateIntervalSeconds"))
    {
        m_updateInterval = pset.getUntrackedParameter<unsigned int>("updateIntervalSeconds");
    }
    if (pset.exists("EMAInterval"))
    {
        m_emaInterval = pset.getUntrackedParameter<double>("EMAInterval");
    }
    if (pset.exists("tag"))
    {
        m_tag = pset.getUntrackedParameter<std::string>("tag");
    }
}


void
CondorStatusService::eventPost(StreamContext const& iContext)
{
    m_events++;
    update();
}


void
CondorStatusService::lumiPost(GlobalContext const&)
{
    m_lumis++;
    update();
}


void
CondorStatusService::runPost(GlobalContext const&)
{
    m_runs++;
    update();
}


void
CondorStatusService::filePost(std::string const & /*lfn*/, bool /*usedFallback*/)
{
    m_files++;
    update();
}


void
CondorStatusService::beginPre(PathsAndConsumesOfModulesBase const&, ProcessContext const& processContext)
{
    if (!m_processParameterSetID.isValid())
    {
        m_processParameterSetID = processContext.parameterSetID();
    }
}


void
CondorStatusService::beginPost()
{
    ParameterSet const& processParameterSet = edm::getParameterSet(m_processParameterSetID);
    const edm::ParameterSet &pset = processParameterSet.getParameterSet("@main_input");
    // PSet info from edm::ScheduleItems
    int maxEvents = processParameterSet.getUntrackedParameterSet("maxEvents", ParameterSet()).getUntrackedParameter<int>("input", -1);
    int maxLumis = processParameterSet.getUntrackedParameterSet("maxLuminosityBlocks", ParameterSet()).getUntrackedParameter<int>("input", -1);

    // lumisToProcess from EventSkipperByID (PoolSource and similar)
    std::vector<edm::LuminosityBlockRange> toProcess = pset.getUntrackedParameter<std::vector<LuminosityBlockRange> >("lumisToProcess", std::vector<LuminosityBlockRange>());
    edm::sortAndRemoveOverlaps(toProcess);
    uint64_t lumiCount = 0;
    for (auto const &range : toProcess)
    {
        if (range.startRun() != range.endRun()) {break;}
        if (range.endLumi() >= edm::LuminosityBlockID::maxLuminosityBlockNumber()) {break;}
        lumiCount += (range.endLumi()-range.startLumi());
    }
    // Handle sources deriving from ProducerSourceBase
    unsigned int eventsPerLumi = pset.getUntrackedParameter<unsigned int>("numberEventsInLuminosityBlock", 0);
    if ((lumiCount == 0) && (maxEvents > 0) && (eventsPerLumi > 0))
    {
        lumiCount = static_cast<unsigned int>(std::ceil(static_cast<float>(maxEvents) / static_cast<float>(eventsPerLumi)));
    }

    std::vector<std::string> fileNames = pset.getUntrackedParameter<std::vector<std::string>>("fileNames", std::vector<std::string>());
    std::stringstream ss_max_files; ss_max_files << fileNames.size();
    updateChirp("MaxFiles", ss_max_files.str());

    if (lumiCount > 0)
    {
        if (maxLumis < 0) {maxLumis = lumiCount;}
        if (maxLumis > static_cast<int>(lumiCount))
        {
            maxLumis = lumiCount;
        }
    }
    if (maxEvents > 0)
    {
        std::stringstream ss_max_events; ss_max_events << maxEvents;
        updateChirp("MaxEvents", ss_max_events.str());
    }
    if (maxLumis > 0)
    {
        std::stringstream ss_max_lumis; ss_max_lumis << maxLumis;
        updateChirp("MaxLumis", ss_max_lumis.str());
    }

    m_beginJob = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
    update();
}


void
CondorStatusService::endPost()
{
    lastUpdate();
}


bool
CondorStatusService::isChirpSupported()
{
    if (m_debug) {return true;}

    return getenv("_CONDOR_CHIRP_CONFIG") && updateChirp("Elapsed", "0");
}


void
CondorStatusService::firstUpdate()
{
    // Note we always update all our statistics to 0 / false / -1
    // This allows us to overwrite the activities of a previous cmsRun process
    // within this HTCondor job.
    updateImpl(0);
    updateChirp("MaxFiles", "-1");
    updateChirp("MaxEvents", "-1");
    updateChirp("MaxLumis", "-1");
    updateChirp("Done", "false");

    edm::Service<edm::CPUServiceBase> cpusvc;
    std::string models;
    double avgSpeed;
    if (cpusvc.isAvailable() && cpusvc->cpuInfo(models, avgSpeed)) {
        updateChirpQuoted("CPUModels", models);
        updateChirp("CPUSpeed", avgSpeed);
    }
}


void
CondorStatusService::lastUpdate()
{
    time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
    updateImpl(now - m_lastUpdate);
    updateChirp("Done", "true");
    edm::Service<edm::CPUServiceBase> cpusvc;
    if (!cpusvc.isAvailable()) {
        std::cout << "At post, CPU service is NOT available.\n";
    }
}


void
CondorStatusService::update()
{
    time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
    if ((now - m_lastUpdate.load(std::memory_order_relaxed)) > m_updateInterval)
    {
        if (!m_shouldUpdate.test_and_set(std::memory_order_acquire))
        {
            try
            {
                time_t sinceLastUpdate = now - m_lastUpdate;
                m_lastUpdate = now;
                updateImpl(sinceLastUpdate);
                m_shouldUpdate.clear(std::memory_order_release);
            }
            catch (...)
            {
                m_shouldUpdate.clear(std::memory_order_release);
                throw;
            }
        }
    }
}


void
CondorStatusService::updateImpl(time_t sinceLastUpdate)
{
    time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
    time_t jobTime = now-m_beginJob;

    edm::Service<edm::TimingServiceBase> timingsvc;
    if (timingsvc.isAvailable()) {
        updateChirp("TotalCPU", timingsvc->getTotalCPU());
    }

    updateChirp("LastUpdate", now);

    if (!m_events || (m_events > m_lastEventCount))
    {
        updateChirp("Events", m_events);
    }

    updateChirp("Lumis", m_lumis);

    updateChirp("Runs", m_runs);

    updateChirp("Files", m_files);

    float ema_coeff = 1 - std::exp(-static_cast<float>(sinceLastUpdate)/std::max(std::min(m_emaInterval, static_cast<float>(jobTime)), 1.0f));
    if (sinceLastUpdate > 0)
    {
        updateChirp("Elapsed", jobTime);
        m_rate = ema_coeff*static_cast<float>(m_events-m_lastEventCount)/static_cast<float>(sinceLastUpdate) + (1.0-ema_coeff)*m_rate;
        m_lastEventCount = m_events;
        updateChirp("EventRate", m_rate);
    }

    // If Xrootd was used, pull the statistics from there.
    edm::Service<XrdAdaptor::XrdStatisticsService> xrdsvc;
    if (xrdsvc.isAvailable()) {
        for (auto const &iter : xrdsvc->condorUpdate()) {
            std::string site = iter.first;
            site.erase(std::remove_if(site.begin(), site.end(),
                       [](char x){return !isalnum(x) && (x != '_');}),
                       site.end());
            auto & iostats = iter.second;
            updateChirp("IOSite_" + site + "_ReadBytes", iostats.bytesRead);
            updateChirp("IOSite_" + site + "_ReadTimeMS",
                        std::chrono::duration_cast<std::chrono::milliseconds>(iostats.transferTime).count());
        }
    }

    // Update storage account information
    auto const& stats = StorageAccount::summary();
    uint64_t readOps = 0;
    uint64_t readVOps = 0;
    uint64_t readSegs = 0;
    uint64_t readBytes = 0;
    uint64_t readTimeTotal = 0;
    uint64_t writeBytes = 0;
    uint64_t writeTimeTotal = 0;
    const auto token = StorageAccount::tokenForStorageClassName("tstoragefile");
    for (const auto & storage : stats)
    {
        // StorageAccount records statistics for both the TFile layer and the
        // StorageFactory layer.  However, the StorageFactory statistics tend to
        // be more accurate as various backends may alter the incoming read requests
        // (such as when lazy-download is used).
        if (storage.first == token.value()) {continue;}
        for (const auto & counter : storage.second)
        {
            if (counter.first == static_cast<int>(StorageAccount::Operation::read))
            {
                readOps += counter.second.successes;
                readSegs++;
                readBytes += counter.second.amount;
                readTimeTotal += counter.second.timeTotal;
            }
            else if (counter.first == static_cast<int>(StorageAccount::Operation::readv))
            {
                readVOps += counter.second.successes;
                readSegs += counter.second.vector_count;
                readBytes += counter.second.amount;
                readTimeTotal += counter.second.timeTotal;
            }
            else if ((counter.first == static_cast<int>(StorageAccount::Operation::write)) || (counter.first == static_cast<int>(StorageAccount::Operation::writev)))
            {
                writeBytes += counter.second.amount;
                writeTimeTotal += counter.second.timeTotal;
            }
        }
    }
    updateChirp("ReadOps", readOps);
    updateChirp("ReadVOps", readVOps);
    updateChirp("ReadSegments", readSegs);
    updateChirp("ReadBytes", readBytes);
    updateChirp("ReadTimeMsecs", readTimeTotal/(1000*1000));
    updateChirp("WriteBytes", writeBytes);
    updateChirp("WriteTimeMsecs", writeTimeTotal/(1000*1000));
}


template<typename T> bool
CondorStatusService::updateChirp(const std::string &key_suffix, const T &value)
{
    std::stringstream ss; ss << value;
    return updateChirpImpl(key_suffix, ss.str());
}

bool
CondorStatusService::updateChirpQuoted(const std::string &key_suffix, const std::string &value)
{
    std::string value_copy = value;
    // Remove double-quotes or the \ character (as it has special escaping semantics in ClassAds).
    // Make sure we have ASCII characters.
    // Otherwise, remainder is allowed (including tabs, newlines, single-quotes).
    value_copy.erase(remove_if(value_copy.begin(), value_copy.end(), [](const char &c) {return !isascii(c) || (c == '"') || (c == '\\');}), value_copy.end());
    return updateChirpImpl(key_suffix, "\"" + value_copy + "\"");
}


bool
CondorStatusService::updateChirpImpl(const std::string &key_suffix, const std::string &value)
{
    std::stringstream ss; ss << "ChirpCMSSW" << m_tag << key_suffix;
    std::string key = ss.str();
    if (m_debug)
    {
        std::cout << "condor_chirp set_job_attr_delayed " << key << " " << value << std::endl;
    }
    int pid = 0;
    posix_spawn_file_actions_t file_actions;
    int devnull_fd = open("/dev/null", O_RDWR);
    if (devnull_fd == -1) {return false;}
    posix_spawn_file_actions_init(&file_actions);
    posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 1);
    posix_spawn_file_actions_adddup2(&file_actions, devnull_fd, 2);
    const std::string chirp_name = "condor_chirp";
    const std::string set_job_attr = "set_job_attr_delayed";
    std::vector<const char *> argv;
    argv.push_back(chirp_name.c_str());
    argv.push_back(set_job_attr.c_str());
    argv.push_back(key.c_str());
    argv.push_back(value.c_str());
    argv.push_back(nullptr);
    int status = posix_spawnp(&pid, "condor_chirp", &file_actions, nullptr, const_cast<char* const*>(&argv[0]), environ);
    close(devnull_fd);
    posix_spawn_file_actions_destroy(&file_actions);
    if (status)
    {
       return false;
    }
    while ((waitpid(pid, &status, 0) == -1) && errno == -EINTR) {}
    return status == 0;
}


void
CondorStatusService::fillDescriptions(ConfigurationDescriptions &descriptions)
{
    ParameterSetDescription desc;
    desc.setComment("Service to update HTCondor with the current CMSSW status.");
    desc.addOptionalUntracked<unsigned int>("updateIntervalSeconds", m_defaultUpdateInterval)
      ->setComment("Interval, in seconds, for HTCondor updates");
    desc.addOptionalUntracked<bool>("debug", false)
      ->setComment("Enable debugging of this service");
    desc.addOptionalUntracked<double>("EMAInterval", m_defaultEmaInterval)
      ->setComment("Interval, in seconds, to calculate event rate over (using EMA)");
    desc.addOptionalUntracked<std::string>("tag")
      ->setComment("Identifier tag for this process (a value of 'Foo' results in ClassAd attributes of the form 'ChirpCMSSWFoo*')");
    descriptions.add("CondorStatusService", desc);
}


typedef edm::serviceregistry::AllArgsMaker<edm::service::CondorStatusService> CondorStatusServiceMaker;
DEFINE_FWK_SERVICE_MAKER(CondorStatusService,CondorStatusServiceMaker);