text:
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
 * Copyright (c) 2011-2015  Regents of the University of California.
 *
 * This file is part of ndnSIM. See AUTHORS for complete list of ndnSIM authors and
 * contributors.
 *
 * ndnSIM is free software: you can redistribute it and/or modify it under the terms
 * of the GNU General Public License as published by the Free Software Foundation,
 * either version 3 of the License, or (at your option) any later version.
 *
 * ndnSIM is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.  See the GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * ndnSIM, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
 **/

// Based on the code by Hajime Tazaki <tazaki@sfc.wide.ad.jp>

#include "annotated-topology-reader.hpp"

#include "ns3/nstime.h"
#include "ns3/log.h"
#include "ns3/assert.h"
#include "ns3/names.h"
#include "ns3/net-device-container.h"
#include "ns3/point-to-point-helper.h"
#include "ns3/point-to-point-net-device.h"
#include "ns3/internet-stack-helper.h"
#include "ns3/ipv4-address-helper.h"
#include "ns3/ipv4-global-routing-helper.h"
#include "ns3/drop-tail-queue.h"
#include "ns3/ipv4-interface.h"
#include "ns3/ipv4.h"
#include "ns3/string.h"
#include "ns3/pointer.h"
#include "ns3/uinteger.h"
#include "ns3/ipv4-address.h"
#include "ns3/error-model.h"
#include "ns3/constant-position-mobility-model.h"
#include "ns3/double.h"

#include "model/ndn-l3-protocol.hpp"

#include <boost/foreach.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/tokenizer.hpp>

#include <boost/graph/adjacency_list.hpp>
#include <boost/graph/graphviz.hpp>

#include <set>

#ifdef NS3_MPI
#include <ns3/mpi-interface.h>
#endif

using namespace std;

