<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/const_table_slice_handle.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/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) {
  if (self->state.results.empty() || self->state.stats.requested == 0)
    return;
  VAST_DEBUG(self, "relays", self->state.results.size(), "events");
  message msg;
  if (self->state.results.size() <= self->state.stats.requested) {
    self->state.stats.requested -= self->state.results.size();
    self->state.stats.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.stats.requested);
    auto begin = self->state.results.begin() + self->state.stats.requested;
    auto end = self->state.results.end();
    std::move(begin, end, std::back_inserter(remainder));
    self->state.results.resize(self->state.stats.requested);
    msg = make_message(std::move(self->state.results));
    self->state.results = std::move(remainder);
    self->state.stats.shipped += self->state.stats.requested;
    self->state.stats.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.stats.runtime = runtime;
  VAST_DEBUG(self, "completed in", runtime);
  self->send(self->state.sink, self->state.id, self->state.stats);
  if (self->state.accountant) {
    auto hits = rank(self->state.hits);
    auto processed = self->state.stats.processed;
    auto shipped = self->state.stats.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) {
  if (rank(self->state.unprocessed) > 0 || !self->state.results.empty()
      || 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) {
  if (!has_historical_option(self->state.options))
    return;
  auto waiting_for_hits =
    self->state.stats.received == self->state.stats.scheduled;
  auto need_more_results = self->state.stats.requested > 0;
  auto have_no_inflight_requests = any<1>(self->state.unprocessed);
  // 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 = self->state.stats.expected - self->state.stats.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(self->state.index, self->state.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) {
      self->send(self->state.sink, sys_atom::value, delete_atom::value);
      self->send_exit(self->state.sink, msg.reason);
      self->quit(msg.reason);
      if (msg.reason != exit_reason::kill)
        report_statistics(self);
    }
  );
  self->set_down_handler(
    [=](const down_msg& msg) {
      VAST_DEBUG(self, "received DOWN from", msg.source);
      if (has_continuous_option(self->state.options)
          && (msg.source == self->state.archive
              || msg.source == self->state.index))
        report_statistics(self);
    }
  );
  auto handle_batch = [=](std::vector<event>& candidates) {
    VAST_DEBUG(self, "got batch of", candidates.size(), "events");
    // Events can arrive in any order: sort them by ID first. Otherwise, we
    // can't compute the bitmap mask as easily.
    std::sort(candidates.begin(), candidates.end(),
              [](auto& x, auto& y) { return x.id() < y.id(); });
    bitmap mask;
    auto sender = self->current_sender();
    for (auto& candidate : candidates) {
      auto& checker = self->state.checkers[candidate.type()];
      // Construct a candidate checker if we don't have one for this type.
      if (caf::holds_alternative<caf::none_t>(checker)) {
        auto x = tailor(expr, candidate.type());
        if (!x) {
          VAST_ERROR(self, "failed to tailor expression:",
                     self->system().render(x.error()));
          ship_results(self);
          self->send_exit(self, exit_reason::normal);
          return;
        }
        checker = std::move(*x);
        VAST_DEBUG(self, "tailored AST to", candidate.type() << ':', checker);
      }
      // Perform candidate check and keep event as result on success.
      if (caf::visit(event_evaluator{candidate}, checker))
        self->state.results.push_back(std::move(candidate));
      else
        VAST_DEBUG(self, "ignores false positive:", candidate);
      if (sender == self->state.archive) {
        mask.append_bits(false, candidate.id() - mask.size());
        mask.append_bit(true);
      }
    }
    self->state.stats.processed += candidates.size();
    if (sender == self->state.archive)
      self->state.unprocessed -= mask;
    ship_results(self);
    request_more_hits(self);
    if (self->state.stats.received == self->state.stats.expected)
      shutdown(self);
  };<commit_msg>Fix ordering to avoid accessing moved-from object<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/const_table_slice_handle.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/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) {
  if (self->state.results.empty() || self->state.stats.requested == 0)
    return;
  VAST_DEBUG(self, "relays", self->state.results.size(), "events");
  message msg;
  if (self->state.results.size() <= self->state.stats.requested) {
    self->state.stats.requested -= self->state.results.size();
    self->state.stats.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.stats.requested);
    auto begin = self->state.results.begin() + self->state.stats.requested;
    auto end = self->state.results.end();
    std::move(begin, end, std::back_inserter(remainder));
    self->state.results.resize(self->state.stats.requested);
    msg = make_message(std::move(self->state.results));
    self->state.results = std::move(remainder);
    self->state.stats.shipped += self->state.stats.requested;
    self->state.stats.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.stats.runtime = runtime;
  VAST_DEBUG(self, "completed in", runtime);
  self->send(self->state.sink, self->state.id, self->state.stats);
  if (self->state.accountant) {
    auto hits = rank(self->state.hits);
    auto processed = self->state.stats.processed;
    auto shipped = self->state.stats.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) {
  if (rank(self->state.unprocessed) > 0 || !self->state.results.empty()
      || 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) {
  if (!has_historical_option(self->state.options))
    return;
  auto waiting_for_hits =
    self->state.stats.received == self->state.stats.scheduled;
  auto need_more_results = self->state.stats.requested > 0;
  auto have_no_inflight_requests = any<1>(self->state.unprocessed);
  // 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 = self->state.stats.expected - self->state.stats.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(self->state.index, self->state.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) {
      self->send(self->state.sink, sys_atom::value, delete_atom::value);
      self->send_exit(self->state.sink, msg.reason);
      self->quit(msg.reason);
      if (msg.reason != exit_reason::kill)
        report_statistics(self);
    }
  );
  self->set_down_handler(
    [=](const down_msg& msg) {
      VAST_DEBUG(self, "received DOWN from", msg.source);
      if (has_continuous_option(self->state.options)
          && (msg.source == self->state.archive
              || msg.source == self->state.index))
        report_statistics(self);
    }
  );
  auto handle_batch = [=](std::vector<event>& candidates) {
    VAST_DEBUG(self, "got batch of", candidates.size(), "events");
    // Events can arrive in any order: sort them by ID first. Otherwise, we
    // can't compute the bitmap mask as easily.
    std::sort(candidates.begin(), candidates.end(),
              [](auto& x, auto& y) { return x.id() < y.id(); });
    bitmap mask;
    auto sender = self->current_sender();
    for (auto& candidate : candidates) {
      auto& checker = self->state.checkers[candidate.type()];
      // Construct a candidate checker if we don't have one for this type.
      if (caf::holds_alternative<caf::none_t>(checker)) {
        auto x = tailor(expr, candidate.type());
        if (!x) {
          VAST_ERROR(self, "failed to tailor expression:",
                     self->system().render(x.error()));
          ship_results(self);
          self->send_exit(self, exit_reason::normal);
          return;
        }
        checker = std::move(*x);
        VAST_DEBUG(self, "tailored AST to", candidate.type() << ':', checker);
      }
      // Append ID to our bitmap mask.
      if (sender == self->state.archive) {
        mask.append_bits(false, candidate.id() - mask.size());
        mask.append_bit(true);
      }
      // Perform candidate check and keep event as result on success.
      if (caf::visit(event_evaluator{candidate}, checker))
        self->state.results.push_back(std::move(candidate));
      else
        VAST_DEBUG(self, "ignores false positive:", candidate);
    }
    self->state.stats.processed += candidates.size();
    if (sender == self->state.archive)
      self->state.unprocessed -= mask;
    ship_results(self);
    request_more_hits(self);
    if (self->state.stats.received == self->state.stats.expected)
      shutdown(self);
  };
{'fraction_non_alphanumeric': 0.16479354237814342, 'fraction_numerical': 0.0012418503570319776, 'mean_word_length': 3.1318624935864547, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 209, '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': '4506287', 'n_tokens_mistral': 5016, 'n_tokens_neox': 4861, 'n_words': 1016}