DataProcessingStats.h

Go to the documentation of this file.

// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
// All rights not expressly granted are reserved.
//
// This software is distributed under the terms of the GNU General Public
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
//
// In applying this license CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
#ifndef O2_FRAMEWORK_DATAPROCESSINGSTATS_H_
#define O2_FRAMEWORK_DATAPROCESSINGSTATS_H_
 
#include "DeviceState.h"
#include "Framework/ServiceSpec.h"
#include <atomic>
#include <cstdint>
#include <array>
#include <memory>
#include <numeric>
#include <mutex>
#include <utility>
 
namespace o2::framework
{
 
enum struct ProcessingStatsId : short {
  ERROR_COUNT,
  EXCEPTION_COUNT,
  PENDING_INPUTS,
  INCOMPLETE_INPUTS,
  TOTAL_INPUTS,
  LAST_ELAPSED_TIME_MS,
  TOTAL_WALL_TIME_MS,
  LAST_PROCESSED_SIZE,
  TOTAL_PROCESSED_SIZE,
  TOTAL_SIGUSR1,
  CONSUMED_TIMEFRAMES,
  AVAILABLE_MANAGED_SHM,
  LAST_SLOW_METRIC_SENT_TIMESTAMP,
  LAST_VERY_SLOW_METRIC_SENT_TIMESTAMP,
  LAST_METRIC_FLUSHED_TIMESTAMP,
  BEGIN_ITERATION_TIMESTAMP,
  PERFORMED_COMPUTATIONS,
  LAST_REPORTED_PERFORMED_COMPUTATIONS,
  TOTAL_BYTES_IN,
  TOTAL_BYTES_OUT,
  LAST_MIN_LATENCY,
  LAST_MAX_LATENCY,
  TOTAL_RATE_IN_MB_S,
  TOTAL_RATE_OUT_MB_S,
  PROCESSING_RATE_HZ,
  MALFORMED_INPUTS,
  DROPPED_COMPUTATIONS,
  DROPPED_INCOMING_MESSAGES,
  RELAYED_MESSAGES,
  CPU_USAGE_FRACTION,
  ARROW_BYTES_CREATED,
  ARROW_BYTES_DESTROYED,
  ARROW_BYTES_EXPIRED,
  ARROW_MESSAGES_CREATED,
  ARROW_MESSAGES_DESTROYED,
  TIMESLICE_OFFER_NUMBER_CONSUMED,
  TIMESLICE_NUMBER_STARTED,
  TIMESLICE_NUMBER_EXPIRED,
  TIMESLICE_NUMBER_DONE,
  RESOURCE_OFFER_EXPIRED,
  SHM_OFFER_BYTES_CONSUMED,
  RESOURCES_MISSING,
  RESOURCES_INSUFFICIENT,
  RESOURCES_SATISFACTORY,
  CCDB_CACHE_HIT,
  CCDB_CACHE_MISS,
  CCDB_CACHE_FAILURE,
  CCDB_CACHE_FETCHED_BYTES,
  CCDB_CACHE_REQUESTED_BYTES,
  AOD_INVALID_READ_SKIPPED_TIMEFRAMES,
  AVAILABLE_MANAGED_SHM_BASE = 512,
};
 
struct DataProcessingStats {
  // Parameters for the default behaviour
  struct DefaultConfig {
    int64_t minOnlinePublishInterval = 0;
  };
 
  DefaultConfig config = {};
 
  DataProcessingStats(std::function<void(int64_t& base, int64_t& offset)> getRealtimeBase,
                      std::function<int64_t(int64_t base, int64_t offset)> getTimestamp,
                      DefaultConfig config);
 
  constexpr static ServiceKind service_kind = ServiceKind::Global;
  constexpr static unsigned short MAX_METRICS = 1 << 15;
  constexpr static short MAX_CMDS = 64;
 
  enum struct Op : char {
    Nop,               
    Set,               
    SetIfPositive,     
    CumulativeRate,    
    InstantaneousRate, 
    Add,               
    Sub,               
    Max,               
    Min                
  };
 
  // Kind of the metric. This is used to know how to interpret the value
  enum struct Kind : char {
    Int,
    UInt64,
    Double,
    Rate, 
    Unknown,
  };
 
  // The scope for a given metric. DPL is used for the DPL Monitoring GUI,
  // Online is used for the online monitoring.
  enum struct Scope : char {
    DPL,
    Online
  };
 
  // This is what the user passes. Notice that there is no
  // need to specify the timestamp, because we calculate it for them
  // using the delta between the last update and the current time.
  struct CommandSpec {
    unsigned short id = 0;
    Op op = Op::Nop;
    int64_t value = 0;
  };
 
