Use plotly to visualize experiments, rework experiment/record logic

1 files changed, 131 insertions, 57 deletions

diff --git a/nemesis/nemesis.py b/nemesis/nemesis.py
index c20d68d..46e8094 100755
--- a/nemesis/nemesis.py
+++ b/nemesis/nemesis.py
@@ -25,17 +25,39 @@ Commentary:
 Code:
 '''
 
+from causal_event_loop import CausalEventLoop
 from collections import defaultdict
-from experiment import Experiment
+from html_gen import plot_results
 import argparse
 import asyncio
 import os
+import random
 import signal
 import sys
+import threading
 import time
 import traceback
 import types
 
+class Experiment:
+
+    # event loops participating in this this experiment
+    _loops = []
+    # a key-value pair where keys represent a handle and values
+    # represent the period of time waiting
+    samples = None
+
+    def __init__(self, loops):
+        self._loops = loops
+        self.samples = defaultdict(lambda: 0)
+
+    def get_loops(self):
+        return [l for l in self._loops if l.is_running()]
+
+    def add_handles(self, handles, loop, time):
+        for h in handles:
+            h.append(loop._thread_id)
+            self.samples[tuple(h)] += time
 
 class Nemesis(object):
 
@@ -43,29 +65,35 @@ class Nemesis(object):
     signal_interval = 0.0
     # the timestamp which the last sample was taken
     last_sample = None
+
     # the current experiment being run
-    curr_experiment = None
-    # results from previous experiments, represented as strings
-    results = []
-    # The duration of each performance experiment
+    experiment_data = None
+    # the coroutine the current experiment is speeding up
+    experiment_coro = None
+    # the speedup of the current experiment
+    experiment_spdp = None
+
+    # results from previous experiments. Keys represent names of coroutines.
+    results = defaultdict(lambda: defaultdict(lambda: []))
+
+    # the file to write results to
+    filename = None
+
+    # The base duration of each performance experiment
     e_duration = None
+
     # The number of seconds remaining in this performance experiment.
     r_duration = None
+
     # A mapping of event loops to the previous running coroutine.
     prev_coro = defaultdict(lambda: None)
 
-    # temp
-    coro = None
-    dilation = 1.0
-
     @staticmethod
-    def __init__(coro, speedup, e_duration, w_time, signal_interval=0.01):
+    def __init__(e_duration, filename, signal_interval=0.01):
         Nemesis.signal_interval = signal_interval
         Nemesis.e_duration = e_duration
+        Nemesis.filename = filename
         Nemesis.r_duration = 0
-        # temporary
-        Nemesis.coro = coro
-        Nemesis.speedup = max(speedup, 1.0)
 
     @staticmethod
     def start():
@@ -79,32 +107,42 @@ class Nemesis(object):
     @staticmethod
     def stop():
         signal.setitimer(signal.ITIMER_REAL, 0)
-        Nemesis._stop_experiment()
-        for r in Nemesis.results:
-            print()
-            print(r)
+        plot_results(Nemesis.results, Nemesis.filename)
+        print(f"Wrote {Nemesis.filename}")
 
     @staticmethod
-    def _start_experiment():
-        Nemesis.r_duration = Nemesis.e_duration
+    def _start_experiment(coro, speedup):
+        Nemesis.r_duration = Nemesis.e_duration * speedup
         Nemesis.prev_coro = defaultdict(lambda: None)
-        Nemesis.curr_experiment = Experiment(Nemesis.coro, Nemesis.speedup)
+        Nemesis.experiment_coro = coro
+        Nemesis.experiment_spdp = speedup
+
+        loops = Nemesis._get_event_loops()
+        for l in loops:
+            if not isinstance(l, CausalEventLoop):
+                raise RuntimeException("Nemesis requires a custom event loop to insert slowdowns. It does not work on programs which change the event loop policy.")
+            l._set_dilation(1.0 / speedup)
+
+        Nemesis.experiment_data = Experiment(loops)
 
     @staticmethod
     def _stop_experiment():
-        if Nemesis.curr_experiment is not None:
-            print(f'finished running {Nemesis.curr_experiment.get_coro()} with speedup {Nemesis.curr_experiment.get_speedup()}')
-            Nemesis.results.append(Nemesis.curr_experiment.get_results())
-            del Nemesis.curr_experiment
+        if Nemesis.experiment_data is not None:
+            print(f'finished running experiment on {Nemesis.experiment_coro}')
+            results = Nemesis.experiment_data.samples
+            Nemesis.results[Nemesis.experiment_coro][Nemesis.experiment_spdp].append(results)
+            del Nemesis.experiment_data
 
     @staticmethod
     def _signal_handler(sig, frame):
         curr_sample = time.perf_counter()
         passed_time = curr_sample - Nemesis.last_sample
         Nemesis.last_sample = curr_sample
-        if Nemesis.curr_experiment:
-            loops = Nemesis.curr_experiment.get_loops()
-            exp_coro = Nemesis.curr_experiment.get_coro()
+
+
+        if getattr(Nemesis, 'experiment_data', None):
+            loops = Nemesis.experiment_data.get_loops()
+            exp_coro = Nemesis.experiment_coro
             for loop in loops:
                 coro = Nemesis._get_current_coro(loop)
                 prev_coro = Nemesis.prev_coro[loop]
@@ -117,31 +155,42 @@ class Nemesis(object):
 
                 loop._update_ready(True)
                 handles = Nemesis._get_waiting_handles(loop)
-                Nemesis.curr_experiment.add_handles(handles, loop, passed_time)
+                Nemesis.experiment_data.add_handles(handles, loop, passed_time)
 
-        Nemesis.r_duration -= passed_time
-        if (Nemesis.r_duration <= 0):
-            Nemesis._stop_experiment()
-            Nemesis._start_experiment()
+            Nemesis.r_duration -= passed_time
+            if (Nemesis.r_duration <= 0):
+                Nemesis._stop_experiment()
 
+        else:
+            coros = []
+            loops = Nemesis._get_event_loops()
+            for loop in loops:
+                coro = Nemesis._get_current_coro(loop)
+                if coro is not None:
+                    coros.append(coro)
+                if coros:
+                    Nemesis._start_experiment(random.choice(coros),
+                                              Nemesis._select_speedup())
+
+    @staticmethod
     def _get_waiting_handles(loop):
         handles = []
         for handle in loop._ready:
-            # no duplicates
             handle_info = Nemesis._parse_handle(handle)
-            if handle_info not in handles:
-                handles.append(handle_info)
+            handles.append(handle_info)
         return handles
 
+    @staticmethod
     def _parse_handle(handle):
         cb = handle._callback
         if isinstance(getattr(cb, '__self__', None), asyncio.tasks.Task):
             task = cb.__self__
             coro = task.get_coro()
-            return [task.get_name(), Nemesis.get_coro_name(coro)]
+            return [task.get_name(), Nemesis._get_coro_name(coro)]
         else:
             return [str(type(handle).__name__), cb.__name__]
 
+    @staticmethod
     def _get_current_coro(loop):
         tid = loop._thread_id
         assert tid, f"{loop} is not running, yet we attempted to sample it!"
@@ -160,6 +209,33 @@ class Nemesis(object):
         return None
 
     @staticmethod
+    def _get_event_loops():
+        '''Returns each thread's event loop, if it exists.'''
+        loops = []
+        for t in threading.enumerate():
+            frame = sys._current_frames().get(t.ident)
+            if frame:
+                loop = Nemesis._walk_back_until_loop(frame)
+                if loop and loop not in loops:
+                    loops.append(loop)
+        return loops
+
+    @staticmethod
+    def _walk_back_until_loop(frame):
+        '''Walks back the callstack until we are in a method named '_run_once'.
+        If this is ever true, we assume we are in an Asyncio event loop method,
+        and check to see if the 'self' variable is indeed and instance of
+        AbstractEventLoop. Return this variable if true.'''
+        while frame:
+            if frame.f_code.co_name == '_run_once' and 'self' in frame.f_locals:
+                loop = frame.f_locals['self']
+                if isinstance(loop, asyncio.AbstractEventLoop):
+                    return loop
+            else:
+                frame = frame.f_back
+        return None
+
+    @staticmethod
     def _should_trace(filename):
         '''Returns FALSE if filename is uninteresting to the user.
         Don't depend on this. It kind of sucks.'''
