AODJAlienReaderHelpers.cxx

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.
 
#include "AODJAlienReaderHelpers.h"
#include <algorithm>
#include <charconv>
#include <cctype>
#include <cstdlib>
#include <exception>
#include <memory>
#include <ranges>
#include <string>
#include <string_view>
#include <vector>
#include "Framework/TableTreeHelpers.h"
#include "Framework/AnalysisHelpers.h"
#include "Framework/DataProcessingStats.h"
#include "Framework/RootArrowFilesystem.h"
#include "Framework/AlgorithmSpec.h"
#include "Framework/ArrowContext.h"
#include "Framework/ConfigParamRegistry.h"
#include "Framework/ControlService.h"
#include "Framework/CallbackService.h"
#include "Framework/EndOfStreamContext.h"
#include "Framework/DeviceSpec.h"
#include "Framework/RawDeviceService.h"
#include "Framework/RuntimeError.h"
#include "Framework/DataSpecUtils.h"
#include "Framework/MessageContext.h"
#include "Framework/Signpost.h"
#include "Framework/StringContext.h"
#include "Framework/ConfigContext.h"
#include "DataInputDirector.h"
#include "Framework/SourceInfoHeader.h"
#include "Framework/ChannelInfo.h"
#include "Framework/Logger.h"
 
#if __has_include(<TJAlienFile.h>)
#include <TJAlienFile.h>
#endif
#include <TGrid.h>
#include <TFile.h>
#include <TTreeCache.h>
#include <TSystem.h>
 
#include <arrow/ipc/reader.h>
#include <arrow/ipc/writer.h>
#include <arrow/io/interfaces.h>
#include <arrow/table.h>
#include <arrow/util/key_value_metadata.h>
#include <arrow/dataset/dataset.h>
#include <arrow/dataset/file_base.h>
 
using namespace o2;
using namespace o2::aod;
 
O2_DECLARE_DYNAMIC_LOG(aod_reader);
 
struct RuntimeWatchdog {
  int numberTimeFrames;
  uint64_t startTime;
  uint64_t lastTime;
  double runTime;
  uint64_t runTimeLimit;
 
  RuntimeWatchdog(Long64_t limit)
  {
    numberTimeFrames = -1;
    startTime = uv_hrtime();
    lastTime = startTime;
    runTime = 0.;
    runTimeLimit = limit;
  }
 
  bool update()
  {
    numberTimeFrames++;
    if (runTimeLimit <= 0) {
      return true;
    }
 
    auto nowTime = uv_hrtime();
 
    // time spent to process the time frame
    double time_spent = numberTimeFrames < 1 ? (double)(nowTime - lastTime) / 1.E9 : 0.;
    runTime += time_spent;
    lastTime = nowTime;
 
    return ((double)(lastTime - startTime) / 1.E9 + runTime / (numberTimeFrames + 1)) < runTimeLimit;
  }
 
  void printOut()
  {
    LOGP(info, "RuntimeWatchdog");
    LOGP(info, "  run time limit: {}", runTimeLimit);
    LOGP(info, "  number of time frames: {}", numberTimeFrames);
    LOGP(info, "  estimated run time per time frame: {}", (numberTimeFrames >= 0) ? runTime / (numberTimeFrames + 1) : 0.);
    LOGP(info, "  estimated total run time: {}", (double)(lastTime - startTime) / 1.E9 + ((numberTimeFrames >= 0) ? runTime / (numberTimeFrames + 1) : 0.));
  }
};
 
using o2::monitoring::Metric;
using o2::monitoring::Monitoring;
using o2::monitoring::tags::Key;
using o2::monitoring::tags::Value;
 
