DaphneV2Interface.cpp

Go to the documentation of this file.

 
#include "DaphneV2Interface.hpp"
#include "logging/Logging.hpp"
#include <sys/time.h>
 
using namespace dunedaq::daphnemodules;
 
DaphneV2Interface::DaphneV2Interface(const char* ipaddr, int port, std::chrono::milliseconds timeout)
  : m_timeout(timeout)
{
 
  m_connection_id = socket(AF_INET, SOCK_DGRAM, 0);
 
  if (m_connection_id < 0)
    throw SocketCreationError(ERS_HERE);
 
  m_target.sin_family = AF_INET;
  m_target.sin_port = htons(port);
  auto ret = inet_pton(AF_INET, ipaddr, &(m_target.sin_addr));
  if (ret <= 0)
    throw InvalidIPAddress(ERS_HERE, ipaddr);
 
  m_ip = ipaddr;
 
  if (!validate_connection())
    throw FailedPing(ERS_HERE, ipaddr, port);
}
 
void
DaphneV2Interface::close()
{
 
  ::close(m_connection_id);
}
 
bool
DaphneV2Interface::validate_connection() const
{
 
  auto ret = read_register(0xaa55, 1);
 
  static const uint64_t good_value = 0xdeadbeef;
  return ret[0] == good_value;
}
 
command_result
DaphneV2Interface::send_command_retry(std::string cmd, size_t retry) const
{
 
  do {
    try {
      auto ret = send_command(cmd);
      return ret;
    } catch (const CommandTimeout& e) {
      ers::warning(e);
      --retry;
    } catch (const ers::Issue& e) {
      throw FailedSocketInteraction(ERS_HERE, cmd, e);
    }
 
  } while (retry > 0);
 
  throw FailedSocketInteraction(ERS_HERE, cmd);
}
 
command_result
DaphneV2Interface::send_command_interruptible(std::string cmd, std::function<bool()> can_retry) const
{
 
  do {
    try {
      auto ret = send_command(cmd);
      return ret;
    } catch (const CommandTimeout& e) {
      ers::warning(e);
    } catch (const ers::Issue& e) {
      throw FailedSocketInteraction(ERS_HERE, cmd, e);
    }
 
  } while (can_retry());
}
 
command_result
DaphneV2Interface::send_command(std::string cmd) const
{
 
  TLOG() << "Board: " << m_ip << ", sending command " << cmd;
  std::vector<uint64_t> bytes;
  for (char ch : cmd) {
    bytes.push_back(static_cast<uint64_t>(ch));
  }
  bytes.push_back(0x0d); // dedicated command flag
 
  const std::lock_guard<std::mutex> lock(m_command_mutex);
 
  // we send the bytes in chunks of 50 words
  for (size_t i = 0; i < (bytes.size() + 49) / 50; ++i) {
    std::vector<uint64_t> part(bytes.begin() + i * 50, bytes.begin() + std::min((i + 1) * 50, bytes.size()));
    write_buffer(0x90000000, std::move(part));
  }
 
  TLOG() << "Board: " << m_ip << ", Command sent, waiting for result";
 
  command_result res;
  std::string* writing_pointer = nullptr;
 
  auto start_time = std::chrono::high_resolution_clock::now();
 
  int more = 40;
  while (more > 0) {
    auto data_block = read_buffer(0x90000000, 50);
    for (size_t i = 0; i < data_block.size(); ++i) {
      if (data_block[i] == 255) {
        break;
      } else if (data_block[i] == 1) {
        // the following data are returning the command that was issued
        writing_pointer = &res.command;
      } else if (data_block[i] == 2) {
        // the following data are the immediate command response
        writing_pointer = &res.result;
      } else if (data_block[i] == 3) {
        // this is the message end
        writing_pointer = nullptr;
      } else if (isprint(static_cast<int>(data_block[i]))) {
        more = 40;
        char c = static_cast<char>(data_block[i]);
        if (writing_pointer) {
          *writing_pointer += static_cast<char>(data_block[i]);
        } else {
          TLOG() << "Failed adding charachter " << c;
        }
      }
    }
    auto now = std::chrono::high_resolution_clock::now();
 
    auto delay = now - start_time;
 
    if (delay > m_timeout) {
      TLOG() << "Details of timeout";
      for (size_t i = 0; i < data_block.size(); ++i) {
        TLOG() << i << "\t" << std::hex << data_block[i] << std::dec;
      }
      TLOG() << "Received so far: " << res.result;
      auto delay_us = std::chrono::duration_cast<std::chrono::microseconds>(delay);
      throw CommandTimeout(ERS_HERE, cmd, delay_us.count());
    }
 
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
    --more;
  }
 
  return res;
}
 
