text:
#pragma once
#include "MappedSet.hpp"
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/sequenced_index.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/random_access_index.hpp>

namespace axis { namespace foundation { namespace collections {

using axis::hash_value;
template struct boost::hash<axis::String>;

template <class T>
struct hash_string
{
  std::size_t operator () (const axis::AxisString<T>& str) const
  {
    return boost::hash_range(str.begin(), str.end());
  }
  std::size_t operator () (const axis::AxisString<T> *str) const
  {
    return boost::hash_range(str->begin(), str->end());
  }
};

template <class T>
struct string_equal_to
{
  bool operator () (const axis::AxisString<T>& s1, const axis::AxisString<T>& s2) const
  {
    return s1 == s2;
  }
  bool operator () (const axis::AxisString<T> *s1, const axis::AxisString<T>& s2) const
  {
    return (*s1) == (s2);
  }
};

template <class Key, class Value>
class MappedSet<Key, Value>::Pimpl
{
public:
  struct mutable_pair
  {
  public:
    typedef KeyType    first_type;
    typedef ValueType *second_type;

    mutable_pair(void);
    mutable_pair(const first_type& f, const second_type s);

    first_type first;
    mutable second_type second;
  };

  typedef boost::multi_index::random_access<>  numbered_index;
  typedef boost::multi_index::hashed_unique<boost::multi_index::member<mutable_pair,KeyType,&mutable_pair::first>, hash_string<wchar_t>, string_equal_to<wchar_t>> unique_index;
  typedef boost::multi_index::indexed_by<unique_index, numbered_index> index;
  typedef boost::multi_index::multi_index_container<mutable_pair, index> collection;

  collection items;
};

} } } // namespace axis::foundation::collections

max_stars_repo_name: renato-yuzup/axis-fem

max_issues_repo_name: renato-yuzup/axis-fem

max_forks_repo_name: renato-yuzup/axis-fem


text:
#ifndef DI_SYSTEMS_INPUT_INPUT_HPP_
#define DI_SYSTEMS_INPUT_INPUT_HPP_

#include <algorithm>
#include <array>
#include <cstddef>
#include <memory>
#include <stdexcept>
#include <string>
#include <vector>

#include <boost/signals2.hpp>
#include <SDL2/SDL.h>

#include <di/systems/input/clipboard.hpp>
#include <di/systems/input/key.hpp>
#include <di/systems/input/game_controller.hpp>
#include <di/systems/input/game_controller_info.hpp>
#include <di/systems/input/haptic_device.hpp>
#include <di/systems/input/joystick.hpp>
#include <di/systems/input/joystick_info.hpp>
#include <di/systems/input/touch_device.hpp>
#include <di/engine.hpp>
#include <di/system.hpp>

namespace di
{
class input_system : public system
{
public:
  input_system           ()
  {
    if (SDL_InitSubSystem(SDL_INIT_EVENTS | SDL_INIT_HAPTIC | SDL_INIT_GAMECONTROLLER | SDL_INIT_JOYSTICK) != 0)
      throw std::runtime_error("Failed to initialize SDL Events / Game Controller / Joystick subsystems. Error: " + std::string(SDL_GetError()));

    game_controller::set_default_mappings();

    for (auto i = 0; i < SDL_GetNumTouchDevices(); ++i)
      touch_devices_.emplace_back(std::make_unique<touch_device>(SDL_GetTouchDevice(i)));
  }
  input_system           (const input_system&  that) = delete ;
  input_system           (      input_system&& temp) = delete ;
  virtual ~input_system  ()
  {
    SDL_QuitSubSystem(SDL_INIT_EVENTS | SDL_INIT_HAPTIC | SDL_INIT_GAMECONTROLLER | SDL_INIT_JOYSTICK);
  }
  input_system& operator=(const input_system&  that) = delete ;
  input_system& operator=(      input_system&& temp) = delete ;

  template<typename... argument_types>
  joystick*                     create_joystick        (argument_types&&... arguments)
  {
    joysticks_.emplace_back(std::make_unique<joystick>(arguments...));
    return joysticks_.back().get();
  }
  void                          destroy_joystick       (joystick* joystick)
  {
    joysticks_.erase(std::remove_if(
      joysticks_.begin(),
      joysticks_.end  (),
      [&joystick] (const std::unique_ptr<di::joystick>& iteratee)
      {
        return iteratee.get() == joystick;
      }),
      joysticks_.end  ());
  }
  std::vector<joystick*>        joysticks              () const
  {
    std::vector<joystick*> joysticks(joysticks_.size());
    std::transform(
      joysticks_.begin(),
      joysticks_.end  (),
      joysticks .begin(),
      [ ] (const std::unique_ptr<joystick>& joystick)
      {
        return joystick.get();
      });
    return joysticks;
  }

  template<typename... argument_types>
  game_controller*              create_game_controller (argument_types&&... arguments)
  {
    game_controllers_.emplace_back(std::make_unique<game_controller>(arguments...));
    return game_controllers_.back().get();
  }
  void                          destroy_game_controller(game_controller* game_controller)
  {
    game_controllers_.erase(std::remove_if(
      game_controllers_.begin(),
      game_controllers_.end  (),
      [&game_controller] (const std::unique_ptr<di::game_controller>& iteratee)
      {
        return iteratee.get() == game_controller;
      }),
      game_controllers_.end  ());
  }
  std::vector<game_controller*> game_controllers       () const
  {
    std::vector<game_controller*> game_controllers(game_controllers_.size());
    std::transform(
      game_controllers_.begin(),
      game_controllers_.end  (),
      game_controllers .begin(),
      [ ] (const std::unique_ptr<game_controller>& game_controller)
      {
        return game_controller.get();
      });
    return game_controllers;
  }

  template<typename... argument_types>
  haptic_device*                create_haptic_device   (argument_types&&... arguments)
  {
    haptic_devices_.emplace_back(std::make_unique<haptic_device>(arguments...));
    return haptic_devices_.back().get();
  }
  void                          destroy_haptic_device  (haptic_device* haptic_device)
  {
    haptic_devices_.erase(std::remove_if(
      haptic_devices_.begin(),
      haptic_devices_.end  (),
      [&haptic_device] (const std::unique_ptr<di::haptic_device>& iteratee)
      {
        return iteratee.get() == haptic_device;
      }),
      haptic_devices_.end  ());
  }
  std::vector<haptic_device*>   haptic_devices         () const
  {
    std::vector<haptic_device*> haptic_devices(haptic_devices_.size());
    std::transform(
      haptic_devices_.begin(),
      haptic_devices_.end  (),
      haptic_devices .begin(),
      [ ] (const std::unique_ptr<haptic_device>& haptic_device)
      {
        return haptic_device.get();
      });
    return haptic_devices;
  }

  std::vector<touch_device*>    touch_devices          () const
  {
    std::vector<touch_device*> touch_devices(touch_devices_.size());
    std::transform(
      touch_devices_.begin(),
      touch_devices_.end  (),
      touch_devices .begin(),
      [ ] (const std::unique_ptr<touch_device>& touch_device)
      {
        return touch_device.get();
      });
    return touch_devices;
  }

  static bool                   has_events             ()
  {
    return SDL_HasEvents(SDL_FIRSTEVENT, SDL_LASTEVENT) != 0;
  }
  static void                   flush_events           ()
  {
    SDL_FlushEvents(SDL_FIRSTEVENT, SDL_LASTEVENT);
  }
  static bool                   quit_pending           ()
  {
    return SDL_QuitRequested();
  }

  static bool                   text_input_enabled     ()
  {
    return SDL_IsTextInputActive() != 0;
  }
  static void                   set_text_input_enabled (bool enabled)
  {
    enabled ? SDL_StartTextInput() : SDL_StopTextInput();
  }
  static void                   set_text_input_area    (const std::array<std::size_t, 2>& position, const std::array<std::size_t, 2>& size)
  {
    SDL_Rect rectangle {
      static_cast<int>(position[0]),
      static_cast<int>(position[1]),
      static_cast<int>(size    [0]),
      static_cast<int>(size    [1])};
    SDL_SetTextInputRect(&rectangle);
  }

