text:
// Boost.Geometry (aka GGL, Generic Geometry Library)
// Unit Test

// Copyright (c) 2010-2012 Barend Gehrels, Amsterdam, the Netherlands.

// Use, modification and distribution is subject to the Boost Software License,
// Version 1.0. (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)

#include <iostream>
#include <string>

#include <geometry_test_common.hpp>

#include <boost/geometry/geometries/geometries.hpp>
#include <boost/geometry/geometries/point_xy.hpp>

#include <boost/geometry/algorithms/buffer.hpp>
#include <boost/geometry/algorithms/num_points.hpp>
#include <boost/geometry/algorithms/unique.hpp>
#include <boost/geometry/extensions/algorithms/connect.hpp>
#include <boost/geometry/multi/io/wkt/read.hpp>
#include <boost/geometry/multi/algorithms/area.hpp>
#include <boost/geometry/multi/algorithms/length.hpp>
#include <boost/geometry/multi/algorithms/num_points.hpp>
#include <boost/geometry/multi/algorithms/unique.hpp>
#include <boost/geometry/multi/geometries/multi_linestring.hpp>
#include <boost/geometry/strategies/strategies.hpp>

#if defined(TEST_WITH_SVG)
#  include <boost/geometry/multi/algorithms/envelope.hpp>
#  include <boost/geometry/extensions/io/svg/svg_mapper.hpp>
#endif

template <typename GeometryOut, typename Geometry>
void test_connect(std::string const& caseid, Geometry const& geometry,
        std::size_t expected_count, std::size_t expected_point_count,
        double expected_length, double limit = -1, double percentage = 0.001)
{
    typedef typename bg::coordinate_type<Geometry>::type coordinate_type;

    std::vector<GeometryOut> connected_vector;
    if (limit > 0)
    {
        bg::connect(geometry, connected_vector, limit);
    }
    else
    {
        bg::connect(geometry, connected_vector);
    }

    typename bg::default_length_result<Geometry>::type length = 0;
    std::size_t count = boost::size(connected_vector);
    std::size_t point_count = 0;

    BOOST_FOREACH(GeometryOut& connected, connected_vector)
    {
        bg::unique(connected);
        length += bg::length(connected);
        point_count += bg::num_points(connected);
    }

    BOOST_CHECK_MESSAGE(count == expected_count,
            "connect: " << caseid
            << " #lines expected: " << expected_count
            << " detected: " << count
            << " type: " << string_from_type<coordinate_type>::name()
            );

    BOOST_CHECK_EQUAL(point_count, expected_point_count);
    BOOST_CHECK_CLOSE(length, expected_length, percentage);

#if defined(TEST_WITH_SVG)
    {
        std::ostringstream filename;
        filename << "connect_"
            << caseid << "_"
            << string_from_type<coordinate_type>::name()
            << ".svg";

        std::ofstream svg(filename.str().c_str());

        bg::svg_mapper
            <
                typename bg::point_type<Geometry>::type
            > mapper(svg, 500, 500);
        mapper.add(geometry);

        bg::model::box<typename bg::point_type<Geometry>::type> extent;
        bg::envelope(geometry, extent);
        bg::buffer(extent, extent, 0.1);
        mapper.add(extent);

        mapper.map(geometry, "opacity:0.6;fill:rgb(0,0,255);stroke:rgb(0,0,0);stroke-width:1");
        BOOST_FOREACH(GeometryOut& connected, connected_vector)
        {
           mapper.map(connected, "opacity:0.6;fill:none;stroke:rgb(255,0,0);stroke-width:5");
        }
    }
#endif
}

template <typename Geometry, typename GeometryOut>
void test_one(std::string const& caseid, std::string const& wkt,
        std::size_t expected_count, std::size_t expected_point_count,
        double expected_length, double limit = -1, double percentage = 0.001)
{
    Geometry geometry;
    bg::read_wkt(wkt, geometry);

    test_connect<GeometryOut>(caseid, geometry,
        expected_count, expected_point_count,
        expected_length, limit, percentage);

#ifdef BOOST_GEOMETRY_TEST_MULTI_PERMUTATIONS
    // Test different combinations of a multi-polygon

    int n = geometry.size();

    // test them in all orders
    std::vector<int> indices;
    for (int i = 0; i < n; i++)
    {
        indices.push_back(i);
    }
    int permutation = 0;
    do
    {
        std::ostringstream out;
        out << caseid;
        Geometry geometry2;
        for (int i = 0; i < n; i++)
        {
            int index = indices[i];
            out << "_" << index;
            geometry2.push_back(geometry[index]);
        }
        test_connect<GeometryOut>(out.str(), geometry2,
                expected_point_count, expected_length, percentage);
    } while (std::next_permutation(indices.begin(), indices.end()));
#endif

}

template <typename P>
void test_all()
{

    typedef bg::model::linestring<P> linestring;
    typedef bg::model::multi_linestring<linestring> multi_linestring;

    test_one<multi_linestring, linestring>("ls_simplex",
        "MULTILINESTRING((0 0,1 1),(1 1,2 2))",
        1, 3, 2 * std::sqrt(2.0));

    // Opposites, forming one line
    test_one<multi_linestring, linestring>("ls_simplex_opposite_to",
        "MULTILINESTRING((0 0,1 1),(2 2,1 1))",
        1, 3, 2 * std::sqrt(2.0));
    test_one<multi_linestring, linestring>("ls_simplex_opposite_from",
        "MULTILINESTRING((1 1,0 0),(1 1,2 2))",
        1, 3, 2 * std::sqrt(2.0));

    // Two output linestrings
    test_one<multi_linestring, linestring>("ls_simplex_two",
        "MULTILINESTRING((0 0,1 1),(1 1,2 2),(3 3,4 4),(4 4,5 5))",
        2, 6, 4 * std::sqrt(2.0));

    // Linestrings forming a ring
    test_one<multi_linestring, linestring>("ls_simplex_ring",
        "MULTILINESTRING((0 0,0 1),(1 1,1 0),(0 1,1 1),(1 0,0 0))",
        1, 5, 4.0);

    // disconnected rings
    test_one<multi_linestring, linestring>("ls_disconnected_ring1",
        "MULTILINESTRING((0 0,0 1.01),(1.02 1.03,0.99 0),(0 0.98,1.001 1),(1.01 0,0 0))",
        1, 8, 4.137147, 0.5);
    test_one<multi_linestring, linestring>("ls_disconnected_ring2",
        "MULTILINESTRING((0 0,0 1.01),(1.02 1.03,0.99 0),(0 0.98,1.001 1),(1.01 0,0 0))",
        1, 8, 4.137147, 0.1);
    test_one<multi_linestring, linestring>("ls_disconnected_ring3",
        "MULTILINESTRING((0 0,0 1.01),(1.02 1.03,0.99 0),(0 0.98,1.001 1),(1.01 0,0 0))",
        3, 7, 4.05163658, 0.01);

    test_one<multi_linestring, linestring>("ls_disconnected_ring4",
        "MULTILINESTRING((0.01 0,0 1.01),(1.02 1.03,0.99 0),(0 0.98,1.001 1),(1.01 0,0.02 0))",
        1, 8, 4.1172, 0.1);
}

int test_main(int, char* [])
{
    test_all<bg::model::d2::point_xy<double> >();

    return 0;
}

max_stars_repo_name: juslee/boost-svn

max_issues_repo_name: smart-make/boost

max_forks_repo_name: smart-make/boost


text:
// MIT License
// Copyright(c) 2020 Futurewei Cloud
//
//     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.

#include "aca_log.h"
#include "aca_config.h"
#include "aca_util.h"
#include "aca_net_config.h"
#include "aca_vlan_manager.h"
#include "aca_ovs_l2_programmer.h"
#include <chrono>
#include <thread>
#include <errno.h>
#include <boost/algorithm/string/classification.hpp> // Include boost::for is_any_of
#include <boost/algorithm/string/split.hpp> // Include for boost::split
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>
#include <ifaddrs.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <linux/if_link.h>

using namespace std;
using namespace aca_vlan_manager;

// mutex for reading and writing to ovs bridges (br-int and br-tun) setups
mutex setup_ovs_bridges_mutex;

extern std::atomic_ulong g_total_execute_ovsdb_time;
extern std::atomic_ulong g_total_execute_openflow_time;
extern bool g_demo_mode;

namespace aca_ovs_l2_programmer
{
static int aca_set_port_vlan_workitem(const string port_name, uint vlan_id)
{
  ACA_LOG_DEBUG("%s", "aca_set_port_vlan_workitem ---> Entering\n");

  ulong not_care_culminative_time = 0;
  int overall_rc;

  if (port_name.empty()) {
    throw std::invalid_argument("port_name is empty");
  }

  if (vlan_id == 0 || vlan_id >= 4095) {
    throw std::invalid_argument("vlan_id is invalid: " + to_string(vlan_id));
  }

  uint retry_times = 0;
  string cmd_string = "set port " + port_name + " tag=" + to_string(vlan_id);

  do {
    std::this_thread::sleep_for(chrono::milliseconds(PORT_SCAN_SLEEP_INTERVAL));

    overall_rc = EXIT_SUCCESS;
    ACA_OVS_L2_Programmer::get_instance().execute_ovsdb_command(
            cmd_string, not_care_culminative_time, overall_rc);

    if (overall_rc == EXIT_SUCCESS)
      break;
  } while (++retry_times < MAX_PORT_SCAN_RETRY);

  if (overall_rc != EXIT_SUCCESS) {
    ACA_LOG_ERROR("Not able to set the vlan tag %d for port %s even after waiting\n",
                  vlan_id, port_name.c_str());
  }

  // TODO: after this workitem thread is done, it should provide the updated success/fail result back to DPM

  ACA_LOG_DEBUG("aca_set_port_vlan_workitem <--- Exiting, overall_rc = %d\n", overall_rc);

  return overall_rc;
}

ACA_OVS_L2_Programmer &ACA_OVS_L2_Programmer::get_instance()
{
  // Instance is destroyed when program exits.
  // It is instantiated on first use.
  static ACA_OVS_L2_Programmer instance;
  return instance;
}

bool ACA_OVS_L2_Programmer::is_ip_on_the_same_host(const std::string host_ip)
{
  return std::find(this->host_ips_vector.begin(), this->host_ips_vector.end(),
                   host_ip) != this->host_ips_vector.end();
}

void ACA_OVS_L2_Programmer::get_local_host_ips()
{
  struct ifaddrs *ifaddr, *ifa;
  int family, s, n;
  char host[NI_MAXHOST];

  if (getifaddrs(&ifaddr) == -1) {
    perror("getifaddrs");
    exit(EXIT_FAILURE);
  }

  /* Walk through linked list, maintaining head pointer so we
       can free list later */

  for (ifa = ifaddr, n = 0; ifa != NULL; ifa = ifa->ifa_next, n++) {
    if (ifa->ifa_addr == NULL)
      continue;

    family = ifa->ifa_addr->sa_family;

    /* Display interface name and family (including symbolic
           form of the latter for the common families) */

    ACA_LOG_INFO("%-8s %s (%d)\n", ifa->ifa_name,
                 (family == AF_PACKET) ? "AF_PACKET" :
                 (family == AF_INET)   ? "AF_INET" :
                 (family == AF_INET6)  ? "AF_INET6" :
                                         "???",
                 family);

    /* For an AF_INET* interface address, display the address */

    if (family == AF_INET || family == AF_INET6) {
      s = getnameinfo(ifa->ifa_addr,
                      (family == AF_INET) ? sizeof(struct sockaddr_in) :
                                            sizeof(struct sockaddr_in6),
                      host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST);
      if (s != 0) {
        ACA_LOG_WARN("getnameinfo() failed: %s\n", gai_strerror(s));
        exit(EXIT_FAILURE);
      }

      ACA_LOG_INFO("\t\taddress: <%s>\n", host);
      std::string host_ip_string = std::string(host);
      host_ips_vector.push_back(host_ip_string);
    }
  }

  freeifaddrs(ifaddr);

  vector<string>::iterator it = host_ips_vector.begin();
  while (it != host_ips_vector.end()) {
    ACA_LOG_DEBUG("Current Host IP: [%s]\n", it->c_str());
    if (it->empty()) {
      ACA_LOG_DEBUG("Removing empty IP: [%s]\n", it->c_str());
      it = host_ips_vector.erase(it);
    } else {
      it++;
    }
  }
}

int ACA_OVS_L2_Programmer::setup_ovs_bridges_if_need()
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::setup_ovs_bridges_if_need ---> Entering\n");

