<commit_before>/*
 * Copyright (C) 2017 ScyllaDB
 */

/*
 * This file is part of Scylla.
 *
 * Scylla is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Scylla is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <boost/algorithm/string/replace.hpp>
#include <boost/range/irange.hpp>
#include "tests/cql_test_env.hh"
#include "tests/perf/perf.hh"
#include "core/app-template.hh"
#include "schema_builder.hh"
#include "database.hh"
#include "db/config.hh"
#include "partition_slice_builder.hh"
#include <seastar/core/reactor.hh>
#include "transport/messages/result_message.hh"

#include "disk-error-handler.hh"

thread_local disk_error_signal_type commit_error;
thread_local disk_error_signal_type general_disk_error;

using namespace std::chrono_literals;
using int_range = nonwrapping_range<int>;

reactor::io_stats s;

static bool errors_found = false;

static void print_error(const sstring& msg) {
    std::cout << "^^^ ERROR: " << msg << "\n";
    errors_found = true;
}

struct metrics_snapshot {
    std::chrono::high_resolution_clock::time_point hr_clock;
    steady_clock_type::duration busy_time;
    steady_clock_type::duration idle_time;
    reactor::io_stats io;
    sstables::shared_index_lists::stats index;
    cache_tracker::stats cache;

    metrics_snapshot() {
        reactor& r = *local_engine;
        io = r.get_io_stats();
        busy_time = r.total_busy_time();
        idle_time = r.total_idle_time();
        hr_clock = std::chrono::high_resolution_clock::now();
        index = sstables::shared_index_lists::shard_stats();
        cache = global_cache_tracker().get_stats();
    }
};

class make_printable {
    using func_type = std::function<void(std::ostream&)>;
    func_type _func;
public:
    make_printable(func_type func) : _func(std::move(func)) {}
    friend std::ostream& operator<<(std::ostream& out, const make_printable& p) {
        p._func(out);
        return out;
    }
};

struct test_result {
    uint64_t fragments_read;
    metrics_snapshot before;
    metrics_snapshot after;

    test_result(metrics_snapshot before, uint64_t fragments_read)
        : fragments_read(fragments_read)
        , before(before)
    { }

    double duration_in_seconds() const {
        return std::chrono::duration<double>(after.hr_clock - before.hr_clock).count();
    }

    double fragment_rate() const { return double(fragments_read) / duration_in_seconds(); }

    uint64_t aio_reads() const { return after.io.aio_reads - before.io.aio_reads; }
    uint64_t aio_read_bytes() const { return after.io.aio_read_bytes - before.io.aio_read_bytes; }
    uint64_t read_aheads_discarded() const { return after.io.fstream_read_aheads_discarded - before.io.fstream_read_aheads_discarded; }
    uint64_t reads_blocked() const { return after.io.fstream_reads_blocked - before.io.fstream_reads_blocked; }

    uint64_t index_hits() const { return after.index.hits - before.index.hits; }
    uint64_t index_misses() const { return after.index.misses - before.index.misses; }
    uint64_t index_blocks() const { return after.index.blocks - before.index.blocks; }

    uint64_t cache_hits() const { return after.cache.partition_hits - before.cache.partition_hits; }
    uint64_t cache_misses() const { return after.cache.partition_misses - before.cache.partition_misses; }
    uint64_t cache_insertions() const { return after.cache.partition_insertions - before.cache.partition_insertions; }

    float cpu_utilization() const {
        auto busy_delta = after.busy_time.count() - before.busy_time.count();
        auto idle_delta = after.idle_time.count() - before.idle_time.count();
        return float(busy_delta) / (busy_delta + idle_delta);
    }

    static auto table_header() {
        return make_printable([] (std::ostream& out) {
            out << sprint("%10s %9s %10s %6s %10s %7s %7s %8s %8s %8s %8s %8s %8s %6s",
                "time [s]", "frags", "frag/s", "aio", "[KiB]", "blocked", "dropped",
                "idx hit", "idx miss", "idx blk",
                "c hit", "c miss", "c ins",
                "cpu");
        });
    }

    auto table_row() {
        return make_printable([this] (std::ostream& out) {
            out << sprint("%10.6f %9d %10.0f %6d %10d %7d %7d %8d %8d %8d %8d %8d %8d %5.1f%%",
                duration_in_seconds(), fragments_read, fragment_rate(),
                aio_reads(), aio_read_bytes() / 1024, reads_blocked(), read_aheads_discarded(),
                index_hits(), index_misses(), index_blocks(),
                cache_hits(), cache_misses(), cache_insertions(),
                cpu_utilization() * 100);
        });
    }
};

static void check_no_disk_reads(const test_result& r) {
    if (r.aio_reads()) {
        print_error("Expected no disk reads");
    }
}

static void check_no_index_reads(const test_result& r) {
    if (r.index_hits() || r.index_misses()) {
        print_error("Expected no index reads");
    }
}

static void check_fragment_count(const test_result& r, uint64_t expected) {
    if (r.fragments_read != expected) {
        print_error(sprint("Expected to read %d fragments", expected));
    }
}

static
uint64_t consume_all(streamed_mutation& sm) {
    class consumer {
        uint64_t _fragments = 0;
    public:
        stop_iteration consume(tombstone) { return stop_iteration::no; }
        stop_iteration consume(...) { _fragments++; return stop_iteration::no; }
        uint64_t consume_end_of_stream() { return _fragments; }
    };
    return consume(sm, consumer()).get0();
}

static
uint64_t consume_all(mutation_reader& rd) {
    uint64_t fragments = 0;
    while (1) {
        streamed_mutation_opt smo = rd().get0();
        if (!smo) {
            break;
        }
        fragments += consume_all(*smo);
    }
    return fragments;
}

// cf should belong to ks.test
static test_result scan_rows_with_stride(column_family& cf, int n_rows, int n_read = 1, int n_skip = 0) {<commit_msg>perf/perf_fast_forward: Don't pass non-pod to varargs function<commit_after>/*
 * Copyright (C) 2017 ScyllaDB
 */

