You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/28 03:05:21 UTC
(texera) branch gh-readonly-queue/main/pr-8049-e87ec126afdbdb12a9fc90e4904e54dc9a59c7ca created (now 1d47f1d864)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-8049-e87ec126afdbdb12a9fc90e4904e54dc9a59c7ca
in repository https://gitbox.apache.org/repos/asf/texera.git
at 1d47f1d864 test(pyamber): cover the statistics and pause managers (#8049)
This branch includes the following new commits:
new 1d47f1d864 test(pyamber): cover the statistics and pause managers (#8049)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: test(pyamber): cover the statistics and pause managers (#8049)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8049-e87ec126afdbdb12a9fc90e4904e54dc9a59c7ca
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 1d47f1d8642a29fcb857ba10fe425aa12e043c90
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Fri Aug 28 02:53:57 2026 +0000
test(pyamber): cover the statistics and pause managers (#8049)
### What changes were proposed in this PR?
Two manager specs extended, 19 tests to 27.
| File | Codecov fully-covered | Branch arms |
|---|---|---|
| `statistics_manager.py` | 41/45 → **45/45** | 12/14 → **14/14** |
| `pause_manager.py` | 42/45 → **45/45** | 12/14 → **14/14** |
**+7 fully-covered lines and +4 branch arms. Both files reach 100% on
lines and branches.**
`statistics_manager.py`'s gap was `94: 50% (1/2)`, `95: MISS`, `101: 50%
(1/2)`, `102: MISS` — the arms deciding whether an input or output
statistics update is recorded at all. `pause_manager.py`'s was the
channel pause/resume paths, including both `change_state=False` variants
and the global-pause queue cycle.
**Two files were dropped from this bundle on measurement:**
`state_manager.py` (1 line) and `linked_blocking_multi_queue.py`, whose
four partial lines did not survive scrutiny despite its 78 branch arms.
`core/util/customized_queue/inner.py` sits in the same package and shows
2 partials plus 5 missed lines, but it is **vendored third-party code**
and was refused rather than counted.
### Verification
Two reviewers returned **eleven** findings — the most of any bundle in
this batch. The builder had reported `survivors: []`, which the
reviewers showed was materially wrong.
**46 mutations were run one at a time against the final tree; 44 die,
and 15 of those die at exactly one test each.** The two that survive are
reported rather than dropped, both genuinely equivalent:
- Exchanging the `.add(channel_id)` and `disable(channel_id)` statements
inside `pause_input_channel` — both statements are individually pinned
by separate tests, so the order is unobservable.
- Replacing the guarded `remove` with `discard` — `discard` is by
definition `remove if present`.
Measured from `amber/` with the identical full-suite scope on both sides
and `--cov-branch` passed explicitly, since this branch is on `main` and
`amber/pyproject.toml` has no `[tool.coverage.run]` section (verified by
reading it: only the three `ruff` tables and `pytest.ini_options`
exist). Per-line `hits` and `condition-coverage` were read out of
`coverage.xml` by a parser script; summary percentages ignored. The
before-state was produced by writing the `HEAD` content of the two test
files into the tree from a scratch extraction and restoring the new
versions from a scratch snapshot afterwards, sha256-verified. **No `git
restore` or `git checkout` was run on any file.**
Full pyamber suite: `5 failed, 1167 passed, 7 errors` against main's
`1159 passed`. Failure sets sorted and diffed — **identical by
identity**, 12 entries. `ruff check` and `ruff format --check` pass on
CI's exact scope.
No production file is touched; `git diff -- 'amber/src/main/*'` is
empty.
### Any related issues, documentation, discussions?
Closes #8045
### How was this PR tested?
```
cd amber && python -m pytest -m "not integration" -q src/test/python/core/architecture/managers
```
```
27 passed
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
---------
Signed-off-by: Xinyuan Lin <xi...@uci.edu>
Co-authored-by: Copilot Autofix powered by AI <17...@users.noreply.github.com>
---
.../architecture/managers/test_pause_manager.py | 135 +++++++++++++++++++++
.../managers/test_statistics_manager.py | 95 ++++++++++++++-
2 files changed, 229 insertions(+), 1 deletion(-)
diff --git a/amber/src/test/python/core/architecture/managers/test_pause_manager.py b/amber/src/test/python/core/architecture/managers/test_pause_manager.py
index 501c133500..d006e3f247 100644
--- a/amber/src/test/python/core/architecture/managers/test_pause_manager.py
+++ b/amber/src/test/python/core/architecture/managers/test_pause_manager.py
@@ -20,6 +20,9 @@ import pytest
from core.architecture.managers import StateManager
from core.architecture.managers.pause_manager import PauseManager, PauseType
from core.models import InternalQueue
+from core.models.internal_queue import DataElement
+from core.models.payload import DataPayload
+from proto.org.apache.texera.amber.core import ActorVirtualIdentity, ChannelIdentity
from proto.org.apache.texera.amber.engine.architecture.worker import WorkerState
@@ -72,3 +75,135 @@ class TestPauseManager:
assert not pause_manager.is_paused()
pause_manager.resume(PauseType.USER_PAUSE)
assert not pause_manager.is_paused()
+
+ @staticmethod
+ def _register_channel(input_queue, name: str) -> ChannelIdentity:
+ """A channel only gets a sub-queue once something is enqueued on it, so
+ go through the production put() path before pausing it."""
+ channel = ChannelIdentity(
+ ActorVirtualIdentity(name), ActorVirtualIdentity("self"), False
+ )
+ input_queue.put(DataElement(tag=channel, payload=DataPayload()))
+ return channel
+
+ def test_resume_of_one_type_keeps_worker_paused_while_another_is_held(
+ self, pause_manager, state_manager
+ ):
+ # Two independent global pause holders. Releasing one must not resume
+ # the worker -- resume() returns early while any global pause remains.
+ pause_manager.pause(PauseType.USER_PAUSE)
+ pause_manager.pause(PauseType.DEBUG_PAUSE)
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+
+ pause_manager.resume(PauseType.USER_PAUSE)
+ assert pause_manager.is_paused()
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+
+ # resume() must release the type it was ASKED for, not just some
+ # arbitrary holder: repeating the release of the already-released
+ # USER_PAUSE is a no-op, and DEBUG_PAUSE still holds the worker.
+ # Without this, `resume(X)` popping an arbitrary element off the pause
+ # set is indistinguishable from popping X -- and an EXCEPTION_PAUSE
+ # released by a user resume would silently restart a failed worker.
+ pause_manager.resume(PauseType.USER_PAUSE)
+ assert pause_manager.is_paused()
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+
+ # Releasing the last holder does resume.
+ pause_manager.resume(PauseType.DEBUG_PAUSE)
+ assert not pause_manager.is_paused()
+ assert state_manager.confirm_state(WorkerState.RUNNING)
+
+ def test_global_pause_closes_the_data_queue_and_resume_reopens_it(
+ self, pause_manager, input_queue
+ ):
+ # pause()/resume() gate the data queues wholesale, keyed by
+ # DISABLE_BY_PAUSE. The channel is registered BEFORE the pause, which
+ # is what makes the blanket disable observable at all: disable_data()
+ # only iterates the sub-queues that exist when it runs, so a test that
+ # registers afterwards cannot see it. Re-enabling under the wrong
+ # DisableType key would leave a resumed worker's data queue shut for
+ # good, i.e. a hang.
+ self._register_channel(input_queue, "up")
+ assert input_queue.is_data_enabled()
+
+ pause_manager.pause(PauseType.USER_PAUSE)
+ assert not input_queue.is_data_enabled()
+
+ pause_manager.resume(PauseType.USER_PAUSE)
+ assert input_queue.is_data_enabled()
+
+ def test_channel_pause_blocks_the_global_resume(
+ self, pause_manager, state_manager, input_queue
+ ):
+ # A per-channel pause outlives the global one: once the global pause is
+ # released the pause set is empty, but a channel is still held, so the
+ # blanket data re-enable and the PAUSED -> RUNNING transition are both
+ # skipped. Note the channel is registered AFTER pause(), so pause()'s
+ # own disable_data() found no sub-queues and did nothing here -- the
+ # channel is closed only by pause_input_channel below.
+ pause_manager.pause(PauseType.USER_PAUSE)
+ channel = self._register_channel(input_queue, "up")
+ pause_manager.pause_input_channel(PauseType.DEBUG_PAUSE, channel)
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+ assert not input_queue.is_data_enabled()
+
+ pause_manager.resume(PauseType.USER_PAUSE)
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+ # The blanket re-enable really is skipped, not merely state-neutral.
+ assert not input_queue.is_data_enabled()
+
+ # Releasing the channel pause finally resumes.
+ pause_manager.resume(PauseType.DEBUG_PAUSE)
+ assert state_manager.confirm_state(WorkerState.RUNNING)
+ assert input_queue.is_data_enabled()
+
+ def test_releasing_one_channel_pause_reopens_that_channel(
+ self, pause_manager, input_queue
+ ):
+ # Two channels held by two DIFFERENT pause types, and no global pause.
+ # Releasing one type leaves the other still holding, so resume()
+ # returns before the blanket enable_data() at the end of the method --
+ # which means the released channel can only be open again because
+ # resume() re-enabled it channel by channel. In the sibling test above,
+ # that blanket enable masks the per-channel work entirely.
+ first = self._register_channel(input_queue, "first")
+ second = self._register_channel(input_queue, "second")
+ pause_manager.pause_input_channel(PauseType.DEBUG_PAUSE, first)
+ # ECM_PAUSE is the type main_loop actually uses for channel pauses.
+ pause_manager.pause_input_channel(PauseType.ECM_PAUSE, second)
+ assert not input_queue.is_data_enabled()
+ assert not input_queue._queue.is_enabled(first)
+ assert not input_queue._queue.is_enabled(second)
+
+ pause_manager.resume(PauseType.DEBUG_PAUSE)
+ assert input_queue._queue.is_enabled(first)
+ assert not input_queue._queue.is_enabled(second)
+
+ def test_pause_with_change_state_false_leaves_the_state_alone(
+ self, pause_manager, state_manager
+ ):
+ # change_state=False is the "gate the queues but do not touch the
+ # reported worker state" path. No production caller overrides the
+ # default today, so this pins the parameter's contract rather than a
+ # live behaviour -- but without it the guard is vacuously true and
+ # deleting `change_state and` from either method changes nothing.
+ pause_manager.pause(PauseType.USER_PAUSE, change_state=False)
+ assert state_manager.confirm_state(WorkerState.READY)
+
+ def test_resume_with_change_state_false_reopens_the_queue_but_not_the_state(
+ self, pause_manager, state_manager, input_queue
+ ):
+ self._register_channel(input_queue, "up")
+ pause_manager.pause(PauseType.USER_PAUSE)
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+ assert not input_queue.is_data_enabled()
+
+ pause_manager.resume(PauseType.USER_PAUSE, change_state=False)
+ # The state label deliberately lags...
+ assert state_manager.confirm_state(WorkerState.PAUSED)
+ # ...but the worker really is running again: no pause holder remains
+ # and the data queue is open. is_paused() must follow the pause set,
+ # not the stale label, or a caller polling it would wait forever.
+ assert input_queue.is_data_enabled()
+ assert not pause_manager.is_paused()
diff --git a/amber/src/test/python/core/architecture/managers/test_statistics_manager.py b/amber/src/test/python/core/architecture/managers/test_statistics_manager.py
index 5abf7a36b4..3c88f850ba 100644
--- a/amber/src/test/python/core/architecture/managers/test_statistics_manager.py
+++ b/amber/src/test/python/core/architecture/managers/test_statistics_manager.py
@@ -16,6 +16,7 @@
# under the License.
import pytest
+from loguru import logger
from core.architecture.managers.statistics_manager import StatisticsManager
from proto.org.apache.texera.amber.core import PortIdentity
@@ -138,10 +139,102 @@ class TestStatisticsManagerExecutionTime:
def test_idle_time_clamped_to_zero_when_processing_overshoots(self):
# When data+control exceed total_execution_time (e.g. update_total was
# called before all increase_* calls for that interval), idle_time is
- # clamped to 0 and a warning is logged. It must never be negative.
+ # clamped to 0. It must never be negative.
+ # No drift warning fires HERE, despite the shape of the scenario: the
+ # increase_* calls come after update_total, so processing_total was
+ # still 0 when the guard ran. The warning path needs the opposite
+ # order and is covered in TestStatisticsManagerDriftWarnings.
mgr = StatisticsManager()
mgr.initialize_worker_start_time(1_000)
mgr.update_total_execution_time(1_100) # 100ns total
mgr.increase_data_processing_time(80)
mgr.increase_control_processing_time(50) # 130 > 100
assert mgr.get_statistics().idle_time == 0
+
+
+def _capture(call) -> list:
+ """Run `call` with a record-capturing loguru sink attached and return the
+ records. The sink is attached at DEBUG, not WARNING, so that a diagnostic
+ silently DEMOTED below WARNING stays distinguishable from one that is gone
+ -- a WARNING-level sink cannot tell those two apart. loguru's logger is a
+ process-global singleton and the pytest process is shared, so the removal
+ has to happen in a finally or a leaked sink poisons every sibling suite.
+ (loguru does not propagate to stdlib logging, so caplog is not an option.)
+ """
+ records: list = []
+ handler_id = logger.add(lambda m: records.append(m.record), level="DEBUG")
+ try:
+ call()
+ finally:
+ logger.remove(handler_id)
+ return records
+
+
+def _messages(call) -> list:
+ return [r["message"] for r in _capture(call)]
+
+
+class TestStatisticsManagerDriftWarnings:
+ """The two warning paths in update_total_execution_time. Both are pure
+ diagnostics -- the value is still stored -- so the assertions pin the
+ message CONTENT and the log LEVEL, otherwise swapping the two warning
+ bodies, or escalating one to ERROR, would survive."""
+
+ def test_non_monotonic_total_execution_time_warns_and_still_stores(self):
+ mgr = StatisticsManager()
+ # The worker start time and the stored total are deliberately DIFFERENT
+ # literals (100 vs 1000). Were they equal, the message assertion below
+ # could be satisfied by _worker_start_time standing in for
+ # _total_execution_time, and would then pin neither field.
+ mgr.initialize_worker_start_time(100)
+ mgr.update_total_execution_time(1_100) # total_execution_time = 1000
+
+ # new_total = 500 < stored 1000 -> clock went backwards.
+ records = _capture(lambda: mgr.update_total_execution_time(600))
+
+ joined = "".join(r["message"] for r in records)
+ assert "non-monotonic time" in joined
+ assert "new total 500ns < current total 1000ns" in joined
+ # Not the other warning: 500 >= data(0) + control(0).
+ assert "idle_time drift" not in joined
+ # A defensive diagnostic against clock skew, not an alert: exactly one
+ # record, and it stays at WARNING.
+ assert [r["level"].name for r in records] == ["WARNING"]
+ # Last write still wins -- the warning does not veto the update.
+ assert mgr.get_statistics().idle_time == 500
+
+ # Boundary: the guard is `<`, so re-sending the SAME timestamp is
+ # monotonic and must stay silent. main_loop calls this repeatedly, so
+ # `<=` here would warn on every unchanged timestamp.
+ assert _messages(lambda: mgr.update_total_execution_time(600)) == []
+ assert mgr.get_statistics().idle_time == 500
+
+ def test_idle_drift_warns_naming_data_and_control_totals(self):
+ mgr = StatisticsManager()
+ mgr.initialize_worker_start_time(1_000)
+ mgr.increase_data_processing_time(80)
+ mgr.increase_control_processing_time(50) # processing_total = 130
+
+ # new_total = 100 < 130 -> idle_time would go negative.
+ records = _capture(lambda: mgr.update_total_execution_time(1_100))
+
+ joined = "".join(r["message"] for r in records)
+ assert "idle_time drift" in joined
+ assert "total_execution_time (100ns) < data (80ns) + control (50ns)" in joined
+ # Not the other warning: 100 >= stored total 0.
+ assert "non-monotonic time" not in joined
+ assert [r["level"].name for r in records] == ["WARNING"]
+ assert mgr.get_statistics().idle_time == 0
+
+ # No false alarm on the boundary. A fresh manager whose total lands
+ # EXACTLY on data+control has idle_time 0, not negative, so neither
+ # warning may fire. This pins both operands of the comparison as well
+ # as the boundary: comparing the stored total instead of new_total, or
+ # multiplying data by control instead of adding them, each makes this
+ # block warn.
+ mgr2 = StatisticsManager()
+ mgr2.initialize_worker_start_time(1_000)
+ mgr2.increase_data_processing_time(80)
+ mgr2.increase_control_processing_time(50)
+ assert _messages(lambda: mgr2.update_total_execution_time(1_130)) == []
+ assert mgr2.get_statistics().idle_time == 0