Repository: smarr/ReBench

Issue:
Memory usage for long running benchmarks with many data points
Currently, we can run out of memory, because we keep all data points and measurements available.
In addition to storing them in memory.

https://math.stackexchange.com/questions/106700/incremental-averageing

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index 7d57c83..3d4f9fc 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -3,13 +3,6 @@ language: python
 matrix:
   include:
     - python: "2.7"
-      virtualenv:
-        system_site_packages: true
-      addons:
-        apt:
-          packages:
-            - python-scipy
-
     - python: "3.6"
     # PyPy versions
     - python: pypy
diff --git a/rebench/executor.py b/rebench/executor.py
index 01c99c4..4d7499e 100644
--- a/rebench/executor.py
+++ b/rebench/executor.py
@@ -35,7 +35,6 @@ from humanfriendly.compat import coerce_string
 
 from . import subprocess_with_timeout as subprocess_timeout
 from .interop.adapter import ExecutionDeliveredNoResults
-from .statistics import StatisticProperties, mean
 
 
 class FailedBuilding(Exception):
@@ -83,14 +82,10 @@ class RunScheduler(object):
         if not self._ui.spinner_initialized():
             return
 
-        totals = run.get_total_values()
         if completed_task:
             self._runs_completed += 1
 
-        if totals:
-            art_mean = mean(run.get_total_values())
-        else:
-            art_mean = 0
+        art_mean = run.get_mean_of_totals()
 
         hour, minute, sec = self._estimate_time_left()
 
@@ -419,15 +414,15 @@ class Executor(object):
             terminate = self._generate_data_point(cmdline, gauge_adapter,
                                                   run_id, termination_check)
 
-        stats = StatisticProperties(run_id.get_total_values())
+        mean_of_totals = run_id.get_mean_of_totals()
         if terminate:
-            run_id.report_run_completed(stats, cmdline)
+            run_id.report_run_completed(cmdline)
             if (not run_id.is_failed() and run_id.min_iteration_time
-                    and stats.mean < run_id.min_iteration_time):
+                    and mean_of_totals < run_id.min_iteration_time):
                 self._ui.warning(
                     ("{ind}Warning: Low mean run time.\n"
                      + "{ind}{ind}The mean (%.1f) is lower than min_iteration_time (%d)\n")
-                    % (stats.mean, run_id.min_iteration_time), run_id, cmdline)
+                    % (mean_of_totals, run_id.min_iteration_time), run_id, cmdline)
 
         return terminate
 
diff --git a/rebench/model/data_point.py b/rebench/model/data_point.py
index f04ed2f..58775ff 100644
--- a/rebench/model/data_point.py
+++ b/rebench/model/data_point.py
@@ -27,6 +27,10 @@ class DataPoint(object):
         self._total = None
         self._invocation = -1
 
+    @property
+    def run_id(self):
+        return self._run_id
+
     @property
     def invocation(self):
         return self._invocation
diff --git a/rebench/model/run_id.py b/rebench/model/run_id.py
index 0d0ca95..846b4c3 100644
--- a/rebench/model/run_id.py
+++ b/rebench/model/run_id.py
@@ -21,6 +21,7 @@ import re
 
 from .benchmark import Benchmark
 from .termination_check import TerminationCheck
+from ..statistics import StatisticProperties
 from ..ui import UIError
 
 
@@ -34,7 +35,8 @@ class RunId(object):
 
         self._reporters = set()
         self._persistence = set()
-        self._data_points = []
+        self._statistics = StatisticProperties()
+        self._total_unit = None
 
         self._termination_check = None
         self._cmdline = None
@@ -136,9 +138,9 @@ class RunId(object):
         for reporter in self._reporters:
             reporter.run_failed(self, cmdline, return_code, output)
 
-    def report_run_completed(self, statistics, cmdline):
+    def report_run_completed(self, cmdline):
         for reporter in self._reporters:
-            reporter.run_completed(self, statistics, cmdline)
+            reporter.run_completed(self, self._statistics, cmdline)
 
     def report_job_completed(self, run_ids):
         for reporter in self._reporters:
@@ -152,6 +154,9 @@ class RunId(object):
         for reporter in self._reporters:
             reporter.start_run(self)
 
+    def is_persisted_by(self, persistence):
+        return persistence in self._persistence
+
     def add_persistence(self, persistence):
         self._persistence.add(persistence)
 
@@ -159,35 +164,34 @@ class RunId(object):
         for persistence in self._persistence:
             persistence.close()
 
-    def loaded_data_point(self, data_point):
+    def _new_data_point(self, data_point):
         self._max_invocation = max(self._max_invocation, data_point.invocation)
