<commit_before>// Copyright 2016 Las Venturas Playground. All rights reserved.
// Use of this source code is governed by the MIT license, a copy of which can
// be found in the LICENSE file.

#include <fstream>
#include <iostream>
#include <mutex>
#include <queue>
#include <stdint.h>
#include <string>
#include <utility>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/program_options.hpp>
#include <boost/thread.hpp>

namespace {

// Maximum time, in seconds, we'll wait for a response from a remote server.
constexpr size_t kSocketTimeoutSeconds = 5;

// Options for the indexer. Sets default values. May be overridden by using the command line flags
// listed next to each option. {@see ParseCommandLine}
struct IndexOptions {
  IndexOptions()
      : threads(boost::thread::hardware_concurrency()),
        user_agent("SAMPIndexer/1.0 (+https://sa-mp.nl/)") {}

  ~IndexOptions() = default;

  std::string logstash;     // --logstash
  std::string server_list;  // --server-list
  uint32_t threads;         // --threads, -t
  std::string user_agent;   // --user-agent

  bool output = true;       // Whether to output the results to stdout.
  bool verbose = false;     // Whether to output exactly what's happening.
};

// Definition for a server entry, which exists of an IP address and a port number.
using ServerEntry = std::pair<std::string, uint16_t>;

// Structure containing the information that can be queried from the server.
struct ServerInfo {
  explicit ServerInfo(const ServerEntry& server)
      : server(server) {}

  ServerEntry server;

  bool has_password = false;

  uint16_t players = 0;
  uint16_t max_players = 0;

