text:
#define BOOST_TEST_MODULE pcraster model_engine usedefanalyzer
#include <boost/test/unit_test.hpp>
#include <iostream>
#include "calc_astpath.h"
#include "calc_astcfgtester.h"
#include "calc_stringparser.h"
#include "calc_usedefanalyzer.h"
#include "calc_parset.h"
#include "calc_astpar.h"
#include "calc_jumpnode.h"
#include "calc_code.h"
#include "calc_iotype.h"

// NOTE use string failureExpected in files expected to fail, see style guide

namespace calc {
  namespace useDefAnalyzerTest {
   class Test {
     ASTCFGTester                  d_a;
     ParSet                        d_input;

     std::string ass(size_t assNr) const {
       std::ostringstream s;
       s << "C/b/l/"<<assNr<<"/a/";
       return s.str();
     }

     bool checkLastUse(const std::string& path, bool lastUse)const {
       bool ok=astCast<ASTPar>(d_a.ast(),path)->lastUse() == lastUse;
       if (!ok) {
         // make sure we do the correct path
         std::cerr << "lastUse failure path: " <<
                      path << " script pos: " <<
                      astCast<ASTPar>(d_a.ast(),path)->shortPosText();
       }
       return ok;
     }

   public:
     typedef std::set<std::string> NameSet;

     Test(const char *code):
       d_a(StringParser::createCodeAsNode(code))
     {
      setLastUse(d_a.cfg());
      d_input = inputSet(d_a.cfg());
     }
     bool input(const NameSet& input) {
       if (!(d_input == input)) {
         std::cerr << "Input mismatch got :" << d_input << std::endl;
         std::cerr << "          expected :"; printSet(std::cerr,input);
         return false;
       }
       return true;
     }
     //! lhs par in simple assignments
     bool lhs(size_t assNr,bool lastUse) const
     {
       return checkLastUse(ass(assNr)+"<",lastUse);
     }
     //! rhs par in simple assignments
     bool rhs(size_t assNr,bool lastUse, size_t argIndex=0) const
     {
       std::ostringstream arg;
       arg << argIndex;
       return checkLastUse(ass(assNr)+">/e/,/"+arg.str()+"/p",lastUse);
     }
     bool par(const char* path,bool lastUse) const {
       return checkLastUse(path,lastUse);
     }
     template <typename T>
       T* cast(const char *path) const {
         return astCast<T>(d_a.ast(),path);
     }
     JumpNode *jump(const char* path) const {
       return cast<JumpNode>(path);
     }

     static void printSet(std::ostream& str, const NameSet& s) {
       str << "(";
       std::copy(s.begin(),s.end(),std::ostream_iterator<std::string>(str,","));
       str << ")" << std::endl;
     }
  };
 }
}

#define UDA_TEST(C)  useDefAnalyzerTest::Test C

BOOST_AUTO_TEST_CASE(testLinear)
{
  using namespace calc;

  // use lookupscalar: unlimited args and no
  // rewriting in CFG as is the case with min/max/cover
  {
   UDA_TEST(test("a=lookupscalar(b);"));
   BOOST_CHECK(test.lhs(0, true));
   JumpNode *j= test.jump("C/}");
   BOOST_CHECK(j);
   BOOST_CHECK(j->deletesOnForward().empty());
   BOOST_CHECK(test.rhs(0, true));
  }
  {
   UDA_TEST(test("a=lookupscalar(b);a=lookupscalar(b)"));
   BOOST_CHECK(test.lhs(0, true));
   BOOST_CHECK(test.rhs(0, false));

   BOOST_CHECK(test.lhs(1, true));
   BOOST_CHECK(test.rhs(1, true));
  }
  {
   UDA_TEST(test("a=lookupscalar(b,c,b);"));
   BOOST_CHECK(test.lhs(0, true));
   BOOST_CHECK(test.rhs(0, true,0));
   BOOST_CHECK(test.rhs(0, true, 1));
   BOOST_CHECK(test.rhs(0, false, 2));

  }
  {
   UDA_TEST(test("a=lookupscalar(b,c,b);"));
   BOOST_CHECK(test.lhs(0, true));
   BOOST_CHECK(test.rhs(0, true,0));
   BOOST_CHECK(test.rhs(0, true, 1));
   BOOST_CHECK(test.rhs(0, false, 2));
  }
  {
   UDA_TEST(test("a=lookupscalar(b,c,b);a=lookupscalar(b)"));
   BOOST_CHECK(test.lhs(0, true));
   BOOST_CHECK(test.rhs(0, false,0));
   BOOST_CHECK(test.rhs(0, true, 1));
   BOOST_CHECK(test.rhs(0, false,2));

   BOOST_CHECK(test.lhs(0, true));
   BOOST_CHECK(test.rhs(1, true));
  }
  {
   UDA_TEST(test("a=lookupscalar(b,c,b);a=lookupscalar(a)"));
   BOOST_CHECK(test.lhs(0, false));
   BOOST_CHECK(test.rhs(0, true,0));
   BOOST_CHECK(test.rhs(0, true, 1));
   BOOST_CHECK(test.rhs(0, false,2));

   BOOST_CHECK(test.lhs(1, true));
   BOOST_CHECK(test.rhs(1, true));
  }
}