  boost::signals2::signal<void(key)>                                   on_key_press              ;
  boost::signals2::signal<void(key)>                                   on_key_release            ;
  boost::signals2::signal<void(std::string, std::size_t, std::size_t)> on_text_edit              ;
  boost::signals2::signal<void(std::string)>                           on_text_input             ;
  boost::signals2::signal<void()>                                      on_key_layout_change      ;
  boost::signals2::signal<void(std::array<std::int32_t, 2>)>           on_mouse_move             ;
  boost::signals2::signal<void(std::array<std::int32_t, 2>)>           on_mouse_move_delta       ;
  boost::signals2::signal<void(std::size_t)>                           on_mouse_press            ;
  boost::signals2::signal<void(std::size_t)>                           on_mouse_release          ;
  boost::signals2::signal<void(std::array<std::int32_t, 2>)>           on_mouse_wheel            ;
  boost::signals2::signal<void(joystick_info)>                         on_joystick_connect       ;
  boost::signals2::signal<void(game_controller_info)>                  on_game_controller_connect;
  boost::signals2::signal<void(std::string)>                           on_clipboard_change       ;
  boost::signals2::signal<void()>                                      on_quit                   ;

protected:
  void initialize() override
  {
    on_quit.connect(std::bind(&engine::stop, engine_));
  }
  void tick      () override
  {
    std::array<SDL_Event, 128> events;
    int                        count ;
    while (SDL_PumpEvents(), count = SDL_PeepEvents(events.data(), static_cast<int>(events.size()), SDL_GETEVENT, SDL_QUIT   , SDL_QUIT        ), count > 0)
    {
      for (auto i = 0; i < count; ++i)
        on_quit();
    }
    while (SDL_PumpEvents(), count = SDL_PeepEvents(events.data(), static_cast<int>(events.size()), SDL_GETEVENT, SDL_KEYDOWN, SDL_DROPCOMPLETE), count > 0)
    {
      for(auto i = 0; i < count; ++i)
      {
        auto& event = events[i];

        if      (event.type == SDL_KEYDOWN                 ) on_key_press        (key{static_cast<key_code>(event.key.keysym.sym), static_cast<key_modifier>(event.key.keysym.mod), static_cast<scan_code>(event.key.keysym.scancode)});
        else if (event.type == SDL_KEYUP                   ) on_key_release      (key{static_cast<key_code>(event.key.keysym.sym), static_cast<key_modifier>(event.key.keysym.mod), static_cast<scan_code>(event.key.keysym.scancode)});
        else if (event.type == SDL_TEXTEDITING             ) on_text_edit        (std::string(event.edit.text), static_cast<std::size_t>(event.edit.start), static_cast<std::size_t>(event.edit.length));
        else if (event.type == SDL_TEXTINPUT               ) on_text_input       (std::string(event.text.text));
        else if (event.type == SDL_KEYMAPCHANGED           ) on_key_layout_change();

        else if (event.type == SDL_MOUSEMOTION             )
        {
          on_mouse_move      ({static_cast<std::int32_t>(event.motion.x   ), static_cast<std::int32_t>(event.motion.y   )});
          on_mouse_move_delta({static_cast<std::int32_t>(event.motion.xrel), static_cast<std::int32_t>(event.motion.yrel)});
        }
        else if (event.type == SDL_MOUSEBUTTONDOWN         ) on_mouse_press  ( static_cast<std::size_t>(event.button.button));
        else if (event.type == SDL_MOUSEBUTTONUP           ) on_mouse_release( static_cast<std::size_t>(event.button.button));
        else if (event.type == SDL_MOUSEWHEEL              ) on_mouse_wheel  ({static_cast<std::int32_t>(event.wheel.x), static_cast<std::int32_t>(event.wheel.y)});

        else if (event.type == SDL_JOYAXISMOTION           )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jaxis.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_axis_motion(static_cast<std::size_t>(event.jaxis.axis), static_cast<float>(event.jaxis.value) / 32768.0F);
        }
        else if (event.type == SDL_JOYBALLMOTION           )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jball.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_trackball_motion(static_cast<std::size_t>(event.jball.ball), {static_cast<std::int32_t>(event.jball.xrel), static_cast<std::int32_t>(event.jball.yrel)});
        }
        else if (event.type == SDL_JOYHATMOTION            )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jhat.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_hat_motion(static_cast<std::size_t>(event.jhat.hat), static_cast<joystick_hat_state>(event.jhat.value));
        }
        else if (event.type == SDL_JOYBUTTONDOWN           )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jbutton.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_button_press(static_cast<std::size_t>(event.jbutton.button));
        }
        else if (event.type == SDL_JOYBUTTONUP             )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jbutton.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_button_press(static_cast<std::size_t>(event.jbutton.button));
        }
        else if (event.type == SDL_JOYDEVICEADDED          )
        {
          auto joysticks = joystick_infos();
          for (auto& joystick : joysticks)
            if (joystick.index == event.cdevice.which)
              on_joystick_connect(joystick);
        }
        else if (event.type == SDL_JOYDEVICEREMOVED        )
        {
          auto joystick = std::find_if(joysticks_.begin(), joysticks_.end(), [&event] (const std::unique_ptr<di::joystick>& iteratee) { return iteratee->instance_id() == event.jdevice.which; });
          if  (joystick == joysticks_.end()) continue;
          joystick->get()->on_close();
        }

        else if (event.type == SDL_CONTROLLERAXISMOTION    )
        {
          auto game_controller = std::find_if(game_controllers_.begin(), game_controllers_.end(), [&event] (const std::unique_ptr<di::game_controller>& iteratee) { return iteratee->instance_id() == event.caxis.which; });
          if  (game_controller == game_controllers_.end()) continue;
          game_controller->get()->on_axis_motion(static_cast<game_controller_axis>(event.caxis.axis), static_cast<float>(event.caxis.value) / 32768.0F);
        }
        else if (event.type == SDL_CONTROLLERBUTTONDOWN    )
        {
          auto game_controller = std::find_if(game_controllers_.begin(), game_controllers_.end(), [&event] (const std::unique_ptr<di::game_controller>& iteratee) { return iteratee->instance_id() == event.cbutton.which; });
          if  (game_controller == game_controllers_.end()) continue;
          game_controller->get()->on_button_press(static_cast<game_controller_button>(event.cbutton.button));
        }
        else if (event.type == SDL_CONTROLLERBUTTONUP      )
        {
          auto game_controller = std::find_if(game_controllers_.begin(), game_controllers_.end(), [&event] (const std::unique_ptr<di::game_controller>& iteratee) { return iteratee->instance_id() == event.cbutton.which; });
          if  (game_controller == game_controllers_.end()) continue;
          game_controller->get()->on_button_press(static_cast<game_controller_button>(event.cbutton.button));
        }
        else if (event.type == SDL_CONTROLLERDEVICEADDED   )
        {
          auto game_controllers = game_controller_infos();
          for (auto& game_controller : game_controllers)
            if (game_controller.index == event.cdevice.which)
              on_game_controller_connect(game_controller);
        }
        else if (event.type == SDL_CONTROLLERDEVICEREMOVED )
        {
          auto game_controller = std::find_if(game_controllers_.begin(), game_controllers_.end(), [&event] (const std::unique_ptr<di::game_controller>& iteratee) { return iteratee->instance_id() == event.cdevice.which; });
          if  (game_controller == game_controllers_.end()) continue;
          game_controller->get()->on_close();
        }
        else if (event.type == SDL_CONTROLLERDEVICEREMAPPED)
        {
          auto game_controller = std::find_if(game_controllers_.begin(), game_controllers_.end(), [&event] (const std::unique_ptr<di::game_controller>& iteratee) { return iteratee->instance_id() == event.cdevice.which; });
          if  (game_controller == game_controllers_.end()) continue;
          game_controller->get()->on_remap();
        }

        else if (event.type == SDL_FINGERDOWN              )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.tfinger.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->on_finger_press  (finger{static_cast<std::size_t>(event.tfinger.fingerId), {event.tfinger.x , event.tfinger.y }, event.tfinger.pressure});
        }
        else if (event.type == SDL_FINGERUP                )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.tfinger.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->on_finger_release(finger{static_cast<std::size_t>(event.tfinger.fingerId), {event.tfinger.x , event.tfinger.y }, event.tfinger.pressure});
        }
        else if (event.type == SDL_FINGERMOTION            )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.tfinger.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->on_finger_motion (finger{static_cast<std::size_t>(event.tfinger.fingerId), {event.tfinger.dx, event.tfinger.dy}, event.tfinger.pressure});
        }
        else if (event.type == SDL_DOLLARGESTURE           )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.dgesture.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->on_gesture(gesture{event.dgesture.gestureId, {event.dgesture.x, event.dgesture.y}, event.dgesture.error, static_cast<std::size_t>(event.dgesture.numFingers)});
        }
        else if (event.type == SDL_DOLLARRECORD            )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.dgesture.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->record_gesture_callback_(gesture{event.dgesture.gestureId});
        }
        else if (event.type == SDL_MULTIGESTURE            )
        {
          auto touch_device = std::find_if(touch_devices_.begin(), touch_devices_.end(), [&event] (const std::unique_ptr<di::touch_device>& iteratee) { return iteratee->id_ == event.mgesture.touchId; });
          if  (touch_device == touch_devices_.end()) continue;
          touch_device->get()->on_multi_gesture(multi_gesture{{event.mgesture.x, event.mgesture.y}, event.mgesture.dTheta, event.mgesture.dDist, event.mgesture.numFingers});
        }

        else if (event.type == SDL_CLIPBOARDUPDATE         ) on_clipboard_change(clipboard::get());
      }
    }

    joystick::       update_all();
    game_controller::update_all();
  }

  std::vector<std::unique_ptr<joystick>>        joysticks_       ;
  std::vector<std::unique_ptr<game_controller>> game_controllers_;
  std::vector<std::unique_ptr<haptic_device>>   haptic_devices_  ;
  std::vector<std::unique_ptr<touch_device>>    touch_devices_   ;
};
}

#endif

ext: hpp


text:
#define BOOST_TEST_MODULE VectorArithmetics
#include <boost/test/unit_test.hpp>
#include <vexcl/mba.hpp>
#include <vexcl/vector.hpp>
#include <vexcl/element_index.hpp>
#include <vexcl/temporary.hpp>
#include <vexcl/function.hpp>
#include "context_setup.hpp"

template <typename T>
inline std::array<T, 2> make_array(T x,  T y) {
    std::array<T, 2> p = {{x, y}};
    return p;
}

BOOST_AUTO_TEST_CASE(mba)
{
    std::vector< std::array<double,2> > p;

    p.push_back(make_array<double>(0.0, 0.0));
    p.push_back(make_array<double>(0.0, 1.0));
    p.push_back(make_array<double>(1.0, 0.0));
    p.push_back(make_array<double>(1.0, 1.0));
    p.push_back(make_array<double>(0.4, 0.4));
    p.push_back(make_array<double>(0.6, 0.6));

    std::vector<double> v;

    v.push_back( 0.2);
    v.push_back( 0.0);
    v.push_back( 0.0);
    v.push_back(-0.2);
    v.push_back(-1.0);
    v.push_back( 1.0);

    vex::mba<2> cloud(ctx,
            make_array<double>(-0.01, -0.01),
            make_array<double>( 1.01,  1.01),
            p, v, make_array<size_t>(2, 2)
            );

    const size_t n = 11;
    vex::vector<double> x(ctx, n);
    vex::vector<double> z(ctx, n);

    x = 1.0 * vex::element_index() / (n - 1.0);
    z = sin(cloud(x, 1.0 * vex::element_index() / (n - 1.0)));

    BOOST_CHECK_CLOSE(static_cast<double>(z[ 0]), sin( 0.2), 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[ 4]), sin(-1.0), 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[ 6]), sin( 1.0), 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[10]), sin(-0.2), 1e-6);

    auto t = vex::make_temp<1>(1.0 * vex::element_index() / (n - 1.0));

    z = cloud(t, t);

    BOOST_CHECK_CLOSE(static_cast<double>(z[ 0]),  0.2, 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[ 4]), -1.0, 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[ 6]),  1.0, 1e-6);
    BOOST_CHECK_CLOSE(static_cast<double>(z[10]), -0.2, 1e-6);

}

BOOST_AUTO_TEST_SUITE_END()

ext: cpp

max_stars_repo_name: lkusch/Kratos

max_issues_repo_name: lkusch/Kratos

max_forks_repo_name: lkusch/Kratos


text:
#include <boost/parameter.hpp>
#include <iostream>

