DataProcessor.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 "Framework/DataProcessor.h"
#include "Framework/DataSender.h"
#include "Framework/DataProcessingStats.h"
#include "Framework/MessageContext.h"
#include "Framework/StringContext.h"
#include "Framework/ArrowContext.h"
#include "Framework/TMessageSerializer.h"
#include "Framework/ServiceRegistry.h"
#include "Framework/FairMQResizableBuffer.h"
#include "Framework/FairMQDeviceProxy.h"
#include "Framework/DeviceState.h"
#include "Headers/DataHeader.h"
#include "Headers/DataHeaderHelpers.h"
 
#include <Monitoring/Monitoring.h>
#include <fairmq/Parts.h>
#include <fairmq/Device.h>
#include <arrow/io/memory.h>
#include <arrow/ipc/writer.h>
#include <cstddef>
 
using namespace o2::framework;
using DataHeader = o2::header::DataHeader;
 
namespace o2::framework
{
 
void DataProcessor::doSend(DataSender& sender, MessageContext& context, ServiceRegistryRef services)
{
  auto& proxy = services.get<FairMQDeviceProxy>();
  std::vector<fair::mq::Parts> outputsPerChannel;
  outputsPerChannel.resize(proxy.getNumOutputChannels());
  auto contextMessages = context.getMessagesForSending();
  for (auto& message : contextMessages) {
    //     monitoringService.send({ message->parts.Size(), "outputs/total" });
    fair::mq::Parts parts = message->finalize();
    assert(message->empty());
    assert(parts.Size() == 2);
    for (auto& part : parts) {
      outputsPerChannel[proxy.getOutputChannelIndex((message->route())).value].AddPart(std::move(part));
    }
  }
  for (int ci = 0; ci < outputsPerChannel.size(); ++ci) {
    auto& parts = outputsPerChannel[ci];
    if (parts.Size() == 0) {
      continue;
    }
    sender.send(parts, {ci});
  }
}
 
void DataProcessor::doSend(DataSender& sender, StringContext& context, ServiceRegistryRef services)
{
  auto& proxy = services.get<FairMQDeviceProxy>();
  for (auto& messageRef : context) {
    fair::mq::Parts parts;
    fair::mq::MessagePtr payload(sender.create(messageRef.routeIndex));
    auto a = messageRef.payload.get();
    // Rebuild the message using the string as input. For now it involves a copy.
    payload->Rebuild(reinterpret_cast<void*>(const_cast<char*>(strdup(a->data()))), a->size(), nullptr, nullptr);
    const DataHeader* cdh = o2::header::get<DataHeader*>(messageRef.header->GetData());
    // sigh... See if we can avoid having it const by not
    // exposing it to the user in the first place.
    auto* dh = const_cast<DataHeader*>(cdh);
    dh->payloadSize = payload->GetSize();
    parts.AddPart(std::move(messageRef.header));
    parts.AddPart(std::move(payload));
    sender.send(parts, proxy.getOutputChannelIndex(messageRef.routeIndex));
  }
}
 
void DataProcessor::doSend(DataSender& sender, ArrowContext& context, ServiceRegistryRef registry)
{
  using o2::monitoring::Metric;
  using o2::monitoring::Monitoring;
  using o2::monitoring::tags::Key;
  using o2::monitoring::tags::Value;
  auto& monitoring = registry.get<Monitoring>();
  auto& stats = registry.get<DataProcessingStats>();
 
  static const std::regex invalid_metric(" ");
  auto& proxy = registry.get<FairMQDeviceProxy>();
  for (auto& messageRef : context) {
    fair::mq::Parts parts;
    // Depending on how the arrow table is constructed, we finalize
    // the writing here.
    messageRef.finalize(messageRef.buffer);
 
    std::unique_ptr<fair::mq::Message> payload = messageRef.buffer->Finalise();
    // FIXME: for the moment we simply send empty bodies.
    const DataHeader* cdh = o2::header::get<DataHeader*>(messageRef.header->GetData());
    // sigh... See if we can avoid having it const by not
    // exposing it to the user in the first place.
    auto* dh = const_cast<DataHeader*>(cdh);
    dh->payloadSize = payload->GetSize();
    dh->serialization = o2::header::gSerializationMethodArrow;
 
    auto origin = std::regex_replace(dh->dataOrigin.as<std::string>(), invalid_metric, "_");
    auto description = std::regex_replace(dh->dataDescription.as<std::string>(), invalid_metric, "_");
    uint64_t version = dh->subSpecification;
    monitoring.send(Metric{(uint64_t)payload->GetSize(),
                           fmt::format("table-bytes-{}-{}-{}-created",
                                       origin,
                                       description,
                                       version)}
                      .addTag(Key::Subsystem, Value::DPL));
    LOGP(detail, "Creating {}MB for table {}/{}/{}.", payload->GetSize() / 1000000., dh->dataOrigin, dh->dataDescription, version);
    context.updateBytesSent(payload->GetSize());
    context.updateMessagesSent(1);
    parts.AddPart(std::move(messageRef.header));
    parts.AddPart(std::move(payload));
    sender.send(parts, proxy.getOutputChannelIndex(messageRef.routeIndex));
  }
  static int64_t previousBytesSent = 0;
  auto disposeResources = [bs = context.bytesSent() - previousBytesSent](int taskId,
                                                                         std::array<ComputingQuotaOffer, 32>& offers,
                                                                         ComputingQuotaStats& stats,
                                                                         std::function<void(ComputingQuotaOffer const&, ComputingQuotaStats&)> accountDisposed) {
    ComputingQuotaOffer disposed;
    disposed.sharedMemory = 0;
    int64_t bytesSent = bs;
    for (auto& offer : offers) {
      if (offer.user != taskId) {
        continue;
      }
      int64_t toRemove = std::min((int64_t)bytesSent, offer.sharedMemory);
      offer.sharedMemory -= toRemove;
      bytesSent -= toRemove;
      disposed.sharedMemory += toRemove;
      if (bytesSent <= 0) {
        break;
      }
    }
    return accountDisposed(disposed, stats);
  };
  registry.get<DeviceState>().offerConsumers.emplace_back(disposeResources);
  previousBytesSent = context.bytesSent();
  stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_BYTES_CREATED), DataProcessingStats::Op::Set, static_cast<int64_t>(context.bytesSent())});
  stats.updateStats({static_cast<short>(ProcessingStatsId::ARROW_MESSAGES_CREATED), DataProcessingStats::Op::Set, static_cast<int64_t>(context.messagesCreated())});
  stats.processCommandQueue();
}
 
} // namespace o2::framework