clb_swiss_knife.cpp

Go to the documentation of this file.

/**

  Program name: CLBSwissKnife

  General description:
    This program aims to provide a powerful debugging
    tool for all the kind of data sent by a KM3NeT DOM.

    It listens to the specified UDP port and prints out to
    the screen some informations contained in the read
    CLB datagrams.

    Since version 5.0, it operates in two main modes:

    - in UDP mode (available since the first release), the
      program reads data directly from a UDP port. In this
      mode it is allowed to perform several tasks, such as
      read all kind of data, dump data on file and send
      monitoring data to a RoyWeb server;

    - in Ligier mode, data are retrieved from a Ligier
      subscription. Only monitoring data are available and
      no secondary tasks are allowed.

  General options:
    The program is configured via command line parameters.
    UDP mode is default and can be forced with the -u/--udp
    command line switch. Ligier mode can be activated with
    the -l/--ligier switch.

    In the systems where the ncurses libraries are present,
    is possible to use the -g/--gui option to display multiple
    DOMs data via a text gui, in conjunction of the UDP-mode
    "-m/--monitoring" option or in the Ligier mode.
    Press the 'h' key to see the help for GUI mode.
    Arroy keys can be used to select a dom and to scroll
    the list.

    With the "--map" option the user can specify a map file to
    be used to match the DOM IDs of incoming data.
    The DOM map file is an ASCII file reporting assignments
    of DOM IDs or MAC addresses with names. Each assignment
    must be separated by a semicolon. An example of map file
    follows:

       08:00:30:BE:B7:A2 = CLB19;
       817805038 = CLB13;

    By default, CLBSwissKnife will try to load the file

      ~/.doms.csk

    as a map file. The user can write the information about
    its setup to this file and let CLBSK use it without
    specifying any "--map" option.

    Use:
     $ CLBSwissKnife -h
    for a detailed help and
     $ CLBSwissKnife -v
    for version information.

  UDP-mode options:
    The user can specify the port to bind and/or the data
    type:

    - if the port is specified using the "-p" or "--port"
      options the type is automatically determined as
      data comes;

    - if the port is specified using one of the other
      options ("-o/--optical", "-a/--acoustic",
      "-m/--monitoring"), the program is forced to parse
      the incoming data with the specified data type;
      with these options it is not mandatory to specify a
      port number, since a default value is pre-defined.

    With the "-f/--dump-file" option is also possible to
    specify a destination file in which the incoming
    data will be stored for offline analysis.

    In the monitoring-mode, with the "-r/--royweb" option
    it is possible to send the hit rates to a ROyWeb
    server. This option accept 3 parameters separated by
    a colon character (":"), which are:

    - the format of the name of the ROy parameter; the
      format string allows the use of the following
      place-holders:
      + "%d": DOM ID
      + "%n": DOM name as in the map file (see below);
              if not present the DOM ID is used.
      + "%c": channel ID (0-30)
    - the ROyWeb server address
    - the ROyWeb server port

    Es: CLBSwissKnife -m -r HitRates-DOM%d-CH%c:localhost:9999

    A default HR_%n.%c:localhost:9999 is used, otherwise
    all three parameters must be set at once.

    Moreover, the time difference between two consecutive
    monitoring packets is sent to the ROy server with the
    tag "Time_difference". This is useful in order to
    highlight possible packet losses when a single CLB is
    used.

  Ligier-mode options:
    The Ligier mode allows the user to monitor CLB data
    using a Ligier connection. To be activated a "-l/--ligier"
    flag must be present in the command line.

    Ligier endpoint can be set with "-s/--server", while
    the tag can be set with the "-t/--tag". Proper default
    values are available and are listed in the on-line
    help.

  E-mail: carmelo.pellegrino@bo.infn.it
  Date:   27 October 2014

  \author  Carmelo Pellegrino
*/

#include <iostream>
#include <stdint.h>
#include <stddef.h>
#include <string>
#include <stdexcept>
#include <fstream>
#include <ctime>
#include <cerrno>
#include <cstring>
#include <cstdlib>

#include <boost/asio.hpp>
#include <boost/program_options.hpp>