BOOST_AUTO_TEST_CASE(testLoops)
{
  using namespace calc;

  { // use of a AFTER repeat
   const char *code=
        "               # test path to assignments\n" \
        "               # a b                     \n" \
        "a=0;           # 1    l/0/a              \n" \
        "repeat {       #      l/1/r              \n" \
        "b=a+1;         # 2 1  l/1/r/b/0/a        \n" \
        "a=b*2;         # 3 2  l/1/r/b/1/a        \n" \
        "} until (3);   #      l/1/r/c            \n" \
        "b=a+4;         # 4 3  l/2                \n" \
        "               #       b:l/2/a/<         \n" \
        "               #       a:l/2/a/>/,/0/p   \n";
   UDA_TEST(test(code));

   // a 1
   BOOST_CHECK(test.par("C/b/0/a/</p",         false));
   // a 2
   BOOST_CHECK(test.par("C/b/1/b/0/a/>/,/0/p", true));
   // a 3
   BOOST_CHECK(test.par("C/b/1/b/1/a/</p",     false));
   // a 4
   BOOST_CHECK(test.par("C/b/2/a/>/,/0/p",     true));

   // b 1
   BOOST_CHECK(test.par("C/b/1/b/0/a/</p",     false));

   // b 2
   BOOST_CHECK(test.par("C/b/1/b/1/a/>/,/0/p", true));

   // b 3
   BOOST_CHECK(test.par("C/b/2/a/<//p",        true));

   JumpNode *j= test.jump("C/b/1/}");
   BOOST_CHECK(j->deletesOnForward().empty());
  }

  { // no use of a AFTER repeat
   const char *code=
        "               # test path to assignments\n" \
        "               # a b                     \n" \
        "a=0;           # 1    l/0/a              \n" \
        "repeat {       #      l/1/r              \n" \
        "b=a+1;         # 2 1  l/1/r/b/0/a        \n" \
        "a=b*2;         # 3 2  l/1/r/b/1/a        \n" \
        "} until (3);   #      l/1/r/c            \n";
   UDA_TEST(test(code));

   // a 1
   BOOST_CHECK(test.par("C/b/0/a/</p",         false));
   // a 2
   BOOST_CHECK(test.par("C/b/1/b/0/a/>/,/0/p", true));
   // a 3
   BOOST_CHECK(test.par("C/b/1/b/1/a/</p",     false));

   // b 1
   BOOST_CHECK(test.par("C/b/1/b/0/a/</p",     false));

   // b 2
   BOOST_CHECK(test.par("C/b/1/b/1/a/>/,/0/p", true));

   JumpNode *j= test.jump("C/b/1/}");
   // should have only a (not b)
   BOOST_CHECK( j->deletesOnForward().count("a"));
   BOOST_CHECK(!j->deletesOnForward().count("b"));
  }
}

// things that f***ed up later
BOOST_AUTO_TEST_CASE(testBugs)
{
  using namespace calc;

 typedef std::set<std::string> Set;
 {
  const char *code=
  "CF=inp1s.map;\n" \
  "tmp.res=CF+inp5s.map*(1-CF);\n" ;
  UDA_TEST(test(code));
  Set i;
  i.insert("inp1s.map");
  i.insert("inp5s.map");
  BOOST_CHECK(test.input(i));
 }
 {
  // Is until condition properly cleaned
  const char *code=
    "CF=inp1s.map;\n" \
    "repeat { report tmpUseDef.res=CF+4*(2-CF); } until (CF);\n";
  UDA_TEST(test(code));
  BOOST_CHECK(test.par("C/b/1/r/c/p", false));
 }
 {
  const char *code=
  "timer 1 2 1; \n " \
  "initial dynamic \n " \
  "tmpTimeOutput.res=timeoutput(nominal(inp5s.map),time()*10);\n";
  UDA_TEST(test(code));
 }
 {
  const char *code=
  " timer 1 3 1; initial tmp2.res=catchmenttotal(inpldd.map,1); "\
  "              dynamic tmp.res = accuflux(inpldd.map,1); \n";
  UDA_TEST(test(code));
  // should delete when Dynamic is ready
  JumpNode *j= test.jump("C/b/l/1/}");
  BOOST_CHECK(j);
  BOOST_CHECK( j->deletesOnForward().count("inpldd.map"));
  BOOST_CHECK(!j->deletesOnForward().count("tmp.res"));
 }
 {
  const char *code=
  " timer 1 3 1; initial tmp.res=1; dynamic tmp.res = accuflux(inpldd.map,1); \n";
  UDA_TEST(test(code));
  // should delete when DynamicSection is ready
  JumpNode *ds= test.jump("C/b/l/1/}");
  BOOST_CHECK(ds);
  BOOST_CHECK( ds->deletesOnForward().count("inpldd.map"));
  // hence not in Code "exit"
  JumpNode *cs= test.jump("C/}");
  BOOST_CHECK(cs);
  BOOST_CHECK(!cs->deletesOnForward().count("inpldd.map"));
 }
 {
   const char *code=
            "timer 1 3 1; dynamic report tmp.res=time();";
   UDA_TEST(test(code));
   JumpNode *ds=test.jump("C/b/0}");
   BOOST_CHECK(ds);
   BOOST_CHECK(!ds->deletesOnForward().count("tmp.res"));
   JumpNode *j= test.jump("C/b/0}");
   BOOST_CHECK(!j->deletesOnForward().count("tmp.res"));
   ASTPar *p=test.cast<ASTPar>("C/b/0/b/l/0/a/</p");
   BOOST_CHECK(p);
   BOOST_CHECK(p->lastUse());
 }
}