-        self._data_points.append(data_point)
+        if self._total_unit is None:
+            self._total_unit = data_point.get_total_unit()
+
+    def loaded_data_point(self, data_point):
+        self._new_data_point(data_point)
+        self._statistics.add_sample(data_point.get_total_value())
 
     def add_data_point(self, data_point, warmup):
-        self._max_invocation = max(self._max_invocation, data_point.invocation)
+        self._new_data_point(data_point)
 
         if not warmup:
-            self._data_points.append(data_point)
+            self._statistics.add_sample(data_point.get_total_value())
         for persistence in self._persistence:
             persistence.persist_data_point(data_point)
 
     def get_number_of_data_points(self):
-        return len(self._data_points)
+        return self._statistics.num_samples
 
-    def get_data_points(self):
-        return self._data_points
+    def get_mean_of_totals(self):
+        return self._statistics.mean
 
-    def discard_data_points(self):
-        self._data_points = []
-        self._max_invocation = 0
-
-    def get_total_values(self):
-        return [dp.get_total_value() for dp in self._data_points]
+    def get_statistics(self):
+        return self._statistics
 
     def get_total_unit(self):
-        if not self._data_points:
-            return None
-        return self._data_points[0].get_total_unit()
+        return self._total_unit
 
     def get_termination_check(self, ui):
         if self._termination_check is None:
@@ -202,7 +206,7 @@ class RunId(object):
     def run_failed(self):
         return (self._termination_check.fails_consecutively() or
                 self._termination_check.has_too_many_failures(
-                    len(self._data_points)))
+                    self.get_number_of_data_points()))
 
     def __hash__(self):
         return hash(self.cmdline())
@@ -236,18 +240,22 @@ class RunId(object):
     def cmdline(self):
         if self._cmdline:
             return self._cmdline
+        return self._construct_cmdline()
 
+    def _construct_cmdline(self):
         cmdline = ""
         if self._benchmark.suite.executor.path:
-            cmdline = "%s/" % (self._benchmark.suite.executor.path, )
+            cmdline = self._benchmark.suite.executor.path + "/"
+
+        cmdline += self._benchmark.suite.executor.executable
 
-        cmdline += "%s %s" % (self._benchmark.suite.executor.executable,
-                              self._benchmark.suite.executor.args or '')
+        if self._benchmark.suite.executor.args:
+            cmdline += " " + str(self._benchmark.suite.executor.args)
 
-        cmdline += self._benchmark.suite.command
+        cmdline += " " + self._benchmark.suite.command
 
-        if self._benchmark.extra_args is not None:
-            cmdline += " %s" % self._benchmark.extra_args
+        if self._benchmark.extra_args:
+            cmdline += " " + str(self._benchmark.extra_args)
 
         cmdline = self._expand_vars(cmdline)
 
diff --git a/rebench/persistence.py b/rebench/persistence.py
index f4cce13..16af47c 100644
--- a/rebench/persistence.py
+++ b/rebench/persistence.py
@@ -18,9 +18,11 @@
 # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 # IN THE SOFTWARE.
 import os
+import shutil
 import subprocess
 import sys
 from datetime import datetime
+from tempfile import NamedTemporaryFile
 from threading import Lock
 
 from .model.data_point  import DataPoint
@@ -36,9 +38,9 @@ class DataStore(object):
         self._bench_cfgs = {}
         self._ui = ui
 
-    def load_data(self):
+    def load_data(self, runs, discard_run_data):
         for persistence in list(self._files.values()):
-            persistence._load_data()
+            persistence.load_data(runs, discard_run_data)
 
     def get(self, filename, discard_old_data):
         if filename not in self._files:
@@ -81,41 +83,6 @@ class DataStore(object):
             self._bench_cfgs[key] = cfg
         return cfg
 
-    @classmethod
-    def get_by_file(cls, runs):
-        by_file = {}
-        for run in runs:
-            points = run.get_data_points()
-            run.discard_data_points()
-            for point in points:
-                measurements = point.get_measurements()
-                for measure in measurements:
-                    if measure.filename in by_file:
-                        by_file[measure.filename].append(measure)
-                    else:
-                        by_file[measure.filename] = [measure]
-        return by_file
-
-    @classmethod
-    def discard_data_of_runs(cls, runs, ui):
-        by_file = cls.get_by_file(runs)
-        for filename, measures in by_file.items():
-            try:
-                with open(filename, 'r') as data_file:
-                    lines = data_file.readlines()
-            except IOError:
-                ui.debug_error_info(
-                    "Tried to discard old data, but file does not seem to exist: %s\n" % filename)
-                continue
-
-            for measure in measures:
-                lines[measure.line_number] = None
-
-            lines = filter(None, lines)
-
-            with open(filename, 'w') as data_file:
-                data_file.writelines(lines)
-
 
 class _DataPointPersistence(object):
 