@@ -181,7 +257,17 @@ class Nemesis(object):
             return False
         return True
 
-    def get_coro_name(coro):
+    def _select_speedup():
+        '''
+        Returns a random speedup between 0% to 100%, in multiples of 5%.
+        Because a baseline is needed to calculate effect on program
+        performance, selects a speedup of 0 with 50% probability.
+        '''
+        r1 = random.randint(-9, 20)
+        return 1 + max(0, r1) / 20
+
+    @staticmethod
+    def _get_coro_name(coro):
         '''
         Stolen from _format_coroutine in cpython/Lib/asyncio/coroutines.py
         '''
@@ -223,26 +309,16 @@ if __name__ == "__main__":
                         metavar='',
                         type=float,
                         default=0.01)
-    parser.add_argument('-s', '--speedup',
-                        help='The amount of virtual speedup. Cannot go below one. Default is 2.0.',
-                        metavar='',
-                        type=float,
-                        default=2.0)
-    parser.add_argument('-c', '--task',
-                        help='The task to virtually speedup.',
-                        metavar='',
-                        type=str,
-                        required=True)
     parser.add_argument('-e', '--experiment-duration',
-                        help='The performance experiment duration. Defaults to 4 seconds.',
+                        help='The performance experiment duration. Defaults to 3 seconds.',
                         metavar='',
                         type=float,
-                        default=4)
-    parser.add_argument('-w', '--warmup-time',
-                        help='Amount of time to wait until the first performance experiment. Default is 0 milliseconds',
+                        default=3)
+    parser.add_argument('-f', '--filename',
+                        help='The filename to write results to.',
                         metavar='',
-                        type=float,
-                        default=0.1)
+                        type=str,
+                        default="results.html")
     parser.add_argument('prog',
                         type=str,
                         nargs='*',
@@ -253,10 +329,8 @@ if __name__ == "__main__":
     try:
         with open(args.prog[0], 'r', encoding='utf-8') as fp:
             code = compile(fp.read(), args.prog[0], "exec")
-            Nemesis(args.task,
-                    args.speedup,
-                    args.experiment_duration,
-                    args.warmup_time,
+            Nemesis(args.experiment_duration,
+                    args.filename,
                     args.interval).start()
             exec(code, the_globals)
             Nemesis.stop()