BOOST_AUTO_TEST_CASE(testXMLReflection)
{
  using namespace calc;

  useDefAnalyzerTest::Test::NameSet mz;
  mz.insert("m");
  mz.insert("z");

  {
   // z should be  deleted after dynamic loop
   const char *code= " timer 1 2 1;"
                     " dynamic"
                     " repeat {"
                     "  x= z eq 0;"
                     " }"
                     " until m le 0;";
   UDA_TEST(test(code));
   JumpNode *dynamicJump=test.jump("C/b/0/}");
   BOOST_CHECK(dynamicJump);
   BOOST_CHECK(dynamicJump->deletesOnForward() == mz);
   JumpNode *repeatJump=test.jump("C/b/0/b/l/0/}");
   BOOST_CHECK(repeatJump);
   BOOST_CHECK(repeatJump->deletesOnForward().empty());
  }

  {
     // z should be  deleted after repeat loop
     const char *code= " repeat {"
                       "  x= z eq 0;"
                       " }"
                       " until m le 0;";
     UDA_TEST(test(code));
     JumpNode *jump=test.jump("C/b/0/}");
     BOOST_CHECK(jump);
     BOOST_CHECK(jump->deletesOnForward() == mz);
  }
}

BOOST_AUTO_TEST_CASE(testIOTypes)
{
  using namespace calc;

  {
  ASTCFGTester  test(StringParser::createCodeAsNode("a=0"));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_CHECK(r.count("a"));
  BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Initial);
  }
  {
  ASTCFGTester  test(StringParser::createCodeAsNode("a=b"));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_CHECK(r.count("a"));
  BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Initial);
  BOOST_CHECK(r.count("b"));
  BOOST_CHECK(r["b"].input() ==pcrxml::ModelInputType::Initial);
  BOOST_CHECK(r["b"].output()==pcrxml::ModelOutputType::Fixed);
  }
  {
    ASTCFGTester  test(StringParser::createCodeAsNode(
          "a=0;         \n" \
          "repeat {     \n" \
          "a=b*2;       \n" \
          "} until (1); \n" ));
    std::map<std::string,IOType> r = ioTypes(test.cfg());
    BOOST_CHECK(r.count("a"));
    BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
    BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Initial);
    BOOST_CHECK(r.count("b"));
    BOOST_CHECK(r["b"].input() ==pcrxml::ModelInputType::Initial);
    BOOST_CHECK(r["b"].output()==pcrxml::ModelOutputType::Fixed);
  }
  {
  ASTCFGTester  test(StringParser::createCodeAsNode(
        "initial  dynamic a=b"));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_CHECK(r.count("a"));
  BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Dynamic);
  BOOST_CHECK(r.count("b"));
  BOOST_CHECK(r["b"].input() ==pcrxml::ModelInputType::Dynamic);
  BOOST_CHECK(r["b"].output()==pcrxml::ModelOutputType::Fixed);
  }
  {
  ASTCFGTester  test(StringParser::createCodeAsNode(
        "initial  dynamic a=b; b=sqrt(a);"));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_CHECK(r.count("a"));
  BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Dynamic);
  BOOST_CHECK(r.count("b"));
  BOOST_CHECK(r["b"].input() ==pcrxml::ModelInputType::Initial);
  BOOST_CHECK(r["b"].output()==pcrxml::ModelOutputType::Dynamic);
  }
  {
  ASTCFGTester  test(StringParser::createCodeAsNode(
        "initial b=3;  dynamic a=b;"));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_CHECK(r.count("a"));
  BOOST_CHECK(r["a"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["a"].output()==pcrxml::ModelOutputType::Dynamic);
  BOOST_CHECK(r.count("b"));
  BOOST_CHECK(r["b"].input() ==pcrxml::ModelInputType::None);
  BOOST_CHECK(r["b"].output()==pcrxml::ModelOutputType::Initial);
  }
 {
  //ASTNode *e=StringParser::createExpr("a+a"); WORKS!
  //ASTNode *e=StringParser::createExpr("a"); // NOT
  ASTNode *e=StringParser::createExpr("not a"); // WORKS
  bool todo_ioTypes_expr=false;
  BOOST_WARN(todo_ioTypes_expr);
  ASTCFGTester  test(new Code(e));
  std::map<std::string,IOType> r = ioTypes(test.cfg());
  BOOST_WARN(r.count("a"));
  BOOST_WARN(r["a"].input() ==pcrxml::ModelInputType::Initial);
  BOOST_WARN(r["a"].output()==pcrxml::ModelOutputType::Fixed);
  }
}

