<commit_before># -*- coding: utf-8 -*-

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import argparse
import datetime
import functools
import logging
import random
import time
import types

from cliff import show
import futurist
from oslo_utils import timeutils
import six.moves

from gnocchiclient.v1 import metric_cli

LOG = logging.getLogger(__name__)


def _pickle_method(m):
    if m.im_self is None:
        return getattr, (m.im_class, m.im_func.func_name)
    else:
        return getattr, (m.im_self, m.im_func.func_name)

six.moves.copyreg.pickle(types.MethodType, _pickle_method)


def grouper(iterable, n, fillvalue=None):
    "Collect data into fixed-length chunks or blocks"
    # grouper('ABCDEFG', 3, 'x') --> ABC DEF
    args = [iter(iterable)] * n
    return six.moves.zip(*args)


def _positive_non_zero_int(argument_value):
    if argument_value is None:
        return None
    try:
        value = int(argument_value)
    except ValueError:
        msg = "%s must be an integer" % argument_value
        raise argparse.ArgumentTypeError(msg)
    if value <= 0:
        msg = "%s must be greater than 0" % argument_value
        raise argparse.ArgumentTypeError(msg)
    return value


class BenchmarkPool(futurist.ProcessPoolExecutor):
    def submit_job(self, times, fn, *args, **kwargs):
        self.sw = timeutils.StopWatch()
        self.sw.start()
        self.times = times
        return [self.submit(fn, *args, **kwargs)
                for i in six.moves.range(times)]

    def map_job(self, fn, iterable, **kwargs):
        self.sw = timeutils.StopWatch()
        self.sw.start()
        r = []
        self.times = 0
        for item in iterable:
            r.append(self.submit(fn, item, **kwargs))
            self.times += 1
        return r

    def _log_progress(self, verb):
        runtime = self.sw.elapsed()
        done = self.statistics.executed
        rate = done / runtime if runtime != 0 else 0
        LOG.info(
            "%d/%d, "
            "total: %.2f seconds, "
            "rate: %.2f %s/second"
            % (done, self.times, runtime, rate, verb))

    def wait_job(self, verb, futures):
        while self.statistics.executed != self.times:
            self._log_progress(verb)
            time.sleep(1)
        self._log_progress(verb)
        self.shutdown(wait=True)
        runtime = self.sw.elapsed()
        results = []
        for f in futures:
            try:
                results.append(f.result())
            except Exception as e:
                LOG.error("Error with %s metric: %s" % (verb, e))
        return results, runtime, {
            'client workers': self._max_workers,
            verb + ' runtime': "%.2f seconds" % runtime,
            verb + ' executed': self.statistics.executed,
            verb + ' speed': (
                "%.2f %s/s" % (self.statistics.executed / runtime, verb)
            ),
            verb + ' failures': self.statistics.failures,
            verb + ' failures rate': (
                "%.2f %%" % (
                    100
                    * self.statistics.failures
                    / float(self.statistics.executed)
                )
            ),
        }
<commit_msg>benchmark: Add more measurements stats<commit_after># -*- coding: utf-8 -*-

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import argparse
import datetime
import functools
import logging
import math
import random
import time
import types

from cliff import show
import futurist
from oslo_utils import timeutils
import six.moves

from gnocchiclient.v1 import metric_cli

LOG = logging.getLogger(__name__)


def _pickle_method(m):
    if m.im_self is None:
        return getattr, (m.im_class, m.im_func.func_name)
    else:
        return getattr, (m.im_self, m.im_func.func_name)

six.moves.copyreg.pickle(types.MethodType, _pickle_method)


def grouper(iterable, n, fillvalue=None):
    "Collect data into fixed-length chunks or blocks"
    # grouper('ABCDEFG', 3, 'x') --> ABC DEF
    args = [iter(iterable)] * n
    return six.moves.zip(*args)


def _positive_non_zero_int(argument_value):
    if argument_value is None:
        return None
    try:
        value = int(argument_value)
    except ValueError:
        msg = "%s must be an integer" % argument_value
        raise argparse.ArgumentTypeError(msg)
    if value <= 0:
        msg = "%s must be greater than 0" % argument_value
        raise argparse.ArgumentTypeError(msg)
    return value


def measure_job(fn, *args, **kwargs):
    # NOTE(sileht): This is not part of BenchmarkPool
    # because we cannot pickle BenchmarkPool class
    sw = timeutils.StopWatch().start()
    return fn(*args, **kwargs), sw.elapsed()


class BenchmarkPool(futurist.ProcessPoolExecutor):
    def submit_job(self, times, fn, *args, **kwargs):
        self.sw = timeutils.StopWatch()
        self.sw.start()
        self.times = times
        return [self.submit(measure_job, fn, *args, **kwargs)
                for i in six.moves.range(times)]

    def map_job(self, fn, iterable, **kwargs):
        self.sw = timeutils.StopWatch()
        self.sw.start()
        r = []
        self.times = 0
        for item in iterable:
            r.append(self.submit(measure_job, fn, item, **kwargs))
            self.times += 1
        return r

    def _log_progress(self, verb):
        runtime = self.sw.elapsed()
        done = self.statistics.executed
        rate = done / runtime if runtime != 0 else 0
        LOG.info(
            "%d/%d, "
            "total: %.2f seconds, "
            "rate: %.2f %s/second"
            % (done, self.times, runtime, rate, verb))

    def wait_job(self, verb, futures):
        while self.statistics.executed != self.times:
            self._log_progress(verb)
            time.sleep(1)
        self._log_progress(verb)
        self.shutdown(wait=True)
        runtime = self.sw.elapsed()
        results = []
        latencies = []
        for f in futures:
            try:
                result, latency = f.result()
                results.append(result)
                latencies.append(latency)
            except Exception as e:
                LOG.error("Error with %s metric: %s" % (verb, e))
        latencies = sorted(latencies)
        return results, runtime, {
            'client workers': self._max_workers,
            verb + ' runtime': "%.2f seconds" % runtime,
            verb + ' executed': self.statistics.executed,
            verb + ' speed': (
                "%.2f %s/s" % (self.statistics.executed / runtime, verb)
            ),
            verb + ' failures': self.statistics.failures,
            verb + ' failures rate': (
                "%.2f %%" % (
                    100
                    * self.statistics.failures
                    / float(self.statistics.executed)
                )
            ),
            verb + ' latency min': min(latencies),
            verb + ' latency max': max(latencies),
            verb + ' latency mean': sum(latencies) / len(latencies),
            verb + ' latency median': self._percentile(latencies, 0.5),
            verb + ' latency 95%\'ile': self._percentile(latencies, 0.95),
            verb + ' latency 99%\'ile': self._percentile(latencies, 0.99),
            verb + ' latency 99.9%\'ile': self._percentile(latencies, 0.999),

        }

    @staticmethod
    def _percentile(sorted_list, percent):
        # NOTE(sileht): we don't to want depends on numpy
        if not sorted_list:
            return None
        k = (len(sorted_list) - 1) * percent
        f = math.floor(k)
        c = math.ceil(k)
        if f == c:
            return sorted_list[int(k)]
        d0 = sorted_list[int(f)] * (c - k)
        d1 = sorted_list[int(c)] * (k - f)
        return d0 + d1

{'fraction_non_alphanumeric': 0.10883731565108037, 'fraction_numerical': 0.007774093975077169, 'mean_word_length': 2.143370463528566, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2820338', 'n_tokens_mistral': 2649, 'n_tokens_neox': 2557, 'n_words': 755}