/*
 * This file is part of Scylla.
 *
 * Scylla is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Scylla is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <boost/algorithm/string/replace.hpp>
#include <boost/range/irange.hpp>
#include "tests/cql_test_env.hh"
#include "tests/perf/perf.hh"
#include "core/app-template.hh"
#include "schema_builder.hh"
#include "database.hh"
#include "db/config.hh"
#include "partition_slice_builder.hh"
#include <seastar/core/reactor.hh>
#include "transport/messages/result_message.hh"

#include "disk-error-handler.hh"

thread_local disk_error_signal_type commit_error;
thread_local disk_error_signal_type general_disk_error;

using namespace std::chrono_literals;
using int_range = nonwrapping_range<int>;

reactor::io_stats s;

static bool errors_found = false;

static void print_error(const sstring& msg) {
    std::cout << "^^^ ERROR: " << msg << "\n";
    errors_found = true;
}

struct metrics_snapshot {
    std::chrono::high_resolution_clock::time_point hr_clock;
    steady_clock_type::duration busy_time;
    steady_clock_type::duration idle_time;
    reactor::io_stats io;
    sstables::shared_index_lists::stats index;
    cache_tracker::stats cache;

    metrics_snapshot() {
        reactor& r = *local_engine;
        io = r.get_io_stats();
        busy_time = r.total_busy_time();
        idle_time = r.total_idle_time();
        hr_clock = std::chrono::high_resolution_clock::now();
        index = sstables::shared_index_lists::shard_stats();
        cache = global_cache_tracker().get_stats();
    }
};

class make_printable {
    using func_type = std::function<void(std::ostream&)>;
    func_type _func;
public:
    make_printable(func_type func) : _func(std::move(func)) {}
    friend std::ostream& operator<<(std::ostream& out, const make_printable& p) {
        p._func(out);
        return out;
    }
};

struct test_result {
    uint64_t fragments_read;
    metrics_snapshot before;
    metrics_snapshot after;

    test_result(metrics_snapshot before, uint64_t fragments_read)
        : fragments_read(fragments_read)
        , before(before)
    { }

    double duration_in_seconds() const {
        return std::chrono::duration<double>(after.hr_clock - before.hr_clock).count();
    }

    double fragment_rate() const { return double(fragments_read) / duration_in_seconds(); }

    uint64_t aio_reads() const { return after.io.aio_reads - before.io.aio_reads; }
    uint64_t aio_read_bytes() const { return after.io.aio_read_bytes - before.io.aio_read_bytes; }
    uint64_t read_aheads_discarded() const { return after.io.fstream_read_aheads_discarded - before.io.fstream_read_aheads_discarded; }
    uint64_t reads_blocked() const { return after.io.fstream_reads_blocked - before.io.fstream_reads_blocked; }

    uint64_t index_hits() const { return after.index.hits - before.index.hits; }
    uint64_t index_misses() const { return after.index.misses - before.index.misses; }
    uint64_t index_blocks() const { return after.index.blocks - before.index.blocks; }

    uint64_t cache_hits() const { return after.cache.partition_hits - before.cache.partition_hits; }
    uint64_t cache_misses() const { return after.cache.partition_misses - before.cache.partition_misses; }
    uint64_t cache_insertions() const { return after.cache.partition_insertions - before.cache.partition_insertions; }

    float cpu_utilization() const {
        auto busy_delta = after.busy_time.count() - before.busy_time.count();
        auto idle_delta = after.idle_time.count() - before.idle_time.count();
        return float(busy_delta) / (busy_delta + idle_delta);
    }

    static auto table_header() {
        return make_printable([] (std::ostream& out) {
            out << sprint("%10s %9s %10s %6s %10s %7s %7s %8s %8s %8s %8s %8s %8s %6s",
                "time [s]", "frags", "frag/s", "aio", "[KiB]", "blocked", "dropped",
                "idx hit", "idx miss", "idx blk",
                "c hit", "c miss", "c ins",
                "cpu");
        });
    }

    auto table_row() {
        return make_printable([this] (std::ostream& out) {
            out << sprint("%10.6f %9d %10.0f %6d %10d %7d %7d %8d %8d %8d %8d %8d %8d %5.1f%%",
                duration_in_seconds(), fragments_read, fragment_rate(),
                aio_reads(), aio_read_bytes() / 1024, reads_blocked(), read_aheads_discarded(),
                index_hits(), index_misses(), index_blocks(),
                cache_hits(), cache_misses(), cache_insertions(),
                cpu_utilization() * 100);
        });
    }
};

static void check_no_disk_reads(const test_result& r) {
    if (r.aio_reads()) {
        print_error("Expected no disk reads");
    }
}

static void check_no_index_reads(const test_result& r) {
    if (r.index_hits() || r.index_misses()) {
        print_error("Expected no index reads");
    }
}

static void check_fragment_count(const test_result& r, uint64_t expected) {
    if (r.fragments_read != expected) {
        print_error(sprint("Expected to read %d fragments", expected));
    }
}

class counting_consumer {
    uint64_t _fragments = 0;
public:
    stop_iteration consume(tombstone) { return stop_iteration::no; }
    template<typename Fragment>
    stop_iteration consume(Fragment&& f) { _fragments++; return stop_iteration::no; }
    uint64_t consume_end_of_stream() { return _fragments; }
};

static
uint64_t consume_all(streamed_mutation& sm) {
    return consume(sm, counting_consumer()).get0();
}

static
uint64_t consume_all(mutation_reader& rd) {
    uint64_t fragments = 0;
    while (1) {
        streamed_mutation_opt smo = rd().get0();
        if (!smo) {
            break;
        }
        fragments += consume_all(*smo);
    }
    return fragments;
}

// cf should belong to ks.test
static test_result scan_rows_with_stride(column_family& cf, int n_rows, int n_read = 1, int n_skip = 0) {
{'fraction_non_alphanumeric': 0.1197221150135083, 'fraction_numerical': 0.014203010420686994, 'mean_word_length': 3.483044982698962, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4988628', 'n_tokens_mistral': 4374, 'n_tokens_neox': 4239, 'n_words': 997}