BOOST_PARAMETER_NAME(graph)
BOOST_PARAMETER_NAME(visitor)
BOOST_PARAMETER_NAME(root_vertex)
BOOST_PARAMETER_NAME(index_map)
BOOST_PARAMETER_NAME(color_map)
#include <boost/graph/depth_first_search.hpp> // for dfs_visitor

BOOST_PARAMETER_FUNCTION(
    (void), depth_first_search, tag

, (required
    (graph, *)
    (visitor, *)
    (root_vertex, *)
    (index_map, *)
    (color_map, *)
  )

)
{
   std::cout << "graph=" << graph << std::endl;
   std::cout << "visitor=" << visitor << std::endl;
   std::cout << "root_vertex=" << root_vertex << std::endl;
   std::cout << "index_map=" << index_map << std::endl;
   std::cout << "color_map=" << color_map << std::endl;
}

int main()
{
    depth_first_search(1, 2, 3, 4, 5);

    depth_first_search(
        "1", '2', _color_map = '5',
        _index_map = "4", _root_vertex = "3");
}


text:
#include <cstdlib>
#include <iostream>
#include <complex>
#define BOOST_UBLAS_NO_ELEMENT_PROXIES
#include <boost/numeric/ublas/matrix.hpp>
#include <boost/numeric/bindings/ublas/matrix.hpp>
#include <boost/numeric/bindings/blas/level3.hpp>
#include <boost/numeric/bindings/lower.hpp>
#include <boost/numeric/bindings/upper.hpp>
#include <boost/numeric/bindings/unit_lower.hpp>
#include <boost/numeric/bindings/unit_upper.hpp>
#include <boost/numeric/bindings/trans.hpp>
#include <boost/numeric/bindings/conj.hpp>
#include <boost/numeric/bindings/left.hpp>
#include <boost/numeric/bindings/right.hpp>
#include "print.hpp"
#include "random.hpp"

namespace ublas=boost::numeric::ublas;
namespace blas=boost::numeric::bindings::blas;

int main(int argc, char *argv[]) {
  {
    typedef std::complex<double> complex;
    typedef ublas::vector<complex> vector;
    typedef ublas::matrix<complex, ublas::column_major> matrix;
    typedef typename vector::size_type size_type;
    rand_normal<complex>::reset();
    size_type n=8;
    matrix A_l(n, n), A_u(n, n), B(n, n);
    for (size_type j=0; j<n; ++j) {
      for (size_type i=0; i<j; ++i) {
    	A_u(i, j)=rand_normal<complex>::get();
	A_l(j, i)=rand_normal<complex>::get();
      }
      A_u(j, j)=rand_normal<complex>::get();
      A_l(j, j)=rand_normal<complex>::get();
      for (size_type i=j+1; i<n; ++i) {
    	A_u(i, j)=0;
	A_l(j, i)=0;
      }
    }
    for (size_type j=0; j<n; ++j)
      for (size_type i=0; i<n; ++i)
	B(j, i)=rand_normal<complex>::get();
    complex alpha=rand_normal<complex>::get();
    {
      matrix B1(alpha*ublas::prod(A_l, B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::lower(A_l), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(A_u, B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::upper(A_u), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::trans(A_l), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::trans(blas::lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::trans(A_u), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::trans(blas::upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::conj(ublas::trans(A_l)), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::conj(blas::lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::conj(ublas::trans(A_u)), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::conj(blas::upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }

    {
      matrix B1(alpha*ublas::prod(B, A_l));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::lower(A_l), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, A_u));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::upper(A_u), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::trans(A_l)));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::trans(blas::lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::trans(A_u)));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::trans(blas::upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::conj(ublas::trans(A_l))));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::conj(blas::lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::conj(ublas::trans(A_u))));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::conj(blas::upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }

    for (size_type i=0; i<n; ++i) {
      A_l(i, i)=1;
      A_u(i, i)=1;
    }

    {
      matrix B1(alpha*ublas::prod(A_l, B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::unit_lower(A_l), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(A_u, B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::unit_upper(A_u), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::trans(A_l), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::trans(blas::unit_lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::trans(A_u), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::trans(blas::unit_upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::conj(ublas::trans(A_l)), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::conj(blas::unit_lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(ublas::conj(ublas::trans(A_u)), B));
      matrix B2(B);
      blas::trmm(blas::left(), alpha, blas::conj(blas::unit_upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (left multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (left multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }

    {
      matrix B1(alpha*ublas::prod(B, A_l));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::unit_lower(A_l), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, A_u));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::unit_upper(A_u), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::trans(A_l)));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::trans(blas::unit_lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::trans(A_u)));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::trans(blas::unit_upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::conj(ublas::trans(A_l))));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::conj(blas::unit_lower(A_l)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_lower):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_lower):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }
    {
      matrix B1(alpha*ublas::prod(B, ublas::conj(ublas::trans(A_u))));
      matrix B2(B);
      blas::trmm(blas::right(), alpha, blas::conj(blas::unit_upper(A_u)), B2);
      std::cout << "testing boost::ublas containers\n"
     		<< "using ublas (right multiply, unit_upper):\n" << print_mat(B1) << '\n'
    		<< "using blas  (right multiply, unit_upper):\n" << print_mat(B2) << '\n'
    		<< '\n';
    }

  }
  return EXIT_SUCCESS;
}

ext: cc


text:
#include <ripple/beast/net/IPEndpoint.h>
#include <ripple/protocol/STLedgerEntry.h>
#include <boost/asio/strand.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl.hpp>
#include <boost/json.hpp>
#include <boost/json/src.hpp>
#include <boost/log/trivial.hpp>
#include <backend/DBHelpers.h>
#include <etl/ETLSource.h>
#include <etl/ReportingETL.h>
#include <thread>

// Create ETL source without grpc endpoint
// Fetch ledger and load initial ledger will fail for this source
// Primarly used in read-only mode, to monitor when ledgers are validated
template <class Derived>
ETLSourceImpl<Derived>::ETLSourceImpl(
    boost::json::object const& config,
    boost::asio::io_context& ioContext,
    std::shared_ptr<BackendInterface> backend,
    std::shared_ptr<SubscriptionManager> subscriptions,
    std::shared_ptr<NetworkValidatedLedgers> networkValidatedLedgers,
    ETLLoadBalancer& balancer)
    : resolver_(boost::asio::make_strand(ioContext))
    , networkValidatedLedgers_(networkValidatedLedgers)
    , backend_(backend)
    , subscriptions_(subscriptions)
    , balancer_(balancer)
    , ioc_(ioContext)
    , timer_(ioContext)
{
    if (config.contains("ip"))
    {
        auto ipJs = config.at("ip").as_string();
        ip_ = {ipJs.c_str(), ipJs.size()};
    }
    if (config.contains("ws_port"))
    {
        auto portjs = config.at("ws_port").as_string();
        wsPort_ = {portjs.c_str(), portjs.size()};
    }
    if (config.contains("grpc_port"))
    {
        auto portjs = config.at("grpc_port").as_string();
        grpcPort_ = {portjs.c_str(), portjs.size()};
        try
        {
            boost::asio::ip::tcp::endpoint endpoint{
                boost::asio::ip::make_address(ip_), std::stoi(grpcPort_)};
            std::stringstream ss;
            ss << endpoint;
            stub_ = org::xrpl::rpc::v1::XRPLedgerAPIService::NewStub(
                grpc::CreateChannel(
                    ss.str(), grpc::InsecureChannelCredentials()));
            BOOST_LOG_TRIVIAL(debug) << "Made stub for remote = " << toString();
        }
        catch (std::exception const& e)
        {
            BOOST_LOG_TRIVIAL(debug)
                << "Exception while creating stub = " << e.what()
                << " . Remote = " << toString();
        }
    }
}

template <class Derived>
void
ETLSourceImpl<Derived>::reconnect(boost::beast::error_code ec)
{
    connected_ = false;
    // These are somewhat normal errors. operation_aborted occurs on shutdown,
    // when the timer is cancelled. connection_refused will occur repeatedly
    std::string err = ec.message();
    // if we cannot connect to the transaction processing process
    if (ec.category() == boost::asio::error::get_ssl_category())
    {
        err = std::string(" (") +
            boost::lexical_cast<std::string>(ERR_GET_LIB(ec.value())) + "," +
            boost::lexical_cast<std::string>(ERR_GET_REASON(ec.value())) + ") ";
        // ERR_PACK /* crypto/err/err.h */
        char buf[128];
        ::ERR_error_string_n(ec.value(), buf, sizeof(buf));
        err += buf;

        std::cout << err << std::endl;
    }

    if (ec != boost::asio::error::operation_aborted &&
        ec != boost::asio::error::connection_refused)
    {
        BOOST_LOG_TRIVIAL(error)
            << __func__ << " : "
            << "error code = " << ec << " - " << toString();
    }
    else
    {
        BOOST_LOG_TRIVIAL(warning)
            << __func__ << " : "
            << "error code = " << ec << " - " << toString();
    }

    // exponentially increasing timeouts, with a max of 30 seconds
    size_t waitTime = std::min(pow(2, numFailures_), 30.0);
    numFailures_++;
    timer_.expires_after(boost::asio::chrono::seconds(waitTime));
    timer_.async_wait([this](auto ec) {
        bool startAgain = (ec != boost::asio::error::operation_aborted);
        BOOST_LOG_TRIVIAL(trace) << __func__ << " async_wait : ec = " << ec;
        derived().close(startAgain);
    });
}

void
PlainETLSource::close(bool startAgain)
{
    timer_.cancel();
    ioc_.post([this, startAgain]() {
        if (closing_)
            return;

        if (derived().ws().is_open())
        {
            // onStop() also calls close(). If the async_close is called twice,
            // an assertion fails. Using closing_ makes sure async_close is only
            // called once
            closing_ = true;
            derived().ws().async_close(
                boost::beast::websocket::close_code::normal,
                [this, startAgain](auto ec) {
                    if (ec)
                    {
                        BOOST_LOG_TRIVIAL(error)
                            << __func__ << " async_close : "
                            << "error code = " << ec << " - " << toString();
                    }
                    closing_ = false;
                    if (startAgain)
                        run();
                });
        }
        else if (startAgain)
        {
            run();
        }
    });
}

void
SslETLSource::close(bool startAgain)
{
    timer_.cancel();
    ioc_.post([this, startAgain]() {
        if (closing_)
            return;

        if (derived().ws().is_open())
        {
            // onStop() also calls close(). If the async_close is called twice,
            // an assertion fails. Using closing_ makes sure async_close is only
            // called once
            closing_ = true;
            derived().ws().async_close(
                boost::beast::websocket::close_code::normal,
                [this, startAgain](auto ec) {
                    if (ec)
                    {
                        BOOST_LOG_TRIVIAL(error)
                            << __func__ << " async_close : "
                            << "error code = " << ec << " - " << toString();
                    }
                    closing_ = false;
                    if (startAgain)
                    {
                        ws_ = std::make_unique<boost::beast::websocket::stream<
                            boost::beast::ssl_stream<
                                boost::beast::tcp_stream>>>(
                            boost::asio::make_strand(ioc_), *sslCtx_);

                        run();
                    }
                });
        }
        else if (startAgain)
        {
            ws_ = std::make_unique<boost::beast::websocket::stream<
                boost::beast::ssl_stream<boost::beast::tcp_stream>>>(
                boost::asio::make_strand(ioc_), *sslCtx_);

            run();
        }
    });
}

template <class Derived>
void
ETLSourceImpl<Derived>::onResolve(
    boost::beast::error_code ec,
    boost::asio::ip::tcp::resolver::results_type results)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    if (ec)
    {
        // try again
        reconnect(ec);
    }
    else
    {
        boost::beast::get_lowest_layer(derived().ws())
            .expires_after(std::chrono::seconds(30));
        boost::beast::get_lowest_layer(derived().ws())
            .async_connect(results, [this](auto ec, auto ep) {
                derived().onConnect(ec, ep);
            });
    }
}

void
PlainETLSource::onConnect(
    boost::beast::error_code ec,
    boost::asio::ip::tcp::resolver::results_type::endpoint_type endpoint)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    if (ec)
    {
        // start over
        reconnect(ec);
    }
    else
    {
        numFailures_ = 0;
        // Turn off timeout on the tcp stream, because websocket stream has it's
        // own timeout system
        boost::beast::get_lowest_layer(derived().ws()).expires_never();

        // Set suggested timeout settings for the websocket
        derived().ws().set_option(
            boost::beast::websocket::stream_base::timeout::suggested(
                boost::beast::role_type::client));

        // Set a decorator to change the User-Agent of the handshake
        derived().ws().set_option(
            boost::beast::websocket::stream_base::decorator(
                [](boost::beast::websocket::request_type& req) {
                    req.set(
                        boost::beast::http::field::user_agent,
                        std::string(BOOST_BEAST_VERSION_STRING) +
                            " clio-client");

                    req.set("X-User", "coro-client");
                }));

        // Update the host_ string. This will provide the value of the
        // Host HTTP header during the WebSocket handshake.
        // See https://tools.ietf.org/html/rfc7230#section-5.4
        auto host = ip_ + ':' + std::to_string(endpoint.port());
        // Perform the websocket handshake
        derived().ws().async_handshake(
            host, "/", [this](auto ec) { onHandshake(ec); });
    }
}

void
SslETLSource::onConnect(
    boost::beast::error_code ec,
    boost::asio::ip::tcp::resolver::results_type::endpoint_type endpoint)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    if (ec)
    {
        // start over
        reconnect(ec);
    }
    else
    {
        numFailures_ = 0;
        // Turn off timeout on the tcp stream, because websocket stream has it's
        // own timeout system
        boost::beast::get_lowest_layer(derived().ws()).expires_never();

        // Set suggested timeout settings for the websocket
        derived().ws().set_option(
            boost::beast::websocket::stream_base::timeout::suggested(
                boost::beast::role_type::client));

        // Set a decorator to change the User-Agent of the handshake
        derived().ws().set_option(
            boost::beast::websocket::stream_base::decorator(
                [](boost::beast::websocket::request_type& req) {
                    req.set(
                        boost::beast::http::field::user_agent,
                        std::string(BOOST_BEAST_VERSION_STRING) +
                            " clio-client");

                    req.set("X-User", "coro-client");
                }));

        // Update the host_ string. This will provide the value of the
        // Host HTTP header during the WebSocket handshake.
        // See https://tools.ietf.org/html/rfc7230#section-5.4
        auto host = ip_ + ':' + std::to_string(endpoint.port());
        // Perform the websocket handshake
        ws().next_layer().async_handshake(
            boost::asio::ssl::stream_base::client,
            [this, endpoint](auto ec) { onSslHandshake(ec, endpoint); });
    }
}

