plesty.lib.test.field_test_concurrency
======================================
.. py:module:: plesty.lib.test.field_test_concurrency
.. autoapi-nested-parse::
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.
:class:`ConcurrentFieldTest` answers the question directly: run the same
:class:`~plesty.lib.test.field_test.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
----------
.. autoapisummary::
plesty.lib.test.field_test_concurrency.SOLO
plesty.lib.test.field_test_concurrency.CONCURRENT
Classes
-------
.. autoapisummary::
plesty.lib.test.field_test_concurrency.LatencyDelta
plesty.lib.test.field_test_concurrency.ConcurrencyComparison
plesty.lib.test.field_test_concurrency.ConcurrentFieldTest
Functions
---------
.. autoapisummary::
plesty.lib.test.field_test_concurrency.compare_reports
plesty.lib.test.field_test_concurrency._ratio
plesty.lib.test.field_test_concurrency._overlap_fraction
Module Contents
---------------
.. py:data:: SOLO
:value: 'solo'
.. py:data:: CONCURRENT
:value: 'concurrent'
.. py:class:: LatencyDelta
How one command's latency changed between the solo and concurrent runs.
:ivar command: Command key, e.g. ``"query:power"``.
:ivar solo: Latency statistics measured with the device running alone.
:ivar concurrent: Latency statistics measured with every device active.
:ivar median_ratio: Concurrent median divided by solo median.
:ivar p95_ratio: Concurrent p95 divided by solo p95.
:ivar solo_failure_rate: Failure rate while running alone.
:ivar concurrent_failure_rate: Failure rate while every device was active.
:ivar degraded: ``True`` when the median ratio exceeds the tolerance, or the
command only started failing under load.
.. py:attribute:: command
:type: str
.. py:attribute:: solo
:type: plesty.lib.test.field_test.LatencyStats
.. py:attribute:: concurrent
:type: plesty.lib.test.field_test.LatencyStats
.. py:attribute:: median_ratio
:type: float
.. py:attribute:: p95_ratio
:type: float
.. py:attribute:: solo_failure_rate
:type: float
.. py:attribute:: concurrent_failure_rate
:type: float
.. py:attribute:: degraded
:type: bool
.. py:method:: to_dict() -> dict[str, Any]
Return a JSON-serialisable mapping of the delta.
.. py:class:: ConcurrencyComparison
Result of a solo-versus-concurrent field test across devices on one host.
:ivar tolerance: Fractional latency growth accepted before a command counts
as degraded (``0.10`` means a 10% slower median is still a pass).
:ivar gates: Field test gates the comparison was built from.
:ivar environment: Host and interpreter identification.
:ivar started_utc: ISO-8601 UTC timestamp of the solo phase.
:ivar finished_utc: ISO-8601 UTC timestamp after the concurrent phase.
:ivar 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.
:ivar solo: Per-device report from the solo phase.
:ivar concurrent: Per-device report from the concurrent phase.
:ivar deltas: Per-device latency deltas, worst median ratio first.
.. py:attribute:: tolerance
:type: float
:value: 0.1
.. py:attribute:: gates
:type: list[str]
:value: []
.. py:attribute:: environment
:type: dict[str, Any]
.. py:attribute:: started_utc
:type: str
:value: ''
.. py:attribute:: finished_utc
:type: str
:value: ''
.. py:attribute:: overlap_fraction
:type: float
:value: 0.0
.. py:attribute:: solo
:type: dict[str, plesty.lib.test.field_test.FieldTestReport]
.. py:attribute:: concurrent
:type: dict[str, plesty.lib.test.field_test.FieldTestReport]
.. py:attribute:: deltas
:type: dict[str, list[LatencyDelta]]
.. py:property:: degraded
:type: dict[str, list[LatencyDelta]]
Per device, the commands that degraded under concurrent load.
.. py:property:: ok
:type: bool
``True`` when no command degraded beyond the tolerance.
.. py:method:: worst(device: str) -> LatencyDelta | None
Return the most degraded command for *device*, or ``None``.
:param device: Device name as given to :class:`ConcurrentFieldTest`.
.. py:method:: to_dict() -> dict[str, Any]
Return the whole comparison as a JSON-serialisable mapping.
.. py:method:: to_markdown() -> str
Return a human-readable summary of the comparison.
.. py:method:: write(report_path: str | pathlib.Path) -> list[pathlib.Path]
Write the comparison as ``.json`` and ``.md``.
:param report_path: Base path; a ``.json`` or ``.md`` suffix is stripped
so both files land beside each other.
:returns: The paths written, JSON first.
.. py:class:: 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
:meth:`~plesty.lib.test.field_test.FieldTestPipeline.reset` before a phase
so the two sets of statistics never mix.
:param pipelines: 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.
:param gates: Field test gates to compare. The default is the stability gate,
which is the one that produces latency statistics.
:param tolerance: Fractional median growth accepted before a command counts
as degraded.
:param report_path: Base path for the JSON and markdown comparison. When
``None``, :meth:`run` returns the comparison without writing it.
:param settle_s: Pause between the solo and concurrent phases.
Store the pipelines and comparison settings.
.. py:attribute:: pipelines
.. py:attribute:: gates
:value: ['stability']
.. py:attribute:: tolerance
:value: 0.1
.. py:attribute:: report_path
:value: None
.. py:attribute:: settle_s
:value: 2.0
.. py:method:: 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.
.. py:method:: 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.
.. py:method:: run() -> ConcurrencyComparison
Run both phases and return the comparison.
:returns: The :class:`ConcurrencyComparison`, written to ``report_path``
when one was given.
.. py:function:: 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.
:param solo: Report from the device running alone.
:param concurrent: Report from the device running alongside the others.
:param tolerance: Fractional median growth accepted before a command counts
as degraded.
:returns: Deltas sorted by median ratio, most degraded first.
.. py:function:: _ratio(numerator: float, denominator: float) -> float
Return ``numerator / denominator``, or ``1.0`` when the baseline is zero.
.. py:function:: _overlap_fraction(spans: collections.abc.Mapping[str, tuple[float, float]]) -> float
Return the share of the phase during which every device was running.
:param spans: Per device, its ``(start, end)`` timestamps from the same clock.