DataInspector.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 "DataInspector.h"
#include "Framework/DataProcessorSpec.h"
#include "Framework/DeviceSpec.h"
#include "Framework/OutputObjHeader.h"
#include "Framework/RawDeviceService.h"
 
#include <algorithm>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <cstdint>
#include <fcntl.h>
#include <iomanip>
#include <ios>
#include <iostream>
#include <iterator>
#include <memory>
#include <sstream>
#include <stdexcept>
#include <string>
#include <unistd.h>
#include <utility>
 
#include <rapidjson/document.h>
#include <rapidjson/prettywriter.h>
#include "boost/asio.hpp"
#include <TBufferJSON.h>
#include <boost/algorithm/string/join.hpp>
#include <arrow/table.h>
#include "Framework/TableConsumer.h"
#include "boost/archive/iterators/base64_from_binary.hpp"
#include "boost/archive/iterators/transform_width.hpp"
#include "boost/predef/other/endian.h"
 
using namespace rapidjson;
 
namespace o2::framework::data_inspector
{
#if BOOST_ENDIAN_BIG_BYTE
static const auto endianness = "BIG";
#elif BOOST_ENDIAN_LITTLE_BYTE
static const auto endianness = "LITTLE";
#else
static const auto endianness = "UNKNOWN";
#endif
 
inline size_t base64PaddingSize(uint64_t dataSize)
{
  return (3 - dataSize % 3) % 3;
}
 
std::string encode64(const char* data, uint64_t size)
{
  auto* begin = data;
  auto* end = data + size;
 
  using namespace boost::archive::iterators;
  using EncodingIt = base64_from_binary<transform_width<const char*, 6, 8>>;
  return std::string(EncodingIt(begin), EncodingIt(end)).append(base64PaddingSize(size), '=');
}
 
void addPayload(Document& message,
                uint64_t payloadSize,
                const char* payload,
                Document::AllocatorType& alloc)
{
  message.AddMember("payload", Value(encode64(payload, payloadSize).c_str(), alloc), alloc);
  message.AddMember("payloadEndianness", Value(endianness, alloc), alloc);
}
 
void addBasicDataHeaderInfo(Document& message, const header::DataHeader* header, Document::AllocatorType& alloc)
{
  std::string origin = header->dataOrigin.as<std::string>();
  std::string description = header->dataDescription.as<std::string>();
  std::string method = header->payloadSerializationMethod.as<std::string>();
 
  message.AddMember("origin", Value(origin.c_str(), alloc), alloc);
  message.AddMember("description", Value(description.c_str(), alloc), alloc);
  message.AddMember("subSpecification", Value(header->subSpecification), alloc);
  message.AddMember("firstTForbit", Value(header->firstTForbit), alloc);
  message.AddMember("tfCounter", Value(header->tfCounter), alloc);
  message.AddMember("runNumber", Value(header->runNumber), alloc);
  message.AddMember("payloadSize", Value(header->payloadSize), alloc);
  message.AddMember("splitPayloadParts", Value(header->splitPayloadParts), alloc);
  message.AddMember("payloadSerialization", Value(method.c_str(), alloc), alloc);
  message.AddMember("payloadSplitIndex", Value(header->splitPayloadIndex), alloc);
}
 
void addBasicDataProcessingHeaderInfo(Document& message, const DataProcessingHeader* header, Document::AllocatorType& alloc)
{
  message.AddMember("startTime", Value(header->startTime), alloc);
  message.AddMember("duration", Value(header->duration), alloc);
  message.AddMember("creationTimer", Value(header->creation), alloc);
}
 
void addBasicOutputObjHeaderInfo(Document& message, const OutputObjHeader* header, Document::AllocatorType& alloc)
{
  message.AddMember("taskHash", Value(header->mTaskHash), alloc);
}
 
void buildDocument(Document& message, std::string sender, const DataRef& ref)
{
  message.SetObject();
  Document::AllocatorType& alloc = message.GetAllocator();
  message.AddMember("sender", Value(sender.c_str(), alloc), alloc);
 
  const header::BaseHeader* baseHeader = header::BaseHeader::get(reinterpret_cast<const std::byte*>(ref.header));
  for (; baseHeader != nullptr; baseHeader = baseHeader->next()) {
    if (baseHeader->description == header::DataHeader::sHeaderType) {
      const auto* header = header::get<header::DataHeader*>(baseHeader->data());
      addBasicDataHeaderInfo(message, header, alloc);
      addPayload(message, header->payloadSize, ref.payload, alloc);
    } else if (baseHeader->description == DataProcessingHeader::sHeaderType) {
      const auto* header = header::get<DataProcessingHeader*>(baseHeader->data());
      addBasicDataProcessingHeaderInfo(message, header, alloc);
    } else if (baseHeader->description == OutputObjHeader::sHeaderType) {
      const auto* header = header::get<OutputObjHeader*>(baseHeader->data());
      addBasicOutputObjHeaderInfo(message, header, alloc);
    }
  }
}
 
/* Callback which transforms each `DataRef` to a JSON object*/
std::vector<DIMessage> serializeO2Messages(const std::vector<DataRef>& refs, const std::string& deviceName)
{
  std::vector<DIMessage> messages{};
 
  for (auto& ref : refs) {
    Document message;
    StringBuffer buffer;
    Writer<StringBuffer> writer(buffer);
    buildDocument(message, deviceName, ref);
    message.Accept(writer);
 
    messages.emplace_back(DIMessage{DIMessage::Header::Type::DATA, std::string{buffer.GetString(), buffer.GetSize()}});
  }
 
  return messages;
}
} // namespace o2::framework::data_inspector