<commit_before>/** Copyright (C) 2016 European Spallation Source ERIC */

/** @file
 *
 *  @brief Implements a command server
 */

#include <algorithm>
#include <arpa/inet.h>
#include <cassert>
#include <cinttypes>
#include <cstdio>
#include <common/Trace.h>
#include <efu/Server.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string>
#include <sys/ioctl.h>
#include <unistd.h>
#include <vector>

//#undef TRC_LEVEL
//#define TRC_LEVEL TRC_L_INF

void Server::server_close() {
  XTRACE(IPC, DEB, "Closing socket fd %d\n", sock_client);
  close(sock_client);
  sock_client = -1;
}

/** @brief Setup socket parameters
 */
void Server::server_open() {
  XTRACE(IPC, INF, "Server::open() called on port %d\n", port_);

  struct sockaddr_in socket_address;
  int ret;
  int option=1; // any nonzero value will do

  sock_server = socket(AF_INET, SOCK_STREAM, 0);
  assert(sock_server >= 0);

  ret = setsockopt(sock_server, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(option));
  assert(ret >= 0);

  ret = ioctl(sock_server, FIONBIO, &option);
  assert(ret >= 0);

  std::fill_n((char*)&socket_address, sizeof(socket_address), 0);
  socket_address.sin_family = AF_INET;
  socket_address.sin_addr.s_addr = htonl(INADDR_ANY);
  socket_address.sin_port = htons(port_);

  ret = bind(sock_server, (struct sockaddr *)&socket_address, sizeof(socket_address));
  assert(ret >= 0);

  ret = listen(sock_server, 32); // backlog == 32
  assert(ret >= 0);

  FD_SET(sock_server, &fd_master);
}


int Server::server_send() {
  XTRACE(IPC, DEB, "server_send() - %d bytes\n", output.bytes);
  if (send(sock_client, output.buffer, output.bytes, 0) < 0) {
    XTRACE(IPC, WAR, "Error sending command reply\n");
    return -1;
  }
  output.bytes = 0;
  output.data = output.buffer;
  return 0;
}

/** @brief Called in main program loop
 */
void Server::server_poll() {

  // Prepare working file desc.
  static_assert(sizeof(fd_master) == sizeof(fd_working), "fd_ struct mismatch");
  memcpy(&fd_working, &fd_master, sizeof(fd_master));
  struct timeval timeout;
  timeout.tv_sec = 0;
  timeout.tv_usec = 1000;

  auto max_socket = std::max(sock_server, sock_client) + 1;

  auto ready = select(max_socket, &fd_working, NULL, NULL, &timeout);
  if (ready < 0) {
    return;
  }

  // Server has activity
  if (ready > 0 && FD_ISSET(sock_server, &fd_working)) {
    if (sock_client < 0) {
      XTRACE(IPC, INF, "Incoming connection\n");
      sock_client = accept(sock_server, NULL, NULL);
      if (sock_client < 0 && errno != EWOULDBLOCK) {
        assert(1 == 0);
      }
      FD_SET(sock_client, &fd_master);
      XTRACE(IPC, DEB, "sock_client: %d, ready: %d\n", sock_client, ready);
      ready--;
    }
  }

  // Client has activity
  if (ready > 0 && FD_ISSET(sock_client, &fd_working)) {
    auto bytes = recv(sock_client, input.data + input.bytes, SERVER_BUFFER_SIZE - input.bytes, 0);

    if ((bytes < 0) && (errno != EWOULDBLOCK || errno != EAGAIN)) {
      XTRACE(IPC, WAR, "recv() failed, errno: %d\n", errno);
      perror("recv() failed");
      server_close();
      return;
    }
    if (bytes == 0) {
      XTRACE(IPC, INF, "Peer closed socket\n");
      server_close();
      return;
    }
    XTRACE(IPC, INF, "Received %ld bytes on socket %d\n", bytes, sock_client);
    input.bytes += bytes;

    auto min = std::min(input.bytes, SERVER_BUFFER_SIZE - 1U);
    input.buffer[min] = '\0';
    XTRACE(IPC, DEB, "buffer[] = %s", input.buffer);

    assert(input.bytes <= SERVER_BUFFER_SIZE);
    XTRACE(IPC, DEB, "input.bytes: %d\n", input.bytes);

    // Parse and generate reply
    server_parse();

    input.bytes = 0;
    input.data = input.buffer;
    if (server_send() < 0) {
      XTRACE(IPC, WAR, "server_send() failed\n");
      server_close();
      return;
    }
    ready--;
  }
}