  ulong not_care_culminative_time;
  int overall_rc = EXIT_SUCCESS;

  // -----critical section starts-----
  // (exclusive access to br-int and br-tun creation status signaled by locking setup_ovs_bridges_mutex):
  setup_ovs_bridges_mutex.lock();

  // check to see if br-int and br-tun is already there
  execute_ovsdb_command("br-exists br-int", not_care_culminative_time, overall_rc);
  bool br_int_existed = (overall_rc == EXIT_SUCCESS);
  overall_rc = EXIT_SUCCESS;

  execute_ovsdb_command("br-exists br-tun", not_care_culminative_time, overall_rc);
  bool br_tun_existed = (overall_rc == EXIT_SUCCESS);
  ACA_LOG_INFO("Environment br-int=%d and br-tun=%d\n", br_int_existed, br_tun_existed);
  overall_rc = EXIT_SUCCESS;

  if (br_int_existed && br_tun_existed) {
    // case 1: both br-int and br-tun existed
    ACA_LOG_DEBUG("%s", "Both br-int and br-tun existed: do nothing\n");
    setup_ovs_bridges_mutex.unlock();
    // -----critical section ends-----

    ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::setup_ovs_bridges_if_need <--- Exiting, overall_rc = %d\n",
                  overall_rc);

    return overall_rc;
  }

  if (!br_int_existed && !br_tun_existed) {
    // case 2: both br-int and br-tun not existed
    ACA_LOG_DEBUG("%s", "Both br-int and br-tun not existed: create them\n");
    ACA_LOG_INFO("Environment br-int=%d and br-tun=%d\n", br_int_existed, br_tun_existed);

    execute_ovsdb_command("add-br br-int", not_care_culminative_time, overall_rc);

    execute_ovsdb_command("add-br br-tun", not_care_culminative_time, overall_rc);

    // create and connect the patch ports between br-int and br-tun
    execute_ovsdb_command("-- add-port br-int patch-tun "
                          "-- set interface patch-tun type=patch options:peer=patch-int "
                          "-- add-port br-tun patch-int "
                          "-- set interface patch-int type=patch options:peer=patch-tun",
                          not_care_culminative_time, overall_rc);

    execute_ovsdb_command(
            string("--may-exist add-port br-tun vxlan-generic -- set interface vxlan-generic ofport_request=") +
                    VXLAN_GENERIC_OUTPORT_NUMBER +
                    " type=vxlan options:df_default=true options:egress_pkt_mark=0 options:in_key=flow options:out_key=flow options:remote_ip=flow",
            not_care_culminative_time, overall_rc);

    setup_ovs_bridges_mutex.unlock();
    // -----critical section ends-----

    ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::setup_ovs_bridges_if_need <--- Exiting, overall_rc = %d\n",
                  overall_rc);

    return overall_rc;
  }

  // case 3: only one of the br-int or br-tun is there,
  // Invalid environment so return an error
  if (br_int_existed) {
    ACA_LOG_DEBUG("You have br-int, but you don't have br-tun, please add br-tun by executing command 'ovs-vsctl add-br br-tun', or delete the existing br-int by executing command 'ovs-vsctl del-br br-int' and try again.\n",
                  br_tun_existed);
  }
  if (br_tun_existed) {
    ACA_LOG_DEBUG("You have br-tun, but you don't have br-int, please add br-int by executing command 'ovs-vsctl add-br br-int', or delete the existing br-tun by executing command 'ovs-vsctl del-br br-tun' and try again.\n",
                  br_tun_existed);
  }
  ACA_LOG_CRIT("Invalid environment br-int=%d and br-tun=%d, cannot proceed\n",
               br_int_existed, br_tun_existed);
  overall_rc = EXIT_FAILURE;

  setup_ovs_bridges_mutex.unlock();
  // -----critical section ends-----

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::setup_ovs_bridges_if_need <--- Exiting, overall_rc = %d\n",
                overall_rc);

  return overall_rc;
}

int ACA_OVS_L2_Programmer::setup_ovs_controller(const std::string ctrler_ip, const int ctrler_port)
{
  int rc = EXIT_SUCCESS;

  const string ctrler_endpoint = " tcp:" + ctrler_ip + ":" + to_string(ctrler_port);
  const string br_int_str = "br-int";
  const string br_tun_str = "br-tun";
  const string setup_br_int_cmd = "set-controller " + br_int_str + ctrler_endpoint;
  const string setup_br_tun_cmd = "set-controller " + br_tun_str + ctrler_endpoint;

  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::setup_ovs_controller ---> Entering\n");

  // get bridge and dpid mappings from ovs
  std::unordered_map<uint64_t, std::string> switch_dpid_map = get_ovs_bridge_mapping();

  // get system port name and ofportid mappings from ovs
  std::unordered_map<std::string, std::string> port_id_map = get_system_port_ids();

  // set bridge controller will clean up flows
  auto ovsdb_client_start = chrono::steady_clock::now();
  string br_int_cmd_string = "ovs-vsctl " + setup_br_int_cmd;
  rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(br_int_cmd_string);
  if (rc != EXIT_SUCCESS) {
    ACA_LOG_ERROR("ACA_OVS_L2_Programmer::setup_ovs_controller - failed to set br-int controller\n");
  }

  string br_tun_cmd_string = "ovs-vsctl " + setup_br_tun_cmd;
  rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(br_tun_cmd_string);
  if (rc != EXIT_SUCCESS) {
    ACA_LOG_ERROR("ACA_OVS_L2_Programmer::setup_ovs_controller - failed to set br-tun controller\n");
  }

  auto ovsdb_client_end = chrono::steady_clock::now();
  auto ovsdb_client_time_total_time =
          cast_to_microseconds(ovsdb_client_end - ovsdb_client_start).count();
  g_total_execute_ovsdb_time += ovsdb_client_time_total_time;

  ACA_LOG_INFO("ACA_OVS_L2_Programmer::setup_ovs_controller - Elapsed time for ovsdb client call took: %ld microseconds or %ld milliseconds\n",
               ovsdb_client_time_total_time, us_to_ms(ovsdb_client_time_total_time));

  // start local ovs server (openflow controller)
  ofctrl = new OFController(switch_dpid_map, port_id_map, ctrler_ip.c_str(), ctrler_port);
  ofctrl->start();

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::setup_ovs_controller <--- Exiting\n");

  return rc;
}

void ACA_OVS_L2_Programmer::clean_up_ovs_controller()
{
  if (ofctrl != NULL)
  {
    ofctrl->stop();
    delete ofctrl;
    ofctrl = NULL;
    ACA_LOG_INFO("%s", "ACA_OVS_L2_Programmer::clean_up_ovs_controller - cleaned up ovs controller.\n");
  }
  else
  {
    ACA_LOG_INFO("%s", "ACA_OVS_L2_Programmer::clean_up_ovs_controller - unable to clean up ovs controller, since it is null.\n");
  }
}

std::string ACA_OVS_L2_Programmer::get_system_port_id(std::string port_name)
{
  return port_id_map[port_name];
}

std::unordered_map<std::string, std::string> ACA_OVS_L2_Programmer::get_system_port_ids()
{
  // these 2 system ports belong to br-tun
  const string patch_int_port = "patch-int";
  const string vxlan_generic_port = "vxlan-generic";

  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::get_system_port_ids ---> Entering\n");
  auto ovsdb_client_start = chrono::steady_clock::now();

  string patch_int_ofport_query = "ovs-vsctl get Interface " + patch_int_port + " ofport";
  string patch_int_ofport_id = aca_net_config::Aca_Net_Config::get_instance().execute_system_command_with_return(patch_int_ofport_query);
  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_system_port_ids - adding %s - %s mapping to port_id_map\n", patch_int_port.c_str(), patch_int_ofport_id.c_str());
  port_id_map[patch_int_port] = patch_int_ofport_id;

  string vxlan_ofport_query = "ovs-vsctl get Interface " + vxlan_generic_port + " ofport";
  string vxlan_ofport_id = aca_net_config::Aca_Net_Config::get_instance().execute_system_command_with_return(vxlan_ofport_query);
  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_system_port_ids - adding %s - %s mapping to port_id_map\n", vxlan_generic_port.c_str(), vxlan_ofport_id.c_str());
  port_id_map[vxlan_generic_port] = vxlan_ofport_id;

  auto ovsdb_client_end = chrono::steady_clock::now();
  auto ovsdb_client_time_total_time =
          cast_to_microseconds(ovsdb_client_end - ovsdb_client_start).count();

  g_total_execute_ovsdb_time += ovsdb_client_time_total_time;

  ACA_LOG_INFO("ACA_OVS_L2_Programmer::get_system_port_ids - Elapsed time for ovsdb client call took: %ld microseconds or %ld milliseconds\n",
               ovsdb_client_time_total_time, us_to_ms(ovsdb_client_time_total_time));

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_system_port_ids <--- Exiting\n");

  return port_id_map;
}

std::unordered_map<uint64_t, std::string> ACA_OVS_L2_Programmer::get_ovs_bridge_mapping()
{
  const string br_int_str = "br-int";
  const string br_tun_str = "br-tun";
  const string get_br_int_dpid = "get Bridge " + br_int_str + " datapath_id";
  const string get_br_tun_dpid = "get Bridge " + br_tun_str + " datapath_id";
  std::unordered_map<uint64_t, std::string> switch_dpid_map;

  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::get_ovs_bridge_mapping ---> Entering\n");
  auto ovsdb_client_start = chrono::steady_clock::now();

  string br_int_cmd_string = "ovs-vsctl " + get_br_int_dpid;
  // raw string output format is like a hex string "00003af45ed7aa45"
  string br_int_dpid_raw = aca_net_config::Aca_Net_Config::get_instance().execute_system_command_with_return(br_int_cmd_string);
  // trim the (") symbol at the start and the end to get 00003af45ed7aa45, and then convert to decimal
  uint64_t br_int_dpid = std::stoul(br_int_dpid_raw.substr(1, br_int_dpid_raw.length() - 3), nullptr, 16);
  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_ovs_bridge_mapping - adding %ld - %s mapping to switch_dpid_map\n", br_int_dpid, br_int_str.c_str());
  switch_dpid_map[br_int_dpid] = br_int_str;

  string br_tun_cmd_string = "ovs-vsctl " + get_br_tun_dpid;
  string br_tun_dpid_raw = aca_net_config::Aca_Net_Config::get_instance().execute_system_command_with_return(br_tun_cmd_string);
  uint64_t br_tun_dpid = std::stoul(br_tun_dpid_raw.substr(1, br_tun_dpid_raw.length() - 3), nullptr, 16);
  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_ovs_bridge_mapping - adding %ld - %s mapping to switch_dpid_map\n", br_tun_dpid, br_tun_str.c_str());
  switch_dpid_map[br_tun_dpid] = br_tun_str;

  auto ovsdb_client_end = chrono::steady_clock::now();
  auto ovsdb_client_time_total_time =
          cast_to_microseconds(ovsdb_client_end - ovsdb_client_start).count();

  g_total_execute_ovsdb_time += ovsdb_client_time_total_time;

  ACA_LOG_INFO("ACA_OVS_L2_Programmer::get_ovs_bridge_mapping - Elapsed time for ovsdb client call took: %ld microseconds or %ld milliseconds\n",
               ovsdb_client_time_total_time, us_to_ms(ovsdb_client_time_total_time));

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::get_ovs_bridge_mapping <--- Exiting\n");

  return switch_dpid_map;
}

