<commit_before>/*
 * Distributed under the OSI-approved Apache License, Version 2.0.  See
 * accompanying file Copyright.txt for details.
 *
 * DataManReader.cpp
 *
 *  Created on: Feb 21, 2017
 *      Author: <NAME>
 */

#include "DataManReader.tcc"
#include "adios2/helper/adiosString.h"

namespace adios2
{
namespace core
{
namespace engine
{

DataManReader::DataManReader(IO &io, const std::string &name,
                             const Mode openMode, helper::Comm comm)
: Engine("DataManReader", io, name, openMode, std::move(comm)),
  m_Serializer(m_Comm, helper::IsRowMajor(io.m_HostLanguage)),
  m_RequesterThreadActive(true), m_SubscriberThreadActive(true),
  m_FinalStep(std::numeric_limits<size_t>::max())
{
    m_MpiRank = m_Comm.Rank();
    m_MpiSize = m_Comm.Size();
    helper::GetParameter(m_IO.m_Parameters, "IPAddress", m_IPAddress);
    helper::GetParameter(m_IO.m_Parameters, "Port", m_Port);
    helper::GetParameter(m_IO.m_Parameters, "Timeout", m_Timeout);
    helper::GetParameter(m_IO.m_Parameters, "Verbose", m_Verbosity);
    helper::GetParameter(m_IO.m_Parameters, "Threading", m_Threading);
    helper::GetParameter(m_IO.m_Parameters, "Monitor", m_MonitorActive);
    helper::GetParameter(m_IO.m_Parameters, "MaxStepBufferSize",
                         m_ReceiverBufferSize);

    if (m_IPAddress.empty())
    {
        throw(std::invalid_argument(
            "IP address not specified in wide area staging"));
    }

    std::string requesterAddress =
        "tcp://" + m_IPAddress + ":" + std::to_string(m_Port);
    std::string subscriberAddress =
        "tcp://" + m_IPAddress + ":" + std::to_string(m_Port + 1);
    m_Requester.OpenRequester(requesterAddress, m_Timeout,
                              m_ReceiverBufferSize);

    auto timeBeforeRequest = std::chrono::system_clock::now();
    auto reply = m_Requester.Request("Handshake", 9);
    auto timeAfterRequest = std::chrono::system_clock::now();
    auto roundLatency = std::chrono::duration_cast<std::chrono::milliseconds>(
                            timeAfterRequest - timeBeforeRequest)
                            .count();

    auto startTime = std::chrono::system_clock::now();

    while (reply == nullptr or reply->empty())
    {
        reply = m_Requester.Request("Handshake", 9);
        auto nowTime = std::chrono::system_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::seconds>(
            nowTime - startTime);
        roundLatency = duration.count() * 1000;
        if (duration.count() > m_Timeout)
        {
            m_InitFailed = true;
            return;
        }
    }

    nlohmann::json message = nlohmann::json::parse(reply->data());
    m_TransportMode = message["Transport"];

    if (m_MonitorActive)
    {
        m_Monitor.SetClockError(roundLatency, message["TimeStamp"]);
        m_Monitor.AddTransport(m_TransportMode);
        if (m_Threading)
        {
            m_Monitor.SetReaderThreading();
        }
        bool writerThreading = message["Threading"];
        if (writerThreading)
        {
            m_Monitor.SetWriterThreading();
        }
        m_Monitor.SetRequiredAccuracy(
            stof(message["FloatAccuracy"].get<std::string>()));
    }

    if (m_TransportMode == "fast")
    {
        m_Subscriber.OpenSubscriber(subscriberAddress, m_ReceiverBufferSize);<commit_msg>added checking for json value<commit_after>/*
 * Distributed under the OSI-approved Apache License, Version 2.0.  See
 * accompanying file Copyright.txt for details.
 *
 * DataManReader.cpp
 *
 *  Created on: Feb 21, 2017
 *      Author: <NAME>
 */

#include "DataManReader.tcc"
#include "adios2/helper/adiosString.h"

namespace adios2
{
namespace core
{
namespace engine
{

DataManReader::DataManReader(IO &io, const std::string &name,
                             const Mode openMode, helper::Comm comm)
: Engine("DataManReader", io, name, openMode, std::move(comm)),
  m_Serializer(m_Comm, helper::IsRowMajor(io.m_HostLanguage)),
  m_RequesterThreadActive(true), m_SubscriberThreadActive(true),
  m_FinalStep(std::numeric_limits<size_t>::max())
{
    m_MpiRank = m_Comm.Rank();
    m_MpiSize = m_Comm.Size();
    helper::GetParameter(m_IO.m_Parameters, "IPAddress", m_IPAddress);
    helper::GetParameter(m_IO.m_Parameters, "Port", m_Port);
    helper::GetParameter(m_IO.m_Parameters, "Timeout", m_Timeout);
    helper::GetParameter(m_IO.m_Parameters, "Verbose", m_Verbosity);
    helper::GetParameter(m_IO.m_Parameters, "Threading", m_Threading);
    helper::GetParameter(m_IO.m_Parameters, "Monitor", m_MonitorActive);
    helper::GetParameter(m_IO.m_Parameters, "MaxStepBufferSize",
                         m_ReceiverBufferSize);

    if (m_IPAddress.empty())
    {
        throw(std::invalid_argument(
            "IP address not specified in wide area staging"));
    }

    std::string requesterAddress =
        "tcp://" + m_IPAddress + ":" + std::to_string(m_Port);
    std::string subscriberAddress =
        "tcp://" + m_IPAddress + ":" + std::to_string(m_Port + 1);
    m_Requester.OpenRequester(requesterAddress, m_Timeout,
                              m_ReceiverBufferSize);

    auto timeBeforeRequest = std::chrono::system_clock::now();
    auto reply = m_Requester.Request("Handshake", 9);
    auto timeAfterRequest = std::chrono::system_clock::now();
    auto roundLatency = std::chrono::duration_cast<std::chrono::milliseconds>(
                            timeAfterRequest - timeBeforeRequest)
                            .count();

    auto startTime = std::chrono::system_clock::now();

    while (reply == nullptr or reply->empty())
    {
        reply = m_Requester.Request("Handshake", 9);
        auto nowTime = std::chrono::system_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::seconds>(
            nowTime - startTime);
        roundLatency = duration.count() * 1000;
        if (duration.count() > m_Timeout)
        {
            m_InitFailed = true;
            return;
        }
    }

    nlohmann::json message = nlohmann::json::parse(reply->data());
    m_TransportMode = message["Transport"];

    if (m_MonitorActive)
    {
        m_Monitor.SetClockError(roundLatency, message["TimeStamp"]);
        m_Monitor.AddTransport(m_TransportMode);
        if (m_Threading)
        {
            m_Monitor.SetReaderThreading();
        }
        bool writerThreading = message["Threading"];
        if (writerThreading)
        {
            m_Monitor.SetWriterThreading();
        }
        auto it = message.find("FloatAccuracy");
        if (it != message.end() && !it->get<std::string>().empty())
        {
            m_Monitor.SetRequiredAccuracy(stof(it->get<std::string>()));
        }
    }

    if (m_TransportMode == "fast")
    {
        m_Subscriber.OpenSubscriber(subscriberAddress, m_ReceiverBufferSize);
{'fraction_non_alphanumeric': 0.13159047758142253, 'fraction_numerical': 0.004965678399298963, 'mean_word_length': 3.0761904761904764, 'pattern_counts': {'":': 4, '<': 14, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5208695', 'n_tokens_mistral': 2117, 'n_tokens_neox': 2080, 'n_words': 355}