int Server::server_parse() {
  auto max = std::max(input.bytes, SERVER_BUFFER_SIZE);
  assert(max > 1);
  if (input.buffer[max - 1] != '\0') {
    XTRACE(IPC, DEB, "Array is NOT null terminated!\n");
    input.buffer[max -1] = '\0';
  }
  if (input.buffer[max - 2] == '\n') {
    XTRACE(IPC, DEB, "Array conatains newline\n");
    input.buffer[max -2] = '\0';
  }


  std::vector<std::string> tokens;
  char * chars_array = strtok((char*)input.buffer, "\n ");
  while(chars_array)
  {
    std::string token(chars_array);
    tokens.push_back(token);
    chars_array = strtok(NULL, "\n ");
  }
  XTRACE(IPC, DEB, "Tokens in command: %d\n", (int)tokens.size());
  for (auto token : tokens) {
    XTRACE(IPC, DEB, "Token: %s\n", token.c_str());
  }

  if (tokens.at(0).compare(std::string("STAT_INPUT")) == 0) {
    XTRACE(IPC, INF, "STAT_INPUT\n");
    output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "STAT_INPUT %" PRIu64 ", %" PRIu64 "\n", 0ULL, 0ULL);
  } else if (tokens.at(0).compare(std::string("STAT_RESET")) == 0) {
    XTRACE(IPC, INF, "STAT_RESET\n");
    output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "<OK>\n");
  } else if (tokens.at(0).compare(std::string("STAT_MASK")) == 0) {
      unsigned int mask = std::stoul(tokens.at(1), nullptr, 0);
      XTRACE(IPC, INF, "STAT_MASK 0x%08x\n", mask);
      output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "<OK>\n");
  } else {
    XTRACE(IPC, WAR, "Unknown command: %s\n", tokens.at(0).c_str());
    output.bytes = snprintf((char*)output.buffer, SERVER_BUFFER_SIZE, "Unknown command: %s\n", tokens.at(0).c_str());
  }
  return 0;
}
<commit_msg>Small changes for cross platform compile<commit_after>/** Copyright (C) 2016 European Spallation Source ERIC */

/** @file
 *
 *  @brief Implements a command server
 */

#include <algorithm>
#include <arpa/inet.h>
#include <cassert>
#include <cinttypes>
#include <cstdio>
#include <cstring>
#include <common/Trace.h>
#include <efu/Server.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string>
#include <sys/ioctl.h>
#include <unistd.h>
#include <vector>

//#undef TRC_LEVEL
//#define TRC_LEVEL TRC_L_INF

void Server::server_close() {
  XTRACE(IPC, DEB, "Closing socket fd %d\n", sock_client);
  close(sock_client);
  sock_client = -1;
}

/** @brief Setup socket parameters
 */
void Server::server_open() {
  XTRACE(IPC, INF, "Server::open() called on port %d\n", port_);

  struct sockaddr_in socket_address;
  int ret;
  int option=1; // any nonzero value will do

  sock_server = socket(AF_INET, SOCK_STREAM, 0);
  assert(sock_server >= 0);

  ret = setsockopt(sock_server, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(option));
  assert(ret >= 0);

  ret = ioctl(sock_server, FIONBIO, &option);
  assert(ret >= 0);

  std::fill_n((char*)&socket_address, sizeof(socket_address), 0);
  socket_address.sin_family = AF_INET;
  socket_address.sin_addr.s_addr = htonl(INADDR_ANY);
  socket_address.sin_port = htons(port_);

  ret = bind(sock_server, (struct sockaddr *)&socket_address, sizeof(socket_address));
  assert(ret >= 0);

  ret = listen(sock_server, 32); // backlog == 32
  assert(ret >= 0);

  FD_SET(sock_server, &fd_master);
}


int Server::server_send() {
  XTRACE(IPC, DEB, "server_send() - %d bytes\n", output.bytes);
  if (send(sock_client, output.buffer, output.bytes, 0) < 0) {
    XTRACE(IPC, WAR, "Error sending command reply\n");
    return -1;
  }
  output.bytes = 0;
  output.data = output.buffer;
  return 0;
}

/** @brief Called in main program loop
 */