int ACA_OVS_L2_Programmer::create_port(const string vpc_id, const string port_name,
                                       const string virtual_ip, const string virtual_mac,
                                       uint tunnel_id, ulong &culminative_time)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::create_port ---> Entering\n");

  int overall_rc = EXIT_SUCCESS;

  if (vpc_id.empty()) {
    throw std::invalid_argument("vpc_id is empty");
  }

  if (port_name.empty()) {
    throw std::invalid_argument("port_name is empty");
  }

  if (virtual_ip.empty()) {
    throw std::invalid_argument("virtual_ip is empty");
  }

  if (virtual_mac.empty()) {
    throw std::invalid_argument("virtual_mac is empty");
  }

  if (tunnel_id == 0) {
    throw std::invalid_argument("tunnel_id is 0");
  }

  // TODO: if ovs_port already exist in vlan manager, that means it is a duplicated
  // port CREATE operation, we have the following options to handle it:
  // 1. Reject call with error message
  // 2. No ops with warning message
  // 3. Do it regardless => if there is anything wrong, controller's fault :-)

  // use vpc_id to query vlan_manager to lookup an existing vpc_id entry to get its
  // internal vlan id or to create a new vpc_id entry to get a new internal vlan id
  int internal_vlan_id = ACA_Vlan_Manager::get_instance().get_or_create_vlan_id(tunnel_id);

  ACA_Vlan_Manager::get_instance().create_ovs_port(vpc_id, port_name, tunnel_id, culminative_time);

  if (g_demo_mode) {
    string cmd_string = "add-port br-int " + port_name +
                        " tag=" + to_string(internal_vlan_id) +
                        " -- set Interface " + port_name + " type=internal";

    execute_ovsdb_command(cmd_string, culminative_time, overall_rc);

    cmd_string = "ip addr add " + virtual_ip + " dev " + port_name;
    int command_rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(
            cmd_string, culminative_time);
    if (command_rc != EXIT_SUCCESS)
      overall_rc = command_rc;

    cmd_string = "ip link set dev " + port_name + " address " + virtual_mac;
    command_rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(
            cmd_string, culminative_time);
    if (command_rc != EXIT_SUCCESS)
      overall_rc = command_rc;

    cmd_string = "ip link set " + port_name + " up";
    command_rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(
            cmd_string, culminative_time);
    if (command_rc != EXIT_SUCCESS)
      overall_rc = command_rc;
  }

  // Disabling this code for testing, as test controller already sets the vlan,
  // and the port names generated by TC's ovs-docker is different from the port_name used here.
  else {
    // non-demo mode is for nova integration, where the vif and ovs port has been
    // created by nova compute agent running on the compute host

    // just need to set the vlan tag on the ovs port, the ovs port may be not created by nova yet
    string cmd_string = "set port " + port_name + " tag=" + to_string(internal_vlan_id);

    execute_ovsdb_command(cmd_string, culminative_time, overall_rc);

    // if the ovs port is not there to set to vlan, we will return PENDING as the result
    // and spin up the new thread to keep trying that in the backgroud
    if (overall_rc != EXIT_SUCCESS) {
      overall_rc = EINPROGRESS;

      // start a new background thread work set the port vlan
      std::thread t(aca_set_port_vlan_workitem, port_name, internal_vlan_id);

      // purposely not wait for this aca_set_port_vlan_workitem
      if (t.joinable()) {
        t.detach();
      }
    }
  }

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::create_port <--- Exiting, overall_rc = %d\n", overall_rc);

  return overall_rc;
}

int ACA_OVS_L2_Programmer::delete_port(const string vpc_id, const string port_name,
                                       uint tunnel_id, ulong &culminative_time)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::delete_port ---> Entering\n");

  if (vpc_id.empty()) {
    throw std::invalid_argument("vpc_id is empty");
  }

  if (port_name.empty()) {
    throw std::invalid_argument("port_name is empty");
  }

  if (tunnel_id == 0) {
    throw std::invalid_argument("tunnel_id is 0");
  }

  int overall_rc = ACA_Vlan_Manager::get_instance().delete_ovs_port(
          vpc_id, port_name, tunnel_id, culminative_time);

  if (g_demo_mode) {
    string cmd_string = "del-port br-int " + port_name;

    execute_ovsdb_command(cmd_string, culminative_time, overall_rc);
  }

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::delete_port <--- Exiting, overall_rc = %d\n", overall_rc);

  return overall_rc;
}

int ACA_OVS_L2_Programmer::create_or_update_l2_neighbor(const string virtual_ip,
                                                        const string virtual_mac,
                                                        const string remote_host_ip,
                                                        uint tunnel_id, ulong &culminative_time)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::create_or_update_l2_neighbor ---> Entering\n");

  if (virtual_ip.empty()) {
    throw std::invalid_argument("virtual_ip is empty");
  }

  if (virtual_mac.empty()) {
    throw std::invalid_argument("virtual_mac is empty");
  }

  if (remote_host_ip.empty()) {
    throw std::invalid_argument("remote_host_ip is empty");
  }

  if (tunnel_id == 0) {
    throw std::invalid_argument("tunnel_id is 0");
  }

  int overall_rc = ACA_Vlan_Manager::get_instance().create_l2_neighbor(
          virtual_ip, virtual_mac, remote_host_ip, tunnel_id, culminative_time);

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::create_or_update_l2_neighbor <--- Exiting, overall_rc = %d\n",
                overall_rc);

  return overall_rc;
}

int ACA_OVS_L2_Programmer::delete_l2_neighbor(const string virtual_ip, const string virtual_mac,
                                              uint tunnel_id, ulong &culminative_time)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::delete_l2_neighbor ---> Entering\n");

  if (virtual_ip.empty()) {
    throw std::invalid_argument("virtual_ip is empty");
  }

  if (virtual_mac.empty()) {
    throw std::invalid_argument("virtual_mac is empty");
  }

  if (tunnel_id == 0) {
    throw std::invalid_argument("tunnel_id is 0");
  }

  int overall_rc = ACA_Vlan_Manager::get_instance().delete_l2_neighbor(
          virtual_ip, virtual_mac, tunnel_id, culminative_time);

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::delete_l2_neighbor <--- Exiting, overall_rc = %d\n",
                overall_rc);

  return overall_rc;
}

void ACA_OVS_L2_Programmer::execute_ovsdb_command(const std::string cmd_string,
                                                  ulong &culminative_time, int &overall_rc)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::execute_ovsdb_command ---> Entering\n");

  auto ovsdb_client_start = chrono::steady_clock::now();

  string ovsdb_cmd_string = "ovs-vsctl " + cmd_string;
  int rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(ovsdb_cmd_string);

  if (rc != EXIT_SUCCESS) {
    overall_rc = rc;
  }

  auto ovsdb_client_end = chrono::steady_clock::now();

  auto ovsdb_client_time_total_time =
          cast_to_microseconds(ovsdb_client_end - ovsdb_client_start).count();

  culminative_time += ovsdb_client_time_total_time;

  g_total_execute_ovsdb_time += ovsdb_client_time_total_time;

  ACA_LOG_INFO("Elapsed time for ovsdb client call took: %ld microseconds or %ld milliseconds. rc: %d, cmd: [%s]\n",
               ovsdb_client_time_total_time, us_to_ms(ovsdb_client_time_total_time),
               rc, ovsdb_cmd_string.c_str());

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::execute_ovsdb_command <--- Exiting, rc = %d\n", rc);
}

void ACA_OVS_L2_Programmer::execute_openflow_command(const std::string cmd_string,
                                                     ulong &culminative_time, int &overall_rc)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::execute_openflow_command ---> Entering\n");

  auto openflow_client_start = chrono::steady_clock::now();

  string openflow_cmd_string = "ovs-ofctl " + cmd_string;
  int rc = aca_net_config::Aca_Net_Config::get_instance().execute_system_command(openflow_cmd_string);

  if (rc != EXIT_SUCCESS) {
    overall_rc = rc;
  }

  auto openflow_client_end = chrono::steady_clock::now();

  auto openflow_client_time_total_time =
          cast_to_microseconds(openflow_client_end - openflow_client_start).count();

  culminative_time += openflow_client_time_total_time;

  g_total_execute_openflow_time += openflow_client_time_total_time;

  ACA_LOG_INFO("Elapsed time for openflow client call took: %ld microseconds or %ld milliseconds. rc: %d\n",
               openflow_client_time_total_time,
               us_to_ms(openflow_client_time_total_time), rc);

  ACA_LOG_DEBUG("ACA_OVS_L2_Programmer::execute_openflow_command <--- Exiting, rc = %d\n", rc);
}

void ACA_OVS_L2_Programmer::execute_openflow(ulong &culminative_time,
                                             const std::string bridge,
                                             const std::string flow_string,
                                             const std::string action)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::execute_openflow ---> Entering\n");
  auto openflow_client_start = chrono::steady_clock::now();

  if (NULL != ofctrl) {
    ofctrl->execute_flow(bridge, flow_string, action);
  } else {
    ACA_LOG_ERROR("%s", "ACA_OVS_L2_Programmer::execute_openflow didn't find OF controller\n");
  }

  auto openflow_client_end = chrono::steady_clock::now();
  auto openflow_client_time_total_time =
          cast_to_microseconds(openflow_client_end - openflow_client_start).count();

  culminative_time += openflow_client_time_total_time;

  g_total_execute_openflow_time += openflow_client_time_total_time;

  ACA_LOG_INFO("Elapsed time for openflow client call took: %ld microseconds or %ld milliseconds.\n",
               openflow_client_time_total_time,
               us_to_ms(openflow_client_time_total_time));

  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::execute_openflow ---> Exiting\n");
}

void ACA_OVS_L2_Programmer::packet_out(const char *bridge, const char *options)
{
  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::packet_out ---> Entering\n");
  auto openflow_client_start = chrono::steady_clock::now();

  if (NULL != ofctrl) {
      ofctrl->packet_out(bridge, options);
  } else {
      ACA_LOG_ERROR("%s", "ACA_OVS_L2_Programmer::packet_out didn't find OF controller\n");
  }

  auto openflow_client_end = chrono::steady_clock::now();
  auto openflow_client_time_total_time =
          cast_to_microseconds(openflow_client_end - openflow_client_start).count();

  g_total_execute_openflow_time += openflow_client_time_total_time;

  ACA_LOG_INFO("Elapsed time for openflow client call took: %ld microseconds or %ld milliseconds.\n",
               openflow_client_time_total_time,
               us_to_ms(openflow_client_time_total_time));

  ACA_LOG_DEBUG("%s", "ACA_OVS_L2_Programmer::packet_out ---> Exiting\n");
}

} // namespace aca_ovs_l2_programmer

ext: cpp

max_stars_repo_name: lfu-ps/alcor-control-agent

max_issues_repo_name: lfu-ps/alcor-control-agent

max_forks_repo_name: lfu-ps/alcor-control-agent


text:
#ifdef __clang__

#include <string>
#include <boost/test/unit_test.hpp>

namespace boost { namespace unit_test { namespace ut_detail {
    std::string normalize_test_case_name(const_string name) {
        return ( name[0] == '&' ? std::string(name.begin()+1, name.size()-1) : std::string(name.begin(), name.size() ));
    }
}}}

#endif

max_stars_repo_name: reuben/semver

max_issues_repo_name: reuben/semver

max_forks_repo_name: reuben/semver


text:
//  tree_suite: Python-land indexing and manipulation for I3Tree
//  (C) 2012 Jakob van Santen <vansanten@wisc.edu>
//           and the IceCube Collaboration
//

#ifndef ICETRAY_PYTHON_TREE_INDEXING_SUITE_HPP_INCLUDED
#define ICETRAY_PYTHON_TREE_INDEXING_SUITE_HPP_INCLUDED

#include <boost/python/iterator.hpp>
#include <boost/python/call_method.hpp>
#include <boost/python/tuple.hpp>

#include <boost/python/object/iterator_core.hpp>

#include <boost/mpl/vector.hpp>
#include <boost/mpl/deref.hpp>
#include <boost/mpl/assert.hpp>
#include <boost/mpl/begin_end.hpp>
#include <boost/mpl/next_prior.hpp>
#include <boost/mpl/at.hpp>
#include <boost/mpl/insert.hpp>
#include <boost/mpl/erase.hpp>
#include <boost/mpl/replace.hpp>

#include <icetray/python/get_class.hpp>

namespace bp = boost::python;

namespace boost { namespace python {

namespace tree_suite_detail {
	template <typename Sig>
	struct LastArg
	{
		typedef typename mpl::deref<typename mpl::prior<
		    typename mpl::end<Sig>::type>::type>::type type;
	};
	template <typename Sig>
	struct ReturnType
	{
		typedef typename mpl::deref<
		    typename mpl::begin<Sig>::type>::type type;
	};
}

template <class Container>
class tree_indexing_suite : public bp::def_visitor<tree_indexing_suite<Container > >{
public:

	typedef typename Container::value_type value_type;
	typedef typename Container::pre_order_iterator pre_order_iterator;
	typedef typename Container::post_order_iterator post_order_iterator;
	typedef typename Container::sibling_iterator sibling_iterator;

	static bp::object get_value_type()
	{
		return get_class<value_type>();
	}

	static void
	index_translation_error(int offset, unsigned depth) throw(python::error_already_set)
	{
		std::ostringstream os;
		os << "Index (" << offset << ") out of range at level " << depth;
		PyErr_SetString(PyExc_IndexError, os.str().c_str());
		throw python::error_already_set();
	}