ext: cc

max_stars_repo_name: quanpands/wflow

max_issues_repo_name: quanpands/wflow

max_forks_repo_name: quanpands/wflow


text:
// Copyright (c) 2019 Graphcore Ltd. All rights reserved.
#define BOOST_TEST_MODULE Priorities0InplaceTest

#include <boost/test/unit_test.hpp>
#include <filereader.hpp>
#include <vector>
#include <popart/builder.hpp>
#include <popart/dataflow.hpp>
#include <popart/inputshapeinfo.hpp>
#include <popart/ir.hpp>
#include <popart/op/identity.hpp>
#include <popart/op/l1.hpp>
#include <popart/op/nll.hpp>
#include <popart/sgd.hpp>
#include <popart/tensor.hpp>
#include <popart/tensorinfo.hpp>
#include <popart/tensornames.hpp>
#include <popart/tensors.hpp>
#include <popart/testdevice.hpp>

using namespace popart;

BOOST_AUTO_TEST_CASE(NegPriorities_concat0) {

  auto runTest = [](float priorityValue) {
    // Two Ops with their inplacing priorities either
    // set negative (so they should NOT be inplaced)
    // or kept positive (so they SHOULD be inplaced if possible)
    //
    // clang-format off
    //
    //
    //                          priority might be negative!
    // in0 -|                       ^
    //      |                       |
    // in1 -|- [Concat] - x1 --|    |
    //      |                  |    |
    // in2 -|--|               |- [Concat] - x3 - [Relu] - o1 -|
    //         |- [Concat] x2 -|                              /
    //         |                                             /
    //         |                                            /
    // in3 ----|                                          [Concat] ---- o2
    //                                                    |              |
    // in4 ---------- [Relu] -- c3 -----------------------|         [ReduceSum]
    //                   ^                                               |
    //                   |                                              out
    //                   |
    //                 priority might be negative!
    //
    //
    // modification happens iff modifyPriorities is true
    //
    // clang-format on

    // Build an onnx model
    auto builder = Builder::create();
    auto aiOnnx  = builder->aiOnnxOpset9();

    TensorInfo shape0{"FLOAT", std::vector<int64_t>{3}};
    auto in0 = builder->addInputTensor(shape0);
    auto in1 = builder->addInputTensor(shape0);
    auto in2 = builder->addInputTensor(shape0);
    auto in3 = builder->addInputTensor(shape0);

    auto x1  = aiOnnx.concat({in0, in1, in2}, 0);
    auto x2  = aiOnnx.concat({in2, in3}, 0);
    auto x3  = aiOnnx.concat({x1, x2}, 0);
    auto o1  = aiOnnx.relu({x3});
    auto in4 = builder->addInputTensor(shape0);
    auto c3  = aiOnnx.relu({in4});
    auto o2  = aiOnnx.concat({o1, c3}, 0);
    auto out = aiOnnx.reducesum({o2}, std::vector<int64_t>{});
    auto l1  = builder->aiGraphcoreOpset1().l1loss({out}, 0.1);

    builder->setInplacePreferences(x3, {{"ConcatInplace", priorityValue}});
    // TODO if  these 2 priorities are large (>10.0f), this test fails. See task
    // T9423
    builder->setInplacePreferences(o2, {{"ConcatInplace", 9.0f}});
    builder->setInplacePreferences(o1, {{"ReluInplace", 9.5f}});

    builder->setInplacePreferences(c3, {{"ReluInplace", priorityValue}});

    auto proto      = builder->getModelProto();
    auto modelProto = io::getModelFromString(proto);

    // Create the IR
    auto dataFlow  = DataFlow(1, {{out, AnchorReturnType("All")}});
    auto optimizer = ConstSGD(0.01);
    auto device    = createTestDevice(TEST_TARGET);

    Ir ir;
    ir.prepare({modelProto,
                InputShapeInfo(),
                dataFlow,
                l1,
                &optimizer,
                *device,
                {},
                Patterns(PatternsLevel::NoPatterns)
                    .enableRuntimeAsserts(false)
                    .enableInPlace(true)});

    // .... Tests ....
    // irrespective of priority x1, x2, o2 should be inplace concats,
    // and x3 should only be inplace if priority > 0.
    std::vector<TensorId> inplaceConcatIds    = {x1, x2, o2};
    std::vector<TensorId> notInplaceConcatIds = {};
    // o1 should always be inplace, c3 only sometimes.
    std::vector<TensorId> inplaceReluIds    = {o1};
    std::vector<TensorId> notInplaceReluIds = {};

    if (priorityValue > 0) {
      inplaceConcatIds.push_back(x3);
      inplaceReluIds.push_back(c3);
    } else {
      notInplaceConcatIds.push_back(x3);
      notInplaceReluIds.push_back(c3);
    }

    for (TensorId id : inplaceConcatIds) {
      BOOST_CHECK(ir.getMainGraphTensors().get(id)->getProducer()->opid.type ==
                  "ConcatInplace");
    }

    for (TensorId id : inplaceReluIds) {
      BOOST_CHECK(ir.getMainGraphTensors().get(id)->getProducer()->opid.type ==
                  "ReluInplace");
    }

    for (TensorId id : notInplaceConcatIds) {
      BOOST_CHECK(ir.getMainGraphTensors().get(id)->getProducer()->opid.type ==
                  "Concat");
    }

    for (TensorId id : notInplaceReluIds) {
      BOOST_CHECK(ir.getMainGraphTensors().get(id)->getProducer()->opid.type ==
                  "Relu");
    }
  };

  // SHOULD all be inplaced:
  runTest(1e07f);
  runTest(1e-5f);

  // should NOT all be inplaced:
  runTest(-1.0f);
}

