test: make benchmarks unittest

This commit is contained in:
Robert Jördens 2015-06-28 20:08:39 -06:00
parent 593dafc118
commit f7427dda39
2 changed files with 109 additions and 44 deletions

View File

@ -1,9 +1,10 @@
from artiq import * from math import sqrt
from artiq.test.hardware_testbench import *
from artiq.language import *
from artiq.test.hardware_testbench import ExperimentCase
from artiq.coredevice.runtime_exceptions import RTIOUnderflow from artiq.coredevice.runtime_exceptions import RTIOUnderflow
class RTTTest(ExperimentCase):
class RTT(Experiment, AutoDB): class RTT(Experiment, AutoDB):
class DBKeys: class DBKeys:
core = Device() core = Device()
@ -14,19 +15,33 @@ class RTTTest(ExperimentCase):
def run(self): def run(self):
self.ttl_inout.output() self.ttl_inout.output()
delay(1*us) delay(1*us)
t0 = now()
with parallel: with parallel:
self.ttl_inout.gate_falling(2*us) self.ttl_inout.gate_rising(2*us)
with sequential:
delay(1*us)
t0 = now()
self.ttl_inout.pulse(1*us) self.ttl_inout.pulse(1*us)
self.rtt = self.ttl_inout.timestamp() - t0 self.rtt = self.ttl_inout.timestamp() - t0
def test_rtt(self):
rtt = self.execute(self.RTT)["rtt"] class Loopback(Experiment, AutoDB):
self.assertGreater(rtt, 0*ns) class DBKeys:
self.assertLess(rtt, 40*ns) core = Device()
loop_in = Device()
loop_out = Device()
rtt = Result()
@kernel
def run(self):
with parallel:
self.loop_in.gate_rising(2*us)
with sequential:
delay(1*us)
t0 = now()
self.loop_out.pulse(1*us)
self.rtt = self.loop_in.timestamp() - t0
class PulseRateTest(ExperimentCase):
class PulseRate(Experiment, AutoDB): class PulseRate(Experiment, AutoDB):
class DBKeys: class DBKeys:
core = Device() core = Device()
@ -48,9 +63,58 @@ class PulseRateTest(ExperimentCase):
self.pulse_rate = cycles_to_time(2*dt) self.pulse_rate = cycles_to_time(2*dt)
break break
def test_rate(self):
rate = self.execute(self.PulseRate)["pulse_rate"] class CoredeviceTest(ExperimentCase):
def test_rtt(self):
rtt = self.execute(RTT)["rtt"]
print(rtt)
self.assertGreater(rtt, 0*ns)
self.assertLess(rtt, 100*ns)
def test_loopback(self):
rtt = self.execute(Loopback)["rtt"]
print(rtt)
self.assertGreater(rtt, 0*ns)
self.assertLess(rtt, 40*ns)
def test_pulse_rate(self):
rate = self.execute(PulseRate)["pulse_rate"]
print(rate)
self.assertGreater(rate, 100*ns) self.assertGreater(rate, 100*ns)
self.assertLess(rate, 2000*ns) self.assertLess(rate, 2500*ns)
class RPCTiming(Experiment, AutoDB):
class DBKeys:
core = Device()
repeats = Argument(100)
rpc_time_mean = Result()
rpc_time_stddev = Result()
def nop(self, x):
pass
@kernel
def bench(self):
self.ts = [0. for _ in range(self.repeats)]
for i in range(self.repeats):
t1 = self.core.get_rtio_time()
self.nop(1)
t2 = self.core.get_rtio_time()
self.ts[i] = t2 - t1
def run(self):
self.bench()
mean = sum(self.ts)/self.repeats
self.rpc_time_stddev = sqrt(
sum([(t - mean)**2 for t in self.ts])/self.repeats)*s
self.rpc_time_mean = mean*s
class RPCTest(ExperimentCase):
def test_rpc_timing(self):
res = self.execute(RPCTiming)
print(res)
self.assertGreater(res["rpc_time_mean"], 100*ns)
self.assertLess(res["rpc_time_mean"], 10*ms)
self.assertLess(res["rpc_time_stddev"], 1*ms)

View File

@ -3,7 +3,7 @@ import sys
import unittest import unittest
import logging import logging
from artiq import * from artiq.language import *
from artiq.protocols.file_db import FlatFileDB from artiq.protocols.file_db import FlatFileDB
from artiq.master.worker_db import DBHub, ResultDB from artiq.master.worker_db import DBHub, ResultDB
from artiq.frontend.artiq_run import ( from artiq.frontend.artiq_run import (
@ -33,6 +33,7 @@ class ExperimentCase(unittest.TestCase):
try: try:
exp = cls(self.dbh, scheduler=sched, **kwargs) exp = cls(self.dbh, scheduler=sched, **kwargs)
except KeyError as e: except KeyError as e:
# skip if ddb does not match requirements
raise unittest.SkipTest(*e.args) raise unittest.SkipTest(*e.args)
self.rdb.build() self.rdb.build()
exp.run() exp.run()