	static typename Container::sibling_iterator
	translate_index(Container &self, bp::tuple &index)
	{
		unsigned maxdepth = len(index)-1;
		sibling_iterator it = self.begin();
		for (unsigned depth=0; depth < maxdepth; depth++) {
			// Advance among the siblings by the given index
			int offset = extract<int>(index[depth]);
			it += offset;

			if (!self.is_valid(it))
				index_translation_error(offset, depth);
			// Drop down a level at the indicated position
			it = self.begin(it);
		}

		int offset = extract<int>(index[maxdepth]);
		it += offset;

		if (!self.is_valid(it))
			index_translation_error(offset, maxdepth);
		return it;
	}

	static bp::tuple
	translate_iterator(Container &self, sibling_iterator it)
	{
		std::vector<int> indices(self.depth(it)+1, 0);
		for (unsigned depth = indices.size()-1; depth > 0; depth--) {
			indices[depth] = self.index(it);
			it = self.parent(it);
		}

		indices[0] = self.index(it);

		return bp::tuple(indices);
	}

	// A basic range wrapper much like boost::python::range
	template <typename Iterator>
	class iterator : public Iterator {
	public:
		iterator(Container &target, Iterator &begin, Iterator &end, bool addr)
		    : self_(target), begin_(begin), end_(end), return_address_(addr) {};
		python::object next()
		{
			if (begin_ == end_)
				objects::stop_iteration_error();
			if (return_address_) {
				python::tuple addr = translate_iterator(self_, begin_);
				return python::make_tuple(addr, *begin_++);
			} else
				return python::object(*begin_++);
		}
	private:
		Container &self_;
		Iterator begin_, end_;
		bool return_address_;
	};

	template <typename Iterator>
	static void
	register_iterator(const char *name)
	{
		typedef iterator<Iterator> IteratorWrapper;
		class_<IteratorWrapper>(name, bp::no_init)
			.def("__iter__", objects::identity_function())
#if PY_MAJOR_VERSION >= 3
			.def("__next__", &IteratorWrapper::next)
#else
			.def("next", &IteratorWrapper::next)
#endif
		;
	}

	template <typename Accessor>
	static inline iterator<Accessor>
	make_iterator(Container &x, Accessor begin, Accessor end, bool addr)
	{
		return iterator<Accessor>(x, begin, end, addr);
	}

	static iterator<pre_order_iterator>
	get_iterator(Container &x)
	{
		return make_iterator(x, x.begin(), x.end(), false);
	};

	static iterator<pre_order_iterator>
	get_keyed_iterator(Container &x)
	{
		return make_iterator(x, x.begin(), x.end(), true);
	};

	static iterator<post_order_iterator>
	get_post_order_iterator(Container &x, bool addr)
	{
		return make_iterator(x, x.begin_post(), x.end_post(), addr);
	};

	static iterator<sibling_iterator>
	get_sibling_iterator(Container &x, tuple pos, bool addr)
	{
		typename Container::sibling_iterator it = translate_index(x, pos);
		return make_iterator(x, x.begin(it), x.end(it), addr);
	};

	static bp::tuple
	add_root(Container &self, const value_type &v)
	{
		if (self.size() == 0)
			return translate_iterator(self, self.set_head(v));
		else
			return translate_iterator(self, self.insert(self.begin(), v));
	}

	static value_type
	getitem(Container &self, bp::tuple index)
	{
		return *translate_index(self, index);
	}

	static value_type
	getitem_simple(Container &self, int index)
	{
		return getitem(self, make_tuple(index));
	}

	static void
	setitem(Container &self, bp::tuple index, const value_type &v)
	{
		*translate_index(self, index) = v;
	}

	static void
	setitem_simple(Container &self, int index, const value_type &v)
	{
		setitem(self, make_tuple(index), v);
	}

	static bp::list
	argfind(Container &self, bp::object comparator)
	{
		bp::list results;
		typename Container::iterator it;
		for (it = self.begin(); it != self.end(); it++)
			if (bp::extract<bool>(comparator(*it)))
				results.append(translate_iterator(self, it));
		return results;
	}

	// iterator_adapter<F,S>::make_function(F f) takes a member-function pointer
	// that takes and returns tree iterators, and automatically generates a wrapper
	// function that converts an address tuple to the equivalent iterator (e.g. (0,1,0)
	// to first_root->second_child->first_child), calls the function, and converts
	// the result back to an address tuple.
	template <typename F, typename S>
	struct iterator_adapter : public bp::object {

		typedef F FunctionPtr;
		typedef S HeldSignature;
		typedef typename tree_suite_detail::ReturnType<S>::type ReturnType;
		typedef typename tree_suite_detail::LastArg<S>::type Arg1;
		typedef typename mpl::replace<S, ReturnType, bp::tuple>::type WrapperSignature;

		typedef iterator_adapter<F,S> Adapter;

		FunctionPtr f_;

		iterator_adapter<F,S>(F f) : f_(f) {};

		// SFINAE dispatch to specializations for different signatures
		template <typename S1>
		static typename bp::object
		make_function(F f, S1 s)
		{
			return Adapter::make_function_impl(f, s);
		}

	private:

		// Methods of the form: iterator (Container::*)(iterator, value_type)
		template <typename S1>
		static typename enable_if<
		    mpl::equal_to<typename mpl::size<S1>::type,
			          typename mpl::long_<4l>::type>,
		    bp::object>::type
		make_function_impl(F f, S1 s)
		{
			return bp::make_function(boost::bind(&Adapter::call_inserter, Adapter(f), _1, _2, _3),
			    default_call_policies(), args("self", "index", "value"), WrapperSignature());
		}

		// Methods of the form: iterator (Container::*)(iterator)
		template <typename S1>
		static typename enable_if<typename mpl::and_<
			                  typename mpl::equal_to<typename mpl::size<S1>::type,
					                         typename mpl::long_<3l>::type>::type,
				          typename is_same<typename mpl::at<S1, typename mpl::long_<2> >::type,
					                   ReturnType>::type>,
		    bp::object>::type
		make_function_impl(F f, S1 s)
		{
			return bp::make_function(boost::bind(&Adapter::call_single_iter, Adapter(f), _1, _2),
			    default_call_policies(), args("self", "index"), WrapperSignature());

		}

		bp::tuple call_inserter(const Container &self, bp::tuple index, Arg1 a1)
		{
			return translate_call(self, boost::bind(f_, const_cast<Container*>(&self), _1, a1), index);
		}

		bp::tuple call_single_iter(const Container &self, bp::tuple index)
		{
			return translate_call(self, boost::bind(f_, const_cast<Container*>(&self), _1), index);
		}

		// Translate tuple to iterator, call functor, translate result back
		template <typename Caller>
		static inline bp::tuple
		translate_call(const Container &selfc, Caller caller, bp::tuple index)
		{
			Container &self = const_cast<Container&>(selfc);
			ReturnType it = translate_index(self, index);
			it = caller(it);
			if (self.is_valid(it))
				return translate_iterator(self, it);
			else
				return bp::make_tuple(0);
		}

	};

	template <typename F>
	static bp::object
	make_adapter(F f)
	{
		return make_adapter_impl(f, detail::get_signature(f));
	}

	template <typename F, typename Sig>
	static bp::object
	make_adapter_impl(F f, Sig sig)
	{
		return iterator_adapter<F,Sig>::make_function(f, sig);
	}

	template <class Class>
	static void
	visit(Class& cl)
	{
		bp::scope outer = cl;

		register_iterator<pre_order_iterator>("pre_order_iterator");
		register_iterator<post_order_iterator>("post_order_iterator");
		register_iterator<sibling_iterator>("sibling_iterator");

		typedef sibling_iterator (Container::*inserter_func)(sibling_iterator, const value_type&);
		typedef sibling_iterator (Container::*itermanip_func)(sibling_iterator);

		cl
			.def("__len__", &Container::size)
			.def("__iter__", &get_iterator)
			.def("__getitem__", &getitem)
			.def("__getitem__", &getitem_simple)
			.def("__setitem__", &setitem)
			.def("__setitem__", &setitem_simple)
			.def("iteritems", &get_keyed_iterator)
			.def("iter_children", &get_sibling_iterator,     arg("self"), arg("index"), arg("return_indices")=true)
			.def("iter_postorder", &get_post_order_iterator, arg("self"), arg("return_indices")=true)
			.def("argfind", &argfind, arg("self"), arg("keyfunc"))
			.def("add_root", &add_root, arg("self"), arg("value"))
			.def("insert",       make_adapter((inserter_func)&Container::insert))
			.def("insert_after", make_adapter((inserter_func)&Container::insert_after))
			.def("replace",      make_adapter((inserter_func)&Container::replace))
			.def("append_child", make_adapter((inserter_func)&Container::append_child))
			.def("erase",        make_adapter((itermanip_func)&Container::erase))
			.def("__value_type__", &get_value_type)
			.staticmethod("__value_type__")
		;

	}

};

} } // namespace boost::python

#endif

ext: hpp


text:
#ifndef MARKDOWN_BLOCK_PARSER_IPP
#define MARKDOWN_BLOCK_PARSER_IPP

#include <boost/spirit/include/qi.hpp>
#include <boost/spirit/include/phoenix_core.hpp>
#include <boost/spirit/include/phoenix_operator.hpp>
#include <boost/spirit/include/phoenix_object.hpp>
#include <boost/spirit/include/phoenix_fusion.hpp>
#include <boost/spirit/include/phoenix_stl.hpp>

#include <boost/phoenix/statement/if.hpp>
#include <boost/phoenix/statement/sequence.hpp>
#include <boost/phoenix/object/construct.hpp>
#include <boost/phoenix/core/reference.hpp>   // For access to ref()
#include <boost/phoenix/stl/container.hpp>    // For access to lazy container empty() and clear().

//#include <boost/spirit/home/phoenix/statement/if.hpp>
//#include <boost/spirit/home/phoenix/statement/sequence.hpp>
//#include <boost/spirit/home/phoenix/object/construct.hpp>
//#include <boost/spirit/home/phoenix/container.hpp>

#include "dom/blocks.hpp"
#include "dom/inlines.hpp"

#include "markdown/blockParser.hpp"

namespace Markdown {

template <typename Iterator>
BlockParser<Iterator>::BlockParser()
:
    BlockParser::base_type(block),
    strings(),
    html(strings),
    inline_(strings, html)
{
    //using namespace boost::spirit;
    using namespace boost::spirit::qi;
    using boost::phoenix::val;
    using boost::phoenix::ref;
    using boost::phoenix::size;
    using boost::phoenix::push_back;
    using boost::phoenix::at_c;
    using boost::phoenix::construct;
    using boost::phoenix::begin;
    using boost::phoenix::end;
    using boost::phoenix::empty;
    using boost::phoenix::clear;
    using boost::spirit::ascii::char_;
    using boost::spirit::ascii::alpha;
    using boost::spirit::ascii::alnum;
    using boost::spirit::ascii::digit;
    using boost::spirit::ascii::string;
    using boost::spirit::ascii::no_case;
    using boost::spirit::ascii::blank;

    //
    // Blocks
    //

    // see description in header file as to why we do this...
    constructHeadingParsers();
    constructListParsers();

    static const std::string nl("\n");

    // each of the subBlock rules assumes that there is a container prefix which has been read from
    // input, but is also passed in as _r1 so that it can be checked at the start of successive
    // lines. That's why each of the subBlock parsers is a bit long-winded, it has to recognise the
    // prefix only on successive lines...

    subPara = !strings.blankLine
        >> +(inline_ - eol) [ push_back(_val, _1) ]
        >> eol [ _a = boost::phoenix::ref(nl) ]
        >> *(
            -lit(_r1)
            >> !(bullet | enumerator | strings.blankLine)
            >> +(inline_ - eol) [
                if_(!empty(_a)) [
                    push_back(_val, _a),
                    clear(_a)
                    ],
                push_back(_val, _1) ]
            >> eol [ _a = boost::phoenix::ref(nl) ] );

    subPara.name("subPara");
    //debug(subPara);
    para %= subPara(std::string());

    subBlockquote =
        raw[ strings.nonIndentSpace
             >> char_('>')
             >> -(strings.nonIndentSpace - strings.indent) ] [ _a = _r1 + construct<std::string>(begin(_1), end(_1)) ]
        >> (
            subVerbatim(_a)
            | subBlockquote(_a)
            | orderedSubList(_a)
            | unorderedSubList(_a)
            | subPara(_a)
            ) [ push_back(_val, _1) ]
        >> *(
            -strings.blankLine
            >> lit(_a)
            >> (strings.blankLine
                | ( subVerbatim(_a)
                    | subBlockquote(_a)
                    | orderedSubList(_a)
                    | unorderedSubList(_a)
                    | subPara(_a)
                    ) [ push_back(_val, _1) ]
                )
            )
        ;

    subBlockquote.name("subBlockquote");
    //debug(subBlockquote);
    blockquote %= subBlockquote(std::string());

    subVerbatim =
        !strings.blankLine
        >> strings.indent
        >> raw[
            +(char_ - (eol|eoi)) >> -eol
            ] [ push_back(_val, construct<std::string>(begin(_1), end(_1))) ]
        >> *(
            -( strings.blankLine [ _a = boost::phoenix::ref(nl) ] )
            >> lit(_r1)
            >> strings.indent
            >> raw[
                +(char_ - (eol|eoi)) >> -eol
                ] [
                    if_(!empty(_a)) [
                        push_back(_val, _a),
                        clear(_a)
                        ],
                    push_back(_val, construct<std::string>(begin(_1), end(_1)))
                    ]
            )
        ;

    subVerbatim.name("subVerbatim");
    //debug(subVerbatim);
    verbatim %= subVerbatim(std::string());

    horizontalRule = ( strings.nonIndentSpace
                       >> ( ( char_('*') >> repeat(2, inf)[ *blank >> char_('*')] )
                            | ( char_('-') >> repeat(2, inf)[ *blank >> char_('-')] )
                            | ( char_('_') >> repeat(2, inf)[ *blank >> char_('_')] ) )
                       >> *blank
                       >> eol
                       >> +strings.blankLine ) [ _val = construct<DOM::HorizontalRule>() ];

    //plain = +inline_ >> -blankLine;

    htmlBlock %= raw[html.blockElement];

    block %= blockquote | verbatim | htmlBlock | heading | orderedList | unorderedList | horizontalRule | para;
}

} // namespace Markdown {