ext: cpp


text:
//----------------------------------------------------------------------------
/// @file   scheduler.hpp
/// @brief  This file contains the implementation of the scheduler for
///         dispatch the works stored
///
/// @author Copyright (c) 2010 2015 Francisco José Tapia (fjtapia@gmail.com )\n
///         Distributed under the Boost Software License, Version 1.0.\n
///         ( See accompanyingfile LICENSE_1_0.txt or copy at
///           http://www.boost.org/LICENSE_1_0.txt  )
/// @version 0.1
///
/// @remarks
//-----------------------------------------------------------------------------
#ifndef __BOOST_SORT_COMMON_SCHEDULER_HPP
#define __BOOST_SORT_COMMON_SCHEDULER_HPP

#include <boost/sort/common/spinlock.hpp>
#include <boost/sort/common/search.hpp>
#include <boost/sort/common/compare_traits.hpp>
#include <scoped_allocator>
#include <utility>
#include <vector>
#include <deque>
#include <iostream>
#include <unordered_map>

namespace boost
{
namespace sort
{
namespace common
{

//
//###########################################################################
//                                                                         ##
//    ################################################################     ##
//    #                                                              #     ##
//    #           C L A S S      S C H E D U L E R                   #     ##
//    #                                                              #     ##
//    ################################################################     ##
//                                                                         ##
//###########################################################################

//
//---------------------------------------------------------------------------
/// @class  scheduler
/// @brief This class is a concurrent stack controled by a spin_lock
/// @remarks
//---------------------------------------------------------------------------
template<typename Func_t, typename Allocator = std::allocator<Func_t> >
struct scheduler
{
    //-----------------------------------------------------------------------
    //                     D E F I N I T I O N S
    //-----------------------------------------------------------------------
    typedef std::scoped_allocator_adaptor <Allocator>   scoped_alloc;
    typedef std::deque <Func_t, scoped_alloc>           deque_t;
    typedef typename deque_t::iterator                  it_deque;
    typedef std::thread::id                             key_t;
    typedef std::hash <key_t>                           hash_t;
    typedef std::equal_to <key_t>                       equal_t;
    typedef std::unique_lock <spinlock_t>               lock_t;
    typedef std::unordered_map <key_t, deque_t, hash_t,
                        equal_t, scoped_alloc>          map_t;
    typedef typename map_t::iterator                    it_map;

    //-----------------------------------------------------------------------
    //                     V A R I A B L E S
    //-----------------------------------------------------------------------
    map_t mp;
    size_t nelem;
    mutable spinlock_t spl;

    //------------------------------------------------------------------------
    //  function : scheduler
    /// @brief  constructor
    //------------------------------------------------------------------------
    scheduler(void) : mp(), nelem(0)  { };
    //
    //-----------------------------------------------------------------------
    //  function : scheduler
    /// @brief  Copy & move constructor
    /// @param [in] VT : stack_cnc from where copy the data
    //-----------------------------------------------------------------------
    scheduler(scheduler && VT) = delete;
    scheduler(const scheduler & VT) = delete;
    //
    //------------------------------------------------------------------------
    //  function : ~scheduler
    /// @brief  Destructor
    //------------------------------------------------------------------------
    virtual ~scheduler(void) {mp.clear();};
    //
    //------------------------------------------------------------------------
    //  function : operator =
    /// @brief Asignation operator
    /// @param [in] VT : stack_cnc from where copy the data
    /// @return Reference to the stack_cnc after the copy
    //------------------------------------------------------------------------
    scheduler & operator=(const scheduler &VT) = delete;
    //
    //------------------------------------------------------------------------
    //  function : size
    /// @brief Asignation operator
    /// @param [in] VT : stack_cnc from where copy the data
    /// @return Reference to the stack_cnc after the copy
    //------------------------------------------------------------------------
    size_t size(void) const
    {
        lock_t s(spl);
        return nelem;
    };
    //
    //------------------------------------------------------------------------
    //  function : clear
    /// @brief Delete all the elements of the stack_cnc.
    //------------------------------------------------------------------------
    void clear_all(void)
    {
        lock_t s(spl);
        mp.clear();
        nelem = 0;
    };