std::vector<uint64_t>
DaphneV2Interface::read(uint8_t command_id, uint64_t addr, uint8_t size) const
{
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
 
  uint8_t cmd[10];
  cmd[0] = command_id;
  cmd[1] = size;
  memcpy(cmd + 2, &addr, sizeof(uint64_t));
  auto result = sendto(m_connection_id, cmd, sizeof(cmd), 0, (struct sockaddr*)&m_target, sizeof(m_target));
 
  if (result < 0)
    throw FailedSocketInteraction(ERS_HERE, "sendto");
 
  struct timeval timeout;
  timeout.tv_sec = m_timeout.count() / 1000;
  timeout.tv_usec = (m_timeout.count() % 1000) * 1000;
  fd_set readfds, masterfds;
 
  FD_ZERO(&masterfds);
  FD_SET(m_connection_id, &masterfds);
 
  memcpy(&readfds, &masterfds, sizeof(fd_set));
 
  auto start_time = std::chrono::high_resolution_clock::now();
 
  if (select(m_connection_id + 1, &readfds, NULL, NULL, &timeout) < 0) {
    throw FailedSocketInteraction(ERS_HERE, "select");
  }
 
  std::vector<uint64_t> ret_value;
 
  if (FD_ISSET(m_connection_id, &readfds)) {
    uint8_t buffer[2 + (8 * size)];
    socklen_t addrlen = sizeof(m_target);
    result = recvfrom(m_connection_id, buffer, sizeof(buffer), 0, (struct sockaddr*)&m_target, &addrlen);
 
    if (result <= 0)
      throw FailedSocketInteraction(ERS_HERE, "recvfrom");
 
    uint8_t fmt[4 + size];
    fmt[0] = '<';
    fmt[1] = 'B';
    fmt[2] = 'B';
    fmt[3] = size;
    for (int i = 0; i < size; i++) {
      fmt[4 + i] = 'Q';
    }
 
    for (int i = 0; i < size; i++) {
      uint64_t value;
      memcpy(&value, buffer + 2 + (8 * i), sizeof(uint64_t));
      ret_value.push_back(value);
    }
 
  } else {
    // the socket timedout
    auto end_time = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time);
 
    throw SocketTimeout(ERS_HERE, duration.count());
  }
 
  return ret_value;
}
 
void
DaphneV2Interface::write(uint8_t command_id, uint64_t addr, std::vector<uint64_t>&& data) const
{
 
  const std::lock_guard<std::mutex> lock(m_access_mutex);
 
  uint8_t cmd[10 + (8 * data.size())];
  cmd[0] = command_id;
  cmd[1] = data.size();
  memcpy(cmd + 2, &addr, sizeof(uint64_t));
  for (size_t i = 0; i < data.size(); i++) {
    memcpy(cmd + 10 + (8 * i), &(data[i]), sizeof(uint64_t));
  }
 
  auto result = sendto(m_connection_id, cmd, sizeof(cmd), 0, (struct sockaddr*)&m_target, sizeof(m_target));
  if (result < 0)
    throw FailedSocketInteraction(ERS_HERE, "sendto");
}