#endif // MARKDOWN_BLOCK_PARSER_IPP

ext: ipp

max_stars_repo_name: rockola/spirit-markdown

max_issues_repo_name: rockola/spirit-markdown

max_forks_repo_name: rockola/spirit-markdown


text:
/**
 * Copyright (c) 2017 Melown Technologies SE
 *
 * 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.
 *
 * 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.
 */

#include <boost/utility/in_place_factory.hpp>

#include "requests.hpp"

ShRaster::ShRaster(const GdalWarper::RasterRequest &other
                   , ManagedBuffer &sm, ShRequestBase *owner)
    : sm_(sm), owner_(owner)
    , operation_(other.operation)
    , dataset_(other.dataset.data()
               , other.dataset.size()
               , sm.get_allocator<char>())
    , srs_(other.srs.srs.data()
           , other.srs.srs.size()
           , sm.get_allocator<char>())
    , srsType_(other.srs.type)
    , extents_(other.extents)
    , size_(other.size)
    , resampling_(other.resampling)
    , mask_(sm.get_allocator<char>())
    , nodata_(other.nodata)
    , response_()
{
    if (other.mask) {
        mask_.assign(other.mask->data(), other.mask->size());
    }
}

ShRaster::~ShRaster() {
    if (response_) { sm_.deallocate(response_); }
}

ShRaster::operator GdalWarper::RasterRequest() const {
    return GdalWarper::RasterRequest
        (operation_
         , std::string(dataset_.data(), dataset_.size())
         , geo::SrsDefinition(asString(srs_), srsType_)
         , extents_, size_, resampling_
         , asOptional(mask_)).setNodata(nodata_);
}

cv::Mat* ShRaster::response() {
    auto response(response_);
    response_ = 0;
    return response;
}

void ShRaster::response(bi::interprocess_mutex &mutex, cv::Mat *response)
{
    Lock lock(mutex);
    if (response_) { return; }
    response_ = response;
    owner_->done();
}

namespace {

void copyLayers(boost::optional<StringVector> &dst
                , const boost::optional<geo::heightcoding
                ::Config::LayerNames> &src
                , ManagedBuffer &sm)
{
    if (!src) { return; }

    dst = boost::in_place(sm.get_allocator<String>());
    for (const auto &str : *src) {
        dst->push_back(String(str.data(), str.size()
                              , sm.get_allocator<char>()));
    }
}

void copyLayers(boost::optional<geo::heightcoding::Config::LayerNames> &dst
                , const boost::optional<StringVector> &src)
{
    if (!src) { return; }

    dst = boost::in_place();
    for (const auto &str : *src) {
        dst->emplace_back(str.data(), str.size());
    }
}

} // namespace

ShHeightCodeConfig
::ShHeightCodeConfig(const geo::heightcoding::Config &config
                     , ManagedBuffer &sm)
    : workingSrs_(sm.get_allocator<char>())
    , workingSrsType_()

    , outputSrs_(sm.get_allocator<char>())
    , outputSrsType_()
    , outputAdjustVertical_()

    , clipWorkingExtents_(config.clipWorkingExtents)
    , format_(config.format)
    , formatConfig_(config.formatConfig)
    , mode_(config.mode)
{
    if (config.workingSrs) {
        workingSrs_.assign(config.workingSrs->srs.data()
                           , config.workingSrs->srs.size());
        workingSrsType_ = config.workingSrs->type;
    }

    if (config.outputSrs) {
        outputSrs_.assign(config.outputSrs->srs.srs.data()
                          , config.outputSrs->srs.srs.size());
        outputSrsType_ = config.outputSrs->srs.type;
        outputAdjustVertical_ = config.outputSrs->adjustVertical;
    }

    copyLayers(layers_, config.layers, sm);
    copyLayers(clipLayers_, config.clipLayers, sm);
}

ShHeightCodeConfig::operator geo::heightcoding::Config() const
{
    geo::heightcoding::Config config;
    config.workingSrs = asOptional(workingSrs_, workingSrsType_);
    if (!outputSrs_.empty()) {
        config.outputSrs
            = boost::in_place
            (geo::SrsDefinition(asString(outputSrs_), outputSrsType_)
             , outputAdjustVertical_);
    }

    config.clipWorkingExtents = clipWorkingExtents_;

    copyLayers(config.layers, layers_);
    copyLayers(config.clipLayers, clipLayers_);

    config.format = format_;
    config.formatConfig = formatConfig_;
    config.mode = mode_;

    return config;
}

ShDemDataset::ShDemDataset(const DemDataset &demDataset
                           , ManagedBuffer &sm)
    : dataset(demDataset.dataset.data(), demDataset.dataset.size()
              , sm.get_allocator<char>())
    , geoidGrid(sm.get_allocator<char>())
{
    if (demDataset.geoidGrid) {
        geoidGrid.assign(demDataset.geoidGrid->data()
                         , demDataset.geoidGrid->size());
    }
}

DemDataset ShDemDataset::demDataset() const
{
    return { asString(dataset), asOptional(geoidGrid) };
}

ShHeightCode
::ShHeightCode(const std::string &vectorDs
               , const DemDataset::list &rasterDs
               , const geo::heightcoding::Config &config
               , const boost::optional<std::string> &vectorGeoidGrid
               , const std::vector<std::string> &openOptions
               , const LayerEnhancer::map &layerEnhancers
               , ManagedBuffer &sm, ShRequestBase *owner)
    : sm_(sm), owner_(owner)
    , vectorDs_(vectorDs.data(), vectorDs.size()
                , sm.get_allocator<char>())
    , rasterDs_(sm.get_allocator<ShDemDataset>())
    , config_(config, sm)
    , vectorGeoidGrid_(sm.get_allocator<char>())
    , layerEnhancers_(sm.get_allocator<char>())
    , response_()
{
    // copy strings to shared memory
    for (const auto &dataset : rasterDs) {
        rasterDs_.push_back(ShDemDataset(dataset, sm));
    }

    if (vectorGeoidGrid) {
        vectorGeoidGrid_.assign(vectorGeoidGrid->data()
                                , vectorGeoidGrid->size());
    }

    if (!openOptions.empty()) {
        openOptions_ = boost::in_place(sm.get_allocator<String>());

        for (const auto &str : openOptions) {
            openOptions_->push_back(String(str.data(), str.size()
                                           , sm.get_allocator<char>()));
        }
    }

    // serialize layer enhancers: 4 strings per element
    for (const auto &item : layerEnhancers) {
        layerEnhancers_.emplace_back
            (item.first.data(), item.first.size()
             , sm.get_allocator<char>());
        layerEnhancers_.emplace_back
            (item.second.key.data(), item.second.key.size()
             , sm.get_allocator<char>());
        layerEnhancers_.emplace_back
            (item.second.databasePath.data(), item.second.databasePath.size()
             , sm.get_allocator<char>());
        layerEnhancers_.emplace_back
            (item.second.table.data(), item.second.table.size()
             , sm.get_allocator<char>());
    }
}

ShHeightCode::~ShHeightCode() {
    if (response_) { sm_.deallocate(response_); }
}

std::string ShHeightCode::vectorDs() const
{
    return asString(vectorDs_);
}

DemDataset::list ShHeightCode::rasterDs() const
{
    DemDataset::list ds;
    for (const auto &dataset : rasterDs_) {
        ds.push_back(dataset.demDataset());
    }
    return ds;
}

geo::heightcoding::Config ShHeightCode::config() const {
    return config_;
}

boost::optional<std::string> ShHeightCode::vectorGeoidGrid() const
{
    if (vectorGeoidGrid_.empty()) { return boost::none; }
    return std::string(vectorGeoidGrid_.data(), vectorGeoidGrid_.size());
}

std::vector<std::string> ShHeightCode::openOptions() const
{
    if (!openOptions_) { return {}; }

    std::vector<std::string> openOptions;
    for (const auto &str : *openOptions_) {
        openOptions.emplace_back(str.data(), str.size());
    }
    return openOptions;
}

LayerEnhancer::map ShHeightCode::layerEnhancers() const
{
    LayerEnhancer::map layerEnhancers;

    auto ilayerEnhancers(layerEnhancers_.begin());

    const auto nextString([&]() -> std::string
    {
        const auto &src(*ilayerEnhancers++);
        return std::string(src.data(), src.size());
    });

    for (auto elayerEnhancers(layerEnhancers_.end());
         ilayerEnhancers != elayerEnhancers; )
    {
        auto layer(nextString());
        auto key(nextString());
        auto db(nextString());
        auto table(nextString());
        layerEnhancers.insert
            (LayerEnhancer::map::value_type
             (std::move(layer)
              , LayerEnhancer
              (std::move(key), std::move(db), std::move(table))));
    }

    return layerEnhancers;
}

/** Steals response.
 */
GdalWarper::Heightcoded* ShHeightCode::response() {
    auto response(response_);
    response_ = 0;
    return response;
}

void ShHeightCode::response(bi::interprocess_mutex &mutex
                            , GdalWarper::Heightcoded *response)
{
    Lock lock(mutex);
    if (response_) { return; }
    response_ = response;
    owner_->done();
}

max_stars_repo_name: melowntech/vts-mapproxy

max_issues_repo_name: melowntech/vts-mapproxy

max_forks_repo_name: melowntech/vts-mapproxy