#include <sys/ioctl.h>

#ifdef CURSES_FOUND
# include "gui.hpp"
#endif

#include "structs.hpp"
#include "datatypes.hpp"
#include "version.hpp"
#include "ligier_helper.hpp"
#include "consumers.hpp"
#include "dom_map.hpp"

int get_terminal_width()
{
  static bool const is_a_tty = isatty(1);

  if (!is_a_tty) {
    return 80;
  } else {
    struct winsize sz;

    ioctl(1, TIOCGWINSZ, &sz);
    return sz.ws_col;
  }
}

namespace po = boost::program_options;

// TODO: use the CLBDataGram instead of a buffer

const static size_t buffer_size = 10000;

const static unsigned int default_opto_port = 56015;
const static unsigned int default_acou_port = 56016;
const static unsigned int default_moni_port = 56017;

void handle_signal(
    boost::asio::signal_set& set,
    int& terminal_width,
    boost::system::error_code const& error,
    int signum
);

template<class Consumer>
class LigierDataHandler
{
  typedef LigierDataHandler<Consumer> this_type;

  boost::asio::ip::tcp::socket m_socket;
  boost::asio::deadline_timer m_timer;

  std::string m_tag;
  boost::asio::ip::tcp::endpoint m_endpoint;

  char* m_buffer;
  std::size_t const m_buffer_size;

  Consumer m_consumer;

  void data_handle(
      boost::system::error_code const& error
    , std::size_t size
  )
  {
    if (error && error != boost::asio::error::operation_aborted) {
      m_consumer(error, 0, 0);
      restart();
    } else if (error == boost::asio::error::operation_aborted) {
    } else {
      m_consumer(error, m_buffer + 16, size);

      boost::asio::async_read(
          m_socket
        , boost::asio::buffer(m_buffer, 16)
        , boost::bind(
              &this_type::header_handle
            , this
            , boost::asio::placeholders::error
            , boost::asio::placeholders::bytes_transferred
          )
      );
    }
  }

  void timeout_handle(boost::system::error_code const& error)
  {
    if (!error) {
      m_socket.async_connect(
          m_endpoint
        , boost::bind(
              &this_type::connect_handle
            , this
            , boost::asio::placeholders::error
          )
      );
    }
  }

  void connect_handle(boost::system::error_code const& error)
  {
    if (error && error != boost::asio::error::operation_aborted) {
      m_consumer(error, 0, 0);
      m_timer.expires_from_now(boost::posix_time::seconds(1));

      m_timer.async_wait(
          boost::bind(
              &this_type::timeout_handle
            , this
            , boost::asio::placeholders::error
          )
      );
    } else if (error == boost::asio::error::operation_aborted) {
    } else {
      std::size_t const size = ligier::prepare_tag(m_buffer, m_buffer_size, m_tag);

      boost::asio::async_write(
          m_socket
        , boost::asio::buffer(m_buffer, size)
        , boost::bind(
              &this_type::subscribe_handle
            , this
            , boost::asio::placeholders::error
            , boost::asio::placeholders::bytes_transferred
          )
      );
    }
  }

  void subscribe_handle(
      boost::system::error_code const& error
    , std::size_t bytes_transferred
  ) {
    if (error && error != boost::asio::error::operation_aborted) {
      m_consumer(error, 0, 0);
      restart();
    } else if (error == boost::asio::error::operation_aborted) {
    } else {
      boost::asio::async_read(
          m_socket
        , boost::asio::buffer(m_buffer, 16)
        , boost::bind(
              &this_type::header_handle
            , this
            , boost::asio::placeholders::error
            , boost::asio::placeholders::bytes_transferred
         )
      );
    }
  }

  void header_handle(
      boost::system::error_code const& error
    , std::size_t bytes_transferred
  ) {
    if (error && error != boost::asio::error::operation_aborted) {
      m_consumer(error, 0, 0);
      restart();
    } else if (error == boost::asio::error::operation_aborted) {
    } else {
      std::size_t const size = ligier::read_data_size(m_buffer);

      boost::asio::async_read(
          m_socket
        , boost::asio::buffer(m_buffer + 16, size)
        , boost::bind(
              &this_type::data_handle
            , this
            , boost::asio::placeholders::error
            , boost::asio::placeholders::bytes_transferred
         )
      );
    }
  }

