plesty.lib.test.field_test_concurrency

Solo-versus-concurrent field test comparison for devices sharing one host.

Running one device server per instrument on a single host is the PLESTY deployment default. Distinct TCP ports never contend, but the instruments behind them share a USB host controller, PCIe lanes, CPU and storage — and that contention is invisible to a single-device test.

ConcurrentFieldTest answers the question directly: run the same FieldTestPipeline gates on each device alone, then on all devices at once, and compare the per-command latency. If a command’s median latency grows beyond the tolerance only when its neighbours are active, the bottleneck is a resource shared below the application — not the device module.

Typical usage:

from plesty.lib.test.field_test import FieldTestPipeline
from plesty.lib.test.field_test_concurrency import ConcurrentFieldTest

comparison = ConcurrentFieldTest(
    {
        "k10cr1": FieldTestPipeline(K10CR1, "COM3", repetitions=50),
        "pm100d": FieldTestPipeline(PM100D, "USB0::0x1313::0x8078::P1::INSTR",
                                    repetitions=50),
        "spectrometer": FieldTestPipeline(LightField, "localhost", repetitions=50),
    },
    report_path="reports/concurrency-labhost",
).run()

raise SystemExit(0 if comparison.ok else 1)

Reports carry host names and instrument addresses; keep them internal.

Attributes

SOLO

CONCURRENT

Classes

LatencyDelta

How one command's latency changed between the solo and concurrent runs.

ConcurrencyComparison

Result of a solo-versus-concurrent field test across devices on one host.

ConcurrentFieldTest

Run the same field test gates per device alone, then all at once.

Functions

compare_reports(→ list[LatencyDelta])

Compare two reports of the same device command by command.

_ratio(→ float)

Return numerator / denominator, or 1.0 when the baseline is zero.

_overlap_fraction(→ float)

Return the share of the phase during which every device was running.

Module Contents

plesty.lib.test.field_test_concurrency.SOLO = 'solo'
plesty.lib.test.field_test_concurrency.CONCURRENT = 'concurrent'
class plesty.lib.test.field_test_concurrency.LatencyDelta

How one command’s latency changed between the solo and concurrent runs.

Variables:

  • command – Command key, e.g. "query:power".

  • solo – Latency statistics measured with the device running alone.

  • concurrent – Latency statistics measured with every device active.

  • median_ratio – Concurrent median divided by solo median.

  • p95_ratio – Concurrent p95 divided by solo p95.

  • solo_failure_rate – Failure rate while running alone.

  • concurrent_failure_rate – Failure rate while every device was active.

  • degraded – True when the median ratio exceeds the tolerance, or the command only started failing under load.

command: str
solo: plesty.lib.test.field_test.LatencyStats
concurrent: plesty.lib.test.field_test.LatencyStats
median_ratio: float
p95_ratio: float
solo_failure_rate: float
concurrent_failure_rate: float
degraded: bool
to_dict() → dict[str, Any]

Return a JSON-serialisable mapping of the delta.

Return type:

dict[str, Any]

class plesty.lib.test.field_test_concurrency.ConcurrencyComparison

Result of a solo-versus-concurrent field test across devices on one host.

Variables:

  • tolerance – Fractional latency growth accepted before a command counts as degraded (0.10 means a 10% slower median is still a pass).

  • gates – Field test gates the comparison was built from.

  • environment – Host and interpreter identification.

  • started_utc – ISO-8601 UTC timestamp of the solo phase.

  • finished_utc – ISO-8601 UTC timestamp after the concurrent phase.

  • overlap_fraction – Share of the concurrent phase during which every device was in fact running. A low value means the devices barely overlapped and the comparison understates contention.

  • solo – Per-device report from the solo phase.

  • concurrent – Per-device report from the concurrent phase.

  • deltas – Per-device latency deltas, worst median ratio first.

tolerance: float = 0.1
gates: list[str] = []
environment: dict[str, Any]
started_utc: str = ''
finished_utc: str = ''
overlap_fraction: float = 0.0
solo: dict[str, plesty.lib.test.field_test.FieldTestReport]
concurrent: dict[str, plesty.lib.test.field_test.FieldTestReport]
deltas: dict[str, list[LatencyDelta]]
property degraded: dict[str, list[LatencyDelta]]