namespace ns3 {

NS_LOG_COMPONENT_DEFINE("AnnotatedTopologyReader");

AnnotatedTopologyReader::AnnotatedTopologyReader(const std::string& path, double scale /*=1.0*/)
  : m_path(path)
  , m_randX(CreateObject<UniformRandomVariable>())
  , m_randY(CreateObject<UniformRandomVariable>())
  , m_scale(scale)
  , m_requiredPartitions(1)
{
  NS_LOG_FUNCTION(this);

  m_randX->SetAttribute("Min", DoubleValue(0));
  m_randX->SetAttribute("Max", DoubleValue(100.0));

  m_randY->SetAttribute("Min", DoubleValue(0));
  m_randY->SetAttribute("Max", DoubleValue(100.0));

  SetMobilityModel("ns3::ConstantPositionMobilityModel");
}

void
AnnotatedTopologyReader::SetBoundingBox(double ulx, double uly, double lrx, double lry)
{
  NS_LOG_FUNCTION(this << ulx << uly << lrx << lry);

  m_randX->SetAttribute("Min", DoubleValue(ulx));
  m_randX->SetAttribute("Max", DoubleValue(lrx));

  m_randY->SetAttribute("Min", DoubleValue(uly));
  m_randY->SetAttribute("Max", DoubleValue(lry));
}

void
AnnotatedTopologyReader::SetMobilityModel(const std::string& model)
{
  NS_LOG_FUNCTION(this << model);
  m_mobilityFactory.SetTypeId(model);
}

AnnotatedTopologyReader::~AnnotatedTopologyReader()
{
  NS_LOG_FUNCTION(this);
}

Ptr<Node>
AnnotatedTopologyReader::CreateNode(const std::string name, uint32_t systemId)
{
  NS_LOG_FUNCTION(this << name);
  m_requiredPartitions = std::max(m_requiredPartitions, systemId + 1);

  Ptr<Node> node = CreateObject<Node>(systemId);

  Names::Add(m_path, name, node);
  m_nodes.Add(node);

  return node;
}

Ptr<Node>
AnnotatedTopologyReader::CreateNode(const std::string name, double posX, double posY,
                                    uint32_t systemId)
{
  NS_LOG_FUNCTION(this << name << posX << posY);
  m_requiredPartitions = std::max(m_requiredPartitions, systemId + 1);

  Ptr<Node> node = CreateObject<Node>(systemId);
  Ptr<MobilityModel> loc = DynamicCast<MobilityModel>(m_mobilityFactory.Create());
  node->AggregateObject(loc);

  loc->SetPosition(Vector(posX, posY, 0));

  Names::Add(m_path, name, node);
  m_nodes.Add(node);

  return node;
}

NodeContainer
AnnotatedTopologyReader::GetNodes() const
{
  return m_nodes;
}

const std::list<TopologyReader::Link>&
AnnotatedTopologyReader::GetLinks() const
{
  return m_linksList;
}

NodeContainer
AnnotatedTopologyReader::Read(void)
{
  ifstream topgen;
  topgen.open(GetFileName().c_str());

  if (!topgen.is_open() || !topgen.good()) {
    NS_FATAL_ERROR("Cannot open file " << GetFileName() << " for reading");
    return m_nodes;
  }

  while (!topgen.eof()) {
    string line;
    getline(topgen, line);

    if (line == "router")
      break;
  }

  if (topgen.eof()) {
    NS_FATAL_ERROR("Topology file " << GetFileName() << " does not have \"router\" section");
    return m_nodes;
  }

  while (!topgen.eof()) {
    string line;
    getline(topgen, line);
    if (line[0] == '#')
      continue; // comments
    if (line == "link")
      break; // stop reading nodes

    istringstream lineBuffer(line);
    string name, city;
    double latitude = 0, longitude = 0;
    uint32_t systemId = 0;

    lineBuffer >> name >> city >> latitude >> longitude >> systemId;
    if (name.empty())
      continue;

    Ptr<Node> node;

    if (abs(latitude) > 0.001 && abs(latitude) > 0.001)
      node = CreateNode(name, m_scale * longitude, -m_scale * latitude, systemId);
    else {
      Ptr<UniformRandomVariable> var = CreateObject<UniformRandomVariable>();
      node = CreateNode(name, var->GetValue(0, 200), var->GetValue(0, 200), systemId);
      // node = CreateNode (name, systemId);
    }
  }

  map<string, set<string>> processedLinks; // to eliminate duplications

  if (topgen.eof()) {
    NS_LOG_ERROR("Topology file " << GetFileName() << " does not have \"link\" section");
    return m_nodes;
  }

  // SeekToSection ("link");
  while (!topgen.eof()) {
    string line;
    getline(topgen, line);
    if (line == "")
      continue;
    if (line[0] == '#')
      continue; // comments

    // NS_LOG_DEBUG ("Input: [" << line << "]");

    istringstream lineBuffer(line);
    string from, to, capacity, metric, delay, maxPackets, lossRate;

    lineBuffer >> from >> to >> capacity >> metric >> delay >> maxPackets >> lossRate;

    if (processedLinks[to].size() != 0
        && processedLinks[to].find(from) != processedLinks[to].end()) {
      continue; // duplicated link
    }
    processedLinks[from].insert(to);

    Ptr<Node> fromNode = Names::Find<Node>(m_path, from);
    NS_ASSERT_MSG(fromNode != 0, from << " node not found");
    Ptr<Node> toNode = Names::Find<Node>(m_path, to);
    NS_ASSERT_MSG(toNode != 0, to << " node not found");

    Link link(fromNode, from, toNode, to);

    link.SetAttribute("DataRate", capacity);
    link.SetAttribute("OSPF", metric);

    if (!delay.empty())
      link.SetAttribute("Delay", delay);
    if (!maxPackets.empty())
      link.SetAttribute("MaxPackets", maxPackets);

    // Saran Added lossRate
    if (!lossRate.empty())
      link.SetAttribute("LossRate", lossRate);

    AddLink(link);
    NS_LOG_DEBUG("New link " << from << " <==> " << to << " / " << capacity << " with " << metric
                             << " metric (" << delay << ", " << maxPackets << ", " << lossRate
                             << ")");
  }

  NS_LOG_INFO("Annotated topology created with " << m_nodes.GetN() << " nodes and " << LinksSize()
                                                 << " links");
  topgen.close();

  ApplySettings();

  return m_nodes;
}

void
AnnotatedTopologyReader::AssignIpv4Addresses(Ipv4Address base)
{
  Ipv4AddressHelper address(base, Ipv4Mask("/24"));

  BOOST_FOREACH (const Link& link, m_linksList) {
    address.Assign(NetDeviceContainer(link.GetFromNetDevice(), link.GetToNetDevice()));

    base = Ipv4Address(base.Get() + 256);
    address.SetBase(base, Ipv4Mask("/24"));
  }
}

void
AnnotatedTopologyReader::ApplyOspfMetric()
{
  BOOST_FOREACH (const Link& link, m_linksList) {
    NS_LOG_DEBUG("OSPF: " << link.GetAttribute("OSPF"));
    uint16_t metric = boost::lexical_cast<uint16_t>(link.GetAttribute("OSPF"));

    {
      Ptr<Ipv4> ipv4 = link.GetFromNode()->GetObject<Ipv4>();
      if (ipv4 != 0) {
        int32_t interfaceId = ipv4->GetInterfaceForDevice(link.GetFromNetDevice());
        NS_ASSERT(interfaceId >= 0);

        ipv4->SetMetric(interfaceId, metric);
      }

      Ptr<ndn::L3Protocol> ndn = link.GetFromNode()->GetObject<ndn::L3Protocol>();
      if (ndn != 0) {
        shared_ptr<ndn::Face> face = ndn->getFaceByNetDevice(link.GetFromNetDevice());
        NS_ASSERT(face != 0);

        face->setMetric(metric);
      }
    }

    {
      Ptr<Ipv4> ipv4 = link.GetToNode()->GetObject<Ipv4>();
      if (ipv4 != 0) {
        int32_t interfaceId = ipv4->GetInterfaceForDevice(link.GetToNetDevice());
        NS_ASSERT(interfaceId >= 0);

        ipv4->SetMetric(interfaceId, metric);
      }

      Ptr<ndn::L3Protocol> ndn = link.GetToNode()->GetObject<ndn::L3Protocol>();
      if (ndn != 0) {
        shared_ptr<ndn::Face> face = ndn->getFaceByNetDevice(link.GetToNetDevice());
        NS_ASSERT(face != 0);

        face->setMetric(metric);
      }
    }
  }
}

void
AnnotatedTopologyReader::ApplySettings()
{
#ifdef NS3_MPI
  if (MpiInterface::IsEnabled() && MpiInterface::GetSize() != m_requiredPartitions) {
    std::cerr << "MPI interface is enabled, but number of partitions (" << MpiInterface::GetSize()
              << ") is not equal to number of partitions in the topology (" << m_requiredPartitions
              << ")";
    exit(-1);
  }
#endif

  PointToPointHelper p2p;

  BOOST_FOREACH (Link& link, m_linksList) {
    // cout << "Link: " << Findlink.GetFromNode () << ", " << link.GetToNode () << endl;
    string tmp;

    ////////////////////////////////////////////////
    if (link.GetAttributeFailSafe("MaxPackets", tmp)) {
      NS_LOG_INFO("MaxPackets = " + link.GetAttribute("MaxPackets"));

      try {
        uint32_t maxPackets = boost::lexical_cast<uint32_t>(link.GetAttribute("MaxPackets"));

        // compatibility mode. Only DropTailQueue is supported
        p2p.SetQueue("ns3::DropTailQueue<Packet>", "MaxPackets", UintegerValue(maxPackets));
      }
      catch (...) {
        typedef boost::tokenizer<boost::escaped_list_separator<char>> tokenizer;
        std::string value = link.GetAttribute("MaxPackets");
        tokenizer tok(value);

        tokenizer::iterator token = tok.begin();
        p2p.SetQueue(*token);

        for (token++; token != tok.end(); token++) {
          boost::escaped_list_separator<char> separator('\\', '=', '\"');
          tokenizer attributeTok(*token, separator);

          tokenizer::iterator attributeToken = attributeTok.begin();

          string attribute = *attributeToken;
          attributeToken++;

          if (attributeToken == attributeTok.end()) {
            NS_LOG_ERROR("Queue attribute [" << *token
                                             << "] should be in form <Attribute>=<Value>");
            continue;
          }

          string value = *attributeToken;

          p2p.SetQueueAttribute(attribute, StringValue(value));
        }
      }
    }

    if (link.GetAttributeFailSafe("DataRate", tmp)) {
      NS_LOG_INFO("DataRate = " + link.GetAttribute("DataRate"));
      p2p.SetDeviceAttribute("DataRate", StringValue(link.GetAttribute("DataRate")));
    }

    if (link.GetAttributeFailSafe("Delay", tmp)) {
      NS_LOG_INFO("Delay = " + link.GetAttribute("Delay"));
      p2p.SetChannelAttribute("Delay", StringValue(link.GetAttribute("Delay")));
    }

    NetDeviceContainer nd = p2p.Install(link.GetFromNode(), link.GetToNode());
    link.SetNetDevices(nd.Get(0), nd.Get(1));

    ////////////////////////////////////////////////
    if (link.GetAttributeFailSafe("LossRate", tmp)) {
      NS_LOG_INFO("LinkError = " + link.GetAttribute("LossRate"));

      typedef boost::tokenizer<boost::escaped_list_separator<char>> tokenizer;
      std::string value = link.GetAttribute("LossRate");
      tokenizer tok(value);

      tokenizer::iterator token = tok.begin();
      ObjectFactory factory(*token);

      for (token++; token != tok.end(); token++) {
        boost::escaped_list_separator<char> separator('\\', '=', '\"');
        tokenizer attributeTok(*token, separator);

        tokenizer::iterator attributeToken = attributeTok.begin();

        string attribute = *attributeToken;
        attributeToken++;

        if (attributeToken == attributeTok.end()) {
          NS_LOG_ERROR("ErrorModel attribute [" << *token
                                                << "] should be in form <Attribute>=<Value>");
          continue;
        }

        string value = *attributeToken;

        factory.Set(attribute, StringValue(value));
      }

      nd.Get(0)->SetAttribute("ReceiveErrorModel", PointerValue(factory.Create<ErrorModel>()));
      nd.Get(1)->SetAttribute("ReceiveErrorModel", PointerValue(factory.Create<ErrorModel>()));
    }
  }
}

void
AnnotatedTopologyReader::SaveTopology(const std::string& file)
{
  ofstream os(file.c_str(), ios::trunc);
  os << "# any empty lines and lines starting with '#' symbol is ignored\n"
     << "\n"
     << "# The file should contain exactly two sections: router and link, each starting with the "
        "corresponding keyword\n"
     << "\n"
     << "# router section defines topology nodes and their relative positions (e.g., to use in "
        "visualizer)\n"
     << "router\n"
     << "\n"
     << "# each line in this section represents one router and should have the following data\n"
     << "# node  comment     yPos    xPos\n";

  for (NodeContainer::Iterator node = m_nodes.Begin(); node != m_nodes.End(); node++) {
    std::string name = Names::FindName(*node);
    Ptr<MobilityModel> mobility = (*node)->GetObject<MobilityModel>();
    Vector position = mobility->GetPosition();

    os << name << "\t"
       << "NA"
       << "\t" << -position.y << "\t" << position.x << "\n";
  }

  os
    << "# link section defines point-to-point links between nodes and characteristics of these "
       "links\n"
    << "\n"
    << "link\n"
    << "\n"
    << "# Each line should be in the following format (only first two are required, the rest can "
       "be omitted)\n"
    << "# srcNode   dstNode     bandwidth   metric  delay   queue\n"
    << "# bandwidth: link bandwidth\n"
    << "# metric: routing metric\n"
    << "# delay:  link delay\n"
    << "# queue:  MaxPackets for transmission queue on the link (both directions)\n"
    << "# error:  comma-separated list, specifying class for ErrorModel and necessary attributes\n";

  for (std::list<Link>::const_iterator link = m_linksList.begin(); link != m_linksList.end();
       link++) {
    os << Names::FindName(link->GetFromNode()) << "\t";
    os << Names::FindName(link->GetToNode()) << "\t";

    string tmp;
    if (link->GetAttributeFailSafe("DataRate", tmp))
      os << link->GetAttribute("DataRate") << "\t";
    else
      NS_FATAL_ERROR("DataRate must be specified for the link");

    if (link->GetAttributeFailSafe("OSPF", tmp))
      os << link->GetAttribute("OSPF") << "\t";
    else
      os << "1\t";

    if (link->GetAttributeFailSafe("Delay", tmp)) {
      os << link->GetAttribute("Delay") << "\t";

      if (link->GetAttributeFailSafe("MaxPackets", tmp)) {
        os << link->GetAttribute("MaxPackets") << "\t";

        if (link->GetAttributeFailSafe("LossRate", tmp)) {
          os << link->GetAttribute("LossRate") << "\t";
        }
      }
    }
    os << "\n";
  }
}

/// @cond include_hidden

template<class Names>
class name_writer {
public:
  name_writer(Names _names)
    : names(_names)
  {
  }