    //
    //------------------------------------------------------------------------
    //  function : insert
    /// @brief Insert one element in the back of the container
    /// @param [in] D : value to insert. Can ve a value, a reference or an
    ///                 rvalue
    /// @return iterator to the element inserted
    /// @remarks This operation is O ( const )
    //------------------------------------------------------------------------
    void insert(Func_t & f)
    {
        lock_t s(spl);
        key_t th_id = std::this_thread::get_id();
        it_map itmp = mp.find(th_id);
        if (itmp == mp.end())
        {
            auto aux = mp.emplace(th_id, deque_t());
            if (aux.second == false) throw std::bad_alloc();
            itmp = aux.first;
        };
        itmp->second.emplace_back(std::move(f));
        nelem++;
    };

    //
    //------------------------------------------------------------------------
    //  function :emplace
    /// @brief Insert one element in the back of the container
    /// @param [in] args :group of arguments for to build the object to insert
    /// @return iterator to the element inserted
    /// @remarks This operation is O ( const )
    //------------------------------------------------------------------------
    template<class ... Args>
    void emplace(Args && ... args)
    {
        lock_t s(spl);
        key_t th_id = std::this_thread::get_id();
        it_map itmp = mp.find(th_id);
        if (itmp == mp.end())
        {
            auto aux = mp.emplace(th_id, deque_t());
            if (aux.second == false) throw std::bad_alloc();
            itmp = aux.first;
        };
        itmp->second.emplace_back(std::forward <Args>(args) ...);
        nelem++;
    };
    //
    //------------------------------------------------------------------------
    //  function : insert
    /// @brief Insert one element in the back of the container
    /// @param [in] D : value to insert. Can ve a value, a reference or an rvalue
    /// @return iterator to the element inserted
    /// @remarks This operation is O ( const )
    //------------------------------------------------------------------------
    template<class it_func>
    void insert_range(it_func first, it_func last)
    {
        //--------------------------------------------------------------------
        //                    Metaprogramming
        //--------------------------------------------------------------------
        typedef value_iter<it_func> value2_t;
        static_assert (std::is_same< Func_t, value2_t >::value,
                        "Incompatible iterators\n");

        //--------------------------------------------------------------------
        //                     Code
        //--------------------------------------------------------------------
        assert((last - first) > 0);

        lock_t s(spl);
        key_t th_id = std::this_thread::get_id();
        it_map itmp = mp.find(th_id);
        if (itmp == mp.end())
        {
            auto aux = mp.emplace(th_id, deque_t());
            if (aux.second == true) throw std::bad_alloc();
            itmp = aux.first;
        };
        while (first != last)
        {
            itmp->second.emplace_back(std::move(*(first++)));
            nelem++;
        };
    };
    //
    //------------------------------------------------------------------------
    //  function : extract
    /// @brief erase the last element of the tree and return a copy
    /// @param [out] V : reference to a variable where copy the element
    /// @return code of the operation
    ///         0- Element erased
    ///         1 - Empty tree
    /// @remarks This operation is O(1)
    //------------------------------------------------------------------------
    bool extract(Func_t & f)
    {
        lock_t s(spl);
        if (nelem == 0) return false;
        key_t th_id = std::this_thread::get_id();
        it_map itmp = mp.find(th_id);
        if (itmp != mp.end() and not itmp->second.empty())
        {
            f = std::move(itmp->second.back());
            itmp->second.pop_back();
            --nelem;
            return true;
        };
        for (itmp = mp.begin(); itmp != mp.end(); ++itmp)
        {
            if (itmp->second.empty()) continue;
            f = std::move(itmp->second.back());
            itmp->second.pop_back();
            --nelem;
            return true;
        }
        return false;
    };
};
// end class scheduler
//*************************************************************************
//               P R I N T      F U N C T I O N S
//************************************************************************
template<class ... Args>
std::ostream & operator <<(std::ostream &out, const std::deque<Args ...> & dq)
{
    for (uint32_t i = 0; i < dq.size(); ++i)
        out << dq[i] << " ";
    out << std::endl;
    return out;
}

template<typename Func_t, typename Allocator = std::allocator<Func_t> >
std::ostream & operator <<(std::ostream &out,
                           const scheduler<Func_t, Allocator> &sch)
{
    std::unique_lock < spinlock_t > s(sch.spl);
    out << "Nelem :" << sch.nelem << std::endl;
    for (auto it = sch.mp.begin(); it != sch.mp.end(); ++it)
    {
        out << it->first << "  :" << it->second << std::endl;
    }
    return out;
}

//***************************************************************************
};// end namespace common
};// end namespace sort
};// end namespace boost
//***************************************************************************
#endif

ext: hpp

max_stars_repo_name: kindlychung/mediasoup-sfu-cpp

max_issues_repo_name: kindlychung/mediasoup-sfu-cpp

max_forks_repo_name: kindlychung/mediasoup-sfu-cpp


text:
/*
 * implementation of marathon interface for tensorflow's distributed runtime
 *
 * ct-clmsn
 * Mar 2016
 */