void
SslETLSource::onSslHandshake(
    boost::beast::error_code ec,
    boost::asio::ip::tcp::resolver::results_type::endpoint_type endpoint)
{
    if (ec)
    {
        reconnect(ec);
    }
    else
    {
        // Perform the websocket handshake
        auto host = ip_ + ':' + std::to_string(endpoint.port());
        // Perform the websocket handshake
        ws().async_handshake(host, "/", [this](auto ec) { onHandshake(ec); });
    }
}

template <class Derived>
void
ETLSourceImpl<Derived>::onHandshake(boost::beast::error_code ec)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    if (ec)
    {
        // start over
        reconnect(ec);
    }
    else
    {
        boost::json::object jv{
            {"command", "subscribe"},
            {"streams",
             {"ledger", "manifests", "validations", "transactions_proposed"}}};
        std::string s = boost::json::serialize(jv);
        BOOST_LOG_TRIVIAL(trace) << "Sending subscribe stream message";

        derived().ws().set_option(
            boost::beast::websocket::stream_base::decorator(
                [](boost::beast::websocket::request_type& req) {
                    req.set(
                        boost::beast::http::field::user_agent,
                        std::string(BOOST_BEAST_VERSION_STRING) +
                            " clio-client");

                    req.set("X-User", "coro-client");
                }));

        // Send the message
        derived().ws().async_write(
            boost::asio::buffer(s),
            [this](auto ec, size_t size) { onWrite(ec, size); });
    }
}

template <class Derived>
void
ETLSourceImpl<Derived>::onWrite(
    boost::beast::error_code ec,
    size_t bytesWritten)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    if (ec)
    {
        // start over
        reconnect(ec);
    }
    else
    {
        derived().ws().async_read(
            readBuffer_, [this](auto ec, size_t size) { onRead(ec, size); });
    }
}

template <class Derived>
void
ETLSourceImpl<Derived>::onRead(boost::beast::error_code ec, size_t size)
{
    BOOST_LOG_TRIVIAL(trace)
        << __func__ << " : ec = " << ec << " - " << toString();
    // if error or error reading message, start over
    if (ec)
    {
        reconnect(ec);
    }
    else
    {
        handleMessage();
        boost::beast::flat_buffer buffer;
        swap(readBuffer_, buffer);

        BOOST_LOG_TRIVIAL(trace)
            << __func__ << " : calling async_read - " << toString();
        derived().ws().async_read(
            readBuffer_, [this](auto ec, size_t size) { onRead(ec, size); });
    }
}

template <class Derived>
bool
ETLSourceImpl<Derived>::handleMessage()
{
    BOOST_LOG_TRIVIAL(trace) << __func__ << " : " << toString();

    setLastMsgTime();
    connected_ = true;
    try
    {
        std::string msg{
            static_cast<char const*>(readBuffer_.data().data()),
            readBuffer_.size()};
        BOOST_LOG_TRIVIAL(trace) << __func__ << msg;
        boost::json::value raw = boost::json::parse(msg);
        BOOST_LOG_TRIVIAL(trace) << __func__ << " parsed";
        boost::json::object response = raw.as_object();

        uint32_t ledgerIndex = 0;
        if (response.contains("result"))
        {
            boost::json::object result = response["result"].as_object();
            if (result.contains("ledger_index"))
            {
                ledgerIndex = result["ledger_index"].as_int64();
            }
            if (result.contains("validated_ledgers"))
            {
                boost::json::string const& validatedLedgers =
                    result["validated_ledgers"].as_string();

                setValidatedRange(
                    {validatedLedgers.c_str(), validatedLedgers.size()});
            }
            BOOST_LOG_TRIVIAL(debug)
                << __func__ << " : "
                << "Received a message on ledger "
                << " subscription stream. Message : " << response << " - "
                << toString();
        }
        else if (
            response.contains("type") && response["type"] == "ledgerClosed")
        {
            BOOST_LOG_TRIVIAL(debug)
                << __func__ << " : "
                << "Received a message on ledger "
                << " subscription stream. Message : " << response << " - "
                << toString();
            if (response.contains("ledger_index"))
            {
                ledgerIndex = response["ledger_index"].as_int64();
            }
            if (response.contains("validated_ledgers"))
            {
                boost::json::string const& validatedLedgers =
                    response["validated_ledgers"].as_string();
                setValidatedRange(
                    {validatedLedgers.c_str(), validatedLedgers.size()});
            }
        }
        else
        {
            if (balancer_.shouldPropagateTxnStream(this))
            {
                if (response.contains("transaction"))
                {
                    subscriptions_->forwardProposedTransaction(response);
                }
                else if (
                    response.contains("type") &&
                    response["type"] == "validationReceived")
                {
                    subscriptions_->forwardValidation(response);
                }
                else if (
                    response.contains("type") &&
                    response["type"] == "manifestReceived")
                {
                    subscriptions_->forwardManifest(response);
                }
            }
        }

        if (ledgerIndex != 0)
        {
            BOOST_LOG_TRIVIAL(trace)
                << __func__ << " : "
                << "Pushing ledger sequence = " << ledgerIndex << " - "
                << toString();
            networkValidatedLedgers_->push(ledgerIndex);
        }
        return true;
    }
    catch (std::exception const& e)
    {
        BOOST_LOG_TRIVIAL(error) << "Exception in handleMessage : " << e.what();
        return false;
    }
}

class AsyncCallData
{
    std::unique_ptr<org::xrpl::rpc::v1::GetLedgerDataResponse> cur_;
    std::unique_ptr<org::xrpl::rpc::v1::GetLedgerDataResponse> next_;

    org::xrpl::rpc::v1::GetLedgerDataRequest request_;
    std::unique_ptr<grpc::ClientContext> context_;

    grpc::Status status_;
    unsigned char nextPrefix_;