  std::string hostname;
  std::string gamemode;
  std::string map;
};

// -------------------------------------------------------------------------------------------------

// Queue and vector used for processing the identified ServerEntries.
std::queue<ServerEntry> server_queue;

std::vector<ServerEntry> failure_vector;
std::vector<ServerInfo> result_vector;

// Mutexes that protect the aforementioned queue and vector.
std::mutex server_queue_mutex, result_mutex, error_mutex;

// -------------------------------------------------------------------------------------------------

// Parses the command line arguments (|argc| and |argv|) in to the IndexOptions (|options|).
bool ParseCommandLine(IndexOptions* options, int argc, const char** argv) {
  namespace po = boost::program_options;

  po::options_description indexer("SA-MP Indexer");
  indexer.add_options()
      ("logstash", po::value<std::string>(), "Path to the logstash UNIX pipe (instead of STDOUT)")
      ("server-list", po::value<std::string>()->required(), "URL of the server list to use")
      ("threads,t", po::value<uint32_t>(), "Number of threads to use")
      ("verbose,v", "Enable verbose output")
      ("user-agent", po::value<std::string>(), "The user agent to use in the request");

  po::variables_map variables;
  try {
    po::store(po::parse_command_line(argc, argv, indexer), variables);
    po::notify(variables);

  } catch (po::error& error) {
    std::cerr << "Unable to parse the command line: " << error.what() << std::endl << std::endl;
    std::cerr << indexer << std::endl;
    return false;
  }

  if (variables.count("logstash")) {
    options->logstash = variables["logstash"].as<std::string>();
    options->output = false;
  }

  if (variables.count("server-list"))
    options->server_list = variables["server-list"].as<std::string>();
  if (variables.count("threads"))
    options->threads = variables["threads"].as<uint32_t>();
  if (variables.count("verbose"))
    options->verbose = true;
  if (variables.count("user-agent"))
    options->user_agent = variables["user-agent"].as<std::string>();

  return true;
}

// Fetches and parses a list of servers from the specified |options.server_list|. Synchronous.
// Supports HTTP (non-HTTPS) URLs, as well as file paths relative to the CHDIR.
bool FetchServerList(std::queue<ServerEntry>* servers, const IndexOptions& options) {
  const std::string& server_list = options.server_list;

  std::vector<std::string> raw_servers;

  if (server_list.substr(0, 7) == "http://") {
    size_t path_separator = server_list.find_first_of('/', 7 /* after the http:// prefix */);
    if (path_separator == std::string::npos) {
      std::cerr << "Invalid hostname given: " << server_list;
      return false;
    }

    using boost::asio::ip::tcp;

    boost::asio::io_service io_service;

    std::string hostname = server_list.substr(7, path_separator - 7);
    std::string path = server_list.substr(path_separator);

    try {
      // (1) Resolve the |hostname| to the IP address associated with it.
      tcp::resolver resolver(io_service);
      tcp::resolver::query query(hostname, "http");
      tcp::resolver::iterator endpoints = resolver.resolve(query);

      // (2) Establish a connection to one of the |endpoints|.
      tcp::socket socket(io_service);
      boost::asio::connect(socket, endpoints);

      boost::asio::streambuf request;
      boost::asio::streambuf response;

      // (3) Build the request headers for the server list.
      std::ostream request_stream(&request);

      request_stream << "GET " << path << " HTTP/1.0\r\n";
      request_stream << "Host: " << hostname << "\r\n";
      request_stream << "User-Agent: " << options.user_agent << "\r\n";
      request_stream << "Connection: close\r\n\r\n";

      boost::asio::write(socket, request);

      std::string http_version;
      uint32_t http_status_code;

      // (4) Read the response received from the server and make that it's OK.
      boost::asio::read_until(socket, response, "\r\n");

      std::istream response_stream(&response);

      response_stream >> http_version;
      response_stream >> http_status_code;
      
      if (!response_stream || http_version.substr(0, 5) != "HTTP/" || http_status_code != 200) {
        std::cerr << "Received an invalid HTTP response (" << http_version << "; "
                  << http_status_code << ")" << std::endl;
        return false;
      }

      boost::system::error_code error;

      // (5) Make sure that we've read the entire response in the |response| stream.
      while (boost::asio::read(socket, response, boost::asio::transfer_at_least(1), error)) {}

      if (error != boost::asio::error::eof)
        throw boost::system::system_error(error);

      // (6) Skip over the headers included in the |response|.
      for (std::string header; std::getline(response_stream, header) && header != "\r";) {}

      // (7) Interpret the rest of the lines as server definitions.
      for (std::string line; std::getline(response_stream, line);)
        raw_servers.push_back(line);

      socket.close();
    }
    catch (std::exception& e) {
      std::cerr << "Unable to fetch the server list: " << e.what() << std::endl;
      return false;
    }
  } else {
    std::ifstream file(server_list, std::ifstream::in);
    if (file.is_open()) {
      for (std::string line; std::getline(file, line);)
        raw_servers.push_back(line);

      file.close();
    }
  }

  if (!raw_servers.size()) {
    std::cerr << "Unable to load the list of servers from: " << server_list;
    return false;
  }

  for (const std::string& input_line : raw_servers) {
    std::string line = boost::trim_copy(input_line);

    size_t colon_position = line.find_first_of(':');
    if (colon_position == std::string::npos) {
      std::cerr << "Unable to parse server entry: " << line;
      continue;
    }

    uint16_t port_number;
    try {
      port_number = boost::lexical_cast<uint16_t>(line.substr(colon_position + 1));
    } catch (boost::bad_lexical_cast const&) {
      std::cerr << "Invalid port number for server entry: " << line;
      continue;
    }

    servers->emplace(line.substr(0, colon_position), port_number);
  }

  return true;
}

// -------------------------------------------------------------------------------------------------

template <typename T>
bool ReadValue(T* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  throw std::exception("Invalid ReadValue() overload reached.");
}

template <>
bool ReadValue(bool* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(bool) > bytes)
    return false;

  *storage = !!buffer[*offset];
  *offset += sizeof(bool);

  return true;
}

template <>
bool ReadValue(uint16_t* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(uint16_t) > bytes)
    return false;

  *storage = static_cast<unsigned char>(buffer[*offset]) |
             static_cast<unsigned char>(buffer[*offset + 1]) << 8;

  *offset += sizeof(uint16_t);

  return true;
}

template <>
bool ReadValue(std::string* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(uint32_t) > bytes)
    return false;

  uint32_t length = static_cast<unsigned char>(buffer[*offset]) |
                    static_cast<unsigned char>(buffer[*offset + 1]) << 8 |
                    static_cast<unsigned char>(buffer[*offset + 2]) << 16 |
                    static_cast<unsigned char>(buffer[*offset + 3]) << 24;

  *offset += sizeof(uint32_t);

  if (*offset + length > bytes)
    return false;

  *storage = std::string(&buffer[*offset], length);
  *offset += length;

  return true;
}

