<commit_before>#include "rpc/rpc_server.h"
#include "log.h"
#include "rpc/rpc_envelope.h"
namespace smf {

rpc_server::rpc_server(distributed<rpc_server_stats> &stats,
                       uint16_t port,
                       uint32_t flags)
  : stats_(stats), port_(port), flags_(flags) {}

void rpc_server::start() {
  LOG_INFO("Starging server. port: {}, flags:{}, limits:{} ", port_, flags_,
           *limits_.get());
  listen_options lo;
  lo.reuse_address = true;
  listener_ = engine().listen(make_ipv4_address({port_}), lo);
  keep_doing([this] {
    return listener_->accept().then(
      [this](connected_socket fd, socket_address addr) mutable {
        auto conn =
          make_lw_shared<rpc_server_connection>(std::move(fd), addr, stats_);
        // DO NOT return the future. Need to execute in parallel
        handle_client_connection(conn);
      });
  })
    .or_terminate(); // end keep_doing
}

future<> rpc_server::stop() {
  // TODO(agallego) - do we need to abort_accept()
  // seastar does this
  // LOG_INFO("Stopping rpc server, aborting_accept()");
  // listener_->abort_accept();
  LOG_INFO("Saving rpc_distpach histogram");
  if((flags_ & RPCFLAGS::RPCFLAGS_PRINT_HISTOGRAM_ON_EXIT)
     == RPCFLAGS::RPCFLAGS_PRINT_HISTOGRAM_ON_EXIT) {
    try {
      sstring s = "rpc_dispatch_histogram_shard_"
                  + to_sstring(engine().cpu_id()) + ".txt";
      FILE *fp = fopen(s.c_str(), "w");
      if(fp) {
        hist_->print(fp);
        fclose(fp);
      }
    } catch(...) {
    }
  }
  LOG_INFO("Closing reply_gate");
  return limits_->reply_gate.close();
}

future<> rpc_server::handle_client_connection(
  lw_shared_ptr<rpc_server_connection> conn) {
  return do_until(
    [conn] { return !conn->is_valid(); },
    [this, conn]() mutable {
      return rpc_recv_context::parse(conn.get(), limits_.get())
        .then([this, conn](auto recv_ctx) mutable {
          if(!recv_ctx) {
            conn->set_error("Could not parse the request");
            return make_ready_future<>();
          }
          auto metric = hist_->auto_measure();
          auto payload_size = recv_ctx.value().body_buf.size();
          return this->dispatch_rpc(conn, std::move(recv_ctx.value()))
            .finally([this, conn, metric = std::move(metric), payload_size] {
              limits_->release_payload_resources(payload_size);
              return conn->ostream.flush().finally([this, conn] {
                if(conn->has_error()) {
                  LOG_ERROR("There was an error with the connection: {}",
                            conn->get_error());
                  stats_.local().bad_requests++;
                  stats_.local().active_connections--;
                  LOG_INFO("Closing connection for client: {}",
                           conn->remote_address);
                  return conn->ostream.close();
                } else {
                  stats_.local().completed_requests++;
                  return make_ready_future<>();
                }
              });
            }); // end finally()
        })      //  parse_rpc_recv_context.then()
        .handle_exception([this, conn](std::exception_ptr eptr) {
          conn->set_error("Exception parsing request ");
          return conn->ostream.close();
        });
    });
}

/// brief - applies a function future<T> apply(T &&t); to all the filters
/// useful for incoming and outgoing filters. Taking a pair of iterators to
/// unique ptrs. see incoming_filter.h and outgoing_filter.h
/// for details
///
template <typename Iterator, typename Arg, typename... Ret>
future<Ret...>
move_filter_apply(const Iterator &b, const Iterator &end, Arg &&arg) {
  auto begin = b;
  if(begin == end) {
    return make_ready_future<Ret...>(std::forward<Arg>(arg));
  }
  return (*begin)
    ->apply(std::forward<Arg>(arg))
    .then([begin = begin++, end](Arg && a) mutable {
      return move_filter_apply<Iterator, Arg, Ret...>(begin, end,<commit_msg>cosmetic: remove 'mutable' from non-mutating fns<commit_after>#include "rpc/rpc_server.h"
#include "log.h"
#include "rpc/rpc_envelope.h"
namespace smf {

rpc_server::rpc_server(distributed<rpc_server_stats> &stats,
                       uint16_t port,
                       uint32_t flags)
  : stats_(stats), port_(port), flags_(flags) {}

void rpc_server::start() {
  LOG_INFO("Starging server. port: {}, flags:{}, limits:{} ", port_, flags_,
           *limits_.get());
  listen_options lo;
  lo.reuse_address = true;
  listener_ = engine().listen(make_ipv4_address({port_}), lo);
  keep_doing([this] {
    return listener_->accept().then(
      [this](connected_socket fd, socket_address addr) mutable {
        auto conn =
          make_lw_shared<rpc_server_connection>(std::move(fd), addr, stats_);
        // DO NOT return the future. Need to execute in parallel
        handle_client_connection(conn);
      });
  })
    .or_terminate(); // end keep_doing
}

future<> rpc_server::stop() {
  // TODO(agallego) - do we need to abort_accept()
  // seastar does this
  // LOG_INFO("Stopping rpc server, aborting_accept()");
  // listener_->abort_accept();
  LOG_INFO("Saving rpc_distpach histogram");
  if((flags_ & RPCFLAGS::RPCFLAGS_PRINT_HISTOGRAM_ON_EXIT)
     == RPCFLAGS::RPCFLAGS_PRINT_HISTOGRAM_ON_EXIT) {
    try {
      sstring s = "rpc_dispatch_histogram_shard_"
                  + to_sstring(engine().cpu_id()) + ".txt";
      FILE *fp = fopen(s.c_str(), "w");
      if(fp) {
        hist_->print(fp);
        fclose(fp);
      }
    } catch(...) {
    }
  }
  LOG_INFO("Closing reply_gate");
  return limits_->reply_gate.close();
}

future<> rpc_server::handle_client_connection(
  lw_shared_ptr<rpc_server_connection> conn) {
  return do_until(
    [conn] { return !conn->is_valid(); },
    [this, conn]() mutable {
      return rpc_recv_context::parse(conn.get(), limits_.get())
        .then([this, conn](auto recv_ctx) mutable {
          if(!recv_ctx) {
            conn->set_error("Could not parse the request");
            return make_ready_future<>();
          }
          auto metric = hist_->auto_measure();
          auto payload_size = recv_ctx.value().body_buf.size();
          return this->dispatch_rpc(conn, std::move(recv_ctx.value()))
            .finally([this, conn, metric = std::move(metric), payload_size] {
              limits_->release_payload_resources(payload_size);
              return conn->ostream.flush().finally([this, conn] {
                if(conn->has_error()) {
                  LOG_ERROR("There was an error with the connection: {}",
                            conn->get_error());
                  stats_.local().bad_requests++;
                  stats_.local().active_connections--;
                  LOG_INFO("Closing connection for client: {}",
                           conn->remote_address);
                  return conn->ostream.close();
                } else {
                  stats_.local().completed_requests++;
                  return make_ready_future<>();
                }
              });
            }); // end finally()
        })      //  parse_rpc_recv_context.then()
        .handle_exception([this, conn](std::exception_ptr eptr) {
          conn->set_error("Exception parsing request ");
          return conn->ostream.close();
        });
    });
}

/// brief - applies a function future<T> apply(T &&t); to all the filters
/// useful for incoming and outgoing filters. Taking a pair of iterators to
/// unique ptrs. see incoming_filter.h and outgoing_filter.h
/// for details
///
template <typename Iterator, typename Arg, typename... Ret>
future<Ret...>
move_filter_apply(const Iterator &b, const Iterator &end, Arg &&arg) {
  Iterator begin = b;
  if(begin == end) {
    return make_ready_future<Ret...>(std::forward<Arg>(arg));
  }
  return (*begin)
    ->apply(std::forward<Arg>(arg))
    .then([begin = begin++, end](Arg && a) {
      return move_filter_apply<Iterator, Arg, Ret...>(begin, end,
{'fraction_non_alphanumeric': 0.1457835867789368, 'fraction_numerical': 0.0012567550584391101, 'mean_word_length': 2.5119152691968227, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 65, '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': '1267668', 'n_tokens_mistral': 2423, 'n_tokens_neox': 2375, 'n_words': 495}