  void restart()
  {
    boost::system::error_code ec;

    m_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
    m_socket.close(ec);

    m_socket.async_connect(
        m_endpoint
      , boost::bind(
            &this_type::connect_handle
          , this
          , boost::asio::placeholders::error
        )
    );
  }

  LigierDataHandler(LigierDataHandler const&);             // non-copyable
  LigierDataHandler& operator =(LigierDataHandler const&); // non-copyable

 public:

  LigierDataHandler(
      boost::asio::io_service& service
    , std::string const& tag
    , boost::asio::ip::tcp::endpoint const& endpoint
    , char* buffer
    , std::size_t buffer_size
    , Consumer const& consumer
  ) : m_socket(service)
    , m_timer(service)
    , m_tag(tag)
    , m_endpoint(endpoint)
    , m_buffer(buffer)
    , m_buffer_size(buffer_size)
    , m_consumer(consumer)
  {
    m_socket.async_connect(
        m_endpoint
      , boost::bind(
            &this_type::connect_handle
          , this
          , boost::asio::placeholders::error
        )
    );
  }
};

template<class Consumer>
class UdpDataHandler
{
  typedef UdpDataHandler<Consumer> this_type;

  boost::asio::ip::udp::socket m_socket;
  char* m_buffer;
  std::size_t m_buffer_size;
  Consumer m_consumer;

  void data_handle(
      boost::system::error_code const& error
    , std::size_t bytes_transferred
  ) {
    if (error) {
      m_consumer(error, 0, 0);
    } else {
      m_consumer(error, m_buffer, bytes_transferred);
    }

    m_socket.async_receive(
        boost::asio::buffer(m_buffer, m_buffer_size)
      , boost::bind(
            &this_type::data_handle
          , this
          , boost::asio::placeholders::error
          , boost::asio::placeholders::bytes_transferred
        )
    );
  }

 public:

  UdpDataHandler(
      boost::asio::io_service& service
    , int port
    , char* buffer
    , std::size_t buffer_size
    , Consumer const& consumer
  ) : m_socket(
          service
        , boost::asio::ip::udp::endpoint(boost::asio::ip::udp::v4(), port)
      )
    , m_buffer(buffer)
    , m_buffer_size(buffer_size)
    , m_consumer(consumer)
  {
    boost::asio::ip::udp::socket::receive_buffer_size option(33554432);
    m_socket.set_option(option);

    m_socket.async_receive(
        boost::asio::buffer(m_buffer, m_buffer_size)
      , boost::bind(
            &this_type::data_handle
          , this
          , boost::asio::placeholders::error
          , boost::asio::placeholders::bytes_transferred
        )
    );
  }
};

inline
boost::asio::ip::tcp::endpoint make_endpoint(
    std::string const& address
) {
  boost::asio::io_service service;
  boost::asio::ip::tcp::resolver resolver(service);

  std::string::size_type const pos = address.find(":");

  std::string const host = pos == std::string::npos ? address : address.substr(0, pos);
  std::string const port = pos == std::string::npos ? "5553"  : address.substr(pos + 1);

  boost::asio::ip::tcp::resolver::query query(
      boost::asio::ip::tcp::v4()
    , host
    , port
  );

  return *resolver.resolve(query);
}