text:
/*
Copyright 2016 Tyler Winters

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

#pragma once

#ifdef __linux__
#define OLD_BOOST // delete when Raspbian has boost update
#endif

#ifndef OLD_BOOST
#include <boost/core/null_deleter.hpp>
#else
#include <boost/utility/empty_deleter.hpp>
#endif
#include <boost/log/core/core.hpp>
#include <boost/log/expressions.hpp>
#include <boost/log/sinks/sync_frontend.hpp>
#include <boost/log/sinks/text_ostream_backend.hpp>
#include <boost/log/sinks/text_file_backend.hpp>
#include <boost/log/sources/global_logger_storage.hpp>
#include <boost/log/sources/severity_logger.hpp>
#include <boost/log/support/date_time.hpp>
#include <boost/log/trivial.hpp>
#include <boost/log/utility/setup.hpp>
#include <boost/log/utility/setup/common_attributes.hpp>

BOOST_LOG_ATTRIBUTE_KEYWORD(timestamp, "TimeStamp", boost::posix_time::ptime)

void logInit()
{
	boost::log::add_common_attributes();
	typedef boost::log::sinks::synchronous_sink<boost::log::sinks::text_ostream_backend> text_sink;
	auto consoleSink = boost::make_shared<text_sink>();
#ifndef OLD_BOOST
	consoleSink->locked_backend()->add_stream(boost::shared_ptr<std::ostream>(&std::clog, boost::null_deleter()));
#else
	consoleSink->locked_backend()->add_stream(boost::shared_ptr<std::ostream>(&std::clog, boost::empty_deleter()));
#endif
	consoleSink->locked_backend()->auto_flush(true);
	boost::log::formatter consoleFormatter = boost::log::expressions::stream << boost::log::expressions::smessage;
	consoleSink->set_formatter(consoleFormatter);
	consoleSink->set_filter(boost::log::trivial::severity >= boost::log::trivial::info);

	boost::log::core::get()->add_sink(consoleSink);

	//TODO: mkdir ./logs
	typedef boost::log::sinks::synchronous_sink<boost::log::sinks::text_file_backend> file_sink;
	auto fileSink = boost::make_shared<file_sink>(
		boost::log::keywords::file_name = "logs/PoloLendingBot_log_%Y%m%d.txt",
		boost::log::keywords::open_mode = std::ios::app,
		boost::log::keywords::rotation_size = 10000000,
		boost::log::keywords::min_free_space = 500000000);
	fileSink->locked_backend()->set_file_collector(boost::log::sinks::file::make_collector(boost::log::keywords::target = "logs", boost::log::keywords::max_size = 100000000));
	fileSink->locked_backend()->scan_for_files();
	fileSink->locked_backend()->auto_flush(true);
	boost::log::formatter fileFormatter = boost::log::expressions::stream
		<< boost::log::expressions::format_date_time(timestamp, "[%Y-%m-%d %H:%M:%S.%f] ")
		<< "[" << boost::log::trivial::severity << "]"
		<< " - " << boost::log::expressions::smessage;
	fileSink->set_formatter(fileFormatter);
	fileSink->set_filter(boost::log::trivial::severity >= boost::log::trivial::info);

	boost::log::core::get()->add_sink(fileSink);

	boost::log::core::get()->set_filter
	(
		boost::log::trivial::severity >= boost::log::trivial::info
	);
}

#define INFO  BOOST_LOG_TRIVIAL(info)
#define WARN  BOOST_LOG_TRIVIAL(warning)
#define ERROR BOOST_LOG_TRIVIAL(error)

max_stars_repo_name: tylawin/PoloniexLendingBot

max_issues_repo_name: tylawin/PoloniexLendingBot

max_forks_repo_name: tylawin/PoloniexLendingBot


text:
#pragma once

#include <sigmaengine/app/plugin.hpp>

#include <sigmaengine/chain/sigmaengine_object_types.hpp>
#include <sigmaengine/protocol/types.hpp>

#include <boost/multi_index/composite_key.hpp>

namespace sigmaengine { namespace dapp_history {
   using namespace std;
   using namespace sigmaengine::chain;
   using namespace boost::multi_index;
   using namespace sigmaengine::protocol;

   struct strcmp_less
   {
      bool operator()(const shared_string& a, const shared_string& b)const
      {
         return less(a.c_str(), b.c_str());
      }

      bool operator()(const shared_string& a, const string& b)const
      {
         return less(a.c_str(), b.c_str());
      }

      bool operator()(const string& a, const shared_string& b)const
      {
         return less(a.c_str(), b.c_str());
      }

   private:
      inline bool less(const char* a, const char* b)const
      {
         return std::strcmp(a, b) < 0;
      }
   };

   struct strcmp_equal
   {
      bool operator()(const shared_string& a, const string& b)
      {
         return a.size() == b.size() || std::strcmp(a.c_str(), b.c_str()) == 0;
      }
   };

   enum dapp_history_by_key_object_type
   {
      dapp_history_object_type               = (DAPP_HISTORY_SPACE_ID << 8),
      nsta602_transfer_history_object_type   = (DAPP_HISTORY_SPACE_ID << 8) + 1
   };

   class dapp_history_object : public object< dapp_history_object_type, dapp_history_object >
   {
      public:
         template< typename Constructor, typename Allocator >
         dapp_history_object( Constructor&& c, allocator< Allocator > a )
         {
            c( *this );
         }

         id_type                 id;

         dapp_name_type          dapp_name;
         uint32_t                sequence = 0;
         uint32_t                all_sequence = 0;
         operation_id_type       op;
   };

   typedef oid< dapp_history_object > dapp_history_id_type;

   class nsta602_transfer_history_object : public object< nsta602_transfer_history_object_type, nsta602_transfer_history_object >
   {
      public:
         template< typename Constructor, typename Allocator >
         nsta602_transfer_history_object( Constructor&& c, allocator< Allocator > a )
         :unique_id(a)
         {
            c( *this );
         }

         id_type                 id;

         dapp_name_type          dapp_name;
         account_name_type       author;
         shared_string           unique_id;

         uint32_t                sequence = 0;
         operation_id_type       op;
   };

   typedef oid< nsta602_transfer_history_object > nsta602_transfer_history_id_type;

   struct by_id;
   struct by_dapp_name;
   struct by_transaction;
   struct by_nsta602;

   typedef multi_index_container<
      dapp_history_object,
      indexed_by<
         ordered_unique< tag< by_id >,
            member< dapp_history_object, dapp_history_id_type, &dapp_history_object::id >
         >,
         ordered_unique< tag< by_dapp_name >,
            composite_key< dapp_history_object,
               member< dapp_history_object, dapp_name_type, &dapp_history_object::dapp_name >,
               member< dapp_history_object, uint32_t, &dapp_history_object::sequence >
            >,
            composite_key_compare< std::less< dapp_name_type >, std::greater< uint32_t > >
         >,
         ordered_unique< tag< by_transaction >,
            composite_key< dapp_history_object,
               member< dapp_history_object, uint32_t, &dapp_history_object::all_sequence >
            >,
            composite_key_compare< std::greater< uint32_t > >
         >
      >,
      allocator< dapp_history_object >
   > dapp_history_index;

   typedef multi_index_container<
      nsta602_transfer_history_object,
      indexed_by<
         ordered_unique< tag< by_id >,
            member< nsta602_transfer_history_object, nsta602_transfer_history_id_type, &nsta602_transfer_history_object::id >
         >,
         ordered_unique< tag< by_nsta602 >,
            composite_key< nsta602_transfer_history_object,
               member< nsta602_transfer_history_object, dapp_name_type, &nsta602_transfer_history_object::dapp_name >,
               member< nsta602_transfer_history_object, account_name_type, &nsta602_transfer_history_object::author >,
               member< nsta602_transfer_history_object, shared_string, &nsta602_transfer_history_object::unique_id >,
               member< nsta602_transfer_history_object, uint32_t, &nsta602_transfer_history_object::sequence >
            >,
            composite_key_compare< std::less< dapp_name_type >, std::less< account_name_type >, strcmp_less, std::greater< uint32_t > >
         >
      >,
      allocator< nsta602_transfer_history_object >
   > nsta602_transfer_history_index;
} } //namespace sigmaengine::dapp_history

FC_REFLECT( sigmaengine::dapp_history::dapp_history_object, (id)(dapp_name)(sequence)(op) )
CHAINBASE_SET_INDEX_TYPE( sigmaengine::dapp_history::dapp_history_object, sigmaengine::dapp_history::dapp_history_index )

FC_REFLECT( sigmaengine::dapp_history::nsta602_transfer_history_object, (id)(dapp_name)( author )( unique_id )(sequence)(op) )
CHAINBASE_SET_INDEX_TYPE( sigmaengine::dapp_history::nsta602_transfer_history_object, sigmaengine::dapp_history::nsta602_transfer_history_index )


text:
//
// unit_test.hpp
// ~~~~~~~~~~~~~
//
// Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//

#ifndef UNIT_TEST_HPP
#define UNIT_TEST_HPP

#include <boost/asio/detail/config.hpp>
#include <iostream>
#include <boost/asio/detail/atomic_count.hpp>

#if defined(__sun)
# include <stdlib.h> // Needed for lrand48.
#endif // defined(__sun)

#if defined(__BORLANDC__)

// Prevent use of intrinsic for strcmp.
# include <cstring>
# undef strcmp

// Suppress error about condition always being true.
# pragma option -w-ccc

#endif // defined(__BORLANDC__)

#if defined(BOOST_ASIO_MSVC)
# pragma warning (disable:4127)
# pragma warning (push)
# pragma warning (disable:4244)
# pragma warning (disable:4702)
#endif // defined(BOOST_ASIO_MSVC)

#if !defined(BOOST_ASIO_TEST_IOSTREAM)
# define BOOST_ASIO_TEST_IOSTREAM std::cerr
#endif // !defined(BOOST_ASIO_TEST_IOSTREAM)

namespace boost {
namespace asio {
namespace detail {

inline const char*& test_name()
{
  static const char* name = 0;
  return name;
}

inline atomic_count& test_errors()
{
  static atomic_count errors(0);
  return errors;
}

inline void begin_test_suite(const char* name)
{
  boost::asio::detail::test_name();
  boost::asio::detail::test_errors();
  BOOST_ASIO_TEST_IOSTREAM << name << " test suite begins" << std::endl;
}

inline int end_test_suite(const char* name)
{
  BOOST_ASIO_TEST_IOSTREAM << name << " test suite ends" << std::endl;
  BOOST_ASIO_TEST_IOSTREAM << "\n*** ";
  long errors = boost::asio::detail::test_errors();
  if (errors == 0)
    BOOST_ASIO_TEST_IOSTREAM << "No errors detected.";
  else if (errors == 1)
    BOOST_ASIO_TEST_IOSTREAM << "1 error detected.";
  else
    BOOST_ASIO_TEST_IOSTREAM << errors << " errors detected." << std::endl;
  BOOST_ASIO_TEST_IOSTREAM << std::endl;
  return errors == 0 ? 0 : 1;
}

template <void (*Test)()>
inline void run_test(const char* name)
{
  test_name() = name;
  long errors_before = boost::asio::detail::test_errors();
  Test();
  if (test_errors() == errors_before)
    BOOST_ASIO_TEST_IOSTREAM << name << " passed" << std::endl;
  else
    BOOST_ASIO_TEST_IOSTREAM << name << " failed" << std::endl;
}

template <void (*)()>
inline void compile_test(const char* name)
{
  BOOST_ASIO_TEST_IOSTREAM << name << " passed" << std::endl;
}

#if defined(BOOST_ASIO_NO_EXCEPTIONS)

template <typename T>
void throw_exception(const T& t)
{
  BOOST_ASIO_TEST_IOSTREAM << "Exception: " << t.what() << std::endl;
  std::abort();
}

#endif // defined(BOOST_ASIO_NO_EXCEPTIONS)

} // namespace detail
} // namespace asio
} // namespace boost

#define BOOST_ASIO_CHECK(expr) \
  do { if (!(expr)) { \
    BOOST_ASIO_TEST_IOSTREAM << __FILE__ << "(" << __LINE__ << "): " \
      << boost::asio::detail::test_name() << ": " \
      << "check '" << #expr << "' failed" << std::endl; \
    ++boost::asio::detail::test_errors(); \
  } } while (0)

#define BOOST_ASIO_CHECK_MESSAGE(expr, msg) \
  do { if (!(expr)) { \
    BOOST_ASIO_TEST_IOSTREAM << __FILE__ << "(" << __LINE__ << "): " \
      << boost::asio::detail::test_name() << ": " \
      << msg << std::endl; \
    ++boost::asio::detail::test_errors(); \
  } } while (0)

#define BOOST_ASIO_WARN_MESSAGE(expr, msg) \
  do { if (!(expr)) { \
    BOOST_ASIO_TEST_IOSTREAM << __FILE__ << "(" << __LINE__ << "): " \
      << boost::asio::detail::test_name() << ": " \
      << msg << std::endl; \
  } } while (0)

#define BOOST_ASIO_ERROR(msg) \
  do { \
    BOOST_ASIO_TEST_IOSTREAM << __FILE__ << "(" << __LINE__ << "): " \
      << boost::asio::detail::test_name() << ": " \
      << msg << std::endl; \
    ++boost::asio::detail::test_errors(); \
  } while (0)

#define BOOST_ASIO_TEST_SUITE(name, tests) \
  int main() \
  { \
    boost::asio::detail::begin_test_suite(name); \
    tests \
    return boost::asio::detail::end_test_suite(name); \
  }

#define BOOST_ASIO_TEST_CASE(test) \
  boost::asio::detail::run_test<&test>(#test);

#define BOOST_ASIO_COMPILE_TEST_CASE(test) \
  boost::asio::detail::compile_test<&test>(#test);

inline void null_test()
{
}

#if defined(__GNUC__) && defined(_AIX)

// AIX needs this symbol defined in asio, even if it doesn't do anything.
int test_main(int, char**)
{
}

#endif // defined(__GNUC__) && defined(_AIX)

#if defined(BOOST_ASIO_MSVC)
# pragma warning (pop)
#endif // defined(BOOST_ASIO_MSVC)

#endif // UNIT_TEST_HPP

max_issues_repo_name: lipper/arangodb

max_forks_repo_name: lipper/arangodb


text:
/**
 * Copyright (C) 2019 Xilinx, Inc
 *
 * Licensed under the Apache License, Version 2.0 (the "License"). You may
 * not use this file except in compliance with the License. A copy of the
 * License is located at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

/*
 * Xilinx Management Proxy Daemon (MPD) for cloud.
 */