namespace o2::framework::readers
{
static bool shouldSkipInvalidReads()
{
  auto const* envValue = getenv("DPL_AOD_READER_SKIP_INVALID");
  if (envValue == nullptr) {
    return false;
  }
 
  std::string value{envValue};
  std::ranges::transform(value, value.begin(), [](unsigned char c) { return std::tolower(c); });
  return !value.empty() && value != "0" && value != "false";
}
 
static std::string describeException(std::exception const& exception)
{
  std::string description{exception.what()};
  try {
    std::rethrow_if_nested(exception);
  } catch (std::exception const& nested) {
    description += fmt::format(": {}", describeException(nested));
  } catch (RuntimeErrorRef const& ref) {
    description += fmt::format(": {}", error_from_ref(ref).what);
  } catch (...) {
    description += ": unknown exception";
  }
  return description;
}
 
AlgorithmSpec AODJAlienReaderHelpers::rootFileReaderCallback(ConfigContext const& ctx)
{
  // aod-parent-base-path-replacement is now a workflow option, so it needs to be
  // retrieved from the ConfigContext. This is because we do not allow workflow options
  // to change over start-stop-start because they can affect the topology generation.
  std::string parentFileReplacement;
  if (ctx.options().isSet("aod-parent-base-path-replacement")) {
    parentFileReplacement = ctx.options().get<std::string>("aod-parent-base-path-replacement");
  }
  int parentAccessLevel = 0;
  if (ctx.options().isSet("aod-parent-access-level")) {
    parentAccessLevel = ctx.options().get<int>("aod-parent-access-level");
  }
  std::vector<std::pair<std::string, int>> originLevelMapping;
  if (ctx.options().isSet("aod-origin-level-mapping")) {
    auto originLevelMappingStr = ctx.options().get<std::string>("aod-origin-level-mapping");
    for (auto pairRange : originLevelMappingStr | std::views::split(',')) {
      std::string_view pair{pairRange.begin(), pairRange.end()};
      auto colonPos = pair.find(':');
      if (colonPos == std::string_view::npos) {
        LOGP(fatal, "Badly formatted aod-origin-level-mapping entry: \"{}\"", pair);
        continue;
      }
      std::string key(pair.substr(0, colonPos));
      std::string_view valueStr = pair.substr(colonPos + 1);
      int value{};
      auto [ptr, ec] = std::from_chars(valueStr.data(), valueStr.data() + valueStr.size(), value);
      if (ec == std::errc{}) {
        originLevelMapping.emplace_back(std::move(key), value);
      } else {
        LOGP(fatal, "Unable to parse level in aod-origin-level-mapping entry: \"{}\"", pair);
      }
    }
  }
  auto callback = AlgorithmSpec{adaptStateful([parentFileReplacement, parentAccessLevel, originLevelMapping](ConfigParamRegistry const& options,
                                                                                                             DeviceSpec const& spec,
                                                                                                             Monitoring& monitoring,
                                                                                                             DataProcessingStats& stats) {
    // FIXME: not actually needed, since data processing stats can specify that we should
    // send the initial value.
    stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_BYTES_CREATED), DataProcessingStats::Op::Set, 0});
    stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_MESSAGES_CREATED), DataProcessingStats::Op::Set, 0});
    stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_BYTES_DESTROYED), DataProcessingStats::Op::Set, 0});
    stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_MESSAGES_DESTROYED), DataProcessingStats::Op::Set, 0});
    stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_BYTES_EXPIRED), DataProcessingStats::Op::Set, 0});
    stats.updateStats({static_cast<short>(ProcessingStatsId::CONSUMED_TIMEFRAMES), DataProcessingStats::Op::Set, 0});
 
    if (!options.isSet("aod-file-private")) {
      LOGP(fatal, "No input file defined!");
      throw std::runtime_error("Processing is stopped!");
    }
 
    auto filename = options.get<std::string>("aod-file-private");
 
    auto maxRate = options.get<float>("aod-max-io-rate");
 
    // create a DataInputDirector
    auto didir = std::make_shared<DataInputDirector>(std::vector<std::string>{filename}, DataInputDirectorContext{&monitoring, parentAccessLevel, parentFileReplacement, originLevelMapping});
    if (options.isSet("aod-reader-json")) {
      auto jsonFile = options.get<std::string>("aod-reader-json");
      if (!didir->readJson(jsonFile)) {
        LOGP(error, "Check the JSON document! Can not be properly parsed!");
      }
    }
 
    // get the run time watchdog
    auto* watchdog = new RuntimeWatchdog(options.get<int64_t>("time-limit"));
 
    // selected the TFN input and
    // create list of requested tables
    bool reportTFN = false;
    bool reportTFFileName = false;
    header::DataHeader TFNumberHeader;
    header::DataHeader TFFileNameHeader;
    std::vector<OutputRoute> requestedTables;
    std::vector<OutputRoute> routes(spec.outputs);
    for (auto route : routes) {
      if (DataSpecUtils::partialMatch(route.matcher, header::DataOrigin("TFN"))) {
        auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
        TFNumberHeader = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec);
        reportTFN = true;
      } else if (DataSpecUtils::partialMatch(route.matcher, header::DataOrigin("TFF"))) {
        auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
        TFFileNameHeader = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec);
        reportTFFileName = true;
      } else {
        requestedTables.emplace_back(route);
      }
    }
    int level = originLevelMapping.empty() ? -1 : 0;
    auto fileCounter = std::make_shared<int>(0);
    auto numTF = std::make_shared<int>(-1);
    bool const skipInvalidReads = shouldSkipInvalidReads();
    return adaptStateless([TFNumberHeader,
                           TFFileNameHeader,
                           requestedTables,
                           fileCounter,
                           numTF,
                           watchdog,
                           maxRate,
                           skipInvalidReads,
                           didir, reportTFN, reportTFFileName, level](Monitoring& monitoring, DataAllocator& outputs, ControlService& control, DeviceSpec const& device, DataProcessingStats& dpstats, ArrowContext& arrowContext, MessageContext& messageContext, StringContext& stringContext) {
      // Each parallel reader device.inputTimesliceId reads the files fileCounter*device.maxInputTimeslices+device.inputTimesliceId
      // the TF to read is numTF
      assert(device.inputTimesliceId < device.maxInputTimeslices);
      int fcnt = (*fileCounter * device.maxInputTimeslices) + device.inputTimesliceId;
      int ntf = *numTF + 1;
      static int currentFileCounter = -1;
      static int filesProcessed = 0;
      if (currentFileCounter != *fileCounter) {
        currentFileCounter = *fileCounter;
        monitoring.send(Metric{(uint64_t)++filesProcessed, "files-opened"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
      }
 
      // loop over requested tables
      static size_t totalSizeUncompressed = 0;
      static size_t totalSizeCompressed = 0;
      static uint64_t totalDFSent = 0;
      static uint64_t totalInvalidReadSkipped = 0;
      static uint64_t totalFileOpenSkipped = 0;
 
      // check if RuntimeLimit is reached
      if (!watchdog->update()) {
        LOGP(info, "Run time exceeds run time limit of {} seconds. Exiting gracefully...", watchdog->runTimeLimit);
        LOGP(info, "Stopping reader {} after time frame {}.", device.inputTimesliceId, watchdog->numberTimeFrames - 1);
        didir->closeInputFiles();
        monitoring.flushBuffer();
        control.endOfStream();
        control.readyToQuit(QuitRequest::Me);
        return;
      }
 
      int64_t startTime = uv_hrtime();
      int64_t startSize = totalSizeCompressed;
      auto discardTimeframe = [&didir, &arrowContext, &messageContext, &stringContext]() {
        clean_all_runtime_errors();
        didir->finishTimeFrame(true);
        arrowContext.clear();
        messageContext.discard();
        stringContext.clear();
      };
      auto skipInvalidRead = [&fcnt, &ntf, &discardTimeframe, &dpstats, &fileCounter, &numTF, &device](ConcreteDataMatcher const& concrete, AODReadError const& e) {
        auto skippedTimeframes = ++totalInvalidReadSkipped;
        LOGP(error, "Invalid AOD read for table {}: fileCounter {}, timeFrame {}. Skipping timeframe (skipped timeframes: {}). Reason: {}",
             concrete.origin.as<std::string>(), fcnt, ntf, skippedTimeframes, describeException(e));
        discardTimeframe();
        dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_INVALID_READ_SKIPPED_TIMEFRAMES), DataProcessingStats::Op::Add, 1});
        *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices;
        *numTF = ntf;
      };
      enum class TFReaderState {
        READ_FIRST_TABLE,
        READ_FIRST_TABLE_FROM_NEXT_FILE,
        READ_NEXT_TABLE,
        TRY_NEXT_FILE,
        TIMEFRAME_READ,
        INVALID_TIMEFRAME,
        INVALID_FILE,
      };
      auto readState = TFReaderState::READ_FIRST_TABLE;
      [[maybe_unused]] auto stateName = [](TFReaderState state) -> char const* {
        switch (state) {
          case TFReaderState::READ_FIRST_TABLE:
            return "READ_FIRST_TABLE";
          case TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE:
            return "READ_FIRST_TABLE_FROM_NEXT_FILE";
          case TFReaderState::READ_NEXT_TABLE:
            return "READ_NEXT_TABLE";
          case TFReaderState::TRY_NEXT_FILE:
            return "TRY_NEXT_FILE";
          case TFReaderState::TIMEFRAME_READ:
            return "TIMEFRAME_READ";
          case TFReaderState::INVALID_TIMEFRAME:
            return "INVALID_TIMEFRAME";
          case TFReaderState::INVALID_FILE:
            return "INVALID_FILE";
        }
        return "UNKNOWN";
      };
      O2_SIGNPOST_ID_FROM_POINTER(readerStateId, aod_reader, &readState);
      auto transitionTo = [&](TFReaderState nextState) {
        O2_SIGNPOST_EVENT_EMIT(aod_reader, readerStateId, "state transition",
                               "%{public}s -> %{public}s (fileCounter %d, timeFrame %d)",
                               stateName(readState), stateName(nextState), fcnt, ntf);
        readState = nextState;
      };
      size_t routeIndex = 0;
      auto reportTimeframe = [&didir, &fcnt, &ntf, &outputs, &TFNumberHeader, &TFFileNameHeader, reportTFN, reportTFFileName](header::DataHeader const& dh) {
        if (reportTFN) {
          // TF number
          auto timeFrameNumber = didir->getTimeFrameNumber(dh, fcnt, ntf);
          auto o = Output(TFNumberHeader);
          outputs.make<uint64_t>(o) = timeFrameNumber;
        }
 
        if (reportTFFileName) {
          // Origin file name for derived output map
          auto o2 = Output(TFFileNameHeader);
          auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf);
          auto rootFS = std::dynamic_pointer_cast<TFileFileSystem>(fileAndFolder.filesystem());
          auto* f = dynamic_cast<TFile*>(rootFS->GetFile());
          std::string currentFilename(f->GetFile()->GetName());
          if (strcmp(f->GetEndpointUrl()->GetProtocol(), "file") == 0 && f->GetEndpointUrl()->GetFile()[0] != '/') {
            // This is not an absolute local path. Make it absolute.
            static std::string pwd = gSystem->pwd() + std::string("/");
            currentFilename = pwd + std::string(f->GetName());
          }
          outputs.make<std::string>(o2) = currentFilename;
        }
      };
      auto tryReadTable = [&device, &didir, &fcnt, &ntf, &outputs, &reportTimeframe, &requestedTables, &routeIndex, &skipInvalidRead, &discardTimeframe, &dpstats, skipInvalidReads](TFReaderState currentState) -> TFReaderState {
        while (routeIndex < requestedTables.size() &&
               (device.inputTimesliceId % requestedTables[routeIndex].maxTimeslices) != requestedTables[routeIndex].timeslice) {
          ++routeIndex;
        }
        if (routeIndex == requestedTables.size()) {
          return TFReaderState::TIMEFRAME_READ;
        }
 
        auto& route = requestedTables[routeIndex];
        auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
        auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec);
        bool wasAOD = std::ranges::any_of(route.matcher.metadata, [](ConfigParamSpec const& p) { return p.name.starts_with("aod-origin-replaced"); });
 
        if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) {
          didir->beginTimeFrame();
        }
        try {
          if (!didir->readTree(outputs, dh, fcnt, ntf, totalSizeCompressed, totalSizeUncompressed, wasAOD)) {
            return TFReaderState::TRY_NEXT_FILE;
          }
        } catch (AODReadError const& e) {
          if (!skipInvalidReads) {
            throw;
          }
          switch (e.getKind()) {
            case AODReadError::Kind::InvalidTimeframe:
              skipInvalidRead(concrete, e);
              return TFReaderState::INVALID_TIMEFRAME;
            case AODReadError::Kind::FileOpen: {
              auto skippedFiles = ++totalFileOpenSkipped;
              LOGP(error, "Unable to open AOD input for table {}: fileCounter {}, timeFrame {}. Skipping file (skipped files: {}). Reason: {}",
                   concrete.origin.as<std::string>(), fcnt, ntf, skippedFiles, describeException(e));
              discardTimeframe();
              dpstats.updateStats({static_cast<short>(ProcessingStatsId::AOD_FILE_OPEN_SKIPPED_FILES), DataProcessingStats::Op::Add, 1});
              return TFReaderState::INVALID_FILE;
            }
          }
          throw;
        }
 
        if (currentState == TFReaderState::READ_FIRST_TABLE || currentState == TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE) {
          reportTimeframe(dh);
        }
        ++routeIndex;
        return TFReaderState::READ_NEXT_TABLE;
      };
      while (readState != TFReaderState::TIMEFRAME_READ) {
        switch (readState) {
          case TFReaderState::READ_FIRST_TABLE:
            transitionTo(tryReadTable(readState));
            break;
          case TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE:
          case TFReaderState::READ_NEXT_TABLE:
            transitionTo(tryReadTable(readState));
            if (readState == TFReaderState::TRY_NEXT_FILE) {
              // Once a file has been selected, every requested table must exist.
              auto concrete = DataSpecUtils::asConcreteDataMatcher(requestedTables[routeIndex].matcher);
              LOGP(fatal, "Can not retrieve tree for table {}: fileCounter {}, timeFrame {}", concrete.origin.as<std::string>(), fcnt, ntf);
              throw std::runtime_error("Processing is stopped!");
            }
            break;
          case TFReaderState::INVALID_FILE:
            // Undo the discard so outputs read from the next file are checked as usual
            messageContext.clear();
            [[fallthrough]];
          case TFReaderState::TRY_NEXT_FILE:
            didir->finishTimeFrame();
            fcnt += device.maxInputTimeslices;
            if (didir->atEnd(fcnt)) {
              LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId);
              didir->closeInputFiles();
              monitoring.flushBuffer();
              control.endOfStream();
              control.readyToQuit(QuitRequest::Me);
              return;
            }
            ntf = 0;
            routeIndex = 0;
            transitionTo(TFReaderState::READ_FIRST_TABLE_FROM_NEXT_FILE);
            break;
          case TFReaderState::INVALID_TIMEFRAME:
            // Returning would leak this callback's timeslice offer, so read the next timeframe.
            messageContext.clear();
            ++ntf;
            routeIndex = 0;
            transitionTo(TFReaderState::READ_FIRST_TABLE);
            break;
          case TFReaderState::TIMEFRAME_READ:
            break;
        }
      }
      didir->finishTimeFrame();
      int64_t stopSize = totalSizeCompressed;
      int64_t bytesDelta = stopSize - startSize;
      int64_t stopTime = uv_hrtime();
      float currentDelta = float(stopTime - startTime) / 1000000000; // in s
      if (ceil(maxRate) > 0.) {
        float extraTime = (bytesDelta / 1000000 - currentDelta * maxRate) / maxRate;
        // We only sleep if we read faster than the max-read-rate.
        if (extraTime > 0.) {
          LOGP(info, "Read {} MB in {} s. Sleeping for {} seconds to stay within {} MB/s limit.", bytesDelta / 1000000, currentDelta, extraTime, maxRate);
          uv_sleep(extraTime * 1000); // in milliseconds
        }
      }
      totalDFSent++;
 
      // Use the new API for sending TIMESLICE_NUMBER_STARTED
      dpstats.updateStats({(int)ProcessingStatsId::TIMESLICE_NUMBER_STARTED, DataProcessingStats::Op::Add, 1});
      dpstats.processCommandQueue();
      monitoring.send(Metric{(uint64_t)totalDFSent, "df-sent"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
      monitoring.send(Metric{(uint64_t)totalSizeUncompressed / 1000, "aod-bytes-read-uncompressed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
      monitoring.send(Metric{(uint64_t)totalSizeCompressed / 1000, "aod-bytes-read-compressed"}.addTag(Key::Subsystem, monitoring::tags::Value::DPL));
 
      // save file number and time frame
      *fileCounter = (fcnt - device.inputTimesliceId) / device.maxInputTimeslices;
      *numTF = ntf;
 
      // Check if the next timeframe is available or
      // if there are more files to be processed. If not, simply exit.
      ntf = *numTF + 1;
      // first route with level 0 or -1 if no mapping requested
      auto firstRoute = std::ranges::find_if(requestedTables, [&didir, level](auto const& route) {
        auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
        return didir->getLevelForOrigin(concrete.origin) == level;
      });
      auto concrete = DataSpecUtils::asConcreteDataMatcher(firstRoute->matcher);
      auto dh = header::DataHeader(concrete.description, concrete.origin, concrete.subSpec);
      auto fileAndFolder = didir->getFileFolder(dh, fcnt, ntf);
 
      // In case the filesource is empty, move to the next one.
      if (fileAndFolder.filesystem() == nullptr) {
        fcnt += 1;
        ntf = 0;
        if (didir->atEnd(fcnt)) {
          LOGP(info, "No input files left to read for reader {}!", device.inputTimesliceId);
          didir->closeInputFiles();
          monitoring.flushBuffer();
          control.endOfStream();
          control.readyToQuit(QuitRequest::Me);
          return;
        }
      }
    });
  })};
 
  return callback;
}
 
} // namespace o2::framework::readers