@@ -141,18 +108,30 @@ class _DataPointPersistence(object):
         with open(filename, 'w'):
             pass
 
-    def _load_data(self):
+    def load_data(self, runs, discard_run_data):
         """
         Loads the data from the configured data file
         """
+        if discard_run_data:
+            current_runs = {run for run in runs if run.is_persisted_by(self)}
+        else:
+            current_runs = None
+
         try:
-            with open(self._data_filename, 'r') as data_file:
-                self._process_lines(data_file)
+            if current_runs:
+                with NamedTemporaryFile('w', delete=False) as target:
+                    with open(self._data_filename, 'r') as data_file:
+                        self._process_lines(data_file, current_runs, target)
+                    os.unlink(self._data_filename)
+                    shutil.move(target.name, self._data_filename)
+            else:
+                with open(self._data_filename, 'r') as data_file:
+                    self._process_lines(data_file, current_runs, None)
         except IOError:
             self._ui.debug_error_info("No data loaded, since %s does not exist.\n"
                                       % self._data_filename)
 
-    def _process_lines(self, data_file):
+    def _process_lines(self, data_file, runs, filtered_data_file):
         """
          The most important assumptions we make here is that the total
          measurement is always the last one serialized for a data point.
@@ -165,6 +144,8 @@ class _DataPointPersistence(object):
         for line in data_file:
             if line.startswith('#'):  # skip comments, and shebang lines
                 line_number += 1
+                if filtered_data_file:
+                    filtered_data_file.write(line)
                 continue
 
             try:
@@ -173,6 +154,13 @@ class _DataPointPersistence(object):
                     line_number, self._data_filename)
 
                 run_id = measurement.run_id
+                if filtered_data_file and runs and run_id in runs:
+                    continue
+
+                # these are all the measurements that are not filtered out
+                if filtered_data_file:
+                    filtered_data_file.write(line)
+
                 if previous_run_id is not run_id:
                     data_point = DataPoint(run_id)
                     previous_run_id = run_id
diff --git a/rebench/rebench.py b/rebench/rebench.py
index d561f84..8f54724 100755
--- a/rebench/rebench.py
+++ b/rebench/rebench.py
@@ -203,22 +203,16 @@ Argument:
         except ConfigurationError as exc:
             raise UIError(exc.message + "\n", exc)
 
-        data_store.load_data()
-        return self.execute_experiment()
+        runs = self._config.get_runs()
+        data_store.load_data(runs, self._config.options.do_rerun)
+        return self.execute_experiment(runs)
 
-    def execute_experiment(self):
+    def execute_experiment(self, runs):
         self._ui.verbose_output_info("Execute experiment: " + self._config.experiment_name + "\n")
 
-        # first load old data if available
-        if self._config.options.clean:
-            pass
-
         scheduler_class = {'batch':       BatchScheduler,
                            'round-robin': RoundRobinScheduler,
                            'random':      RandomScheduler}.get(self._config.options.scheduler)
-        runs = self._config.get_runs()
-        if self._config.options.do_rerun:
-            DataStore.discard_data_of_runs(runs, self._ui)
 
         executor = Executor(runs, self._config.use_nice, self._config.do_builds,
                             self._ui,
diff --git a/rebench/reporter.py b/rebench/reporter.py
index 6674a3a..4b8b197 100644
--- a/rebench/reporter.py
+++ b/rebench/reporter.py
@@ -36,8 +36,6 @@ except ImportError:
     from urllib import urlencode # pylint: disable=ungrouped-imports
     from urllib2 import urlopen
 
-from .statistics import StatisticProperties
-
 
 class Reporter(object):
 
@@ -80,13 +78,14 @@ class TextReporter(Reporter):
         rows = []
 
         for run_id in run_ids:
-            stats = StatisticProperties(run_id.get_total_values())
+            mean = run_id.get_mean_of_totals()
+            num_samples = run_id.get_number_of_data_points()
             out = run_id.as_str_list()
-            out.append(stats.num_samples)
-            if stats.num_samples == 0:
+            out.append(num_samples)
+            if num_samples == 0:
                 out.append("Failed")
             else:
-                out.append(int(round(stats.mean, 0)))
+                out.append(int(round(mean, 0)))
             rows.append(out)
 
         return sorted(rows, key=itemgetter(2, 1, 3, 4, 5, 6, 7))
@@ -248,8 +247,7 @@ class CodespeedReporter(Reporter):
                     + "{ind}{ind}" + msg + "\n", run_id)
 
     def _prepare_result(self, run_id):
-        stats = StatisticProperties(run_id.get_total_values())
-        return self._format_for_codespeed(run_id, stats)
+        return self._format_for_codespeed(run_id, run_id.get_statistics())
 
     def report_job_completed(self, run_ids):
         if self._incremental_report:
diff --git a/rebench/statistics.py b/rebench/statistics.py
index bcccbc6..945243a 100644
--- a/rebench/statistics.py
+++ b/rebench/statistics.py
@@ -17,81 +17,64 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 # IN THE SOFTWARE.
-import copy
 import math
-import operator
-from functools import reduce # pylint: disable=redefined-builtin
-
-
-def mean(values):
-    return sum(values) / float(len(values))
-
-
-def median(values):
-    values_ = sorted(copy.deepcopy(values))
-    index = int(len(values_) / 2)
-    if len(values_) % 2 == 0:
-        return float(values_[index] + values_[index - 1]) / 2
-    else:
-        return values_[index]
-
-
-def geo_mean(values):
-    product = reduce(operator.mul, values, 1)
-    return product ** (1.0 / len(values))
-
-
-def variance(values):
-    avg = mean(values)
-    var = [(x - avg) ** 2 for x in values]
-    return mean(var)
-
-
-def stddev(values):
-    return math.sqrt(variance(values))
 
 
 class StatisticProperties(object):
     """