#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <unistd.h>
#include <syslog.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <libudev.h>
#include <boost/algorithm/string.hpp>

#include <fstream>
#include <vector>
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <map>
#include <atomic>
#include <cstdlib>
#include <cstring>
#include <functional>
#include <exception>
#include <dlfcn.h>

#include "pciefunc.h"
#include "sw_msg.h"
#include "common.h"
#include "mpd_plugin.h"

enum Hotplug_state {
    MAILBOX_REMOVED,
    MAILBOX_ADDED,
};
static bool quit = false;
static const std::string plugin_path("/opt/xilinx/xrt/lib/libmpd_plugin.so");
static struct mpd_plugin_callbacks plugin_cbs;
static std::map<std::string, std::atomic<bool>> threads_handling;
static std::map<std::string, enum Hotplug_state> state_machine;
static std::map<std::string, std::shared_ptr<Msgq<queue_msg>>> threads_msgq;
udev* mpd_hotplug;
udev_monitor* mpd_hotplug_monitor;

class Mpd : public Common
{
public:
    Mpd(const std::string name, const std::string plugin_path, bool for_user) :
        Common(name, plugin_path, for_user), plugin_init(nullptr), plugin_fini(nullptr)
    {
    }

    ~Mpd()
    {
    }

    void start();
    void run();
    void stop();
    static void mpd_getMsg(size_t index);
    static void mpd_handleMsg(size_t index);
    static int localMsgHandler(const pcieFunc& dev,
        std::unique_ptr<sw_msg>& orig,
        std::unique_ptr<sw_msg>& processed);
    static std::string getIP(std::string host);
    static int connectMsd(const pcieFunc& dev, std::string &ip,
        uint16_t port, int id);
    init_fn plugin_init;
    fini_fn plugin_fini;
    std::map<std::string, std::thread> threads_getMsg;
    std::map<std::string, std::thread> threads_handleMsg;

private:
};

void Mpd::start()
{
    mpd_hotplug = udev_new();
    if (!mpd_hotplug)
        throw std::runtime_error("mpd: can't create udev object");
    mpd_hotplug_monitor = udev_monitor_new_from_netlink(mpd_hotplug, "udev");
    udev_monitor_enable_receiving(mpd_hotplug_monitor);

    if (plugin_handle != nullptr) {
        plugin_init = (init_fn) dlsym(plugin_handle, INIT_FN_NAME);
        plugin_fini = (fini_fn) dlsym(plugin_handle, FINI_FN_NAME);
        if (plugin_init == nullptr || plugin_fini == nullptr) {
            syslog(LOG_ERR, "failed to find init/fini symbols in mpd plugin");
            return;
        }
        int ret = (*plugin_init)(&plugin_cbs);
        if (ret != 0)
            syslog(LOG_ERR, "mpd plugin_init failed: %d", ret);
    }
}

void Mpd::run()
{
    /*
     * Fire up 2 threads for each board - one is to get msg and the other is to
     * handle msg. The reason is, in some cases, handle msg may take a relative
     * long time, eg. downloading a large xclbin, and in this case, one thead
     * implementation makes the next mailbox msg not read out promptly and ends
     * up a tx timeout
     *
     * MPD, running as a daemon, will open mailbox subdevice. As a result, removing
     * the xocl module before mailbox is closed is impossible, this will make
     * the hotplug hang.(hotplug is required in some cases, eg, auzre hotreset,
     * aws device id change, etc. In order to handle that, mpd is to monitor udev
     * events, which hotplug will produce. For each hotplug, a bunch of events will
     * be produced, here we need to monitor mailbox remove and add events.
     * We maintain a state machine for each fpga. After mpd get started, the state is
     * initialized as MAILBOX_ADDED, we create pair of threads for each fpga. Whenever
     * a mailbox remove event is monitored, the state machine changes to MAILBOX_REMOVED,
     * and the pair of threads will exit and the mailbox will be closed. After a
     * mailbox add event is monitored, new pair of threads will be created.
     *
     */
    for (size_t i = 0; i < total; i++) {
        std::string sysfs_name = pcidev::get_dev(i, true)->sysfs_name;
        state_machine[sysfs_name] = MAILBOX_ADDED;
    }

    int udev_fd = udev_monitor_get_fd(mpd_hotplug_monitor);
    do
    {
        if (total == 0)
            syslog(LOG_INFO, "no device found");
        for (size_t i = 0; i < total; i++) {
            std::string sysfs_name = pcidev::get_dev(i, true)->sysfs_name;

            if (state_machine[sysfs_name] != MAILBOX_ADDED)
                continue;
            if (threads_getMsg.find(sysfs_name) != threads_getMsg.end() &&
                threads_handleMsg.find(sysfs_name) != threads_handleMsg.end())
                continue;

            threads_handling[sysfs_name] = true;

            /*
             * create the thread pair for it.
             */
            syslog(LOG_INFO, "create thread pair for %s", sysfs_name.c_str());
            std::shared_ptr<Msgq<queue_msg>> msgq = std::make_shared<Msgq<queue_msg>>();
            threads_msgq[sysfs_name] = msgq;
            threads_getMsg.insert(std::pair<std::string, std::thread>(sysfs_name,
                std::move(std::thread(Mpd::mpd_getMsg, i))));
            threads_handleMsg.insert(std::pair<std::string, std::thread>(sysfs_name,
                std::move(std::thread(Mpd::mpd_handleMsg, i))));
            syslog(LOG_INFO, "%ld pairs of threads running...", threads_getMsg.size());
        }

        std::string sysfs_name = "";
        int ret = waitForMsg(udev_fd, 3);
        if (ret) { //timeout
            continue;
        } else { //udev events
            udev_device* udev_dev = udev_monitor_receive_device(mpd_hotplug_monitor);
            if (!udev_dev)
                continue;
            const char *subsystem = udev_device_get_subsystem(udev_dev);
            if (!subsystem || (strcmp(subsystem, "xrt_user"))) {
                udev_device_unref(udev_dev);
                continue;
            }
            const char *devpath = udev_device_get_devpath(udev_dev);
            if (!devpath) {
                udev_device_unref(udev_dev);
                continue;
            }
            std::string pathStr = devpath;
            size_t pos_e = pathStr.find("mailbox.u");
            if (pos_e == std::string::npos) {
                udev_device_unref(udev_dev);
                continue;
            }
            size_t pos_s = pathStr.substr(0, pos_e - 1).rfind("/");
            sysfs_name = pathStr.substr(pos_s + 1, pos_e - pos_s - 2);

            const char *action = udev_device_get_action(udev_dev);
            if (action && strcmp(action, "remove") == 0) {
                state_machine[sysfs_name] = MAILBOX_REMOVED;
                threads_handling[sysfs_name] = false;
                if (threads_getMsg.find(sysfs_name) != threads_getMsg.end()) {
                    threads_getMsg[sysfs_name].join();
                    threads_getMsg.erase(sysfs_name);
                }
                if (threads_handleMsg.find(sysfs_name) != threads_handleMsg.end()) {
                    threads_handleMsg[sysfs_name].join();
                    threads_handleMsg.erase(sysfs_name);
                }
                syslog(LOG_INFO, "udev: %s %s. Close mailbox", action, devpath);
            } else if (action && strcmp(action, "add") == 0 &&
                state_machine[sysfs_name] == MAILBOX_REMOVED) {
                state_machine[sysfs_name] = MAILBOX_ADDED;
                syslog(LOG_INFO, "udev: %s %s. Open mailbox", action, devpath);
            }
            udev_device_unref(udev_dev);
        }

    } while (!quit);
}

void Mpd::stop()
{
    // Wait for all threads to finish before quit.
    for (auto& t : threads_handleMsg) {
        syslog(LOG_INFO, "%s handleMsg thread exit", t.first.c_str());
        t.second.join();
    }
    for (auto& t : threads_getMsg) {
        syslog(LOG_INFO, "%s getMsg thread exit", t.first.c_str());
        t.second.join();
    }

    if (mpd_hotplug_monitor)
        udev_monitor_unref(mpd_hotplug_monitor);
    if (mpd_hotplug)
        udev_unref(mpd_hotplug);

    if (plugin_fini)
        (*plugin_fini)(plugin_cbs.mpc_cookie);
}

std::string Mpd::getIP(std::string host)
{
    struct hostent *hp = gethostbyname(host.c_str());

    if (hp == NULL)
        return "";

    char dst[INET_ADDRSTRLEN + 1] = { 0 };
    const char *d = inet_ntop(AF_INET, (struct in_addr *)(hp->h_addr),
        dst, sizeof(dst));
    return d;
}

int Mpd::connectMsd(const pcieFunc& dev, std::string &ip, uint16_t port, int id)
{
    int msdfd;
    struct sockaddr_in msdaddr = { 0 };

    msdfd = socket(AF_INET, SOCK_STREAM, 0);
    if (msdfd < 0) {
        dev.log(LOG_ERR, "failed to create socket: %m");
        return -1;
    }

    msdaddr.sin_family = AF_INET;
    msdaddr.sin_addr.s_addr = inet_addr(ip.c_str());
    msdaddr.sin_port = htons(port);
    if (connect(msdfd, (struct sockaddr *)&msdaddr, sizeof(msdaddr)) != 0) {
        dev.log(LOG_ERR, "failed to connect to msd: %m");
        close(msdfd);
        return -1;
    }

    id = htonl(id);
    if (write(msdfd, &id, sizeof(id)) != sizeof(id)) {
        dev.log(LOG_ERR, "failed to send id to msd: %m");
        close(msdfd);
        return -1;
    }

    int ret = 0;
    if (recv(msdfd, &ret, sizeof(ret), MSG_WAITALL) != sizeof(ret) || ret) {
        dev.log(LOG_ERR, "id not recognized by msd");
        close(msdfd);
        return -1;
    }

    dev.log(LOG_INFO, "successfully connected to msd");
    return msdfd;
}

/*
 * Local mailbox msg handler which is used to interpret the msg and handle it.
 * If libmpd_plugin.so is found, which means the users don't need the software
 * mailbox in the mgmt side, instead, users want to inperpret and handle the
 * the mailbox msg from user PF themselves, this local mailbox msg handler is
 * required. A typical use case is, xclbin download, when the users want have
 * their own control.
 */
