<commit_before>/******************************************************************************
 *                    _   _____   __________                                  *
 *                   | | / / _ | / __/_  __/     Visibility                   *
 *                   | |/ / __ |_\ \  / /          Across                     *
 *                   |___/_/ |_/___/ /_/       Space and Time                 *
 *                                                                            *
 * This file is part of VAST. It is subject to the license terms in the       *
 * LICENSE file found in the top-level directory of this distribution and at  *
 * http://vast.io/license. No part of VAST, including this file, may be       *
 * copied, modified, propagated, or distributed except according to the terms *
 * contained in the LICENSE file.                                             *
 ******************************************************************************/

#include <caf/all.hpp>

#include "vast/concept/printable/std/chrono.hpp"
#include "vast/concept/printable/to_string.hpp"
#include "vast/concept/printable/vast/event.hpp"
#include "vast/concept/printable/vast/expression.hpp"
#include "vast/concept/printable/vast/uuid.hpp"
#include "vast/detail/assert.hpp"
#include "vast/event.hpp"
#include "vast/expression_visitors.hpp"
#include "vast/logger.hpp"
#include "vast/table_slice.hpp"
#include "vast/to_events.hpp"

#include "vast/system/archive.hpp"
#include "vast/system/atoms.hpp"
#include "vast/system/exporter.hpp"

#include "vast/detail/assert.hpp"

using namespace std::chrono;
using namespace std::string_literals;
using namespace caf;