  // This is the structure to keep track of local updates to the stats.
  // Each command will be queued in a buffer and then flushed to the
  // global stats either when the buffer is full (after MAX_CMDS commands)
  // or when the queue is flushed explicitly via the processQueue() method.
  struct Command {
    unsigned short id = 0; // StatsId of the metric to update
    int64_t value = 0;     // Value to update the metric with
    int64_t timestamp = 0; // Timestamp of the update
    Op op = Op::Nop;       // Operation to perform to do the update
  };
 
  // This structure is used to keep track of the last updates
  // for each of the metrics. This can be used to defer the need
  // to flush the buffers to the remote end, so that we do not need to
  // send metrics synchronously but we can do e.g. as a batch update.
  // It also prevents that we send the same metric multiple times, because
  // we keep track of the time of the last update.
  struct UpdateInfo {
    int64_t timestamp = 0;     // When the update actually took place
    int64_t lastPublished = 0; // When the update was last published
  };
 
  struct MetricSpec {
    // Id of the metric. It must match the index in the metrics array.
    // Name of the metric
    std::string name = "";
    // Wether or not the metric is enabled
    bool enabled = true;
    int metricId = -1;
    Kind kind = Kind::Int;
    Scope scope = Scope::DPL;
    int64_t defaultValue = 0;
    int64_t minPublishInterval = 0;
    uint64_t maxRefreshLatency = -1;
    bool sendInitialValue = false;
  };
 
  void registerMetric(MetricSpec const& spec);
 
  // Update some stats as specified by the @cmd cmd
  void updateStats(CommandSpec cmd);
 
  char const* findMetricNameById(ProcessingStatsId id) const;
  void processCommandQueue();
 
  void flushChangedMetrics(std::function<void(MetricSpec const&, int64_t, int64_t)> const& callback);
 
  std::atomic<size_t> statesSize = 0;
 
  std::array<Command, MAX_CMDS> cmds = {};
  std::array<int64_t, MAX_METRICS> metrics = {};
  std::array<bool, MAX_METRICS> updated = {};
  std::array<std::string, MAX_METRICS> metricsNames = {};
  std::array<UpdateInfo, MAX_METRICS> updateInfos = {};
  std::array<MetricSpec, MAX_METRICS> metricSpecs = {};
  std::array<int64_t, MAX_METRICS> lastPublishedMetrics = {};
  std::vector<int> availableMetrics;
  // for fast check for AVAILABLE_MANAGED_SHM metric which is only provided for readout-proxy
  bool hasAvailSHMMetric = false;
  // How many commands have been committed to the queue.
  std::atomic<int> insertedCmds = 0;
  // The insertion point for the next command.
  std::atomic<int> nextCmd = 0;
  // How many commands are currently in flight.
  std::atomic<int> pendingCmds = 0;
  int64_t lastFlushedToRemote = 0;
  int64_t lastMetrics = 0;
  // This is the mutex to protect the queue of commands.
  std::mutex mMutex;
 
  // Function to retrieve an aritrary base for the realtime clock.
  std::function<void(int64_t& base, int64_t& offset)> getRealtimeBase;
  // Function to retrieve the timestamp from the value returned by getRealtimeBase.
  std::function<int64_t(int64_t base, int64_t offset)> getTimestamp;
  // The value of the uv_hrtime() at the last update.
  int64_t realTimeBase = 0;
  // The value of the uv_now() at the last update.
  int64_t initialTimeOffset = 0;
 
  // Invoke to make sure that the updatedMetricsTotal is updated.
  void lapseTelemetry()
  {
    updatedMetricsTotal += updatedMetricsLapse.load();
    pushedMetricsTotal += pushedMetricsLapse;
    publishedMetricsTotal += publishedMetricsLapse;
    updatedMetricsLapse = 0;
    pushedMetricsLapse = 0;
    publishedMetricsLapse = 0;
  }
 
  // Telemetry for the metric updates and pushes
  std::atomic<int64_t> updatedMetricsLapse = 0;
  int64_t updatedMetricsTotal = 0;
  int64_t pushedMetricsTotal = 0;
  int64_t pushedMetricsLapse = 0;
  int64_t publishedMetricsTotal = 0;
  int64_t publishedMetricsLapse = 0;
  int64_t publishingInvokedTotal = 0;
  int64_t publishingDoneTotal = 0;
};
 
} // namespace o2::framework
 
#endif // O2_FRAMEWORK_DATAPROCESSINGSTATS_H_