Per device, the commands that degraded under concurrent load.

Return type:

dict[str, list[LatencyDelta]]

property ok: bool

True when no command degraded beyond the tolerance.

Return type:

bool

worst(device: str) → LatencyDelta | None

Return the most degraded command for device, or None.

Parameters:

device (str) – Device name as given to ConcurrentFieldTest.

Return type:

LatencyDelta | None

to_dict() → dict[str, Any]

Return the whole comparison as a JSON-serialisable mapping.

Return type:

dict[str, Any]

to_markdown() → str

Return a human-readable summary of the comparison.

Return type:

str

write(report_path: str | pathlib.Path) → list[pathlib.Path]

Write the comparison as <base>.json and <base>.md.

Parameters:

report_path (str | pathlib.Path) – Base path; a .json or .md suffix is stripped so both files land beside each other.

Returns:

The paths written, JSON first.

Return type:

list[pathlib.Path]

class plesty.lib.test.field_test_concurrency.ConcurrentFieldTest(pipelines: collections.abc.Mapping[str, plesty.lib.test.field_test.FieldTestPipeline], *, gates: collections.abc.Sequence[str] = ('stability',), tolerance: float = 0.1, report_path: str | None = None, settle_s: float = 2.0)

Run the same field test gates per device alone, then all at once.

The pipelines are reused across both phases: each is reset() before a phase so the two sets of statistics never mix.

Parameters:

  • pipelines (collections.abc.Mapping[str, plesty.lib.test.field_test.FieldTestPipeline]) – Device name to its configured pipeline. Give each pipeline enough repetitions that its solo run takes a few seconds — otherwise the concurrent phase barely overlaps.

  • gates (collections.abc.Sequence[str]) – Field test gates to compare. The default is the stability gate, which is the one that produces latency statistics.

  • tolerance (float) – Fractional median growth accepted before a command counts as degraded.

  • report_path (str | None) – Base path for the JSON and markdown comparison. When None, run() returns the comparison without writing it.

  • settle_s (float) – Pause between the solo and concurrent phases.

Store the pipelines and comparison settings.

pipelines
gates = ['stability']
tolerance = 0.1
report_path = None
settle_s = 2.0
run_solo() → dict[str, plesty.lib.test.field_test.FieldTestReport]

Run the gates on each device in turn, one device at a time.

Returns:

The per-device report of the solo phase.

Return type:

dict[str, plesty.lib.test.field_test.FieldTestReport]

run_concurrent() → tuple[dict[str, plesty.lib.test.field_test.FieldTestReport], float]

Run the gates on every device at the same time.

All workers wait on a barrier so the runs start together, and each worker’s start and end are timed to measure how much of the phase was genuinely concurrent.

Returns:

The per-device report of the concurrent phase, and the fraction of the phase during which every device was active.

Return type:

tuple[dict[str, plesty.lib.test.field_test.FieldTestReport], float]

run() → ConcurrencyComparison

Run both phases and return the comparison.

Returns:

The ConcurrencyComparison, written to report_path when one was given.

Return type:

ConcurrencyComparison

plesty.lib.test.field_test_concurrency.compare_reports(solo: plesty.lib.test.field_test.FieldTestReport, concurrent: plesty.lib.test.field_test.FieldTestReport, tolerance: float = 0.1) → list[LatencyDelta]

Compare two reports of the same device command by command.

Only commands measured in both runs are compared. A command counts as degraded when its median latency grew by more than tolerance, or when it only started failing under concurrent load.

Parameters:
Returns:

Deltas sorted by median ratio, most degraded first.

Return type:

list[LatencyDelta]

plesty.lib.test.field_test_concurrency._ratio(numerator: float, denominator: float) → float

Return numerator / denominator, or 1.0 when the baseline is zero.

Parameters:

  • numerator (float)

  • denominator (float)

Return type:

float

plesty.lib.test.field_test_concurrency._overlap_fraction(spans: collections.abc.Mapping[str, tuple[float, float]]) → float

Return the share of the phase during which every device was running.

Parameters:

spans (collections.abc.Mapping[str, tuple[float, float]]) – Per device, its (start, end) timestamps from the same clock.

Return type:

float