void Server::server_poll() {

  // Prepare working file desc.
  static_assert(sizeof(fd_master) == sizeof(fd_working), "fd_ struct mismatch");
  memcpy(&fd_working, &fd_master, sizeof(fd_master));
  struct timeval timeout;
  timeout.tv_sec = 0;
  timeout.tv_usec = 1000;

  auto max_socket = std::max(sock_server, sock_client) + 1;

  auto ready = select(max_socket, &fd_working, NULL, NULL, &timeout);
  if (ready < 0) {
    return;
  }

  // Server has activity
  if (ready > 0 && FD_ISSET(sock_server, &fd_working)) {
    if (sock_client < 0) {
      XTRACE(IPC, INF, "Incoming connection\n");
      sock_client = accept(sock_server, NULL, NULL);
      if (sock_client < 0 && errno != EWOULDBLOCK) {
        assert(1 == 0);
      }
      FD_SET(sock_client, &fd_master);
      XTRACE(IPC, DEB, "sock_client: %d, ready: %d\n", sock_client, ready);
      ready--;
    }
  }

  // Client has activity
  if (ready > 0 && FD_ISSET(sock_client, &fd_working)) {
    auto bytes = recv(sock_client, input.data + input.bytes, SERVER_BUFFER_SIZE - input.bytes, 0);

    if ((bytes < 0) && (errno != EWOULDBLOCK || errno != EAGAIN)) {
      XTRACE(IPC, WAR, "recv() failed, errno: %d\n", errno);
      perror("recv() failed");
      server_close();
      return;
    }
    if (bytes == 0) {
      XTRACE(IPC, INF, "Peer closed socket\n");
      server_close();
      return;
    }
    XTRACE(IPC, INF, "Received %ld bytes on socket %d\n", bytes, sock_client);
    input.bytes += bytes;

    auto min = std::min(input.bytes, SERVER_BUFFER_SIZE - 1U);
    input.buffer[min] = '\0';
    XTRACE(IPC, DEB, "buffer[] = %s", input.buffer);

    assert(input.bytes <= SERVER_BUFFER_SIZE);
    XTRACE(IPC, DEB, "input.bytes: %d\n", input.bytes);

    // Parse and generate reply
    server_parse();

    input.bytes = 0;
    input.data = input.buffer;
    if (server_send() < 0) {
      XTRACE(IPC, WAR, "server_send() failed\n");
      server_close();
      return;
    }
    ready--;
  }
}

int Server::server_parse() {
  auto max = std::max(input.bytes, SERVER_BUFFER_SIZE);
  assert(max > 1);
  if (input.buffer[max - 1] != '\0') {
    XTRACE(IPC, DEB, "Array is NOT null terminated!\n");
    input.buffer[max -1] = '\0';
  }
  if (input.buffer[max - 2] == '\n') {
    XTRACE(IPC, DEB, "Array conatains newline\n");
    input.buffer[max -2] = '\0';
  }


  std::vector<std::string> tokens;
  char * chars_array = strtok((char*)input.buffer, "\n ");
  while(chars_array)
  {
    std::string token(chars_array);
    tokens.push_back(token);
    chars_array = strtok(NULL, "\n ");
  }
  XTRACE(IPC, DEB, "Tokens in command: %d\n", (int)tokens.size());
  for (auto token : tokens) {
    XTRACE(IPC, DEB, "Token: %s\n", token.c_str());
  }

  if (tokens.at(0).compare(std::string("STAT_INPUT")) == 0) {
    XTRACE(IPC, INF, "STAT_INPUT\n");
    output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "STAT_INPUT %" PRIu64 ", %" PRIu64 "\n", (uint64_t)0, (uint64_t)0);
  } else if (tokens.at(0).compare(std::string("STAT_RESET")) == 0) {
    XTRACE(IPC, INF, "STAT_RESET\n");
    output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "<OK>\n");
  } else if (tokens.at(0).compare(std::string("STAT_MASK")) == 0) {
      unsigned int mask = std::stoul(tokens.at(1), nullptr, 0);
      XTRACE(IPC, INF, "STAT_MASK 0x%08x\n", mask);
      output.bytes = snprintf((char *)output.buffer, SERVER_BUFFER_SIZE, "<OK>\n");
  } else {
    XTRACE(IPC, WAR, "Unknown command: %s\n", tokens.at(0).c_str());
    output.bytes = snprintf((char*)output.buffer, SERVER_BUFFER_SIZE, "Unknown command: %s\n", tokens.at(0).c_str());
  }
  return 0;
}

{'fraction_non_alphanumeric': 0.17350644634102053, 'fraction_numerical': 0.013074269112039223, 'mean_word_length': 3.667372881355932, 'pattern_counts': {'":': 0, '<': 52, '<?xml version=': 0, '>': 54, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1650302', 'n_tokens_mistral': 4257, 'n_tokens_neox': 4078, 'n_words': 927}