    std::string lastKey_;

public:
    AsyncCallData(
        uint32_t seq,
        ripple::uint256 const& marker,
        std::optional<ripple::uint256> const& nextMarker)
    {
        request_.mutable_ledger()->set_sequence(seq);
        if (marker.isNonZero())
        {
            request_.set_marker(marker.data(), marker.size());
        }
        request_.set_user("ETL");
        nextPrefix_ = 0x00;
        if (nextMarker)
            nextPrefix_ = nextMarker->data()[0];

        unsigned char prefix = marker.data()[0];

        BOOST_LOG_TRIVIAL(debug)
            << "Setting up AsyncCallData. marker = " << ripple::strHex(marker)
            << " . prefix = " << ripple::strHex(std::string(1, prefix))
            << " . nextPrefix_ = "
            << ripple::strHex(std::string(1, nextPrefix_));

        assert(nextPrefix_ > prefix || nextPrefix_ == 0x00);

        cur_ = std::make_unique<org::xrpl::rpc::v1::GetLedgerDataResponse>();

        next_ = std::make_unique<org::xrpl::rpc::v1::GetLedgerDataResponse>();

        context_ = std::make_unique<grpc::ClientContext>();
    }

    enum class CallStatus { MORE, DONE, ERRORED };
    CallStatus
    process(
        std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub>& stub,
        grpc::CompletionQueue& cq,
        BackendInterface& backend,
        bool abort,
        bool cacheOnly = false)
    {
        BOOST_LOG_TRIVIAL(trace) << "Processing response. "
                                 << "Marker prefix = " << getMarkerPrefix();
        if (abort)
        {
            BOOST_LOG_TRIVIAL(error) << "AsyncCallData aborted";
            return CallStatus::ERRORED;
        }
        if (!status_.ok())
        {
            BOOST_LOG_TRIVIAL(error)
                << "AsyncCallData status_ not ok: "
                << " code = " << status_.error_code()
                << " message = " << status_.error_message();
            return CallStatus::ERRORED;
        }
        if (!next_->is_unlimited())
        {
            BOOST_LOG_TRIVIAL(warning)
                << "AsyncCallData is_unlimited is false. Make sure "
                   "secure_gateway is set correctly at the ETL source";
        }

        std::swap(cur_, next_);

        bool more = true;

        // if no marker returned, we are done
        if (cur_->marker().size() == 0)
            more = false;

        // if returned marker is greater than our end, we are done
        unsigned char prefix = cur_->marker()[0];
        if (nextPrefix_ != 0x00 && prefix >= nextPrefix_)
            more = false;

        // if we are not done, make the next async call
        if (more)
        {
            request_.set_marker(std::move(cur_->marker()));
            call(stub, cq);
        }

        BOOST_LOG_TRIVIAL(trace) << "Writing objects";
        std::vector<Backend::LedgerObject> cacheUpdates;
        cacheUpdates.reserve(cur_->ledger_objects().objects_size());
        for (int i = 0; i < cur_->ledger_objects().objects_size(); ++i)
        {
            auto& obj = *(cur_->mutable_ledger_objects()->mutable_objects(i));
            if (!more && nextPrefix_ != 0x00)
            {
                if (((unsigned char)obj.key()[0]) >= nextPrefix_)
                    continue;
            }
            cacheUpdates.push_back(
                {*ripple::uint256::fromVoidChecked(obj.key()),
                 {obj.mutable_data()->begin(), obj.mutable_data()->end()}});
            if (!cacheOnly)
            {
                if (lastKey_.size())
                    backend.writeSuccessor(
                        std::move(lastKey_),
                        request_.ledger().sequence(),
                        std::string{obj.key()});
                lastKey_ = obj.key();
                backend.writeLedgerObject(
                    std::move(*obj.mutable_key()),
                    request_.ledger().sequence(),
                    std::move(*obj.mutable_data()));
            }
        }
        backend.cache().update(
            cacheUpdates, request_.ledger().sequence(), cacheOnly);
        BOOST_LOG_TRIVIAL(trace) << "Wrote objects";

        return more ? CallStatus::MORE : CallStatus::DONE;
    }

    void
    call(
        std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub>& stub,
        grpc::CompletionQueue& cq)
    {
        context_ = std::make_unique<grpc::ClientContext>();

        std::unique_ptr<grpc::ClientAsyncResponseReader<
            org::xrpl::rpc::v1::GetLedgerDataResponse>>
            rpc(stub->PrepareAsyncGetLedgerData(context_.get(), request_, &cq));

        rpc->StartCall();

        rpc->Finish(next_.get(), &status_, this);
    }

    std::string
    getMarkerPrefix()
    {
        if (next_->marker().size() == 0)
            return "";
        else
            return ripple::strHex(std::string{next_->marker().data()[0]});
    }

    std::string
    getLastKey()
    {
        return lastKey_;
    }
};

template <class Derived>
bool
ETLSourceImpl<Derived>::loadInitialLedger(
    uint32_t sequence,
    uint32_t numMarkers,
    bool cacheOnly)
{
    if (!stub_)
        return false;

    grpc::CompletionQueue cq;

    void* tag;

    bool ok = false;

    std::vector<AsyncCallData> calls;
    auto markers = getMarkers(numMarkers);

    for (size_t i = 0; i < markers.size(); ++i)
    {
        std::optional<ripple::uint256> nextMarker;
        if (i + 1 < markers.size())
            nextMarker = markers[i + 1];
        calls.emplace_back(sequence, markers[i], nextMarker);
    }

    BOOST_LOG_TRIVIAL(debug) << "Starting data download for ledger " << sequence
                             << ". Using source = " << toString();

    for (auto& c : calls)
        c.call(stub_, cq);

    size_t numFinished = 0;
    bool abort = false;
    size_t incr = 500000;
    size_t progress = incr;
    std::vector<std::string> edgeKeys;
    while (numFinished < calls.size() && cq.Next(&tag, &ok))
    {
        assert(tag);

        auto ptr = static_cast<AsyncCallData*>(tag);

        if (!ok)
        {
            BOOST_LOG_TRIVIAL(error) << "loadInitialLedger - ok is false";
            return false;
            // handle cancelled
        }
        else
        {
            BOOST_LOG_TRIVIAL(trace)
                << "Marker prefix = " << ptr->getMarkerPrefix();
            auto result = ptr->process(stub_, cq, *backend_, abort, cacheOnly);
            if (result != AsyncCallData::CallStatus::MORE)
            {
                numFinished++;
                BOOST_LOG_TRIVIAL(debug)
                    << "Finished a marker. "
                    << "Current number of finished = " << numFinished;
                std::string lastKey = ptr->getLastKey();
                if (lastKey.size())
                    edgeKeys.push_back(ptr->getLastKey());
            }
            if (result == AsyncCallData::CallStatus::ERRORED)
            {
                abort = true;
            }
            if (backend_->cache().size() > progress)
            {
                BOOST_LOG_TRIVIAL(info)
                    << "Downloaded " << backend_->cache().size()
                    << " records from rippled";
                progress += incr;
            }
        }
    }
    BOOST_LOG_TRIVIAL(info)
        << __func__ << " - finished loadInitialLedger. cache size = "
        << backend_->cache().size();
    size_t numWrites = 0;
    if (!abort)
    {
        backend_->cache().setFull();
        if (!cacheOnly)
        {
            auto start = std::chrono::system_clock::now();
            for (auto& key : edgeKeys)
            {
                BOOST_LOG_TRIVIAL(debug)
                    << __func__
                    << " writing edge key = " << ripple::strHex(key);
                auto succ = backend_->cache().getSuccessor(
                    *ripple::uint256::fromVoidChecked(key), sequence);
                if (succ)
                    backend_->writeSuccessor(
                        std::move(key), sequence, uint256ToString(succ->key));
            }
            ripple::uint256 prev = Backend::firstKey;
            while (auto cur = backend_->cache().getSuccessor(prev, sequence))
            {
                assert(cur);
                if (prev == Backend::firstKey)
                {
                    backend_->writeSuccessor(
                        uint256ToString(prev),
                        sequence,
                        uint256ToString(cur->key));
                }

                if (isBookDir(cur->key, cur->blob))
                {
                    auto base = getBookBase(cur->key);
                    // make sure the base is not an actual object
                    if (!backend_->cache().get(cur->key, sequence))
                    {
                        auto succ =
                            backend_->cache().getSuccessor(base, sequence);
                        assert(succ);
                        if (succ->key == cur->key)
                        {
                            BOOST_LOG_TRIVIAL(debug)
                                << __func__ << " Writing book successor = "
                                << ripple::strHex(base) << " - "
                                << ripple::strHex(cur->key);

                            backend_->writeSuccessor(
                                uint256ToString(base),
                                sequence,
                                uint256ToString(cur->key));
                        }
                    }
                    ++numWrites;
                }
                prev = std::move(cur->key);
                if (numWrites % 100000 == 0 && numWrites != 0)
                    BOOST_LOG_TRIVIAL(info) << __func__ << " Wrote "
                                            << numWrites << " book successors";
            }

            backend_->writeSuccessor(
                uint256ToString(prev),
                sequence,
                uint256ToString(Backend::lastKey));

            ++numWrites;
            auto end = std::chrono::system_clock::now();
            auto seconds =
                std::chrono::duration_cast<std::chrono::seconds>(end - start)
                    .count();
            BOOST_LOG_TRIVIAL(info)
                << __func__
                << " - Looping through cache and submitting all writes took "
                << seconds
                << " seconds. numWrites = " << std::to_string(numWrites);
        }
    }
    return !abort;
}

template <class Derived>
std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
ETLSourceImpl<Derived>::fetchLedger(
    uint32_t ledgerSequence,
    bool getObjects,
    bool getObjectNeighbors)
{
    org::xrpl::rpc::v1::GetLedgerResponse response;
    if (!stub_)
        return {{grpc::StatusCode::INTERNAL, "No Stub"}, response};

    // ledger header with txns and metadata
    org::xrpl::rpc::v1::GetLedgerRequest request;
    grpc::ClientContext context;
    request.mutable_ledger()->set_sequence(ledgerSequence);
    request.set_transactions(true);
    request.set_expand(true);
    request.set_get_objects(getObjects);
    request.set_get_object_neighbors(getObjectNeighbors);
    request.set_user("ETL");
    grpc::Status status = stub_->GetLedger(&context, request, &response);
    if (status.ok() && !response.is_unlimited())
    {
        BOOST_LOG_TRIVIAL(warning)
            << "ETLSourceImpl::fetchLedger - is_unlimited is "
               "false. Make sure secure_gateway is set "
               "correctly on the ETL source. source = "
            << toString() << " status = " << status.error_message();
    }
    // BOOST_LOG_TRIVIAL(debug)
    //    << __func__ << " Message size = " << response.ByteSizeLong();
    return {status, std::move(response)};
}

