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 501c1335005..d006e3f247e 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 @@ 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 @@ def test_it_can_be_resumed_when_resumed(self, pause_manager): 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 5abf7a36b45..3c88f850baa 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 @@ def test_total_execution_time_before_start_raises(self): 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