int main(int argc, char* argv[])
{
  bool dump = false;
  uint32_t type = unkn;
  bool uses_roy = false;
  bool use_gui = false;
  bool lm_mode = false;

  std::string dom_names = std::getenv("HOME") + std::string("/.doms.csk");
  dom_map_type dom_names_map;

  std::ofstream output_file;

  unsigned int port = 0;

  std::string filename;
  std::string roy_setup;

  std::string lserver = "localhost:5553";
  std::string tag;
  boost::asio::ip::tcp::endpoint endpoint;

  po::options_description desc("Generic options");
  desc.add_options()
    ("help,h",    "Print this help and exit.")
    ("version,v", "Print the version and exit.")
    (
        "map"
      , po::value<std::string>(&dom_names)->value_name("filename")
      , "File that specifies a conversion from DOM IDs to a human readable name."
    )
#ifdef CURSES_FOUND
    ("gui,g", "Use GUI (avalilable in UDP-monitoring and Ligier modes only).")
#endif
    ("ligier,l", "Set Ligier mode.")
    ("udp,u", "Force UDP mode (default).");

  po::options_description lmdesc("Ligier-mode options");
  lmdesc.add_options()
    (
        "server,s"
      , po::value<std::string>(&lserver)
          ->default_value("localhost:5553")
          ->value_name("server:[port]")
      , "Set the Ligier server endpoint to read data from."
    )
    (
        "tag,t"
      , po::value<std::string>(&tag)
          ->default_value("IO_MONIT")
          ->value_name("tag")
      , "Set the Ligier tag to read data from."
    );

  po::options_description umdesc("UDP-mode options");
  umdesc.add_options()
      ("optical,o",
          po::value<unsigned int>(&port)
              ->implicit_value(default_opto_port)
              ->value_name("port"),
          "Set the expected data type to optical. If a port is not \
provided the default one is used (56015).")
      ("acoustic,a",
          po::value<unsigned int>(&port)
              ->implicit_value(default_acou_port)
              ->value_name("port"),
          "Set the expected data type to acoustic. If a port is not \
provided the default one is used (56016).")
      ("monitoring,m",
          po::value<unsigned int>(&port)
              ->implicit_value(default_moni_port)
              ->value_name("port"),
          "Set the expected data type to monitoring. If a port is \
not provided the default one is used (56017).")
      ("port,p",
          po::value<unsigned int>(&port)->value_name("port"),
          "Listen on the specified port, automatically determining the \
data type.")
      ("dumpfile,f",
          po::value<std::string>(&filename)->value_name("filename"),
          "Dumps the acquired data to a file.")
      ("royweb,r",
          po::value<std::string>(&roy_setup)
              ->implicit_value("HR_%n.%c:localhost:9999")
              ->value_name("format:host:port"),
          "Send the monitoring hit rates to the specified ROyWeb \
server. The syntax is format:server_ip:server_port. \
The format parameter can contain the following placeholders: %d for \
DOM ID, %n for DOM name (if available, otherwise the DOM ID is used) \
and %c for channel ID.");

  std::string templ;
  std::string roy_server;
  int roy_port = 0;

  try {
    po::options_description global;
    global.add(desc).add(lmdesc).add(umdesc);

    po::variables_map vm;
    po::store(
        po::command_line_parser(argc, argv).options(global).run()
      , vm
    );

    if (vm.count("help")) {
      std::cout
        << desc   << '\n'
        << umdesc << '\n'
        << lmdesc << '\n';
      return EXIT_SUCCESS;
    }

    if (vm.count("version")) {
      std::cout << clbsk::version::v() << std::endl;
      return EXIT_SUCCESS;
    }

    po::notify(vm);

#ifdef CURSES_FOUND
    use_gui = vm.count("gui");
#endif

    lm_mode = vm.count("ligier");

    if (lm_mode && vm.count("udp")) {
      throw std::runtime_error("You can specify only one mode. Both provided.");
    }

    if (vm.count("map")) {
      std::ifstream map(dom_names.c_str());

      if (! map) {
        throw std::runtime_error(
            std::string("Error reading map file: ")
            + std::strerror(errno)
        );
      }

      dom_names_map = load_dom_map(map);
    } else {
      std::ifstream map(dom_names.c_str());

      dom_names_map = load_dom_map(map);
    }

    if (lm_mode) {
      po::variables_map vm;
      po::options_description options;
      options.add(lmdesc).add(desc);

      po::store(
          po::command_line_parser(argc, argv).options(options).run()
        , vm
      );

      po::notify(vm);

      endpoint = make_endpoint(lserver);

      type = tmch;
    } else {
      po::variables_map vm;
      po::options_description options;
      options.add(desc).add(umdesc);

      po::store(
          po::command_line_parser(argc, argv).options(options).run()
        , vm
      );

      po::notify(vm);

      unsigned int counter = 0;
      if (vm.count("port")) {
        type = unkn;
        ++counter;
      }
      if (vm.count("optical")) {
        type = ttdc;
        ++counter;
      }
      if (vm.count("acoustic")) {
        type = taes;
        ++counter;
      }
      if (vm.count("monitoring")) {
        type = tmch;
        ++counter;
      }

      if (counter != 1) {
        if (counter) {
          throw std::runtime_error("More than one port provided.");
        } else {
          throw std::runtime_error("You must specify at least one port.");
        }
      }

      if (use_gui && type != tmch) {
        throw std::runtime_error("GUI is only available in monitoring mode");
      }

      if (vm.count("royweb")) {
        if (type != tmch) {
          throw std::runtime_error("you can use ROyWeb only with the"
          " monitoring channel");
        }

        uses_roy = true;
        std::replace(roy_setup.begin(), roy_setup.end(), ':', ' ');
        std::istringstream ss(roy_setup);
        int param_count = 0;
        if (ss >> templ) {
          ++param_count;
        }
        if (ss >> roy_server) {
          ++param_count;
        }
        if (ss >> roy_port) {
          ++param_count;
        }

        if (param_count != 3) {
          throw std::runtime_error("you must specify all the parameters"
          " or accept all the default one to use with ROyWeb.");
        }
      }

      if (vm.count("dumpfile")) {
        output_file.open(filename.c_str(), std::ios::binary);
        if (! output_file) {
          throw std::runtime_error(
              "Unable to open file "
              + filename
              + " for writing: "
              + strerror(errno)
          );
        }
        dump = true;
      }
    }
  } catch (const po::error& e) {
    std::cerr
      << "CLBSwissKnife: Error: " << e.what() << '\n'
      << desc   << '\n'
      << umdesc << '\n'
      << lmdesc << '\n';

    return EXIT_FAILURE;
  } catch (const std::runtime_error& e) {
    std::cerr << "CLBSwissKnife: Error: " << e.what() << std::endl;
    return EXIT_FAILURE;
  }

  boost::asio::io_service io_service;
  char buffer[buffer_size] __attribute__((aligned(8)));

  // Initialise the terminal width
  int terminal_width = get_terminal_width();

  if (lm_mode) { // Ligier
    if (use_gui) { // Ligier-GUI
  #ifdef CURSES_FOUND
      ChList stats;

      boost::asio::deadline_timer timer(io_service, boost::posix_time::seconds(1));

      boost::asio::posix::stream_descriptor keyb(io_service, ::dup(STDIN_FILENO));

      boost::asio::signal_set window_resize(io_service, SIGWINCH);

      char key = 0;
      gui::ScreenInitialiser init;

      gui::Screen screen(io_service, lserver, stats);

      keyb.async_read_some(
          boost::asio::buffer(&key, sizeof(key)),
          boost::bind(
              async_cin,
              boost::ref(keyb),
              boost::ref(key),
              boost::ref(screen),
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred
          )
      );

      timer.async_wait(
          boost::bind(
              async_commit,
              boost::ref(timer),
              boost::ref(screen),
              boost::asio::placeholders::error
          )
      );

      LigierDataHandler<clbsk::consumer::Gui> dh(
          io_service
        , tag
        , endpoint
        , buffer
        , buffer_size
        , clbsk::consumer::Gui(terminal_width, dom_names_map, stats, screen)
      );

      window_resize.async_wait(
          boost::bind(
              async_win_resize,
              boost::ref(window_resize),
              boost::ref(screen),
              boost::asio::placeholders::error,
              boost::asio::placeholders::signal_number
          )
      );

      io_service.run();
  #endif
    } else { // Ligier-Scrollback
      std::cout << "Listening from: " << endpoint << ':' << tag << '\n';
      boost::asio::signal_set signals_handler(io_service, SIGWINCH, SIGINT);

      signals_handler.async_wait(
          boost::bind(
              handle_signal,
              boost::ref(signals_handler),
              boost::ref(terminal_width),
              boost::asio::placeholders::error,
              boost::asio::placeholders::signal_number
          )
      );

      LigierDataHandler<clbsk::consumer::Scrollback> dh(
          io_service
        , tag
        , endpoint
        , buffer
        , buffer_size
        , clbsk::consumer::Scrollback(terminal_width, tmch, dom_names_map)
      );

      io_service.run();
    }
  } else { // UDP
    std::cout << "Listening port: " << port << '\n' << "Data type: ";

    if (type == taes) {
      std::cout << "Acoustic\n";
    } else if (type == ttdc) {
      std::cout << "Optical\n";
    } else if (type == tmch) {
      std::cout << "Monitoring\n";
    } else {
      std::cout << "Automatically determined\n";
    }

    clbsk::consumer::ConsumerSet consumer_set;

    trm::MonStreamer ms(roy_server, roy_port);

    if (uses_roy) {
      std::cout
        << "Using ROyWeb ("
        << roy_server
        << ", "
        << roy_port
        << ") with tag format "
        << templ << std::endl;

      consumer_set.add(
          clbsk::consumer::Roy(
              dom_names_map
            , templ
            , ms
          )
      );
    }

    if (dump) {
      consumer_set.add(clbsk::consumer::File(output_file));
    }

    if (use_gui) { // UDP-GUI
  #ifdef CURSES_FOUND
      ChList stats;

      boost::asio::deadline_timer timer(io_service, boost::posix_time::seconds(1));

      boost::asio::posix::stream_descriptor keyb(io_service, ::dup(STDIN_FILENO));

      boost::asio::signal_set window_resize(io_service, SIGWINCH);

      char key = 0;
      gui::ScreenInitialiser init;

      std::string const heading
        = "UDP " + boost::lexical_cast<std::string>(port);

      gui::Screen screen(io_service, heading, stats);

      keyb.async_read_some(
          boost::asio::buffer(&key, sizeof(key)),
          boost::bind(
              async_cin,
              boost::ref(keyb),
              boost::ref(key),
              boost::ref(screen),
              boost::asio::placeholders::error,
              boost::asio::placeholders::bytes_transferred
          )
      );

      timer.async_wait(
          boost::bind(
              async_commit,
              boost::ref(timer),
              boost::ref(screen),
              boost::asio::placeholders::error
          )
      );

      window_resize.async_wait(
          boost::bind(
              async_win_resize,
              boost::ref(window_resize),
              boost::ref(screen),
              boost::asio::placeholders::error,
              boost::asio::placeholders::signal_number
          )
      );

      consumer_set.add(
          clbsk::consumer::Gui(
              terminal_width
            , dom_names_map
            , stats
            , screen
          )
      );

      UdpDataHandler<clbsk::consumer::ConsumerSet> dh(
          io_service
        , port
        , buffer
        , buffer_size
        , consumer_set
      );

      io_service.run();
  #endif
    } else { // UDP-Scrollback
      consumer_set.add(
          clbsk::consumer::Scrollback(
              terminal_width
            , type
            , dom_names_map
          )
      );

      boost::asio::signal_set signals_handler(io_service, SIGWINCH, SIGINT);

      signals_handler.async_wait(
          boost::bind(
              handle_signal
            , boost::ref(signals_handler)
            , boost::ref(terminal_width)
            , boost::asio::placeholders::error
            , boost::asio::placeholders::signal_number
          )
      );

      UdpDataHandler<clbsk::consumer::ConsumerSet> dh(
          io_service
        , port
        , buffer
        , buffer_size
        , consumer_set
      );

      io_service.run();
    }
  }
}

void handle_signal(
    boost::asio::signal_set& set,
    int& terminal_width,
    boost::system::error_code const& error,
    int signum
)
{
  if (!error) {
    if (signum == SIGINT) {
      set.get_io_service().stop();
      if (isatty(1)) {
        std::cout << "\033]2;thank you for flying with CLBSwissKnife!\007";
        std::cout << "\rBye bye!\n";
      }
      return;
    } else if (signum == SIGWINCH) {
      terminal_width = get_terminal_width();
    }

    set.async_wait(
        boost::bind(
            handle_signal,
            boost::ref(set),
            boost::ref(terminal_width),
            boost::asio::placeholders::error,
            boost::asio::placeholders::signal_number
        )
    );
  }
}