From 094c0b1ea4594050c199d8e584d0107a389c48e0 Mon Sep 17 00:00:00 2001
From: timotheyca <tim@ongoteam.yaconnect.com>
Date: Fri, 15 Jul 2022 21:37:52 +0300
Subject: [PATCH] better trace

---
 trace_common.py |  9 ++++++---
 trace_flow.py   | 52 ++++++++++++++++++++++++++++++++++---------------
 2 files changed, 42 insertions(+), 19 deletions(-)

diff --git a/trace_common.py b/trace_common.py
index 4d59d27..eac0e16 100644
--- a/trace_common.py
+++ b/trace_common.py
@@ -3,6 +3,7 @@ import os
 import random
 import shutil
 import time
+from pathlib import Path
 
 from rainbowadn.core import *
 from rainbowadn.instrument import *
@@ -11,10 +12,11 @@ from rainbowadn.testing.resolvers import *
 __all__ = ('get_dr', 'target_str', 'jsonify', 'get_fn', 'jsonify_list', 'dump', 'copy', 'DeintrumentationSize',)
 
 
-def get_dr() -> ExtendableResolver:
+def get_dr(mean_delay: float, caching: bool) -> ExtendableResolver:
     dr = DictResolver()
-    dr = DelayedResolver(dr, lambda: random.uniform(0.200, 0.500))
-    dr = CachingResolver(dr)
+    dr = DelayedResolver(dr, lambda: mean_delay * random.gammavariate(10.0, 0.1))
+    if caching:
+        dr = CachingResolver(dr)
     return dr
 
 
@@ -43,6 +45,7 @@ def jsonify(dumped: Instrumentation) -> dict:
 
 
 def get_fn() -> str:
+    Path('trace').mkdir(exist_ok=True)
     return f'trace/{int(time.time())}-{os.urandom(2).hex()}.json'
 
 
diff --git a/trace_flow.py b/trace_flow.py
index 9484d71..1758f8a 100644
--- a/trace_flow.py
+++ b/trace_flow.py
@@ -1,6 +1,7 @@
 import asyncio
 import random
 from contextlib import ExitStack
+from typing import Any, Callable, Coroutine
 
 from nacl.signing import SigningKey
 
@@ -20,6 +21,7 @@ def get_instrumentations() -> list[Instrumentation]:
         sleep_cc,
         Concurrency(ActiveBinaryTree, 'add'),
         Concurrency(ActiveBinaryTree, 'contains'),
+        Concurrency(FlowStandard, 'verify_subset'),
     ]
 
 
@@ -51,48 +53,58 @@ async def _generate(
     return bank
 
 
-async def _migrate(bank: BankBlock) -> BankBlock:
+async def _migrate(bank: BankBlock, params) -> BankBlock:
     assert_true(await bank.verify())
-    bank = BankBlock(await get_dr().migrate_resolved(bank.reference))
+    bank = BankBlock(await get_dr(params['delay'], params['caching']).migrate_resolved(bank.reference))
     print('migrated')
     return bank
 
 
-async def _instrument(bank: BankBlock) -> list[Instrumentation]:
+async def _instrument(process: Callable[[], Coroutine[Any, Any, None]]) -> list[Instrumentation]:
     with ExitStack() as estack:
         instrumentations: list[Instrumentation] = get_instrumentations()
         for stacked in instrumentations:
             stacked.enter(estack)
-        assert_true(await bank.verify())
+        try:
+            await process()
+        except Terminated:
+            pass
     print('deinstrumentation (should be empty):', Instrumentation.deinstrumentation)
     print('instrumented')
     return instrumentations
 
 
-params = (
-    16,
-    8, 16,
-    8, 16,
-)
+async def _process(bank: BankBlock) -> None:
+    assert_true(await bank.verify())
+    print('processed')
 
 
-async def _trace():
+async def _pre_measure(bank: BankBlock):
+    with Measure(bank, 'verify') as measurement:
+        await _process(bank)
+    print('pre-measured', *(f'{t:.3f}' for t in measurement.log))
+
+
+async def _trace(params):
     set_gather_linear()
     bank = await _generate(
-        *params
+        params['blocks'],
+        *params['subjects'],
+        *params['transactions'],
     )
-    bank = await _migrate(bank)
+    await _pre_measure(bank)
+    bank = await _migrate(bank, params)
     set_gather_asyncio()
     with DeintrumentationSize(Instrumentation, 'deinstrument'):
         with Counter(DeintrumentationSize, 'instrument') as de_ctr:
-            instrumentations = await _instrument(bank)
+            instrumentations = await _instrument(lambda: _process(bank))
             print(jsonify(de_ctr))
     print('traced')
     return instrumentations
 
 
-async def main():
-    instrumentations = await _trace()
+async def trace(params):
+    instrumentations = await _trace(params)
     fn = get_fn()
     jsonified = jsonify_list(instrumentations)
     dump(fn, jsonified | {'params': params})
@@ -101,5 +113,13 @@ async def main():
     print('plotted')
 
 
+preset_long = dict(blocks=64, subjects=(8, 16), transactions=(8, 16), caching=True, delay=.35)
+preset_short = dict(blocks=16, subjects=(8, 16), transactions=(8, 16), caching=True, delay=.35)
+preset_old = dict(blocks=16, subjects=(8, 15), transactions=(8, 15), caching=False, delay=.35)
+
 if __name__ == '__main__':
-    asyncio.run(main())
+    asyncio.run(
+        trace(
+            preset_short
+        )
+    )