#include "marathonimpl.hpp"

#include <iostream>
#include <string>
#include <thread>
#include <chrono>

#include <unistd.h>
#include <curl/curl.h>

#ifndef CXX15
   #include <boost/lexical_cast.hpp>
   #include <boost/regex.hpp>
   using namespace boost;
#else
   #include <regex>
#endif

using namespace std;

/*
 * curl callback function that copies response from marathon server into the "data" buffer
 */
static int writer(char* data, size_t size, size_t nmemb, string *writerData) {
   if (writerData == NULL) {
      return 0;
   }

   writerData->append(data, size*nmemb);

   return size*nmemb;
}

/*
 * bool return code, did this succeed (true) or fail (false)
 *    marathon_curl_cmd(
 * 	const string marathon_dns_url_str = base url to the marathon server,
 * 	string& marathon_task_json = response json string from marathon server,
 * 	const string usr = marathon user name
 * 	const string pwd = marathon user password
 */
bool marathon_curl_cmd(const string marathon_dns_url_str, string& marathon_task_json, const string usr, const string pwd) {
   int retcode = 0;
   CURL* curl;
   CURLcode res;
   char errorBuffer[CURL_ERROR_SIZE];

   curl_global_init(CURL_GLOBAL_DEFAULT);
   curl = curl_easy_init();

   if(curl) {
      res = curl_easy_setopt(curl, CURLOPT_URL, marathon_dns_url_str.c_str());
      if(res != CURLE_OK) {
         cout << "failed to set url" << endl;
         retcode = -1;
      }

      res = curl_easy_setopt(curl, CURLOPT_HEADER, 1L);
      if(res != CURLE_OK) {
         cout << "failed to set http header" << endl;
         retcode = -1;
      }

      res = curl_easy_setopt(curl, CURLOPT_HTTPGET, 1L);
      if(res != CURLE_OK) {
         cout << "failed to set http get" << endl;
         retcode = -1;
      }

      res = curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, errorBuffer);
      if(res != CURLE_OK) {
         cout << "failed to set error buffer" << endl;
         retcode = -1;
      }

      res = curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writer);
      if(res != CURLE_OK) {
         cout << "failed to set writer" << endl;
         retcode = -1;
      }

      res = curl_easy_setopt(curl, CURLOPT_WRITEDATA, &marathon_task_json);
      if(res != CURLE_OK) {
         cout << "failed to set write data" << endl;
         retcode = -1;
      }

      const string usrpwd = usr + ":" + pwd;
      res = curl_easy_setopt(curl, CURLOPT_USERPWD, "marathon:st0rmtr00per");
      if(res != CURLE_OK) {
         cout << "failed to set usrname" << endl;
         retcode = -1;
      }

      errorBuffer[0] = 0;
      res = curl_easy_perform(curl);
      if (res != CURLE_OK) {
         cout << "curl_easy_perform failed" << endl;
         retcode = -1;
      }

   }

   curl_easy_cleanup(curl);
   return retcode != -1 ? true : false;
}

/*
 * bool success (true) fails (false)
 *    taskIdIsAvailable
 *       const string& marathon_results = data from the marathon server
 *       const string task_id = the application's task id/name
 */
bool taskIdIsAvailable(const string& marathon_results, const string task_id) {
   regex tasks_regex("\"id\":\"" + task_id + "\"");
   smatch tasks_match;
   regex_search(marathon_results, tasks_match, tasks_regex);
   return !tasks_match.empty();
}

/*
 * bool success(true) failure (false)
 *    getHostsPorts(
 *       const string& marathon_dns_str = the marathon server's url
 *       const string task_id = the task id/name
 *       const string usr = marathon user name
 *       const string pwd = marathon user's password
 *       string& cluster_spec = storage for the result
 */
bool getHostsPorts(const string& marathon_dns_str, const string task_id, const string usr, const string pwd, string& cluster_spec) {
   const string marathon_dns_url_str = marathon_dns_str + "/v2/apps/" + task_id;

   string marathon_results;
   marathon_curl_cmd(marathon_dns_url_str, marathon_results, usr, pwd);

   regex hostport_regex("\"host\":\"[a-zA-Z0-9.]+\",\"ports\":\\[([0-9]+)(,)?\\]");
   smatch hostport_match;
   const bool found_hostport = regex_search(marathon_results, hostport_match, hostport_regex);
   if(!found_hostport) { cout << "host+port not found" << endl; return false; }
   if(hostport_match.empty()) { cout << "host+port not found" << endl; return false; }
   const string hostportstr = hostport_match.str();

   regex host_regex("\"host\":\"[a-zA-Z0-9.]+\"");
   smatch host_match;
   const bool found_host = regex_search(hostportstr, host_match, host_regex);
   if(!found_host) { cout << "hosts not found" << endl; return false; }
   if(host_match.empty()) { cout << "hosts not found" << endl; return false; }
   const string jsonhoststr = host_match.str();

   regex hostname_regex(":\"([a-zA-Z0-9.]+)+\"");
   smatch hostname_match;
   const bool found_hostname = regex_search(jsonhoststr, hostname_match, hostname_regex);
   if(!found_hostname) { cout << "hostname not found" << endl; return false; }
   if(hostname_match.empty()) { cout << "hostname not found" << endl; return false; }
   const string hostname_str = hostname_match.str();
   const string hostname = hostname_str.substr(2,hostname_str.length()-3);

   regex port_regex("\"ports\":\\[([0-9]+)(,)?\\]");
   smatch port_match;
   const bool found_ports = regex_search(hostportstr, port_match, port_regex);
   if(!found_ports) { cout << "ports not found" << endl; return false; }
   if(port_match.empty() || port_match.size() < 2) { cout << "ports not found" << endl; return false; }
   const string port_str = port_match[1].str();

   regex portstr_regex("([0-9]+)+");
   smatch portstr_match; // final pair
   const bool found_portstr = regex_search(port_str, portstr_match, portstr_regex);
   if(!found_portstr) { cout << "port not found" << endl; return false; }
   if(portstr_match.empty()) { cout << "port not found" << endl; return false; }
   const string portstr = portstr_match.str();

   // append result for cluster_spec string
   cluster_spec = hostname + ":" + portstr;
   return true;
}