namespace vast {
namespace system {

namespace {

void ship_results(stateful_actor<exporter_state>* self) {
  VAST_TRACE("");
  if (self->state.results.empty() || self->state.query.requested == 0) {
    return;
  }
  VAST_INFO(self, "relays", self->state.results.size(), "events");
  message msg;
  if (self->state.results.size() <= self->state.query.requested) {
    self->state.query.requested -= self->state.results.size();
    self->state.query.shipped += self->state.results.size();
    msg = make_message(std::move(self->state.results));
    self->state.results = {};
  } else {
    std::vector<event> remainder;
    remainder.reserve(self->state.results.size() - self->state.query.requested);
    auto begin = self->state.results.begin() + self->state.query.requested;
    auto end = self->state.results.end();
    std::move(begin, end, std::back_inserter(remainder));
    self->state.results.resize(self->state.query.requested);
    msg = make_message(std::move(self->state.results));
    self->state.results = std::move(remainder);
    self->state.query.shipped += self->state.query.requested;
    self->state.query.requested = 0;
  }
  self->send(self->state.sink, msg);
}

void report_statistics(stateful_actor<exporter_state>* self) {
  timespan runtime = steady_clock::now() - self->state.start;
  self->state.query.runtime = runtime;
  VAST_INFO(self, "completed in", runtime);
  self->send(self->state.sink, self->state.id, self->state.query);
  if (self->state.accountant) {
    auto hits = rank(self->state.hits);
    auto processed = self->state.query.processed;
    auto shipped = self->state.query.shipped;
    auto results = shipped + self->state.results.size();
    auto selectivity = double(results) / hits;
    self->send(self->state.accountant, "exporter.hits", hits);
    self->send(self->state.accountant, "exporter.processed", processed);
    self->send(self->state.accountant, "exporter.results", results);
    self->send(self->state.accountant, "exporter.shipped", shipped);
    self->send(self->state.accountant, "exporter.selectivity", selectivity);
    self->send(self->state.accountant, "exporter.runtime", runtime);
  }
}

void shutdown(stateful_actor<exporter_state>* self, caf::error err) {
  VAST_DEBUG(self, "initiates shutdown with error", self->system().render(err));
  self->send_exit(self, std::move(err));
}

void shutdown(stateful_actor<exporter_state>* self) {
  if (has_continuous_option(self->state.options))
    return;
  VAST_DEBUG(self, "initiates shutdown");
  self->send_exit(self, exit_reason::normal);
}

void request_more_hits(stateful_actor<exporter_state>* self) {
  const auto& st = self->state;
  if (!has_historical_option(st.options))
    return;
  auto waiting_for_hits =
    st.query.received == st.query.scheduled;
  auto need_more_results = st.query.requested > 0;
  auto have_no_inflight_requests =
    st.query.lookups_issued == st.query.lookups_complete;
  // If we're (1) no longer waiting for index hits, (2) still need more
  // results, and (3) have no inflight requests to the archive, we ask
  // the index for more hits.
  if (!waiting_for_hits && need_more_results && have_no_inflight_requests) {
    auto remaining = st.query.expected - st.query.received;
    VAST_ASSERT(remaining > 0);
    // TODO: Figure out right number of partitions to ask for. For now, we
    // bound the number by an arbitrary constant.
    auto n = std::min(remaining, size_t{2});
    VAST_DEBUG(self, "asks index to process", n, "more partitions");
    self->send(st.index, st.id, n);
  }
}

} // namespace <anonymous>

behavior exporter(stateful_actor<exporter_state>* self, expression expr,
                  query_options options) {
  auto eu = self->system().dummy_execution_unit();
  self->state.sink = actor_pool::make(eu, actor_pool::broadcast());
  if (auto a = self->system().registry().get(accountant_atom::value))
    self->state.accountant = actor_cast<accountant_type>(a);
  self->state.options = options;
  if (has_continuous_option(options))
    VAST_DEBUG(self, "has continuous query option");
  self->set_exit_handler(
    [=](const exit_msg& msg) {
      VAST_DEBUG(self, "received exit from", msg.source, "with reason:", msg.reason);
      self->send<message_priority::high>(self->state.index, self->state.id, 0);
      self->send(self->state.sink, sys_atom::value, delete_atom::value);<commit_msg>Make exporter announce itself to the accountant<commit_after>/******************************************************************************
 *                    _   _____   __________                                  *
 *                   | | / / _ | / __/_  __/     Visibility                   *
 *                   | |/ / __ |_\ \  / /          Across                     *
 *                   |___/_/ |_/___/ /_/       Space and Time                 *
 *                                                                            *
 * This file is part of VAST. It is subject to the license terms in the       *
 * LICENSE file found in the top-level directory of this distribution and at  *
 * http://vast.io/license. No part of VAST, including this file, may be       *
 * copied, modified, propagated, or distributed except according to the terms *
 * contained in the LICENSE file.                                             *
 ******************************************************************************/

#include <caf/all.hpp>

#include "vast/concept/printable/std/chrono.hpp"
#include "vast/concept/printable/to_string.hpp"
#include "vast/concept/printable/vast/event.hpp"
#include "vast/concept/printable/vast/expression.hpp"
#include "vast/concept/printable/vast/uuid.hpp"
#include "vast/detail/assert.hpp"
#include "vast/event.hpp"
#include "vast/expression_visitors.hpp"
#include "vast/logger.hpp"
#include "vast/table_slice.hpp"
#include "vast/to_events.hpp"

#include "vast/system/archive.hpp"
#include "vast/system/atoms.hpp"
#include "vast/system/exporter.hpp"

#include "vast/detail/assert.hpp"

using namespace std::chrono;
using namespace std::string_literals;
using namespace caf;

namespace vast {
namespace system {

namespace {

void ship_results(stateful_actor<exporter_state>* self) {
  VAST_TRACE("");
  if (self->state.results.empty() || self->state.query.requested == 0) {
    return;
  }
  VAST_INFO(self, "relays", self->state.results.size(), "events");
  message msg;
  if (self->state.results.size() <= self->state.query.requested) {
    self->state.query.requested -= self->state.results.size();
    self->state.query.shipped += self->state.results.size();
    msg = make_message(std::move(self->state.results));
    self->state.results = {};
  } else {
    std::vector<event> remainder;
    remainder.reserve(self->state.results.size() - self->state.query.requested);
    auto begin = self->state.results.begin() + self->state.query.requested;
    auto end = self->state.results.end();
    std::move(begin, end, std::back_inserter(remainder));
    self->state.results.resize(self->state.query.requested);
    msg = make_message(std::move(self->state.results));
    self->state.results = std::move(remainder);
    self->state.query.shipped += self->state.query.requested;
    self->state.query.requested = 0;
  }
  self->send(self->state.sink, msg);
}

void report_statistics(stateful_actor<exporter_state>* self) {
  timespan runtime = steady_clock::now() - self->state.start;
  self->state.query.runtime = runtime;
  VAST_INFO(self, "completed in", runtime);
  self->send(self->state.sink, self->state.id, self->state.query);
  if (self->state.accountant) {
    auto hits = rank(self->state.hits);
    auto processed = self->state.query.processed;
    auto shipped = self->state.query.shipped;
    auto results = shipped + self->state.results.size();
    auto selectivity = double(results) / hits;
    self->send(self->state.accountant, "exporter.hits", hits);
    self->send(self->state.accountant, "exporter.processed", processed);
    self->send(self->state.accountant, "exporter.results", results);
    self->send(self->state.accountant, "exporter.shipped", shipped);
    self->send(self->state.accountant, "exporter.selectivity", selectivity);
    self->send(self->state.accountant, "exporter.runtime", runtime);
  }
}

void shutdown(stateful_actor<exporter_state>* self, caf::error err) {
  VAST_DEBUG(self, "initiates shutdown with error", self->system().render(err));
  self->send_exit(self, std::move(err));
}

void shutdown(stateful_actor<exporter_state>* self) {
  if (has_continuous_option(self->state.options))
    return;
  VAST_DEBUG(self, "initiates shutdown");
  self->send_exit(self, exit_reason::normal);
}

void request_more_hits(stateful_actor<exporter_state>* self) {
  const auto& st = self->state;
  if (!has_historical_option(st.options))
    return;
  auto waiting_for_hits =
    st.query.received == st.query.scheduled;
  auto need_more_results = st.query.requested > 0;
  auto have_no_inflight_requests =
    st.query.lookups_issued == st.query.lookups_complete;
  // If we're (1) no longer waiting for index hits, (2) still need more
  // results, and (3) have no inflight requests to the archive, we ask
  // the index for more hits.
  if (!waiting_for_hits && need_more_results && have_no_inflight_requests) {
    auto remaining = st.query.expected - st.query.received;
    VAST_ASSERT(remaining > 0);
    // TODO: Figure out right number of partitions to ask for. For now, we
    // bound the number by an arbitrary constant.
    auto n = std::min(remaining, size_t{2});
    VAST_DEBUG(self, "asks index to process", n, "more partitions");
    self->send(st.index, st.id, n);
  }
}

} // namespace <anonymous>

behavior exporter(stateful_actor<exporter_state>* self, expression expr,
                  query_options options) {
  auto eu = self->system().dummy_execution_unit();
  self->state.sink = actor_pool::make(eu, actor_pool::broadcast());
  if (auto a = self->system().registry().get(accountant_atom::value)) {
    self->state.accountant = actor_cast<accountant_type>(a);
    self->send(self->state.accountant, "announce", self->name());
  }
  self->state.options = options;
  if (has_continuous_option(options))
    VAST_DEBUG(self, "has continuous query option");
  self->set_exit_handler(
    [=](const exit_msg& msg) {
      VAST_DEBUG(self, "received exit from", msg.source, "with reason:", msg.reason);
      self->send<message_priority::high>(self->state.index, self->state.id, 0);
      self->send(self->state.sink, sys_atom::value, delete_atom::value);
{'fraction_non_alphanumeric': 0.1685264194289465, 'fraction_numerical': 0.0014768624876928126, 'mean_word_length': 3.416304347826087, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 162, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4128468', 'n_tokens_mistral': 3858, 'n_tokens_neox': 3724, 'n_words': 762}