// -------------------------------------------------------------------------------------------------

// Queries the server defined in |info|. Writes the results to |info| as well.
bool QueryServer(ServerInfo* info, boost::asio::io_service& io_service) {
  unsigned char packet[11];

  boost::asio::ip::address_v4 address = boost::asio::ip::address_v4::from_string(info->server.first);
  unsigned short port = info->server.second;

  // Compile the information in |packet| necessary to query the server.
  {
    unsigned long long_address = address.to_ulong();
    unsigned short port = info->server.second;

    packet[0] = 'S';
    packet[1] = 'A';
    packet[2] = 'M';
    packet[3] = 'P';

    static_assert(sizeof(long_address) == 4, "unsigned long must be four bytes");
    memcpy(&packet[4], &long_address, sizeof(long_address));

    packet[8] = port & 0xFF;
    packet[9] = (port >> 8) & 0xFF;

    packet[10] = 'i';  // information packet identifier
  }

  using boost::asio::ip::udp;
  try {
    udp::endpoint endpoint;
    endpoint.address(address);
    endpoint.port(port);

    udp::socket socket(io_service);

    // Set send and receive timeouts on the |socket| for |kSocketTimeoutSeconds| seconds.
#if defined(WIN32)
    int32_t timeout = kSocketTimeoutSeconds * 1000;
    setsockopt(socket.native(), SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout, sizeof(timeout));
    setsockopt(socket.native(), SOL_SOCKET, SO_SNDTIMEO, (const char*)&timeout, sizeof(timeout));
#else
    struct timeval tv;
    tv.tv_sec = kSocketTimeoutSeconds;
    tv.tv_usec = 0;
    setsockopt(socket.native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
    setsockopt(socket.native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
#endif

    socket.open(endpoint.protocol());

    boost::array<char, 256> receive_buffer;

    // Send the |packet| to the |endpoint| over the opened |socket|.
    socket.send_to(boost::asio::buffer(packet), endpoint);

    size_t bytes = socket.receive_from(boost::asio::buffer(receive_buffer), endpoint);
    size_t offset = 11;  // the |packet| that we sent

    if (bytes <= offset)
      throw std::exception("Received packet size too short");

    return ReadValue(&info->has_password, receive_buffer, bytes, &offset) &&
           ReadValue(&info->players, receive_buffer, bytes, &offset) &&
           ReadValue(&info->max_players, receive_buffer, bytes, &offset) &&
           ReadValue(&info->hostname, receive_buffer, bytes, &offset) &&
           ReadValue(&info->gamemode, receive_buffer, bytes, &offset) &&
           ReadValue(&info->map, receive_buffer, bytes, &offset);

  } catch (std::exception& e) {
    std::lock_guard<std::mutex> guard(error_mutex);
    std::cerr << "Exception: " << e.what() << std::endl;
    return false;
  }

  return true;
}

// Thread responsible for querying servers. It will pick a ServerEntry from the vector of parsed
// server entries, query it, and then write the results to the vector of queried servers. Both
// vectors are protected by mutexes
void QueryThread() {
  boost::asio::io_service io_service;
  ServerEntry server;

  while (true) {
    {
      std::lock_guard<std::mutex> guard(server_queue_mutex);
      if (server_queue.empty())
        return;

      server = server_queue.front();
      server_queue.pop();
    }

    ServerInfo info(server);

    const bool success = QueryServer(&info, io_service);
    {
      std::lock_guard<std::mutex> guard(result_mutex);
      if (success)
        result_vector.push_back(std::move(info));
      else
        failure_vector.push_back(server);
    }
  }
}

}  // namespace

// -------------------------------------------------------------------------------------------------

int main(int argc, const char** argv) {
  IndexOptions options;
  if (!ParseCommandLine(&options, argc, argv))
    return 1;

  if (options.verbose)
    std::cout << "Parsing servers from '" << options.server_list << "'..." << std::endl;

  if (!FetchServerList(&server_queue, options))
    return 1;

  if (options.verbose)
    std::cout << "Found " << server_queue.size() << " servers..." << std::endl;

  if (!server_queue.size())
    return 0;  // there are no servers to index

  // Now query all servers in the list in parallel, using |options.threads| threads.
  {
    boost::thread_group query_threads;
    for (size_t i = 0; i < options.threads; ++i)
      query_threads.create_thread(QueryThread);

    query_threads.join_all();
  }

  // TODO: Write the results to `logstash` when this has been configured.
  for (const ServerInfo& info : result_vector) {
    std::cout << info.server.first << ":" << info.server.second << " -- " << info.hostname << " -- "<commit_msg>Add debugging output to see what requests are doing<commit_after>// Copyright 2016 Las Venturas Playground. All rights reserved.
// Use of this source code is governed by the MIT license, a copy of which can
// be found in the LICENSE file.

#include <fstream>
#include <iostream>
#include <mutex>
#include <queue>
#include <stdint.h>
#include <string>
#include <utility>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/program_options.hpp>
#include <boost/thread.hpp>

#if defined(WIN32)
#include <Windows.h>
#elif defined(LINUX)
#include <sys/time.h>
#include <time.h>
#endif

namespace {

// Options for the indexer. Sets default values. May be overridden by using the command line flags
// listed next to each option. {@see ParseCommandLine}
struct IndexOptions {
  IndexOptions()
      : threads(boost::thread::hardware_concurrency()),
        timeout(4 /* seconds */),
        user_agent("SAMPIndexer/1.0 (+https://sa-mp.nl/)") {}

  ~IndexOptions() = default;

  std::string logstash;     // --logstash
  std::string server_list;  // --server-list
  uint32_t threads;         // --threads, -t
  uint32_t timeout;         // --timeout
  std::string user_agent;   // --user-agent

  bool output = true;       // Whether to output the results to stdout.
  bool verbose = false;     // Whether to output exactly what's happening.
};

// Definition for a server entry, which exists of an IP address and a port number.
using ServerEntry = std::pair<std::string, uint16_t>;

// Structure containing the information that can be queried from the server.
struct ServerInfo {
  explicit ServerInfo(const ServerEntry& server)
      : server(server) {}

  ServerEntry server;

  bool has_password = false;

  uint16_t players = 0;
  uint16_t max_players = 0;

  std::string hostname;
  std::string gamemode;
  std::string map;
};

// -------------------------------------------------------------------------------------------------

IndexOptions options;

// Queue and vector used for processing the identified ServerEntries.
std::queue<ServerEntry> server_queue;

std::vector<ServerEntry> failure_vector;
std::vector<ServerInfo> result_vector;

// Mutexes that protect the aforementioned queue and vector, as well as verbose output.
std::mutex server_queue_mutex, result_mutex, verbose_output_mutex;

// -------------------------------------------------------------------------------------------------

std::string ToDisplay(const ServerEntry& server) {
  return server.first + ":" + std::to_string(server.second);
}

// -------------------------------------------------------------------------------------------------

#if defined(WIN32)

// Difference between January 1st, 1601 and January 1st, 1970 in nanoseconds, to convert from
// the Windows file time epoch to the UNIX timestamp epoch.
const uint64_t kEpochDelta = 116444736000000000ull;

double monotonicallyIncreasingTime() {
  static uint64_t begin_time = 0ull;
  static bool set_begin_time = false;

  FILETIME tm;

  // Use precise (<1us) timing for Windows 8 and above, normal (~1ms) on other versions.
#if defined(NTDDI_WIN8) && NTDDI_VERSION >= NTDDI_WIN8
  GetSystemTimePreciseAsFileTime(&tm);
#else
  GetSystemTimeAsFileTime(&tm);
#endif

  uint64_t time = 0;
  time |= tm.dwHighDateTime;
  time <<= 32;
  time |= tm.dwLowDateTime;
  time -= kEpochDelta;

  if (!set_begin_time) {
    set_begin_time = true;
    begin_time = time;
  }

  return static_cast<double>(time - begin_time) / 10000.0;
}

#elif defined(LINUX)

double monotonicallyIncreasingTime() {
  static uint64_t begin_time = 0ull;
  static bool set_begin_time = false;

  struct timespec ts;
  clock_gettime(CLOCK_MONOTONIC, &ts);

  uint64_t time = static_cast<uint64_t>(ts.tv_sec) * 1000000000u + static_cast<uint64_t>(ts.tv_nsec);

  if (!set_begin_time) {
    set_begin_time = true;
    begin_time = time;
  }

  return static_cast<double>(time - begin_time) / 1000000.0;
}

#endif

// -------------------------------------------------------------------------------------------------

// Parses the command line arguments (|argc| and |argv|) in to the IndexOptions (|options|).
bool ParseCommandLine(IndexOptions* options, int argc, const char** argv) {
  namespace po = boost::program_options;

  po::options_description indexer("SA-MP Indexer");
  indexer.add_options()
      ("logstash", po::value<std::string>(), "Path to the logstash UNIX pipe (instead of STDOUT)")
      ("server-list", po::value<std::string>()->required(), "URL of the server list to use")
      ("threads,t", po::value<uint32_t>(), "Number of threads to use")
      ("timeout", po::value<uint32_t>(), "Request timeout for an individual server, in seconds")
      ("verbose,v", "Enable verbose output")
      ("user-agent", po::value<std::string>(), "The user agent to use in the request");

  po::variables_map variables;
  try {
    po::store(po::parse_command_line(argc, argv, indexer), variables);
    po::notify(variables);

  } catch (po::error& error) {
    std::cerr << "Unable to parse the command line: " << error.what() << std::endl << std::endl;
    std::cerr << indexer << std::endl;
    return false;
  }

  if (variables.count("logstash")) {
    options->logstash = variables["logstash"].as<std::string>();
    options->output = false;
  }

  if (variables.count("server-list"))
    options->server_list = variables["server-list"].as<std::string>();
  if (variables.count("threads"))
    options->threads = variables["threads"].as<uint32_t>();
  if (variables.count("verbose"))
    options->verbose = true;
  if (variables.count("user-agent"))
    options->user_agent = variables["user-agent"].as<std::string>();

  return true;
}

// Fetches and parses a list of servers from the specified |options.server_list|. Synchronous.
// Supports HTTP (non-HTTPS) URLs, as well as file paths relative to the CHDIR.
bool FetchServerList(std::queue<ServerEntry>* servers) {
  const std::string& server_list = options.server_list;

  std::vector<std::string> raw_servers;

  if (server_list.substr(0, 7) == "http://") {
    size_t path_separator = server_list.find_first_of('/', 7 /* after the http:// prefix */);
    if (path_separator == std::string::npos) {
      std::cerr << "Invalid hostname given: " << server_list;
      return false;
    }

    using boost::asio::ip::tcp;

    boost::asio::io_service io_service;

    std::string hostname = server_list.substr(7, path_separator - 7);
    std::string path = server_list.substr(path_separator);

    try {
      // (1) Resolve the |hostname| to the IP address associated with it.
      tcp::resolver resolver(io_service);
      tcp::resolver::query query(hostname, "http");
      tcp::resolver::iterator endpoints = resolver.resolve(query);

      // (2) Establish a connection to one of the |endpoints|.
      tcp::socket socket(io_service);
      boost::asio::connect(socket, endpoints);

      boost::asio::streambuf request;
      boost::asio::streambuf response;

      // (3) Build the request headers for the server list.
      std::ostream request_stream(&request);

      request_stream << "GET " << path << " HTTP/1.0\r\n";
      request_stream << "Host: " << hostname << "\r\n";
      request_stream << "User-Agent: " << options.user_agent << "\r\n";
      request_stream << "Connection: close\r\n\r\n";

      boost::asio::write(socket, request);

      std::string http_version;
      uint32_t http_status_code;

      // (4) Read the response received from the server and make that it's OK.
      boost::asio::read_until(socket, response, "\r\n");

      std::istream response_stream(&response);

      response_stream >> http_version;
      response_stream >> http_status_code;
      
      if (!response_stream || http_version.substr(0, 5) != "HTTP/" || http_status_code != 200) {
        std::cerr << "Received an invalid HTTP response (" << http_version << "; "
                  << http_status_code << ")" << std::endl;
        return false;
      }

      boost::system::error_code error;

      // (5) Make sure that we've read the entire response in the |response| stream.
      while (boost::asio::read(socket, response, boost::asio::transfer_at_least(1), error)) {}

      if (error != boost::asio::error::eof)
        throw boost::system::system_error(error);

      // (6) Skip over the headers included in the |response|.
      for (std::string header; std::getline(response_stream, header) && header != "\r";) {}

      // (7) Interpret the rest of the lines as server definitions.
      for (std::string line; std::getline(response_stream, line);)
        raw_servers.push_back(line);

      socket.close();
    }
    catch (std::exception& e) {
      std::cerr << "Unable to fetch the server list: " << e.what() << std::endl;
      return false;
    }
  } else {
    std::ifstream file(server_list, std::ifstream::in);
    if (file.is_open()) {
      for (std::string line; std::getline(file, line);)
        raw_servers.push_back(line);

      file.close();
    }
  }

  if (!raw_servers.size()) {
    std::cerr << "Unable to load the list of servers from: " << server_list;
    return false;
  }

  for (const std::string& input_line : raw_servers) {
    std::string line = boost::trim_copy(input_line);

    size_t colon_position = line.find_first_of(':');
    if (colon_position == std::string::npos) {
      std::cerr << "Unable to parse server entry: " << line;
      continue;
    }

    uint16_t port_number;
    try {
      port_number = boost::lexical_cast<uint16_t>(line.substr(colon_position + 1));
    } catch (boost::bad_lexical_cast const&) {
      std::cerr << "Invalid port number for server entry: " << line;
      continue;
    }

    servers->emplace(line.substr(0, colon_position), port_number);
  }

  return true;
}

// -------------------------------------------------------------------------------------------------

template <typename T>
bool ReadValue(T* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  throw std::exception("Invalid ReadValue() overload reached.");
}

template <>
bool ReadValue(bool* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(bool) > bytes)
    return false;

  *storage = !!buffer[*offset];
  *offset += sizeof(bool);

  return true;
}

template <>
bool ReadValue(uint16_t* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(uint16_t) > bytes)
    return false;

  *storage = static_cast<unsigned char>(buffer[*offset]) |
             static_cast<unsigned char>(buffer[*offset + 1]) << 8;

  *offset += sizeof(uint16_t);

  return true;
}

template <>
bool ReadValue(std::string* storage, const boost::array<char, 256>& buffer, const size_t& bytes, size_t* offset) {
  if (*offset + sizeof(uint32_t) > bytes)
    return false;

  uint32_t length = static_cast<unsigned char>(buffer[*offset]) |
                    static_cast<unsigned char>(buffer[*offset + 1]) << 8 |
                    static_cast<unsigned char>(buffer[*offset + 2]) << 16 |
                    static_cast<unsigned char>(buffer[*offset + 3]) << 24;

  *offset += sizeof(uint32_t);

  if (*offset + length > bytes)
    return false;

  *storage = std::string(&buffer[*offset], length);
  *offset += length;

  return true;
}

// -------------------------------------------------------------------------------------------------

// Queries the server defined in |info|. Writes the results to |info| as well.
bool QueryServer(ServerInfo* info, boost::asio::io_service& io_service) {
  unsigned char packet[11];

  boost::asio::ip::address_v4 address = boost::asio::ip::address_v4::from_string(info->server.first);
  unsigned short port = info->server.second;

  // Compile the information in |packet| necessary to query the server.
  {
    unsigned long long_address = address.to_ulong();
    unsigned short port = info->server.second;

    packet[0] = 'S';
    packet[1] = 'A';
    packet[2] = 'M';
    packet[3] = 'P';

    static_assert(sizeof(long_address) == 4, "unsigned long must be four bytes");
    memcpy(&packet[4], &long_address, sizeof(long_address));

    packet[8] = port & 0xFF;
    packet[9] = (port >> 8) & 0xFF;

    packet[10] = 'i';  // information packet identifier
  }

  using boost::asio::ip::udp;
  try {
    udp::endpoint endpoint;
    endpoint.address(address);
    endpoint.port(port);

    udp::socket socket(io_service);

#if defined(WIN32)
    uint32_t timeout = options.timeout * 1000;
    setsockopt(socket.native(), SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout, sizeof(timeout));
    setsockopt(socket.native(), SOL_SOCKET, SO_SNDTIMEO, (const char*)&timeout, sizeof(timeout));
#else
    struct timeval tv;
    tv.tv_sec = options.timeout;
    tv.tv_usec = 0;
    setsockopt(socket.native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
    setsockopt(socket.native(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
#endif

    socket.open(endpoint.protocol());

    boost::array<char, 256> receive_buffer;

    // Send the |packet| to the |endpoint| over the opened |socket|.
    if (socket.send_to(boost::asio::buffer(packet), endpoint) != sizeof(packet))
      return false;

    size_t bytes = socket.receive_from(boost::asio::buffer(receive_buffer), endpoint);
    size_t offset = 11;  // the |packet| that we sent

    if (bytes <= offset)
      throw std::exception("Received packet size too short");

    return ReadValue(&info->has_password, receive_buffer, bytes, &offset) &&
           ReadValue(&info->players, receive_buffer, bytes, &offset) &&
           ReadValue(&info->max_players, receive_buffer, bytes, &offset) &&
           ReadValue(&info->hostname, receive_buffer, bytes, &offset) &&
           ReadValue(&info->gamemode, receive_buffer, bytes, &offset) &&
           ReadValue(&info->map, receive_buffer, bytes, &offset);

  } catch (std::exception& e) {
    if (options.verbose) {
      std::lock_guard<std::mutex> guard(verbose_output_mutex);
      std::cerr << "[" << ToDisplay(info->server) << "] ERROR: " << e.what() << std::endl;
    }
    return false;
  }

  return true;
}

// Thread responsible for querying servers. It will pick a ServerEntry from the vector of parsed
// server entries, query it, and then write the results to the vector of queried servers. Both
// vectors are protected by mutexes
void QueryThread() {
  boost::asio::io_service io_service;
  ServerEntry server;

  while (true) {
    {
      std::lock_guard<std::mutex> guard(server_queue_mutex);
      if (server_queue.empty())
        return;

      server = server_queue.front();
      server_queue.pop();
    }

    double begin = monotonicallyIncreasingTime();

    ServerInfo info(server);
    if (options.verbose) {
      std::lock_guard<std::mutex> guard(verbose_output_mutex);
      std::cout << "[" << ToDisplay(server) << "] Querying the server..." << std::endl;
    }

    const bool success = QueryServer(&info, io_service);

    if (options.verbose) {
      double delta = monotonicallyIncreasingTime() - begin;

      std::lock_guard<std::mutex> guard(verbose_output_mutex);
      std::cout << "[" << ToDisplay(server) << "] Finished the query in "
                << std::setprecision(2) << delta << "ms" << std::endl;
    }

    {
      std::lock_guard<std::mutex> guard(result_mutex);
      if (success)
        result_vector.push_back(std::move(info));
      else
        failure_vector.push_back(server);
    }
  }
}

}  // namespace

// -------------------------------------------------------------------------------------------------

int main(int argc, const char** argv) {
  if (!ParseCommandLine(&options, argc, argv))
    return 1;

  if (options.verbose)
    std::cout << "Parsing servers from '" << options.server_list << "'..." << std::endl;

  if (!FetchServerList(&server_queue))
    return 1;

  if (options.verbose)
    std::cout << "Found " << server_queue.size() << " servers..." << std::endl;

  if (!server_queue.size())
    return 0;  // there are no servers to index

  // Now query all servers in the list in parallel, using |options.threads| threads.
  {
    boost::thread_group query_threads;
    for (size_t i = 0; i < options.threads; ++i)
      query_threads.create_thread(QueryThread);

    query_threads.join_all();
  }

  // TODO: Write the results to `logstash` when this has been configured.
  for (const ServerInfo& info : result_vector) {
    std::cout << info.server.first << ":" << info.server.second << " -- " << info.hostname << " -- "
{'fraction_non_alphanumeric': 0.17720943818766025, 'fraction_numerical': 0.010418357080263542, 'mean_word_length': 3.061692591616135, 'pattern_counts': {'":': 3, '<': 354, '<?xml version=': 0, '>': 161, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '1719430', 'n_tokens_mistral': 9667, 'n_tokens_neox': 9185, 'n_words': 2470}