  template<class VertexOrEdge>
  void
  operator()(std::ostream& out, const VertexOrEdge& v) const
  {
    // out << "[label=\"" << names[v] << "\",style=filled,fillcolor=\"" << colors[v] << "\"]";
    out << "[shape=\"circle\",width=0.1,label=\"\",style=filled,fillcolor=\"green\"]";
  }

private:
  Names names;
};

template<class Names>
inline name_writer<Names>
make_name_writer(Names n)
{
  return name_writer<Names>(n);
}

/// @endcond

void
AnnotatedTopologyReader::SaveGraphviz(const std::string& file)
{
  typedef boost::adjacency_list_traits<boost::setS, boost::setS, boost::undirectedS> Traits;

  typedef boost::property<boost::vertex_name_t, std::string,
                          boost::property<boost::vertex_index_t, uint32_t>> nodeProperty;

  typedef boost::no_property edgeProperty;

  typedef boost::adjacency_list<boost::setS, boost::setS, boost::undirectedS, nodeProperty,
                                edgeProperty> Graph;

  typedef map<string, Traits::vertex_descriptor> node_map_t;
  node_map_t graphNodes;
  Graph graph;

  for (NodeContainer::Iterator node = m_nodes.Begin(); node != m_nodes.End(); node++) {
    std::pair<node_map_t::iterator, bool> retval = graphNodes.insert(
      make_pair(Names::FindName(*node), add_vertex(nodeProperty(Names::FindName(*node)), graph)));
    // NS_ASSERT (ok == true);

    put(boost::vertex_index, graph, retval.first->second, (*node)->GetId());
  }

  for (std::list<Link>::const_iterator link = m_linksList.begin(); link != m_linksList.end();
       link++) {
    node_map_t::iterator from = graphNodes.find(Names::FindName(link->GetFromNode()));
    node_map_t::iterator to = graphNodes.find(Names::FindName(link->GetToNode()));

    // add_edge (node->second, otherNode->second, m_graph);
    boost::add_edge(from->second, to->second, graph);
  }

  ofstream of(file.c_str());
  boost::property_map<Graph, boost::vertex_name_t>::type names = get(boost::vertex_name, graph);
  write_graphviz(of, graph, make_name_writer(names));
}

}

max_stars_repo_name: NDNLink/NDN-Chord

max_issues_repo_name: NDNLink/NDN-Chord

max_forks_repo_name: NDNLink/NDN-Chord


text:
// -*- mode: c++ -*-

#ifndef LBP_EXAMPLES_RUN_HPP
#define LBP_EXAMPLES_RUN_HPP

#include <iostream>

#include <boost/filesystem.hpp>
namespace fs = boost::filesystem;

#include <options.hpp>

template< typename Function >
void
run_from_stream (Function f, cv::VideoCapture& cap, const options_t& opts) {
    f (cap, opts);
}

template< typename Function >
void
run_from_file_with (Function f, const options_t& opts) {
    const fs::path& filename (opts ["input"].as< std::string > ());
    const fs::path ext = filename.extension ();

    if (ext == ".avi" || ext == ".mp4" || ext == ".mpg" || ext == ".mov") {
        cv::VideoCapture cap;

        if (cap.open (filename.generic_string ())) {
            cap.set (cv::CAP_PROP_FPS, 25);
            run_from_stream (f, cap, opts);
        }
    }
    else
        std::cerr << "unsupported file type" << std::endl;
}

template< typename Function >
void
run_from_camera_with (Function f, const options_t& opts) {
    cv::VideoCapture cap;

    const int stream = std::stoi (opts ["input"].as< std::string > ());

    if (cap.open (stream)) {
        cap.set (cv::CAP_PROP_FPS, 25);
        run_from_stream (f, cap, opts);
    }
}

template< typename Function >
void
run_with (Function f, const options_t& opts) {
    const auto input = opts ["input"].as< std::string > ();

    if (fs::exists (input))
        run_from_file_with (f, opts);
    else
        run_from_camera_with (f, opts);
}

#endif // LBP_EXAMPLES_RUN_HPP

ext: hpp

max_stars_repo_name: thinkoid/lbp

max_issues_repo_name: thinkoid/lbp

max_forks_repo_name: thinkoid/lbp


text:
/************************************************************************
MIT License

Copyright (c) 2021 Deqi Tang

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
************************************************************************/

/// @file src/atomsciflow/server/job_scheduler.cpp
/// @author DeqiTang
/// Mail: deqitang@gmail.com
/// Created Time: Thu 17 Mar 2022 08:22:23 PM CST

#include "atomsciflow/server/job_scheduler.h"