-    The statistics class calculates the statistic
-    properties of a given set of data samples, i.e., the chosen values
-    from a set of data points
+    The class maintains running statistics for the added data points.
+    Data points can be added one by one, or as lists of values.
     """
 
-    def __init__(self, data_samples):
-        self._data_samples = data_samples
-
-        self.mean = 0
-        self.geom_mean = 0
-        self.median = 0
-        self.std_dev = 0
+    def __init__(self):
         self.num_samples = 0
+
+        self.mean = 0.0
+        self.geom_mean = 0.0
+        self.std_dev = 0.0
         self.min = 0
         self.max = 0
 
-        if self._data_samples:
-            self.num_samples = len(self._data_samples)
-            self._calc_basic_statistics()
+        # used to calculate std_dev
+        # Variance (sigma^2) * num_samples
+        self._variance_times_num_samples = 0
+        # used to calculate geomean
+        self._product_of_samples = 1.0
+
+    def add(self, samples):
+        for sample in samples:
+            self.add_sample(sample)
+
+    def add_sample(self, sample):
+        if self.num_samples == 0:
+            self.mean = float(sample)
+            self.geom_mean = float(sample)
+            self._product_of_samples = float(sample)
+
+            self.min = sample
+            self.max = sample
+            self.num_samples = 1
         else:
-            self.num_samples = 0
+            self.num_samples += 1
+            prev_mean = self.mean
+            self.mean = prev_mean + ((sample - prev_mean) / self.num_samples)
+
+            self._product_of_samples = self._product_of_samples * float(sample)
+            self.geom_mean = self._product_of_samples ** (1/float(self.num_samples))
+
+            self._variance_times_num_samples = (self._variance_times_num_samples +
+                                                ((sample - prev_mean) * (sample - self.mean)))
+            self.std_dev = math.sqrt(self._variance_times_num_samples / self.num_samples)
 
-    def _calc_basic_statistics(self):
-        """This function determines the mean and the standard deviation
-           of the data sample.
-           Furthermore, several other simple properties are determined.
-        """
-        self.mean = mean(self._data_samples)
-        self.geom_mean = geo_mean(self._data_samples)
-        self.median = median(self._data_samples)
-        self.std_dev = stddev(self._data_samples)
+            if self.min > sample:
+                self.min = sample
 
-        self.min = min(self._data_samples)
-        self.max = max(self._data_samples)
+            if self.max < sample:
+                self.max = sample
 
     def as_tuple(self):
         return (self.mean,
                 self.geom_mean,
-                self.median,
                 self.std_dev,
                 self.num_samples,
                 self.min,
@@ -99,5 +82,5 @@ class StatisticProperties(object):
 
     @classmethod
     def tuple_mapping(cls):
-        return ('arithmetic mean', 'geometric mean', 'median', 'stdDev',
+        return ('arithmetic mean', 'geometric mean', 'stdDev',
                 '#samples', 'min', 'max')
diff --git a/setup.py b/setup.py
index 725b243..48f47ce 100644
--- a/setup.py
+++ b/setup.py
@@ -39,7 +39,6 @@ setup(name='ReBench',
       entry_points = {
           'console_scripts' : ['rebench = rebench.rebench:main_func']
       },
-      tests_require = ['scipy>=0.8.0'],
       test_suite = 'rebench.tests',
       license = 'MIT'
 )


Test patch:
diff --git a/rebench/tests/executor_test.py b/rebench/tests/executor_test.py
index 297b5a4..01e55ba 100644
--- a/rebench/tests/executor_test.py
+++ b/rebench/tests/executor_test.py
@@ -21,6 +21,7 @@ import unittest
 import subprocess
 import os
 
+from .persistence import TestPersistence
 from .rebench_test_case import ReBenchTestCase
 from ..rebench           import ReBench
 from ..executor          import Executor
@@ -54,19 +55,6 @@ class ExecutorTest(ReBenchTestCase):
         finally:
             subprocess.Popen = old_popen
 
-# TODO: should test more details
-#        (mean, sdev, (interval, interval_percentage),
-#                (interval_t, interval_percentage_t)) = ex.result['test-executor']['test-bench']
-#
-#        self.assertEqual(31, len(ex.benchmark_data['test-executor']['test-bench']))
-#        self.assertAlmostEqual(45870.4193548, mean)
-#        self.assertAlmostEqual(2.93778711485, sdev)
-#
-#        (i_low, i_high) = interval
-#        self.assertAlmostEqual(45869.385195243565, i_low)
-#        self.assertAlmostEqual(45871.453514433859, i_high)
-#        self.assertAlmostEqual(0.00450904792104, interval_percentage)
-
     def test_broken_command_format_with_ValueError(self):
         with self.assertRaises(UIError) as err:
             options = ReBench().shell_options().parse_args(['dummy'])
@@ -92,10 +80,16 @@ class ExecutorTest(ReBenchTestCase):
     def _basic_execution(self, cnf):
         runs = cnf.get_runs()
         self.assertEqual(8, len(runs))
-        ex = Executor(cnf.get_runs(), cnf.use_nice, cnf.do_builds, TestDummyUI())
+
+        runs = cnf.get_runs()
+        persistence = TestPersistence()
+        for run in runs:
+            run.add_persistence(persistence)
+
+        ex = Executor(runs, cnf.use_nice, cnf.do_builds, TestDummyUI())
         ex.execute()
         for run in runs:
-            data_points = run.get_data_points()
+            data_points = persistence.get_data_points(run)
             self.assertEqual(10, len(data_points))
             for data_point in data_points:
                 measurements = data_point.get_measurements()
diff --git a/rebench/tests/features/issue_15_warm_up_support_test.py b/rebench/tests/features/issue_15_warm_up_support_test.py
index af7e26a..68c2043 100644
--- a/rebench/tests/features/issue_15_warm_up_support_test.py
+++ b/rebench/tests/features/issue_15_warm_up_support_test.py
@@ -57,4 +57,3 @@ class Issue15WarmUpSupportTest(ReBenchTestCase):
         ex.execute()
 
         self.assertEqual(runs[0].get_number_of_data_points(), 10)
-        self.assertLessEqual(runs[0].get_total_values()[0], 850)
diff --git a/rebench/tests/features/issue_16_multiple_data_points_test.py b/rebench/tests/features/issue_16_multiple_data_points_test.py
index b9e55c7..c672930 100644
--- a/rebench/tests/features/issue_16_multiple_data_points_test.py
+++ b/rebench/tests/features/issue_16_multiple_data_points_test.py
@@ -20,6 +20,7 @@
 from ...configurator           import Configurator, load_config
 from ...executor               import Executor
 from ...persistence            import DataStore
+from ..persistence import TestPersistence
 from ..rebench_test_case import ReBenchTestCase
 
 
@@ -38,12 +39,16 @@ class Issue16MultipleDataPointsTest(ReBenchTestCase):
         cnf = Configurator(load_config(self._path + '/issue_16.conf'), DataStore(self._ui),
                            self._ui, exp_name=exp_name,
                            data_file=self._tmp_file)
+        runs = cnf.get_runs()
+        persistence = TestPersistence()
+        for run in runs:
+            run.add_persistence(persistence)
         ex = Executor(cnf.get_runs(), False, False, self._ui)
         ex.execute()
         self.assertEqual(1, len(cnf.get_runs()))
         run = next(iter(cnf.get_runs()))
-        self.assertEqual(num_data_points, len(run.get_data_points()))
-        return run.get_data_points()
+        self.assertEqual(num_data_points, run.get_number_of_data_points())
+        return persistence.get_data_points()
 
     def test_records_multiple_data_points_from_single_execution_10(self):
         self._records_data_points('Test1', 10)
diff --git a/rebench/tests/features/issue_31_multivariate_data_points_test.py b/rebench/tests/features/issue_31_multivariate_data_points_test.py
index 8359fbc..72e9420 100644
--- a/rebench/tests/features/issue_31_multivariate_data_points_test.py
+++ b/rebench/tests/features/issue_31_multivariate_data_points_test.py
@@ -22,6 +22,7 @@ from __future__ import print_function
 from ...configurator           import Configurator, load_config
 from ...executor               import Executor
 from ...persistence            import DataStore
+from ..persistence import TestPersistence
 from ..rebench_test_case import ReBenchTestCase
 
 
@@ -38,12 +39,17 @@ class Issue31MultivariateDataPointsTest(ReBenchTestCase):
         cnf = Configurator(load_config(self._path + '/issue_31.conf'), DataStore(self._ui),
                            self._ui, exp_name=exp_name,
                            data_file=self._tmp_file)
-        ex = Executor(cnf.get_runs(), False, False, self._ui)
+        runs = cnf.get_runs()
+        persistence = TestPersistence()
+        for run in runs:
+            run.add_persistence(persistence)
+
+        ex = Executor(runs, False, False, self._ui)
         ex.execute()
         self.assertEqual(1, len(cnf.get_runs()))
         run = next(iter(cnf.get_runs()))
-        self.assertEqual(num_data_points, len(run.get_data_points()))
-        return run.get_data_points()
+        self.assertEqual(num_data_points, run.get_number_of_data_points())
+        return persistence.get_data_points()
 
     def test_records_multiple_data_points_from_single_execution_10(self):
         self._records_data_points('Test1', 10)
diff --git a/rebench/tests/persistence.py b/rebench/tests/persistence.py
new file mode 100644
index 0000000..5675014
--- /dev/null
+++ b/rebench/tests/persistence.py
@@ -0,0 +1,15 @@
+class TestPersistence(object):
+
+    def __init__(self):
+        self._data_points = []
+
+    def get_data_points(self, run_id=None):
+        if run_id:
+            return [dp for dp in self._data_points if dp.run_id is run_id]
+        return self._data_points
+
+    def persist_data_point(self, data_point):
+        self._data_points.append(data_point)
+
+    def close(self):
+        pass
diff --git a/rebench/tests/persistency_test.py b/rebench/tests/persistency_test.py
index 9f8a72c..2e6d373 100644
--- a/rebench/tests/persistency_test.py
+++ b/rebench/tests/persistency_test.py
@@ -66,11 +66,11 @@ class PersistencyTest(ReBenchTestCase):
         self.assertEqual(num_invocations, run.completed_invocations)
 
     def test_iteration_invocation_semantics(self):
-        ## Executes first time
+        # Executes first time
         ds = DataStore(self._ui)
         cnf = Configurator(load_config(self._path + '/persistency.conf'),
                            ds, self._ui, data_file=self._tmp_file)
-        ds.load_data()
+        ds.load_data(None, False)
 
         self._assert_runs(cnf, 1, 0, 0)
 
@@ -79,12 +79,12 @@ class PersistencyTest(ReBenchTestCase):
 
         self._assert_runs(cnf, 1, 10, 10)
 
-        ## Execute a second time, should not add any data points,
-        ## because goal is already reached
+        # Execute a second time, should not add any data points,
+        # because goal is already reached
         ds2 = DataStore(self._ui)
         cnf2 = Configurator(load_config(self._path + '/persistency.conf'),
                             ds2, self._ui, data_file=self._tmp_file)
-        ds2.load_data()
+        ds2.load_data(None, False)
 
         self._assert_runs(cnf2, 1, 10, 10)
 
@@ -92,3 +92,31 @@ class PersistencyTest(ReBenchTestCase):
         ex2.execute()
 
         self._assert_runs(cnf2, 1, 10, 10)
+
+    def test_data_discarding(self):
+        # Executes first time
+        ds = DataStore(self._ui)
+        cnf = Configurator(load_config(self._path + '/persistency.conf'),
+                           ds, self._ui, data_file=self._tmp_file)
+        ds.load_data(None, False)
+
+        self._assert_runs(cnf, 1, 0, 0)
+
+        ex = Executor(cnf.get_runs(), False, False, self._ui)
+        ex.execute()
+
+        self._assert_runs(cnf, 1, 10, 10)
+
+        # Execute a second time, this time, discard the data first, and then rerun
+        ds2 = DataStore(self._ui)
+        cnf2 = Configurator(load_config(self._path + '/persistency.conf'),
+                            ds2, self._ui, data_file=self._tmp_file)
+        run2 = cnf2.get_runs()
+        ds2.load_data(run2, True)
+
+        self._assert_runs(cnf2, 1, 0, 0)
+
+        ex2 = Executor(run2, False, False, self._ui)
+        ex2.execute()
+
+        self._assert_runs(cnf2, 1, 10, 10)
diff --git a/rebench/tests/stats_test.py b/rebench/tests/stats_test.py
index d233076..2ca8471 100644
--- a/rebench/tests/stats_test.py
+++ b/rebench/tests/stats_test.py
@@ -18,20 +18,8 @@
 # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 # IN THE SOFTWARE.
 
-
-## This code is inspired by https://code.google.com/p/python-statlib
-
 import unittest
-from .. import statistics as stats
-
-NUMPY_AND_SCIPY_AVAILABLE = False
-try:
-    import numpy
-    import scipy.stats
-    # import scipy.stats.distributions -- not yet implemented
-    NUMPY_AND_SCIPY_AVAILABLE = True
-except ImportError:
-    NUMPY_AND_SCIPY_AVAILABLE = False
+from ..statistics import StatisticProperties
 
 
 class StatsTest(unittest.TestCase):
@@ -44,136 +32,39 @@ class StatsTest(unittest.TestCase):
         self._mixed = [x if x % 2 == 0 else float(x) + 4.5
                        for x in self._integers]
 
-    def test_mean_simple(self):
-        self.assertEqual(2, stats.mean([1, 2, 3]))
-        self.assertAlmostEqual(2, stats.mean([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(25, stats.mean(self._integers))
-        self.assertAlmostEqual(25, stats.mean(self._floats))
-        self.assertAlmostEqual(25 + 2.31, stats.mean(self._floats2))
-        self.assertAlmostEqual(27.295918367, stats.mean(self._mixed))
-
-    @unittest.skipUnless(NUMPY_AND_SCIPY_AVAILABLE,
-                         "NumPy or SciPy is not available")
-    def test_mean_vs_numpy(self):
-        self.assertEqual(numpy.mean([1, 2, 3]),
-                         stats.mean([1, 2, 3]))
-        self.assertAlmostEqual(numpy.mean([1.0, 2.0, 3.0]),
-                               stats.mean([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(numpy.mean(self._integers),
-                               stats.mean(self._integers))
-
-        self.assertAlmostEqual(numpy.mean(self._floats),
-                               stats.mean(self._floats))
-
-        self.assertAlmostEqual(numpy.mean(self._floats2),
-                               stats.mean(self._floats2))
-
-        self.assertAlmostEqual(numpy.mean(self._mixed),
-                               stats.mean(self._mixed))
-
-    def test_median_simple(self):
-        self.assertEqual(2.5, stats.median([1, 2, 3, 4]))
-        self.assertAlmostEqual(2.5, stats.median([1.0, 2.0, 3.0, 4.0]))
-
-        self.assertAlmostEqual(25, stats.median(self._integers))
-        self.assertAlmostEqual(25, stats.median(self._floats))
-        self.assertAlmostEqual(25 + 2.31, stats.median(self._floats2))
-        self.assertAlmostEqual(27.5, stats.median(self._mixed))
-
-    @unittest.skipUnless(NUMPY_AND_SCIPY_AVAILABLE,
-                         "NumPy or SciPy is not available")
-    def test_median_vs_numpy(self):
-        self.assertEqual(numpy.median([1, 2, 3, 4]),
-                         stats.median([1, 2, 3, 4]))
-        self.assertAlmostEqual(numpy.median([1.0, 2.0, 3.0, 4.0]),
-                               stats.median([1.0, 2.0, 3.0, 4.0]))
-
-        self.assertAlmostEqual(numpy.median(self._integers),
-                               stats.median(self._integers))
-
-        self.assertAlmostEqual(numpy.median(self._floats),
-                               stats.median(self._floats))
-
-        self.assertAlmostEqual(numpy.median(self._floats2),
-                               stats.median(self._floats2))
-
-        self.assertAlmostEqual(numpy.median(self._mixed),
-                               stats.median(self._mixed))
-
-    def test_geomean_simple(self):
-        self.assertAlmostEqual(1.817120592, stats.geo_mean([1, 2, 3]))
-        self.assertAlmostEqual(1.817120592, stats.geo_mean([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(19.112093553, stats.geo_mean(self._integers))
-        self.assertAlmostEqual(19.112093553, stats.geo_mean(self._floats))
-        self.assertAlmostEqual(22.533409416, stats.geo_mean(self._floats2))
-        self.assertAlmostEqual(22.245044799, stats.geo_mean(self._mixed))
-
-    @unittest.skipUnless(NUMPY_AND_SCIPY_AVAILABLE,
-                         "NumPy or SciPy is not available")
-    def test_geomean_vs_scipy(self):
-        self.assertAlmostEqual(scipy.stats.gmean([1, 2, 3]),
-                               stats.geo_mean([1, 2, 3]))
-        self.assertAlmostEqual(scipy.stats.gmean([1.0, 2.0, 3.0]),
-                               stats.geo_mean([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(scipy.stats.gmean(self._integers),
-                               stats.geo_mean(self._integers))
-
-        self.assertAlmostEqual(scipy.stats.gmean(self._floats),
-                               stats.geo_mean(self._floats))
-
-        self.assertAlmostEqual(scipy.stats.gmean(self._floats2),
-                               stats.geo_mean(self._floats2))
-
-        self.assertAlmostEqual(scipy.stats.gmean(self._mixed),
-                               stats.geo_mean(self._mixed))
-
-    def test_stddev_simple(self):
-        self.assertAlmostEqual(0.8164965809,
-                               stats.stddev([1, 2, 3]))
-        self.assertAlmostEqual(0.8164965809,
-                               stats.stddev([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(14.142135623,
-                               stats.stddev(self._integers))
-
-        self.assertAlmostEqual(14.142135623,
-                               stats.stddev(self._floats))
-
-        self.assertAlmostEqual(14.142135623,
-                               stats.stddev(self._floats2))
-
-        self.assertAlmostEqual(14.319929870,
-                               stats.stddev(self._mixed))
-
-    @unittest.skipUnless(NUMPY_AND_SCIPY_AVAILABLE,
-                         "NumPy or SciPy is not available")
-    def test_stddev_vs_numpy(self):
-        self.assertAlmostEqual(numpy.std([1, 2, 3]),
-                               stats.stddev([1, 2, 3]))
-        self.assertAlmostEqual(numpy.std([1.0, 2.0, 3.0]),
-                               stats.stddev([1.0, 2.0, 3.0]))
-
-        self.assertAlmostEqual(numpy.std(self._integers),
-                               stats.stddev(self._integers))
-
-        self.assertAlmostEqual(numpy.std(self._floats),
-                               stats.stddev(self._floats))
-
-        self.assertAlmostEqual(numpy.std(self._floats2),
-                               stats.stddev(self._floats2))
-
-        self.assertAlmostEqual(numpy.std(self._mixed),
-                               stats.stddev(self._mixed))
-
-# TODO: implement
-#    @unittest.skipUnless(are_numpy_and_scipy_available, "NumPy or SciPy is not available")
-#    def test_norm_distribution(self):
-#        self.fail("not yet implemented")
-#
-#    @unittest.skipUnless(are_numpy_and_scipy_available, "NumPy or SciPy is not available")
-#    def test_t_distribution(self):
-#        self.fail("not yet implemented")
+    def _assert(self, stats, mean_val, geo_mean, min_val, max_val, std_dev):
+        self.assertAlmostEqual(mean_val, stats.mean)
+        self.assertAlmostEqual(geo_mean, stats.geom_mean)
+
+        self.assertEqual(min_val, stats.min)
+        self.assertEqual(max_val, stats.max)
+        self.assertAlmostEqual(std_dev, stats.std_dev)
+
+    def test_123(self):
+        stats = StatisticProperties()
+        stats.add([1, 2, 3])
+        self._assert(stats, 2, 1.817120592, 1, 3, 0.816496580927726)
+
+        stats = StatisticProperties()
+        stats.add([1.0, 2.0, 3.0])
+        self._assert(stats, 2, 1.817120592, 1.0, 3.0, 0.816496580927726)
+
+    def test_1to49(self):
+        stats = StatisticProperties()
+        stats.add(self._integers)
+        self._assert(stats, 25, 19.112093553, 1, 49, 14.142135623730951)
+
+        stats = StatisticProperties()
+        stats.add(self._floats)
+        self._assert(stats, 25, 19.112093553, 1.0, 49.0, 14.142135623730951)
+
+    def test_shifted(self):
+        stats = StatisticProperties()
+        stats.add(self._floats2)
+        self._assert(stats, 25 + 2.31, 22.533409416, 1.0 + 2.31, 49.0 + 2.31, 14.142135623730951)
+
+    def test_mixed(self):
+        stats = StatisticProperties()
+        stats.add(self._mixed)
+        self.assertAlmostEqual(27.295918367, stats.mean)
+        self._assert(stats, 27.295918367, 22.245044799, 2, 53.5, 14.319929870761944)