static std::unique_ptr<ETLSource>
make_ETLSource(
    boost::json::object const& config,
    boost::asio::io_context& ioContext,
    std::optional<std::reference_wrapper<boost::asio::ssl::context>> sslCtx,
    std::shared_ptr<BackendInterface> backend,
    std::shared_ptr<SubscriptionManager> subscriptions,
    std::shared_ptr<NetworkValidatedLedgers> networkValidatedLedgers,
    ETLLoadBalancer& balancer)
{
    std::unique_ptr<ETLSource> src = nullptr;
    if (sslCtx)
    {
        src = std::make_unique<SslETLSource>(
            config,
            ioContext,
            sslCtx,
            backend,
            subscriptions,
            networkValidatedLedgers,
            balancer);
    }
    else
    {
        src = std::make_unique<PlainETLSource>(
            config,
            ioContext,
            backend,
            subscriptions,
            networkValidatedLedgers,
            balancer);
    }

    src->run();

    return src;
}

ETLLoadBalancer::ETLLoadBalancer(
    boost::json::object const& config,
    boost::asio::io_context& ioContext,
    std::optional<std::reference_wrapper<boost::asio::ssl::context>> sslCtx,
    std::shared_ptr<BackendInterface> backend,
    std::shared_ptr<SubscriptionManager> subscriptions,
    std::shared_ptr<NetworkValidatedLedgers> nwvl)
{
    if (config.contains("num_markers") && config.at("num_markers").is_int64())
    {
        downloadRanges_ = config.at("num_markers").as_int64();

        downloadRanges_ = std::clamp(downloadRanges_, {1}, {256});
    }
    else if (backend->fetchLedgerRange())
    {
        downloadRanges_ = 4;
    }

    for (auto& entry : config.at("etl_sources").as_array())
    {
        std::unique_ptr<ETLSource> source = make_ETLSource(
            entry.as_object(),
            ioContext,
            sslCtx,
            backend,
            subscriptions,
            nwvl,
            *this);

        sources_.push_back(std::move(source));
        BOOST_LOG_TRIVIAL(info) << __func__ << " : added etl source - "
                                << sources_.back()->toString();
    }
}

void
ETLLoadBalancer::loadInitialLedger(uint32_t sequence, bool cacheOnly)
{
    execute(
        [this, &sequence, cacheOnly](auto& source) {
            bool res =
                source->loadInitialLedger(sequence, downloadRanges_, cacheOnly);
            if (!res)
            {
                BOOST_LOG_TRIVIAL(error) << "Failed to download initial ledger."
                                         << " Sequence = " << sequence
                                         << " source = " << source->toString();
            }
            return res;
        },
        sequence);
}

std::optional<org::xrpl::rpc::v1::GetLedgerResponse>
ETLLoadBalancer::fetchLedger(
    uint32_t ledgerSequence,
    bool getObjects,
    bool getObjectNeighbors)
{
    org::xrpl::rpc::v1::GetLedgerResponse response;
    bool success = execute(
        [&response, ledgerSequence, getObjects, getObjectNeighbors](
            auto& source) {
            auto [status, data] = source->fetchLedger(
                ledgerSequence, getObjects, getObjectNeighbors);
            response = std::move(data);
            if (status.ok() && (response.validated() || true))
            {
                BOOST_LOG_TRIVIAL(info)
                    << "Successfully fetched ledger = " << ledgerSequence
                    << " from source = " << source->toString();
                return true;
            }
            else
            {
                BOOST_LOG_TRIVIAL(warning)
                    << "Error getting ledger = " << ledgerSequence
                    << " Reply : " << response.DebugString()
                    << " error_code : " << status.error_code()
                    << " error_msg : " << status.error_message()
                    << " source = " << source->toString();
                return false;
            }
        },
        ledgerSequence);
    if (success)
        return response;
    else
        return {};
}

std::optional<boost::json::object>
ETLLoadBalancer::forwardToRippled(
    boost::json::object const& request,
    std::string const& clientIp,
    boost::asio::yield_context& yield) const
{
    srand((unsigned)time(0));
    auto sourceIdx = rand() % sources_.size();
    auto numAttempts = 0;
    while (numAttempts < sources_.size())
    {
        if (auto res =
                sources_[sourceIdx]->forwardToRippled(request, clientIp, yield))
            return res;

        sourceIdx = (sourceIdx + 1) % sources_.size();
        ++numAttempts;
    }
    return {};
}

template <class Derived>
std::optional<boost::json::object>
ETLSourceImpl<Derived>::forwardToRippled(
    boost::json::object const& request,
    std::string const& clientIp,
    boost::asio::yield_context& yield) const
{
    BOOST_LOG_TRIVIAL(debug) << "Attempting to forward request to tx. "
                             << "request = " << boost::json::serialize(request);

    boost::json::object response;
    if (!connected_)
    {
        BOOST_LOG_TRIVIAL(error)
            << "Attempted to proxy but failed to connect to tx";
        return {};
    }
    namespace beast = boost::beast;          // from <boost/beast.hpp>
    namespace http = beast::http;            // from <boost/beast/http.hpp>
    namespace websocket = beast::websocket;  // from
    namespace net = boost::asio;             // from
    using tcp = boost::asio::ip::tcp;        // from
    try
    {
        boost::beast::error_code ec;
        // These objects perform our I/O
        tcp::resolver resolver{ioc_};

        BOOST_LOG_TRIVIAL(debug) << "Creating websocket";
        auto ws = std::make_unique<websocket::stream<beast::tcp_stream>>(ioc_);

        // Look up the domain name
        auto const results = resolver.async_resolve(ip_, wsPort_, yield[ec]);
        if (ec)
            return {};

        ws->next_layer().expires_after(std::chrono::seconds(3));

        BOOST_LOG_TRIVIAL(debug) << "Connecting websocket";
        // Make the connection on the IP address we get from a lookup
        ws->next_layer().async_connect(results, yield[ec]);
        if (ec)
            return {};

        // Set a decorator to change the User-Agent of the handshake
        // and to tell rippled to charge the client IP for RPC
        // resources. See "secure_gateway" in
        //
        // https://github.com/ripple/rippled/blob/develop/cfg/rippled-example.cfg
        ws->set_option(websocket::stream_base::decorator(
            [&clientIp](websocket::request_type& req) {
                req.set(
                    http::field::user_agent,
                    std::string(BOOST_BEAST_VERSION_STRING) +
                        " websocket-client-coro");
                req.set(http::field::forwarded, "for=" + clientIp);
            }));
        BOOST_LOG_TRIVIAL(debug) << "client ip: " << clientIp;

        BOOST_LOG_TRIVIAL(debug) << "Performing websocket handshake";
        // Perform the websocket handshake
        ws->async_handshake(ip_, "/", yield[ec]);
        if (ec)
            return {};

        BOOST_LOG_TRIVIAL(debug) << "Sending request";
        // Send the message
        ws->async_write(
            net::buffer(boost::json::serialize(request)), yield[ec]);
        if (ec)
            return {};

        beast::flat_buffer buffer;
        ws->async_read(buffer, yield[ec]);
        if (ec)
            return {};

        auto begin = static_cast<char const*>(buffer.data().data());
        auto end = begin + buffer.data().size();
        auto parsed = boost::json::parse(std::string(begin, end));

        if (!parsed.is_object())
        {
            BOOST_LOG_TRIVIAL(error)
                << "Error parsing response: " << std::string{begin, end};
            return {};
        }
        BOOST_LOG_TRIVIAL(debug) << "Successfully forward request";

        response = parsed.as_object();

        response["forwarded"] = true;
        return response;
    }
    catch (std::exception const& e)
    {
        BOOST_LOG_TRIVIAL(error) << "Encountered exception : " << e.what();
        return {};
    }
}

template <class Func>
bool
ETLLoadBalancer::execute(Func f, uint32_t ledgerSequence)
{
    srand((unsigned)time(0));
    auto sourceIdx = rand() % sources_.size();
    auto numAttempts = 0;

    while (true)
    {
        auto& source = sources_[sourceIdx];

        BOOST_LOG_TRIVIAL(debug)
            << __func__ << " : "
            << "Attempting to execute func. ledger sequence = "
            << ledgerSequence << " - source = " << source->toString();
        if (source->hasLedger(ledgerSequence) || true)
        {
            bool res = f(source);
            if (res)
            {
                BOOST_LOG_TRIVIAL(debug)
                    << __func__ << " : "
                    << "Successfully executed func at source = "
                    << source->toString()
                    << " - ledger sequence = " << ledgerSequence;
                break;
            }
            else
            {
                BOOST_LOG_TRIVIAL(warning)
                    << __func__ << " : "
                    << "Failed to execute func at source = "
                    << source->toString()
                    << " - ledger sequence = " << ledgerSequence;
            }
        }
        else
        {
            BOOST_LOG_TRIVIAL(warning)
                << __func__ << " : "
                << "Ledger not present at source = " << source->toString()
                << " - ledger sequence = " << ledgerSequence;
        }
        sourceIdx = (sourceIdx + 1) % sources_.size();
        numAttempts++;
        if (numAttempts % sources_.size() == 0)
        {
            BOOST_LOG_TRIVIAL(error)
                << __func__ << " : "
                << "Error executing function "
                << " - ledger sequence = " << ledgerSequence
                << " - Tried all sources. Sleeping and trying again";
            std::this_thread::sleep_for(std::chrono::seconds(2));
        }
    }
    return true;
}

max_stars_repo_name: crypticrabbit/clio

max_issues_repo_name: crypticrabbit/clio

max_forks_repo_name: crypticrabbit/clio