#include <iostream>
#include <fstream>
#include <boost/filesystem.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/format.hpp>

#include "atomsciflow/server/submit_script.h"
#include "atomsciflow/remote/server.h"
#include "atomsciflow/server/utils.h"

namespace atomsciflow {

namespace fs = boost::filesystem;

JobScheduler::JobScheduler() {
    this->set_run_default("bash");
    this->set_run_default("llhpc");
    this->set_run_default("pbs");
    this->set_run_default("bash");
    this->set_run_default("lsf_sz");
    this->set_run_default("lsf_sustc");
    this->set_run_default("cdcloud");
}

template <typename T>
void JobScheduler::set_run(std::string key, T value) {
    this->run_params[key] = boost::lexical_cast<std::string>(value);
}

void JobScheduler::py_set_run(std::string key, std::string value) {
    this->set_run(key, value);
}

void JobScheduler::py_set_run(std::string key, int value) {
    this->set_run(key, value);
}

void JobScheduler::py_set_run(std::string key, float value) {
    this->set_run(key, value);
}

void JobScheduler::py_set_run(std::string key, double value) {
    this->set_run(key, value);
}

/// @param runopt the running options
///  candidates:
///     gen    -> only generate the files for calculation
///     run    -> for previously generated files run them
///     genrun -> generate the needed files and run it
/// @param auto_level
///     0 -> nothing to do
///     1 -> tansferring the files to the server
void JobScheduler::set_run_default(std::string platform) {
    this->set_run("runopt", "genrun");
    this->set_run("auto_level", "0");

    if ("llhpc" == platform) {
        this->set_run_default_llhpc();
    } else if ("yh" == platform) {
        this->set_run_default_yh();
    } else if ("pbs" == platform) {
        this->set_run_default_pbs();
    } else if ("bash" == platform) {
        this->set_run_default_pbs();
    } else if ("lsf_sz" == platform) {
        this->set_run_default_lsf_sz();
    } else if ("lsf_sustc" == platform) {
        this->set_run_default_lsf_sustc();
    } else if ("cdcloud" == platform) {
        this->set_run_default_cdcloud();
    }
}

void JobScheduler::set_run_default_llhpc() {
    this->run_params["partition"] = "free";
    this->run_params["jobname"] = "running";
    this->run_params["nodes"] = std::to_string(1);
    this->run_params["ntask"] = boost::lexical_cast<std::string>(24);
    this->run_params["stdout"] = "slurm.out";
    this->run_params["stderr"] = "slurm.err";
}

void JobScheduler::set_run_default_yh() {
}

void JobScheduler::set_run_default_pbs() {
    this->run_params["nodes"] = boost::lexical_cast<std::string>(1);
    this->run_params["ppn"] = boost::lexical_cast<std::string>(32);
}

void JobScheduler::set_run_default_bash() {
    this->run_params["cmd"] = "$ELK_BIN";
    this->run_params["mpi"] = "";
}

void JobScheduler::set_run_default_lsf_sz() {
    this->run_params["np"] = "24";
    this->run_params["np_per_node"] = "12";
    this->run_params["queue"] = "intelY_mid";
}

void JobScheduler::set_run_default_lsf_sustc() {
    this->run_params["jobname"] = "running-job";
    this->run_params["np" ] = "24";
    this->run_params["np_per_node"] = "12";
    this->run_params["queue"] = "medium";
}

void JobScheduler::set_run_default_cdcloud() {
    this->run_params["partition"] = "normal";
    this->run_params["nodes"] = "1";
    this->run_params["ntask"] = "32";
    this->run_params["jobname"] = "running-job";
    this->run_params["stdout"] = "slurm.out";
    this->run_params["stderr"] = "slurm.err";
}

void JobScheduler::gen_llhpc(const std::string& script = "vasp.sub") {

    std::ofstream outfile;
    outfile.open(script);
    outfile << "";
    outfile.setf(std::ios::fixed);

    outfile << submit_header_llhpc(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_yh(const std::string& script) {
    std::ofstream outfile;
    outfile.open(script);
    outfile.setf(std::ios::fixed);

    outfile << submit_header_yh(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_pbs(const std::string& script) {
    std::ofstream outfile;
    outfile.open(script);

    outfile << submit_header_pbs(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_bash(const std::string& script) {
    std::ofstream outfile;
    outfile.open(script);

    outfile << submit_header_bash(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_lsf_sz(const std::string& script) {
    std::ofstream outfile;
    outfile.open(script);

    outfile << submit_header_lsf_sz(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_lsf_sustc(const std::string& script = "vasp.lsf_sustc") {
    std::ofstream outfile;
    outfile.open(script);
    outfile.setf(std::ios::fixed);

    outfile << submit_header_lsf_sustc(this->run_params);
    outfile << "cd ${ABSOLUTE_WORK_DIR}" << "\n";

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

void JobScheduler::gen_cdcloud(const std::string& script = "vasp.sub") {
    std::ofstream outfile;
    outfile.open(script);
    outfile.setf(std::ios::fixed);

    outfile << submit_header_cdcloud(this->run_params);

    int i = 0;
    for (const auto& step : this->steps) {
        outfile << script_step_header(i);
        outfile << step << "\n";
        i++;
    }

    outfile.close();
}

/**
 * @brief Overall control of the running of the job
 *
 * @param directory
 *
 * run_params["auto_level"]:
 *  0 -> doing nothing;
 *  1 -> generate files only;
 *  2 -> generate and run locally using bash directly;
 *  3 -> generate and submit job to job scheduler locally;
 *  4 -> generate and copy to remote server;
 *  5 -> generate and copy to remote server, then run remotely using bash directly;
 *  6 -> generate and copy to remote server, then submit it to remote server job scheduler;
 */
void JobScheduler::run(const std::string& directory) {
    this->set_run("absolute_work_dir", fs::absolute(fs::path(directory)).string());

    int auto_level = boost::lexical_cast<int>(run_params["auto_level"]);

    // generate files
    if (auto_level >= 1 && auto_level <= 6) {
        if (fs::exists(directory)) {
            fs::remove_all(directory);
        }
        fs::create_directory(directory);
        fs::copy(run_params["xyz_file"], fs::path(directory));

        gen_llhpc((
            fs::path(directory) / (run_params["script_name_head"] + ".slurm")
            ).string()
        );
        gen_bash((
            fs::path(directory) / (run_params["script_name_head"] + ".sh")
            ).string()
        );
        gen_lsf_sz((
            fs::path(directory) / (run_params["script_name_head"] + ".lsf_sz")
            ).string()
        );
        gen_pbs((
            fs::path(directory) / (run_params["script_name_head"] + ".pbs")
            ).string()
        );
        gen_lsf_sustc((
            fs::path(directory) / (run_params["script_name_head"] + ".lsf_sustc")
            ).string()
        );
        gen_cdcloud((
            fs::path(directory) / (run_params["script_name_head"] + ".slurm_cd")
            ).string()
        );
    }
    // run locally using bash directly
    if (2 == auto_level) {
        std::string cmd = "";
        cmd += "bash ";
        cmd += (
            fs::path(directory) /
            (run_params["script_name_head"] + ".sh")
        ).string();
        std::system(cmd.c_str());
    }
    server_handle(
        directory,
        run_params["script_name_head"],
        run_params["server"],
        boost::lexical_cast<int>(run_params["auto_level"])
    );
}

// explicit template instantiation
template void JobScheduler::set_run<std::string>(std::string key, std::string value);
template void JobScheduler::set_run<int>(std::string key, int value);
template void JobScheduler::set_run<float>(std::string key, float value);
template void JobScheduler::set_run<double>(std::string key, double value);

} // namespace atomsciflow

ext: cpp

max_stars_repo_name: DeqiTang/build-test-atomsciflow

max_issues_repo_name: DeqiTang/build-test-atomsciflow

max_forks_repo_name: DeqiTang/build-test-atomsciflow


text:
//This code is under MIT licence, you can find the complete file here: https://github.com/kwiato88/WinApi/blob/master/LICENSE
#include <boost/lexical_cast.hpp>
#include <windowsx.h>
#include "Dialog.hpp"
#include "WinApiLastErrorException.hpp"
#include "MessageDialog.hpp"
#include "StringConversion.hpp"

namespace WinApi
{

Handle Dialog::getFocus()
{
	return Handle(GetFocus());
}

Dialog::Dialog(InstanceHandle p_hInstance, Handle p_parent, ResourceId p_resourceId, const std::string& p_title)
    : m_hInstance(p_hInstance), m_parent(p_parent), m_self(0), m_resourceId(p_resourceId), m_defaultTitle(p_title)
{
}

BOOL CALLBACK Dialog::DialogFun(HWND hwnd, UINT Msg, WPARAM wParam, LPARAM lParam)
{
    Dialog *l_dlg=reinterpret_cast<Dialog*>(GetWindowLongPtr(hwnd, GWLP_USERDATA));
    if (!l_dlg)
    {
        if (Msg == WM_INITDIALOG)
        {
            l_dlg = reinterpret_cast<Dialog*>(lParam);
            l_dlg->m_self = hwnd;
			if (!l_dlg->m_defaultTitle.empty())
				l_dlg->setTitle(l_dlg->m_defaultTitle);
            l_dlg->onInit();
            SetWindowLongPtr(hwnd,GWLP_USERDATA,lParam);
        }
        else
        {
            return 0; //let system deal with message
        }
    }
    //forward message to member function handler
    return l_dlg->handleDialogMsg(Msg, wParam, lParam);
}

BOOL Dialog::handleDialogMsg(UINT p_msgCode, WPARAM p_wParam, LPARAM p_lParam)
{
    if(p_msgCode == WM_CLOSE)
    {
        onClose();
        return TRUE;
    }
    if(p_msgCode == WM_COMMAND)
        return dispatchCommandMsg(p_wParam);
    if(p_msgCode == WM_NOTIFY)
        return dispatchNotifyMsg(p_lParam);
	if (p_msgCode == WM_CONTEXTMENU)
		return handleContextMenuMsg(p_lParam);
    return FALSE;
}

BOOL Dialog::dispatchCommandMsg(WPARAM p_wParam)
{
    for(const auto& handler : m_commandMsgHandlers)
    {
        if(handler.first(LOWORD(p_wParam), HIWORD(p_wParam)))
        {
            handler.second();
            return TRUE;
        }
    }
    return FALSE;
}

int Dialog::adjustPosX(int p_x) const
{
	if (p_x != -1)
		return p_x;
	RECT rec;
	if (GetWindowRect(m_self, &rec) == 0)
		return 0;
	return rec.left;
}

int Dialog::adjustPosY(int p_y) const
{
	if (p_y != -1)
		return p_y;
	RECT rec;
	if (GetWindowRect(m_self, &rec) == 0)
		return 0;
	return rec.top;
}

BOOL Dialog::handleContextMenuMsg(LPARAM p_lParam)
{
	try
	{
		bool isHandled = showContextMenu(adjustPosX(GET_X_LPARAM(p_lParam)), adjustPosY(GET_Y_LPARAM(p_lParam)));
		return isHandled ? TRUE : FALSE;
	}
	catch (Exception& e)
	{
		MessageDialog dlg(m_self);
		dlg.withTitle("Error");
		dlg.withContent(std::string("Filed to execute context menu. ") + e.what());
		dlg.with(MessageDialog::Icon::Error);
		dlg.with(MessageDialog::Buttons::Ok);
		dlg.show();
		return FALSE;
	}
}

void Dialog::registerHandler(CommandMsgMatcher p_matcher, MsgHandler p_handler)
{
    m_commandMsgHandlers.push_back(std::make_pair(p_matcher, p_handler));
}

void Dialog::registerHandler(NotifyMsgMatcher p_matcher, MsgHandler p_handler)
{
    m_notifyMsgHandlers.push_back(std::make_pair(p_matcher, p_handler));
}

BOOL Dialog::dispatchNotifyMsg(LPARAM p_lParam)
{
    for(const auto& handler : m_notifyMsgHandlers)
    {
        if(handler.first(p_lParam))
        {
            handler.second();
            return TRUE;
        }
    }
    return FALSE;
}

int Dialog::show()
{
    auto exitCode = DialogBoxParam(
        m_hInstance,                     // app instance: HISTANCE
        MAKEINTRESOURCE( m_resourceId ), // dialog template id
        m_parent,                        // parent: HWND
        DialogFun,                       // dialog func
        reinterpret_cast<LPARAM>(this)
    );
	if (exitCode == -1 || exitCode == 0)
		throw LastErrorException("Failed to show dialog. Exit code: " + boost::lexical_cast<std::string>(exitCode));
	return exitCode;
}

void Dialog::close(int p_returnCode)
{
    EndDialog(m_self, p_returnCode);
}

void Dialog::redraw()
{
	InvalidateRect(m_self, NULL, true);
	UpdateWindow(m_self);
}

void Dialog::setTitle(const std::string& p_title)
{
	TCHAR title[512] = TEXT("");
	stringToArray(p_title, title);
	SetWindowText(m_self, title);
}

Handle Dialog::getItem(ResourceId p_itemId) const
{
    return Handle(GetDlgItem(m_self, p_itemId));
}

bool Dialog::showContextMenu(int, int)
{
	return false;
}

void Dialog::onClose()
{
    close(RESULT_CANCEL);
}

} // namespace WinApi


text:
/*    Copyright (c) 2010-2015, Delft University of Technology
 *    All rights reserved.
 *
 *    Redistribution and use in source and binary forms, with or without modification, are
 *    permitted provided that the following conditions are met:
 *      - Redistributions of source code must retain the above copyright notice, this list of
 *        conditions and the following disclaimer.
 *      - Redistributions in binary form must reproduce the above copyright notice, this list of
 *        conditions and the following disclaimer in the documentation and/or other materials
 *        provided with the distribution.
 *      - Neither the name of the Delft University of Technology nor the names of its contributors
 *        may be used to endorse or promote products derived from this software without specific
 *        prior written permission.
 *
 *    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
 *    OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 *    MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
 *    COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 *    EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE
 *    GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
 *    AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 *    NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 *    OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 *    Changelog
 *      YYMMDD    Author            Comment
 *      100910    J. Melman         First creation of code.
 *      110111    J. Melman         Adapted to the offical Tudat standards.
 *      110124    J. Melman         Further adapted to the offical Tudat standards.
 *      110201    J. Melman         Made the tests for obliquity and astronomical unit more
 *                                  accurate.
 *      120127    D. Dirkx          Moved to Tudat core.
 *      120127    K. Kumar          Transferred unit tests over to Boost unit test framework.
 *      120128    K. Kumar          Changed BOOST_CHECK to BOOST_CHECK_CLOSE_FRACTION for unit test
 *                                  comparisons.
 *      121205    K. Kumar          Updated license in file header.
 *      150417    D. Dirkx          Added tests for floating ints.
 *
 *    References
 *
 *    Notes
 *
 */

#define BOOST_TEST_MAIN

#include <iostream>
#include <iomanip>
#include <limits>

#include <boost/math/special_functions/fpclassify.hpp>
#include <boost/test/floating_point_comparison.hpp>
#include <boost/test/unit_test.hpp>

#include "Tudat/Mathematics/BasicMathematics/mathematicalConstants.h"

namespace tudat
{
namespace unit_tests
{

// Define Boost test suite.
BOOST_AUTO_TEST_SUITE( test_mathematical_constants )

//! Check correct pi using PI = circumference / diameter:
BOOST_AUTO_TEST_CASE( test_PI )
{
    const double radius = 5.0;

    // Circumference circle with radius 5, 32 digits precision, see
    // http://www.wolframalpha.com/input/?i=Circumference+of+a+circle+with+radius+5
    // http://www.wolframalpha.com/input/?i=N[10*PI,66]
    double circumference = 31.4159265358979323846264338327950288419716939937510582097494459230;
    BOOST_CHECK_CLOSE(  mathematical_constants::PI,
                        circumference / ( 2.0 * radius ) ,
                        std::numeric_limits< double >::epsilon( ) );
}

//! Check correct E using E Wolfram alpha as reference
BOOST_AUTO_TEST_CASE( test_E )
{
    // Numerical value from:
    // http://www.wolframalpha.com/input/?i=e+72+digits
    BOOST_CHECK_CLOSE( mathematical_constants::E,
        2.71828182845904523536028747135266249775724709369995957496696762772407663,
        std::numeric_limits< double >::epsilon( ) );
}

//! Check correct GOLDEN_RATIO using GOLDEN_RATIO Wolfram alpha as reference
BOOST_AUTO_TEST_CASE( test_GOLDEN_RATIO )
{
    // Numerical value from:
    // http://www.wolframalpha.com/input/?i=golden+ratio+72+digits
    BOOST_CHECK_CLOSE(  mathematical_constants::GOLDEN_RATIO,
        1.618033988749894848204586834365638117720309179805762862135448622705260463,
        std::numeric_limits< double >::epsilon( ) );
}

//! Check correct NAN using boost Floating Point Classification (fpclassify)
BOOST_AUTO_TEST_CASE( test_NAN )
{
    // Numerical value from:
    // http://www.wolframalpha.com/input/?i=golden+ratio+72+digits
    BOOST_CHECK( boost::math::isnan( TUDAT_NAN ) );
}

BOOST_AUTO_TEST_CASE( test_TemplatedValues )
{
    double one = 1.0;
    long double longOne = 1.00000000000000000000L;

    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingInteger< double >( 1 ),
                        one, std::numeric_limits< double >::epsilon( ) );
    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingInteger< long double >( 1 ),
                        longOne, std::numeric_limits< long  double >::epsilon( ) );

    BOOST_CHECK_SMALL(  mathematical_constants::getFloatingInteger< double >( 0 ),
                        std::numeric_limits< double >::epsilon( ) );
    BOOST_CHECK_SMALL(  mathematical_constants::getFloatingInteger< long double >( 0 ),
                        std::numeric_limits< long  double >::epsilon( ) );

    double two = 2.0;
    long double longTwo = 2.00000000000000000000L;

    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingInteger< double >( 2 ),
                        two, std::numeric_limits< double >::epsilon( ) );
    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingInteger< long double >( 2 ),
                        longTwo, std::numeric_limits< long  double >::epsilon( ) );

    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingFraction< double >( 1, 2 ),
                        mathematical_constants::getFloatingInteger< double >( 1 ) /
                        mathematical_constants::getFloatingInteger< double >( 2 ),
                        std::numeric_limits< double >::epsilon( ) );
    BOOST_CHECK_CLOSE(  mathematical_constants::getFloatingFraction< long double >( 1, 2 ),
                        mathematical_constants::getFloatingInteger< long double >( 1 ) /
                        mathematical_constants::getFloatingInteger< long double >( 2 )
                        , std::numeric_limits< long double >::epsilon( ) );

    BOOST_CHECK_CLOSE(  mathematical_constants::getPi< double >( ),
                        mathematical_constants::PI, std::numeric_limits< double >::epsilon( ) );
    BOOST_CHECK_CLOSE(  mathematical_constants::getPi< long double >( ),
                        mathematical_constants::LONG_PI, std::numeric_limits< long  double >::epsilon( ) );

}

// Close Boost test suite.
BOOST_AUTO_TEST_SUITE_END( ) // End test_mathematical_constants

} // namespace unit_tests
} // namespace tudat

max_stars_repo_name: JPelamatti/ThesisTUDAT

max_issues_repo_name: JPelamatti/ThesisTUDAT

max_forks_repo_name: JPelamatti/ThesisTUDAT


text:
/*
 * Copyright 2013 Stanford University.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 *
 * - Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the
 *   distribution.
 *
 * - Neither the name of the copyright holders nor the names of
 *   its contributors may be used to endorse or promote products derived
 *   from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
 * THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 */

 /*
  * Client (worker) side interface of the Nimbus scheduler protocol.
  *
  * This interface is a simple message passing interface to the scheduler.
  * It's responsible for transforming commands between object and wire
  * formats. On the reception side, this means instantiating the proper
  * command subclass using a factory.

  * Calls to the network should be non-blocking. In the case of
  * sending, this means we need to have error return values for
  * whether the send was successful. It should also properly handle no
  * messages to receive through non-blocking read calls.
  *
  * Author: Omid Mashayekhi <omidm@stanford.edu>
  * Author: Philip Levis <pal@cs.stanford.edu>
  */

#include <boost/tokenizer.hpp>
#include <string>
#include <sstream>
#include <iostream>  // NOLINT
#include "src/shared/dbg.h"
#include "src/shared/scheduler_client.h"

using boost::asio::ip::tcp;

// Note: you may have to increase the OS limits first.
// Look at the nimbus/scripts/configure_tcp.sh for help.
#ifndef __MACH__
#define CLIENT_TCP_SEND_BUF_SIZE 134217728  // 128MB
#define CLIENT_TCP_RECEIVE_BUF_SIZE 134217728  // 128MB
#endif

#define CLIENT_BUF_SIZE 40960000
#define SEPARATE_COMMAND_TEMPLATE_THREAD true

namespace nimbus {

SchedulerClient::SchedulerClient(std::string scheduler_ip,
                                 port_t scheduler_port)
  : scheduler_ip_(scheduler_ip),
    scheduler_port_(scheduler_port) {
      filling_template_ = false;
      execution_template_active_ = true;
}

SchedulerClient::~SchedulerClient() {
  delete socket_;
  delete read_buffer_;
  delete io_service_;
}

bool SchedulerClient::Initialize() {
  io_service_ = new boost::asio::io_service();
  socket_ = new tcp::socket(*io_service_);
  read_buffer_ = new char[CLIENT_BUF_SIZE];;
  max_read_buffer_length_ = CLIENT_BUF_SIZE;
  existing_bytes_ = 0;
  return true;
}

bool SchedulerClient::ReceiveCommands(SchedulerCommandList* storage,
                                      size_t maxCommands) {
  storage->clear();
  boost::unique_lock<boost::recursive_mutex> lock(*command_processor_mutex_);
  uint32_t pending = received_commands_.size();
  if (pending == 0) {
    return false;
  } else if (pending < maxCommands) {
    maxCommands = pending;
  }
  for (uint32_t i = 0; i < maxCommands; i++) {
    SchedulerCommand* command = received_commands_.front();
    received_commands_.pop_front();
    dbg(DBG_NET, "Copying command %s to user buffer.\n", command->ToString().c_str());
    storage->push_back(command);
  }
  return true;
}

void SchedulerClient::SendCommand(SchedulerCommand* command) {
  dbg(DBG_NET, "Sending command %s .\n", command->ToString().c_str());
  std::string data = command->ToNetworkData();
  SchedulerCommand::length_field_t len;
  len = htonl((uint32_t)data.length() + sizeof(len));
  std::string msg;
  msg.append((const char*)&len, sizeof(len));
  msg.append(data.c_str(), data.length());

  boost::mutex::scoped_lock lock(send_command_mutex_);
  // dbg(DBG_NET, "Sending command %s.\n", command->ToString().c_str());
  boost::system::error_code ignored_error;
  // Why are we IGNORING ERRORS!??!?!?
  boost::asio::write(*socket_, boost::asio::buffer(msg),
                     boost::asio::transfer_all(), ignored_error);
}

void SchedulerClient::SendCommands(SchedulerCommandList* commands) {
  SchedulerCommandList::iterator iter = commands->begin();
  for (; iter != commands->end(); ++iter) {
    SchedulerCommand* command = *iter;
    SendCommand(command);
  }
}

using boost::tokenizer;
using boost::char_separator;

  /** Reads commands from buffer with size bytes. Puts commands into
   * internal command list for later reads. Returns the number of
   * bytes read into commands. If this is less than size, the
   * caller must save the unread bytes. */
size_t SchedulerClient::EnqueueCommands(char* buffer, size_t size) {
  size_t total_read = 0;
  char* read_ptr = buffer;
  size_t bytes_remaining = size;

  dbg(DBG_NET, "Read %i bytes from worker.\n", size);
  size_t header_len = sizeof(SchedulerCommand::length_field_t);

  while (bytes_remaining >= header_len) {
    // Read the header length field
    SchedulerCommand::length_field_t len;
    char* ptr = reinterpret_cast<char*>(&len);
    memcpy(ptr, read_ptr, header_len);
    len = ntohl(len);

    if (bytes_remaining < len) {  // Don't have a complete command
      return total_read;
    } else {
      std::string input(read_ptr + header_len, len - header_len);
      SchedulerCommand* command;
      if (SchedulerCommand::GenerateSchedulerCommandChild(input,
                                                          scheduler_command_table_,
                                                          command)) {
        if (execution_template_active_) {
          dbg(DBG_NET, "Enqueueing command %s.\n", command->ToString().c_str());
          boost::unique_lock<boost::recursive_mutex> lock(*command_processor_mutex_);
          received_commands_.push_back(command);
          command_processor_cond_->notify_all();
        } else {
          switch (command->type()) {
            case SchedulerCommand::START_COMMAND_TEMPLATE:
              HandleStartCommandTemplateCommand(
                  reinterpret_cast<StartCommandTemplateCommand*>(command));
              break;
            case SchedulerCommand::END_COMMAND_TEMPLATE:
              HandleEndCommandTemplateCommand(
                  reinterpret_cast<EndCommandTemplateCommand*>(command));
              break;
            case SchedulerCommand::SPAWN_COMMAND_TEMPLATE:
              HandleSpawnCommandTemplateCommand(
                  reinterpret_cast<SpawnCommandTemplateCommand*>(command));
              break;
            default:
              if (filling_template_) {
                TemplateMap::iterator iter = template_map_.find(template_name_in_progress_);
                assert(iter != template_map_.end());
                CommandTemplate *ct = iter->second;
                switch (command->type()) {
                  case SchedulerCommand::EXECUTE_COMPUTE:
                    ct->AddComputeJobCommand(reinterpret_cast<ComputeJobCommand*>(command));
                    break;
                  case SchedulerCommand::EXECUTE_COMBINE:
                    ct->AddCombineJobCommand(reinterpret_cast<CombineJobCommand*>(command));
                    break;
                  case SchedulerCommand::LOCAL_COPY:
                    ct->AddLocalCopyCommand(reinterpret_cast<LocalCopyCommand*>(command));
                    break;
                  case SchedulerCommand::REMOTE_SEND:
                    ct->AddRemoteCopySendCommand(reinterpret_cast<RemoteCopySendCommand*>(command));
                    break;
                  case SchedulerCommand::REMOTE_RECEIVE:
                    ct->AddRemoteCopyReceiveCommand(reinterpret_cast<RemoteCopyReceiveCommand*>(command)); // NOLINT
                    break;
                  case SchedulerCommand::MEGA_RCR:
                    ct->AddMegaRCRCommand(reinterpret_cast<MegaRCRCommand*>(command)); // NOLINT
                    break;
                  default:
                    assert(false);
                    break;
                }
              }
              dbg(DBG_NET, "Enqueueing command %s.\n", command->ToString().c_str());
              boost::unique_lock<boost::recursive_mutex> lock(*command_processor_mutex_);
              received_commands_.push_back(command);
              command_processor_cond_->notify_all();
              break;
          }
        }
      } else {
        dbg(DBG_NET, "Ignored unknown command: %s.\n", input.c_str());
        assert(false);
      }
      dbg(DBG_NET, "Read %i bytes of %i available.\n", len, bytes_remaining);
      bytes_remaining -= len;
      total_read += len;
      read_ptr += len;
    }
  }

  return total_read;
}

void SchedulerClient::HandleStartCommandTemplateCommand(StartCommandTemplateCommand *cm) {
  std::string key = cm->command_template_name();
  assert(template_map_.find(key) == template_map_.end());
  CommandTemplate *ct = new CommandTemplate(key,
                                            cm->inner_job_ids(),
                                            cm->outer_job_ids(),
                                            cm->phy_ids());
  template_map_[key] = ct;
  filling_template_ = true;
  template_name_in_progress_ = key;
}

void SchedulerClient::HandleEndCommandTemplateCommand(EndCommandTemplateCommand *cm) {
  assert(filling_template_);
  std::string key = cm->command_template_name();
  assert(template_name_in_progress_ == key);
  TemplateMap::iterator iter = template_map_.find(key);
  assert(iter != template_map_.end());
  iter->second->Finalize();
  filling_template_ = false;
}

void SchedulerClient::HandleSpawnCommandTemplateCommand(SpawnCommandTemplateCommand *cm) {
  std::string key = cm->command_template_name();
  TemplateMap::iterator iter = template_map_.find(key);
  assert(iter != template_map_.end());
  if (SEPARATE_COMMAND_TEMPLATE_THREAD) {
    CommandTemplateSeed *cts =
      new CommandTemplateSeed(iter->second,
                              cm->inner_job_ids(),
                              cm->outer_job_ids(),
                              cm->extra_dependency(),
                              cm->parameters(),
                              cm->phy_ids());
    {
      boost::unique_lock<boost::mutex> command_template_seeds_lock(command_template_seeds_mutex_);
      command_template_seeds_.push_back(cts);
      command_template_seeds_cond_.notify_all();
    }
  } else {
  iter->second->Instantiate(cm->inner_job_ids(),
                            cm->outer_job_ids(),
                            cm->extra_dependency(),
                            cm->parameters(),
                            cm->phy_ids(),
                            this);
  }
}

void SchedulerClient::HandleRead(const boost::system::error_code& error,
                                 size_t bytes_transferred) {
  if (error) {
    dbg(DBG_NET|DBG_ERROR,
        "Error %s receiving %i bytes from scheduler.\n",
        error.message().c_str(), bytes_transferred);
    return;
  }

  size_t real_length = bytes_transferred + existing_bytes_;
  size_t len = EnqueueCommands(read_buffer_, real_length);
  assert(real_length >= len);
  size_t remaining = (real_length - len);
  dbg(DBG_NET,
      "Worker received %i bytes from scheduler, enqueued %i bytes as commands, %i remaining.\n",
      bytes_transferred, len, remaining);

  // This is the case when the buffer is full with incomplete message.
  // Then it cannot make progress anymore.
  if (remaining == max_read_buffer_length_) {
    std::string err_msg = "ERROR: ";
    err_msg += "scheduler client buffer is full with incomplete command. ";
    err_msg += "You need to increase the CLIENT_BUF_SIZE in scheduler_client.cc.\n";
    dbg(DBG_ERROR, "%s\n.", err_msg.c_str());
    exit(-1);
  }

  // This is for the case when the string buffer had an incomplete
  // command at its end. Copy the fragement of the command to the beginning
  // of the buffer, mark how many bytes are valid with existing_bytes.
  if (remaining > 0) {
    char* buffer = read_buffer_;
    memmove(buffer, (buffer + len), remaining);
    char* read_start_ptr = buffer + remaining;
    int read_buffer_length = max_read_buffer_length_ - remaining;

    existing_bytes_ = remaining;
    boost::asio::async_read(*socket_,
                            boost::asio::buffer(read_start_ptr,
                                                read_buffer_length),
                            boost::asio::transfer_at_least(1),
                            boost::bind(&SchedulerClient::HandleRead,
                                        this,
                                        boost::asio::placeholders::error,
                                        boost::asio::placeholders::bytes_transferred));
  } else {
    existing_bytes_ = 0;
    boost::asio::async_read(*socket_,
                            boost::asio::buffer(read_buffer_,
                                                max_read_buffer_length_),
                            boost::asio::transfer_at_least(1),
                            boost::bind(&SchedulerClient::HandleRead,
                                        this,
                                        boost::asio::placeholders::error,
                                        boost::asio::placeholders::bytes_transferred));
  }
}

void SchedulerClient::HandleWrite(const boost::system::error_code& error,
                                  size_t bytes_transferred) {}

void SchedulerClient::CreateNewConnections() {
  dbg(DBG_NET, "Attempting to open a connection to server.\n");
  tcp::resolver resolver(*io_service_);
  tcp::resolver::query query(scheduler_ip_,
      boost::to_string(scheduler_port_));
  tcp::resolver::iterator iterator = resolver.resolve(query);

  while (true) {
    boost::system::error_code error = boost::asio::error::host_not_found;
    socket_->connect(*iterator, error);
    if (error) {
      dbg(DBG_ERROR, "Could not connect to server: %s , retrying in 1 second ... \n", error.message().c_str()); // NOLINT
      delete socket_;
      socket_ = new tcp::socket(*io_service_);
      sleep(1);
    } else {
      dbg(DBG_NET, "Successfully connected to the server.\n");
      break;
    }
  }

  // Set the tcp send and receive buf size.
  // Note: you may have to increase the OS limits first.
  // Look at the nimbus/scripts/configure_tcp.sh for help.
#ifndef __MACH__
  boost::asio::socket_base::send_buffer_size s_option(CLIENT_TCP_SEND_BUF_SIZE);
  boost::asio::socket_base::receive_buffer_size r_option(CLIENT_TCP_RECEIVE_BUF_SIZE);
  socket_->set_option(s_option);
  socket_->set_option(r_option);
#endif
  // Turn of Nagle algorithm.
  boost::asio::ip::tcp::no_delay nd_option(TCP_NODELAY_OPTION);
  socket_->set_option(nd_option);
  boost::asio::async_read(*socket_,
                          boost::asio::buffer(read_buffer_,
                                              max_read_buffer_length_),
                          boost::asio::transfer_at_least(1),
                          boost::bind(&SchedulerClient::HandleRead,
                                      this,
                                      boost::asio::placeholders::error,
                                      boost::asio::placeholders::bytes_transferred));
}

void SchedulerClient::Run() {
  dbg(DBG_NET, "Running the scheduler client.\n");
  Initialize();
  if (SEPARATE_COMMAND_TEMPLATE_THREAD) {
    command_template_thread_ =
      new boost::thread(boost::bind(&SchedulerClient::CommandTemplateThread, this));
  }
  CreateNewConnections();
  io_service_->run();
}

void SchedulerClient::CommandTemplateThread() {
  while (true) {
    CommandTemplateSeed *cts;
    {
      boost::unique_lock<boost::mutex> command_template_seeds_lock(command_template_seeds_mutex_);

      while (command_template_seeds_.size() == 0) {
        command_template_seeds_cond_.wait(command_template_seeds_lock);
      }

      CommandTemplateSeedList::iterator iter = command_template_seeds_.begin();
      cts = *iter;
      command_template_seeds_.erase(iter);
    }

    cts->command_template_->Instantiate(cts->inner_job_ids_,
                                        cts->outer_job_ids_,
                                        cts->extra_dependency_,
                                        cts->parameters_,
                                        cts->physical_ids_,
                                        this);
    delete cts;
  }
}

void
SchedulerClient::set_scheduler_command_table(SchedulerCommand::PrototypeTable* cmt) {
  scheduler_command_table_ = cmt;
}

void SchedulerClient::set_command_processor_mutex(boost::recursive_mutex *mutex) {
  command_processor_mutex_ = mutex;
}

void SchedulerClient::set_command_processor_cond(boost::condition_variable_any *cond) {
  command_processor_cond_ = cond;
}

void SchedulerClient::PushCommandToTheQueue(SchedulerCommand *command) {
  boost::unique_lock<boost::recursive_mutex> lock(*command_processor_mutex_);
  received_commands_.push_back(command);
  command_processor_cond_->notify_all();
}

void SchedulerClient::set_execution_template_active(bool flag) {
  execution_template_active_ = flag;
}

}  // namespace nimbus

ext: cc

max_stars_repo_name: schinmayee/nimbus

max_issues_repo_name: schinmayee/nimbus

max_forks_repo_name: schinmayee/nimbus