bool fetchClusterSpec(const string marathon_dns_str, const string task_id, const string usr, const string pwd, string& cluster_spec) {
   const string marathon_dns_url_str = marathon_dns_str + "/v2/apps/" + task_id;
   string marathon_results;
   const bool curl_success = marathon_curl_cmd(marathon_dns_url_str, marathon_results, usr, pwd);
   const bool fetchedspec = (curl_success == true && marathon_results.size() > 0 && taskIdIsAvailable(marathon_results, task_id) && getHostsPorts(marathon_dns_str, task_id, usr, pwd, cluster_spec));
   return fetchedspec;
}

bool getMarathonClusterSpec(const string marathon_dns_str, const string task_id, const string usr, const string pwd, const int num_tasks, string& cluster_spec, const int MAXATTEMPTS) {
   cluster_spec = task_id + "|";

   int tasks_attempts [num_tasks];

   for(int i = 0; i < num_tasks; i++) {
      bool addedspec = true;

#ifndef CXX15
      const string task_count = boost::lexical_cast<string>(i+1);
#else
      const string task_count = to_string(i+1);
#endif

      tasks_attempts[i] = 0;
      // busy loop that keeps polling marathon site for task info
      while((addedspec == true) && (tasks_attempts[i] < MAXATTEMPTS)) {
         string node_info_str;
         const string task_id_str = "/" + task_id + task_count;

         // exponential backoff
         this_thread::sleep_for(chrono::seconds((int)pow(2, tasks_attempts[i])));
         const bool successful_fetch = fetchClusterSpec(marathon_dns_str, task_id_str, usr, pwd, node_info_str);

         if(successful_fetch == false && node_info_str.size() < 1) {
            tasks_attempts[i] += 1;
            continue;
         }
         else {
            cluster_spec += node_info_str;
            if (i < num_tasks-1) {
               cluster_spec+=";";
            }
            addedspec = false;
         }
      }
   }

   int tasks_attempts_check = 0;
   for(int i = 0; i < num_tasks; i++) {
      tasks_attempts_check += (tasks_attempts[i] < MAXATTEMPTS) ? 0 : 1;
   }

   return (tasks_attempts_check < num_tasks) ? true : false;
}

/*

#define MAXATTEMPTS 10

void getClusterSpec(const string marathon_dns_str, const string task_id, const string usr, const string pwd, const int num_tasks, string& cluster_spec) {
   cluster_spec = task_id + "|";

   for(int i = 0; i < num_tasks; i++) {
      bool addedspec = true;

#ifndef CXX15
      const string task_count = boost::lexical_cast<string>(i+1);
#else
      const string task_count = to_string(i+1);
#endif

      int attempts = 0;
      // busy loop that keeps polling marathon site for task info
      while((addedspec == true) && (attempts < MAXATTEMPTS)) {
         string node_info_str;
         const string task_id_str = "/" + task_id + task_count;

         // exponential backoff
         this_thread::sleep_for(chrono::seconds((int)pow(2, attempts)));
         const bool successful_fetch = fetchClusterSpec(marathon_dns_str, task_id_str, usr, pwd, node_info_str);

         if(successful_fetch == false && node_info_str.size() < 1) {
            attempts += 1;
            continue;
         }
         else {
            cluster_spec += node_info_str;
            if (i < num_tasks-1) {
               cluster_spec+=";";
            }
            addedspec = false;
         }
      }
   }
}
*/

/*
 * test driver
 *
int main(int argc, char** argv) {
   const char* marathon_uri = argv[1];
   const string tsk_id(argv[2]);
   const string usr(argv[3]);
   const string pwd(argv[4]);

   const int ntasks = argc > 5 ? stoi(argv[5]) : 1;
   string cluster_spec;
   getClusterSpec(marathon_uri, tsk_id, usr, pwd, ntasks, cluster_spec);
   cout << cluster_spec << endl;
}
*/