text:
/*
 * MIT License
 *
 * Copyright (c) 2020 Robert Grupp
 *
 * 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 "xregRayCastOccContourOCL.h"

#include <boost/compute/utility/source.hpp>
#include <boost/compute/types/struct.hpp>

#include "xregAssert.h"
#include "xregExceptionUtils.h"

//////////////////////////////////////////////////////////////////////

struct ContourSurCollArgs
{
  float sur_coll_thresh;
  float occluding_ang_thresh_rad;
};

BOOST_COMPUTE_ADAPT_STRUCT(ContourSurCollArgs, ContourSurCollArgs,
                           (sur_coll_thresh, occluding_ang_thresh_rad))

//////////////////////////////////////////////////////////////////////

namespace  // un-named
{

const char* kRAY_CASTING_OCCLUDING_CONTOUR_OPENCL_SRC = BOOST_COMPUTE_STRINGIZE_SOURCE(

__kernel void xregOccludingContourKernel(const RayCastArgs args,
                                         const ContourSurCollArgs sur_coll_args,
                                         __global const float4* det_pts,
                                         image3d_t vol_tex,
                                         __global const float16* cam_to_itk_phys_xforms,
                                         __global float* dst_contours_buf,
                                         __global const ulong* cam_model_for_proj,
                                         __global const float4* cam_focal_pts)
{
  const ulong idx = get_global_id(0);

  const ulong num_rays = args.num_projs * args.num_det_pts;

  if (idx < num_rays)
  {
    const ulong proj_idx   = idx / args.num_det_pts;
    const ulong det_pt_idx = idx - (proj_idx * args.num_det_pts);
    const ulong cam_idx    = cam_model_for_proj[proj_idx];

    const float4 focal_pt_wrt_cam = cam_focal_pts[cam_idx];

    const sampler_t sampler = CLK_NORMALIZED_COORDS_FALSE | CLK_ADDRESS_CLAMP | CLK_FILTER_LINEAR;

    const float4 tex_coords_off = (float4) (0.5f, 0.5f, 0.5f, 0);

    const float16 xform_cam_to_itk_idx = xregFrm4x4Composition(args.itk_phys_pt_to_itk_idx_xform, cam_to_itk_phys_xforms[proj_idx]);

    const float4 pinhole_wrt_itk_idx = xregFrm4x4XformFloat4Pt(xform_cam_to_itk_idx, focal_pt_wrt_cam);

    const float4 cur_det_pt_wrt_cam = det_pts[(cam_idx * args.num_det_pts) + det_pt_idx];

    const float4 pinhole_to_det_wrt_itk_idx = xregFrm4x4XformFloat4Pt(xform_cam_to_itk_idx, cur_det_pt_wrt_cam) - pinhole_wrt_itk_idx;

    // check intersection with the image/volume boundary
    // these pointer were passed as float4's since boost::compute does not have a float3
    float2 t = xregRayRectIntersect(xregFloat4HmgToFloat3(args.img_aabb_min),
                                    xregFloat4HmgToFloat3(args.img_aabb_max),
                                    xregFloat4HmgToFloat3(pinhole_wrt_itk_idx),
                                    xregFloat4HmgToFloat3(pinhole_to_det_wrt_itk_idx));
    // t.x indicates the first intersection with the volume along the source to detector ray,
    // t.y indicates the exit of the volume along the source to detector ray
    // t.x, t.y are in [0,inf)

    const float4 start_pt_wrt_itk_idx = pinhole_wrt_itk_idx + (t.x * pinhole_to_det_wrt_itk_idx);

    const float pinhole_to_det_len_wrt_itk_idx = xregFloat4HmgNorm(pinhole_to_det_wrt_itk_idx);
    const float intersect_len_wrt_itk_idx = (t.y - t.x) * pinhole_to_det_len_wrt_itk_idx;

    const float4 focal_pt_to_det_wrt_cam = cur_det_pt_wrt_cam - focal_pt_wrt_cam;
    // NOTE: since we are transforming a vector with a scaling transform (points to indices),
    //       the output vector may not have the same norm, thus we take the norm
    const float step_len_wrt_itk_idx = xregFloat4HmgNorm(
                                          xregFrm4x4XformFloat4Vec(xform_cam_to_itk_idx,
                                                                   (focal_pt_to_det_wrt_cam / xregFloat4HmgNorm(focal_pt_to_det_wrt_cam)) * args.step_size));

    const ulong num_steps = (ulong)(intersect_len_wrt_itk_idx / step_len_wrt_itk_idx);

    float4 cur_cont_vol_idx = (float4) (start_pt_wrt_itk_idx.x, start_pt_wrt_itk_idx.y, start_pt_wrt_itk_idx.z, 0);

    cur_cont_vol_idx += tex_coords_off;

    // "/ pinhole_to_det_len_wrt_itk_idx" makes pinhole_to_det_wrt_itk_idx a unit vector
    const float scale_to_step = step_len_wrt_itk_idx / pinhole_to_det_len_wrt_itk_idx;

    const float4 step_vec_wrt_itk_idx = (float4) (pinhole_to_det_wrt_itk_idx.x * scale_to_step, pinhole_to_det_wrt_itk_idx.y * scale_to_step, pinhole_to_det_wrt_itk_idx.z * scale_to_step, 0);

    int is_edge = 0;

    for (ulong step_idx = 0; step_idx <= num_steps; ++step_idx, cur_cont_vol_idx += step_vec_wrt_itk_idx)
    {
      if (read_imagef(vol_tex, sampler, cur_cont_vol_idx).x >= sur_coll_args.sur_coll_thresh)
      {
        // approximate the derivative at this point with finite differencing adjacent indices
        // NOTE: these are wrt image (index) axes.

        float3 grad_vec;

        // Gradient in X Direction:
        grad_vec.x = read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x + 1, cur_cont_vol_idx.y, cur_cont_vol_idx.z, cur_cont_vol_idx.w)).x -
                     read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x - 1, cur_cont_vol_idx.y, cur_cont_vol_idx.z, cur_cont_vol_idx.w)).x;

        // Gradient in Y Direction:
        grad_vec.y = read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x, cur_cont_vol_idx.y + 1, cur_cont_vol_idx.z, cur_cont_vol_idx.w)).x -
                     read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x, cur_cont_vol_idx.y - 1, cur_cont_vol_idx.z, cur_cont_vol_idx.w)).x;

        // Gradient in Z Direction:
        grad_vec.z = read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x, cur_cont_vol_idx.y, cur_cont_vol_idx.z + 1, cur_cont_vol_idx.w)).x -
                     read_imagef(vol_tex, sampler, (float4) (cur_cont_vol_idx.x, cur_cont_vol_idx.y, cur_cont_vol_idx.z - 1, cur_cont_vol_idx.w)).x;

        grad_vec /= xregFloat3Norm(grad_vec);

        // 1.5707963267948966 == pi/2
        is_edge = fabs(acos(xregFloat3Inner(grad_vec,
                     xregFloat3Normalize(xregFloat4HmgToFloat3(pinhole_to_det_wrt_itk_idx))))
                                            - 1.5707963267948966f) < sur_coll_args.occluding_ang_thresh_rad;

        break;
      }
    }

    dst_contours_buf[idx] += is_edge ? 1 : 0;
  }  // if (idx < num_rays)
}

);

//////////////////////////////////////////////////////////////////////

}  // un-named

xreg::RayCasterOccludingContoursOCL::RayCasterOccludingContoursOCL(const boost::compute::device& dev)
  : RayCasterOCL(dev)
{ }

xreg::RayCasterOccludingContoursOCL::RayCasterOccludingContoursOCL(const boost::compute::context& ctx,
                                                                   const boost::compute::command_queue& queue)
  : RayCasterOCL(ctx, queue)
{ }

void xreg::RayCasterOccludingContoursOCL::compute(const size_type vol_idx)
{
  xregASSERT(this->resources_allocated_);

  compute_helper_pre_kernels(vol_idx);

  // setup kernel arguments and launch
  ContourSurCollArgs coll_args = { this->render_thresh(), this->occlusion_angle_thresh_rad() };

  dev_kernel_.set_arg(0, sizeof(ray_cast_kernel_args_), &ray_cast_kernel_args_);

  dev_kernel_.set_arg(1, sizeof(coll_args), &coll_args);

  dev_kernel_.set_arg(2, det_pts_dev_);

  dev_kernel_.set_arg(3, vol_texs_dev_[vol_idx]);

  dev_kernel_.set_arg(4, cam_to_itk_phys_xforms_dev_);

  dev_kernel_.set_arg(5, *proj_pixels_dev_to_use_);

  dev_kernel_.set_arg(6, cam_model_for_proj_dev_);

  dev_kernel_.set_arg(7, focal_pts_dev_);

  std::size_t global_work_size = ray_cast_kernel_args_.num_det_pts * this->num_projs_;

  cmd_queue_.enqueue_nd_range_kernel(dev_kernel_,
                                     1, // dim
                                     0, // null offset -> start at 0
                                     &global_work_size,
                                     0  // passing null lets open CL pick a local size
                                    ).wait();

  compute_helper_post_kernels(vol_idx);
}

void xreg::RayCasterOccludingContoursOCL::allocate_resources()
{
  namespace bc = boost::compute;

  if (this->num_backtracking_steps() > 0)
  {
    xregThrow("For GPU Occluding Contours - Backtracking is NOT supported!");
  }

  if (!this->stop_after_collision())
  {
    xregThrow("For GPU Occluding Contours - Continuing after collision is NOT supported!");
  }

  RayCasterOCL::allocate_resources();

  // Build the surface rendering ray casting program
  std::stringstream ss;
  ss << RayCastBaseOCLStr()
     << boost::compute::type_definition<ContourSurCollArgs>()
     << kRAY_CASTING_OCCLUDING_CONTOUR_OPENCL_SRC;

  bc::program prog = bc::program::create_with_source(ss.str(), ctx_);

  try
  {
    prog.build();
  }
  catch (bc::opencl_error &)
  {
    std::cerr << "OpenCL Kernel Compile Error (RayCasterOccludingContoursOCL):\n"
              << prog.build_log() << std::endl;
    throw;
  }

  dev_kernel_ = prog.create_kernel("xregOccludingContourKernel");
}


text:
/// @file
/// @copyright The code is licensed under the BSD License
///            <http://opensource.org/licenses/BSD-2-Clause>,
///            Copyright (c) 2013-2015 Alexandre Hamez.
/// @author Alexandre Hamez

#include <algorithm> // copy, shuffle, sort, transform
#include <cassert>
#include <deque>
#include <chrono>
#include <set>
#include <sstream>
#include <vector>

#include <boost/filesystem/fstream.hpp>

#include <sdd/order/strategies/force.hh>
#include <sdd/order/strategies/identifiers_per_hierarchy.hh>
#include <sdd/tools/order.hh>

#include "mc/classic/make_order.hh"
#include "mc/shared/export.hh"
#include "mc/shared/step.hh"
#include "support/util/paths.hh"

namespace pnmc { namespace mc { namespace classic {

/*------------------------------------------------------------------------------------------------*/