int Mpd::localMsgHandler(const pcieFunc& dev, std::unique_ptr<sw_msg>& orig,
    std::unique_ptr<sw_msg>& processed)
{
    int ret = 0;
    xcl_mailbox_req *req = reinterpret_cast<xcl_mailbox_req *>(orig->payloadData());
    size_t reqSize;
    if (orig->payloadSize() < sizeof(xcl_mailbox_req)) {
        dev.log(LOG_ERR, "local request dropped, wrong size");
        ret = -EINVAL;
        processed = std::make_unique<sw_msg>(&ret, sizeof(ret), orig->id(),
            XCL_MB_REQ_FLAG_RESPONSE);
        dev.log(LOG_INFO, "mpd daemon: response %d sent ret = %d", req->req, ret);
        return FOR_LOCAL;
    }
    reqSize = orig->payloadSize() - sizeof(xcl_mailbox_req);

    dev.log(LOG_INFO, "mpd daemon: request %d received(reqSize: %d)",
           req->req, reqSize);

    switch (req->req) {
    case XCL_MAILBOX_REQ_LOAD_XCLBIN: {//mandatory for every plugin
        Sw_mb_container c(sizeof(int), orig->id());
        if (plugin_cbs.mb_req.load_xclbin) {
            int *resp = reinterpret_cast<int *>(c.get_payload_buf());
            const axlf *xclbin = reinterpret_cast<axlf *>(req->data);
            c.set_hook(std::bind(plugin_cbs.mb_req.load_xclbin, dev.getIndex(), xclbin, resp));
        }
        processed = c.get_response();
        dev.log(LOG_INFO, "mpd daemon: response %d sent ret = %d", req->req,
             *((int *)(processed->payloadData())));
        break;
    }
    case XCL_MAILBOX_REQ_PEER_DATA: {//optional. aws plugin need to implement this.
        xcl_mailbox_subdev_peer *subdev_req =
            reinterpret_cast<xcl_mailbox_subdev_peer *>(req->data);
        switch (subdev_req->kind) {
        case XCL_ICAP: {
            Sw_mb_container c(sizeof(xcl_pr_region), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_icap_data) {
                xcl_pr_region *resp = reinterpret_cast<xcl_pr_region *>(c.get_payload_buf());
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_icap_data, dev.getIndex(), resp));
            }
            processed = c.get_response();
            break;
        }
        case XCL_SENSOR: {
            Sw_mb_container c(sizeof(xcl_sensor), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_sensor_data) {
                xcl_sensor *resp = reinterpret_cast<xcl_sensor *>(c.get_payload_buf());
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_sensor_data, dev.getIndex(), resp));
            }
            processed = c.get_response();
            break;
        }
        case XCL_BDINFO: {
            Sw_mb_container c(sizeof(xcl_board_info), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_board_info) {
                xcl_board_info *resp = reinterpret_cast<xcl_board_info *>(c.get_payload_buf());
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_board_info, dev.getIndex(), resp));
            }
            processed = c.get_response();
            break;
        }
        case XCL_MIG_ECC: {
            Sw_mb_container c(subdev_req->entries * sizeof(xcl_mig_ecc), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_mig_data) {
                char *resp = c.get_payload_buf();
                size_t actualSz = subdev_req->entries * sizeof(xcl_mig_ecc);
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_mig_data, dev.getIndex(), resp, actualSz));
            }
            processed = c.get_response();
            break;
        }
        case XCL_FIREWALL: {
            Sw_mb_container c(sizeof(xcl_mig_ecc), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_firewall_data) {
                xcl_mig_ecc *resp = reinterpret_cast<xcl_mig_ecc *>(c.get_payload_buf());
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_firewall_data, dev.getIndex(), resp));
            }
            processed = c.get_response();
            break;
        }
        case XCL_DNA: {
            Sw_mb_container c(sizeof(xcl_dna), orig->id());
            if (plugin_cbs.mb_req.peer_data.get_dna_data) {
                xcl_dna *resp = reinterpret_cast<xcl_dna *>(c.get_payload_buf());
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_dna_data, dev.getIndex(), resp));
            }
            processed = c.get_response();
            break;
        }
        case XCL_SUBDEV: {
            Sw_mb_container c(subdev_req->size, orig->id());
            if (plugin_cbs.mb_req.peer_data.get_subdev_data) {
                char *resp = c.get_payload_buf();
                size_t actualSz = subdev_req->size;
                c.set_hook(std::bind(plugin_cbs.mb_req.peer_data.get_subdev_data, dev.getIndex(), resp, actualSz));
            }
            processed = c.get_response();
            break;
        }
        default:
            ret = -ENOTSUP;
            processed = std::make_unique<sw_msg>(&ret, sizeof(ret), orig->id(),
                XCL_MB_REQ_FLAG_RESPONSE);
            break;
        }

        return FOR_LOCAL;
    }
    case XCL_MAILBOX_REQ_USER_PROBE: {//mandary for aws
        Sw_mb_container c(sizeof(xcl_mailbox_conn_resp), orig->id());
        if (plugin_cbs.mb_req.user_probe) {
            xcl_mailbox_conn_resp *resp = reinterpret_cast<xcl_mailbox_conn_resp *>(c.get_payload_buf());
            c.set_hook(std::bind(plugin_cbs.mb_req.user_probe, dev.getIndex(), resp));
        }
        processed = c.get_response();
        break;
    }
    case XCL_MAILBOX_REQ_HOT_RESET: {//optional
        //do reset
        Sw_mb_container c(sizeof(int), orig->id());
        if (plugin_cbs.mb_req.hot_reset) {
            int *resp = reinterpret_cast<int *>(c.get_payload_buf());
            c.set_hook(std::bind(plugin_cbs.mb_req.hot_reset, dev.getIndex(), resp));
        }
        processed = c.get_response();
        dev.log(LOG_INFO, "mpd daemon: response %d sent ret = %d", req->req,
             *((int *)(processed->payloadData())));
        break;
    }
    case XCL_MAILBOX_REQ_RECLOCK: {//optional
        Sw_mb_container c(sizeof(int), orig->id());
        if (plugin_cbs.mb_req.reclock2) {
            int *resp = reinterpret_cast<int *>(c.get_payload_buf());
            struct xclmgmt_ioc_freqscaling *obj =
                reinterpret_cast<struct xclmgmt_ioc_freqscaling *>(req->data);
            c.set_hook(std::bind(plugin_cbs.mb_req.reclock2, dev.getIndex(), obj, resp));
        }
        processed = c.get_response();
        break;
    }
    case XCL_MAILBOX_REQ_PROGRAM_SHELL: {//optional
        Sw_mb_container c(sizeof(int), orig->id());
        if (plugin_cbs.mb_req.program_shell) {
            int *resp = reinterpret_cast<int *>(c.get_payload_buf());
            c.set_hook(std::bind(plugin_cbs.mb_req.program_shell, dev.getIndex(), resp));
        }
        processed = c.get_response();
        break;
    }
    case XCL_MAILBOX_REQ_READ_P2P_BAR_ADDR: {//optional
        Sw_mb_container c(sizeof(int), orig->id());
        if (plugin_cbs.mb_req.read_p2p_bar_addr) {
            const xcl_mailbox_p2p_bar_addr *addr = reinterpret_cast<xcl_mailbox_p2p_bar_addr *>(req->data);
            int *resp = reinterpret_cast<int *>(c.get_payload_buf());
            c.set_hook(std::bind(plugin_cbs.mb_req.read_p2p_bar_addr, dev.getIndex(), addr, resp));
        }
        processed = c.get_response();
        break;
    }
    default:
        processed = std::make_unique<sw_msg>(&ret, sizeof(ret), orig->id(),
            XCL_MB_REQ_FLAG_RESPONSE);
        break;
    }

    return FOR_LOCAL;
}

// Client of MPD getting msg. Will quit on any error from either local mailbox or socket fd.
// No retry is ever conducted.
void Mpd::mpd_getMsg(size_t index)
{
    std::string sysfs_name = pcidev::get_dev(index, true)->sysfs_name;
    std::shared_ptr<Msgq<queue_msg>> msgq = threads_msgq[sysfs_name];
    int msdfd = -1, mbxfd = -1;
    int ret = 0;
    std::string ip;
    msgHandler cb = nullptr;

    pcieFunc dev(index);

    /*
     * If there is user plugin, then we assume the users either don't want to
     * use the communication channel we setup by default, or they even don't
     * want to use the software mailbox at all. In this case, we interpret the
     * mailbox msg and process the msg with the hook function the plugin provides.
     */
    if (plugin_cbs.get_remote_msd_fd) {
        ret = (*plugin_cbs.get_remote_msd_fd)(dev.getIndex(), &msdfd);
        if (ret) {
            dev.log(LOG_ERR, "failed to get remote fd in plugin");
            threads_handling[sysfs_name] = false;
            return;
        }
        cb = Mpd::localMsgHandler;
    } else {
        if (!dev.loadConf()) {
            threads_handling[sysfs_name] = false;
            return;
        }

        ip = getIP(dev.getHost());
        if (ip.empty()) {
            dev.log(LOG_ERR, "Can't find out IP from host: %s", dev.getHost());
            threads_handling[sysfs_name] = false;
            return;
        }

        dev.log(LOG_INFO, "peer msd ip=%s, port=%d, id=0x%x",
            ip.c_str(), dev.getPort(), dev.getId());

        if ((msdfd = connectMsd(dev, ip, dev.getPort(), dev.getId())) < 0) {
            threads_handling[sysfs_name] = false;
            return;
        }
    }

    mbxfd = dev.getMailbox();
    if (mbxfd == -1) {
        threads_handling[sysfs_name] = false;
        return;
    }

   /*
    * Notify software mailbox online
    * This is usefull for aws. Since there is no mgmt, when xocl driver is loaded,
    * and before the mpd daemon is running, sending MAILBOX_REQ_USER_PROBE msg
    * will timeout and get no response, so there is no chance to know the card is
    * ready.
    * With this notification, when the mpd open/close the mailbox instance, a fake
    * MAILBOX_REQ_MGMT_STATE msg is sent to mailbox in xocl, pretending a mgmt is
    * ready, then xocl will send a MAILBOX_REQ_USER_PROBE again. This time, mpd
    * will get and msg and send back a MB_PEER_READY response.
    */
    if (plugin_cbs.mb_notify) {
        ret = (*plugin_cbs.mb_notify)(index, mbxfd, true);
        if (ret)
            dev.log(LOG_ERR, "failed to mark mgmt as online");
    }

    struct queue_msg msg = {
        .localFd = mbxfd,
        .remoteFd = msdfd,
        .cb = cb,
        .data = nullptr,
    };

    int retfd[2];
    for ( ;; ) {
        retfd[0] = retfd[1] = -100;
        ret = waitForMsg(dev, mbxfd, msdfd, 3, retfd);

        if (quit)
            break;
        if (!threads_handling[sysfs_name])
            break;

        if (ret < 0) {
            if (ret == -EAGAIN)
                continue;
            else
                break;
        }

        bool broken = false;
        for (int i = 0; i < 2; i++) {
            if (retfd[i] == mbxfd) {
                msg.type = LOCAL_MSG;
                msg.data = std::move(getLocalMsg(dev, mbxfd));
            } else if (retfd[i] == msdfd) {
                msg.type = REMOTE_MSG;
                msg.data = std::move(getRemoteMsg(dev, msdfd));
            } else
                continue;

            if (msg.data == nullptr) {
                broken = true;
                break;
            }

            msgq->addMsg(msg);
        }

        if (broken)
            break;
    }

    threads_handling[sysfs_name] = false;

    //notify mailbox driver the daemon is offline
    if (plugin_cbs.mb_notify) {
        ret = (*plugin_cbs.mb_notify)(index, mbxfd, false);
        if (ret)
            dev.log(LOG_ERR, "failed to mark mgmt as offline");
    }

    if (msdfd > 0)
        close(msdfd);

    dev.log(LOG_INFO, "mpd_getMsg thread for %s exit!!",
	pcidev::get_dev(index)->sysfs_name.c_str());
}

// Client of MPD handling msg. Will quit on any error from either local mailbox or socket fd.
// No retry is ever conducted.
void Mpd::mpd_handleMsg(size_t index)
{
    pcieFunc dev(index);
    std::string sysfs_name = pcidev::get_dev(index, true)->sysfs_name;
    std::shared_ptr<Msgq<queue_msg>> msgq = threads_msgq[sysfs_name];
    for ( ;; ) {
        struct queue_msg msg;
        if (quit)
                break;
        if (!threads_handling[sysfs_name])
                break;
        int ret = msgq->getMsg(3, msg);
        if (ret) //timeout
            continue;
        if (handleMsg(dev, msg) != 0)
            break;
    }
    threads_handling[sysfs_name] = false;

    dev.log(LOG_INFO, "mpd_handleMsg thread for %s exit!!",
	pcidev::get_dev(index)->sysfs_name.c_str());
}

/*
 * 'kill -15' is sent. or 'crtl c' on the terminal for debug.
 * so far 'kill -9' is not handled.
 */
static void signalHandler(int signum)
{
    if (signum == SIGINT || signum == SIGTERM) {
        syslog(LOG_INFO, "mpd caught signal %d", signum);
        quit = true;
    }
}

int main(void)
{
    signal(SIGINT, signalHandler);
    signal(SIGTERM, signalHandler);
    try {
        Mpd mpd("mpd", plugin_path, true);
        mpd.preStart();
        mpd.start();
        mpd.run();
        mpd.stop();
        mpd.postStop();
    } catch (std::exception& e) {
        syslog(LOG_ERR, "mpd: %s", e.what());
    }

    return 0;
}

max_stars_repo_name: matwey/XRT

max_issues_repo_name: matwey/XRT

max_forks_repo_name: matwey/XRT