order_builder
make_hierarchical_order(const std::vector<pn::module>& modules)
{
  struct mk_order_visitor
    : public boost::static_visitor<order_builder>
  {
    using result_type = order_builder;

    // Place: base case of the recursion, there's no more possible nested hierarchies.
    result_type
    operator()(const pn::place& p)
    const
    {
      return p.connected() ? order_builder(p.name) : order_builder();
    }

    // Hierarchy.
    result_type
    operator()(const pn::module_node& m)
    const
    {
      assert(not m.all().empty());
      auto local_ob = order_builder{};
      if (m.all().size() == 1)
      {
        return boost::apply_visitor(*this, m.all().front().variant());
      }
      else
      {
        for (const auto& nested_module : m.all())
        {
          local_ob << boost::apply_visitor(*this, nested_module.variant());
        }
        return order_builder(m.id(), local_ob);
      }
    }
  };

  order_builder current;
  for (const auto& sub : modules)
  {
    current << boost::apply_visitor(mk_order_visitor{}, sub.variant());
  }
  return current;
}

/*------------------------------------------------------------------------------------------------*/

order
make_order(const conf::configuration& conf, statistics& stats, const pn::net& net)
{
  if (net.places().empty() and net.transitions().empty())
  {
    return order{order_builder{}};
  }

  for (const auto& place : net.places_by_insertion())
  {
    if (not place.connected())
    {
      std::cerr << "Warning: place " << place.name << " is not connected";
      if (place.marking > 0)
      {
        std::cerr << " and has an initial marking";
      }
      std::cerr << ".\n";
    }
  }

  using identifier_type = sdd_conf::Identifier;

  // Load a pre-computed order from a JSON file and check if it matches the Petri net.
  // We don't try to apply any heuristic on this order.
  if (conf.order_file)
  {
    boost::filesystem::ifstream file(*conf.order_file);
    if (not file.is_open())
    {
      throw std::runtime_error("Can't open JSON order file " + conf.order_file->string());
    }

    const auto ob = sdd::tools::load_order<sdd_conf>(file);
    if (not ob)
    {
      throw std::runtime_error("Empty JSON order file " + conf.order_file->string());
    }
    auto o = order{*ob};

    // Check if loaded order corresponds to the Petri net.
    auto order_identifiers = std::set<std::string>{};
    o.flat(std::inserter(order_identifiers, order_identifiers.end()));

    auto pn_identifiers = boost::container::flat_set<std::string>{};
    std::transform( begin(net.places()), end(net.places())
                  , std::inserter(pn_identifiers, pn_identifiers.end())
                  , [](const pn::place& p){return p.name;});

    if (net.timed())
    {
      for (const auto& t : net.transitions())
      {
        if (t.timed())
        {
          pn_identifiers.insert(t.name);
        }
      }
    }

    auto diff = std::vector<std::string>{};
    diff.reserve(order_identifiers.size());

    std::set_difference( order_identifiers.cbegin(), order_identifiers.cend()
                       , pn_identifiers.cbegin(), pn_identifiers.cend()
                       , std::back_inserter(diff));

    if (not diff.empty())
    {
      std::stringstream ss{};
      ss << "The following identifiers from " << conf.order_file->string()
         << " don't exist in PN: ";
      std::copy( diff.cbegin(), std::prev(diff.cend())
               , std::ostream_iterator<std::string>(ss, ", "));
      ss << diff.back();
      throw std::runtime_error(ss.str());
    }

    std::set_difference( pn_identifiers.cbegin(), pn_identifiers.cend()
                       , order_identifiers.cbegin(), order_identifiers.cend()
                       , std::back_inserter(diff));

    if (not diff.empty())
    {
      std::stringstream ss{};
      ss << "The following identifiers from PN don't exist in " << conf.order_file->string()
         << ": ";
      std::copy( diff.cbegin(), std::prev(diff.cend())
                , std::ostream_iterator<std::string>(ss, ", "));
      ss << diff.back();
      throw std::runtime_error(ss.str());
    }

    return o;
  }

  // FORCE heuristic.
  auto ob = order_builder{};
  if (conf.order_ordering_force)
  {
    stats.force_duration.emplace();
    auto s = shared::step{"force", &*stats.force_duration};

    // Temporary placeholder for identifiers.
    auto identifiers = std::vector<identifier_type>{};

    // Collect identifiers.
    identifiers.reserve(net.places_by_insertion().size());
    for (const auto& place : net.places_by_insertion())
    {
      if (place.connected())
      {
        identifiers.emplace_back(place.name);
      }
    }

    // Add clocks of timed transitions.
    for (const auto& transition : net.transitions())
    {
      if (transition.timed())
      {
        identifiers.emplace_back(transition.name);
      }
    }

    // The hypergraph that stores connections between the places of the Petri net.
    auto graph = sdd::force::hypergraph<sdd_conf>{identifiers.cbegin(), identifiers.cend()};

    // This container will be used again.
    identifiers.clear();

    // A new connection is created for each transition of the Petri net.
    for (const auto& transition : net.transitions())
    {
      for (const auto& arc : transition.pre)
      {
        identifiers.emplace_back(arc.first);
      }

      for (const auto& arc : transition.post)
      {
        identifiers.emplace_back(arc.first);
      }

      if (transition.timed())
      {
        identifiers.emplace_back(transition.name);
      }

      graph.add_hyperedge(identifiers.cbegin(), identifiers.cend());

      // We use this container again in the next loop.
      identifiers.clear();
    }
    // Apply the FORCE ordering strategy.
    auto force = sdd::force::worker<sdd_conf>(graph, conf.order_reverse);
    ob = force(conf.order_force_iterations);
    stats.force_spans = force.spans();

    // Dump the hypergraph to a DOT file if required by the configuration.
    shared::export_dot(conf, conf::filename::dot_force, graph);
  }
  // Lexical order
  else if (conf.order_lexical)
  {
    auto to_sort = std::vector<identifier_type>{};

    for (const auto& place : net.places())
    {
      to_sort.push_back(place.name);
    }

    if (net.timed())
    {
      for (const auto& transition : net.transitions())
      {
        if (transition.timed())
        {
          to_sort.push_back(transition.name);
        }
      }
    }

    if (conf.order_reverse)
    {
      std::sort(to_sort.begin(), to_sort.end());
    }
    else
    {
      std::sort( to_sort.begin(), to_sort.end()
               , [](const auto& lhs, const auto& rhs)
                   {
                     return lhs > rhs;
                   });
    }

    for (const auto& id : to_sort)
    {
      ob.push(id);
    }
  }
  // Use model's hierarchy, if any.
  else if (not conf.order_flat and not net.modules.empty())
  {
    if (net.timed())
    {
      throw std::runtime_error("Hierarchical order for timed PN is not supported yet");
    }
    ob = make_hierarchical_order(net.root_modules);
  }
  // Flat reversed order.
  else if (conf.order_reverse)
  {
    for ( auto rcit = net.places_by_insertion().rbegin(); rcit != net.places_by_insertion().rend()
        ; ++rcit)
    {
      if (rcit->connected())
      {
        ob.push(rcit->name);
      }
    }

    if (net.timed())
    {
      for (const auto& transition : net.transitions())
      {
        if (transition.timed())
        {
          ob.push(transition.name);
        }
      }
    }
  }
  // Flat order.
  else
  {
    for (const auto& place : net.places_by_insertion())
    {
      if (place.connected())
      {
        ob.push(place.name);
      }
    }

    if (net.timed())
    {
      for (const auto& transition : net.transitions())
      {
        if (transition.timed())
        {
          ob.push(transition.name);
        }
      }
    }
  }

  // Constrain hierarchy with a given number of variables per level.
  if (conf.order_id_per_hierarchy > 0 and not conf.order_flat)
  {
    ob = sdd::identifiers_per_hierarchy<sdd_conf>(conf.order_id_per_hierarchy)(ob);
  }
  return order{ob};
}

/*------------------------------------------------------------------------------------------------*/

}}} // namespace pnmc::mc::classic

max_stars_repo_name: ahamez/pnmc

max_issues_repo_name: ahamez/pnmc

max_forks_repo_name: ahamez/pnmc


text:
/*
Copyright (c) Members of the EGEE Collaboration. 2004.
See http://www.eu-egee.org/partners/ for details on the copyright
holders.

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.
*/
// File: jobdir_reader.cpp
// Author: Francesco Giacomini

// $Id$

#include "glite/wms/common/utilities/jobdir_reader.h"
#include <boost/filesystem/path.hpp>
#include <boost/filesystem/operations.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/bind.hpp>
#include <classad_distribution.h>
#include "glite/wms/common/logger/logger_utils.h"
#include "glite/wms/common/utilities/scope_guard.h"
#include "glite/wms/common/utilities/jobdir.h"

namespace utilities = glite::wms::common::utilities;
namespace fs = boost::filesystem;

namespace glite {
namespace wms {
namespace common {
namespace utilities {

struct JobDirItem::Impl
{
  fs::path item_path;
};

JobDirItem::JobDirItem(fs::path const& p)
  : m_impl(new Impl)
{
  m_impl->item_path = p;
}

std::string
JobDirItem::value() const
{
  std::string result;
  result.reserve(fs::file_size(m_impl->item_path));
  fs::ifstream is(m_impl->item_path);
  getline(is, result, '\0');
  return result;
}

void
JobDirItem::remove_from_input()
{
  fs::remove(m_impl->item_path);
}

struct JobDirReader::Impl
{
  Impl(std::string const& source)
    : jd(fs::path(source, fs::native))
  {
  }
  JobDir jd;
};

JobDirReader::JobDirReader(std::string const& source)
  : m_impl(new Impl(source))
{
}

std::string JobDirReader::name() const
{
  return "jobdir";
}

std::string JobDirReader::source() const
{
  return m_impl->jd.base_dir().native_file_string();
}

JobDirReader::InputItems JobDirReader::read()
{
  InputItems result;

  std::pair<utilities::JobDir::iterator, utilities::JobDir::iterator> p(
    m_impl->jd.new_entries()
  );
  utilities::JobDir::iterator b = p.first;
  utilities::JobDir::iterator const e = p.second;

  for ( ; b != e; ++b) {
    fs::path const& new_file = *b;
    fs::path const old_file = m_impl->jd.set_old(new_file);
    InputItemPtr item(new JobDirItem(old_file));
    result.push_back(item);
  }

  return result;
}

}}}}

max_stars_repo_name: italiangrid/wms

max_issues_repo_name: italiangrid/wms

max_forks_repo_name: italiangrid/wms