Yicong-Huang commented on code in PR #6011:
URL: https://github.com/apache/texera/pull/6011#discussion_r3649989280
##########
amber/src/main/scala/org/apache/texera/amber/engine/common/statetransition/StateManager.scala:
##########
@@ -43,6 +43,16 @@ class StateManager[T](
private var currentState: T = initialState
+ // Monotonically increasing version, bumped on every successful state
transition.
+ // It is the state machine's logical clock: because a single owner drives all
+ // transitions, the version totally orders them in causal order. Reporting
this
+ // version alongside the state lets remote observers (e.g. the controller)
reject
+ // stale state reports that arrive out of order, without relying on
wall-clock
+ // timestamps that cannot be compared across processes.
+ private var stateVersion: Long = 0L
+
+ def getStateVersion: Long = stateVersion
Review Comment:
Done in d142028 — added `getStateWithVersion: (T, Long)` (and
`get_state_with_version()` on the Python side, with mirror pointers) and
switched all 10 report sites (6 Scala, 4 Python) to it, so a report can never
pair a state with another transition's version. Added pair-consistency tests in
WorkerStateManagerSpec / test_state_manager.py.
##########
amber/src/main/scala/org/apache/texera/amber/engine/architecture/deploysemantics/layer/WorkerExecution.scala:
##########
@@ -37,50 +41,56 @@ case class WorkerExecution() extends Serializable {
private var stats: WorkerStatistics = {
WorkerStatistics(Seq.empty, Seq.empty, 0, 0, 0)
}
- private var lastUpdateTimeStamp = 0L
+ // Logical version of the last applied state, sourced from the worker's
+ // WorkerStateManager. Starts below any real version so the first report
applies.
+ private var lastStateVersion = -1L
+ // Wall-clock (controller-side nanoTime) of the last applied stats snapshot.
+ private var lastStatsTimeStamp = 0L
+
+ private def isTerminal(s: WorkerState): Boolean = s == COMPLETED || s ==
TERMINATED
/**
- * Updates both the worker state and statistics if the provided timestamp
is newer
- * than the last recorded update timestamp. This ensures that only the most
recent
- * data is reflected in the execution state.
+ * Applies a worker state report, ordered causally by the worker's monotonic
+ * `stateVersion` rather than by wall-clock time. A report is applied only
when it
+ * is strictly newer than the last applied one, so a stale state that
arrives late
+ * (e.g. the RUNNING snapshot carried by a slow startWorker response)
cannot clobber
+ * a newer state. In addition, terminal states (COMPLETED/TERMINATED) are
absorbing:
+ * once reached, no later report can move the worker out of them.
*
- * @param timeStamp the nanosecond-timestamp of this update
- * @param state the new WorkerState to set
- * @param stats the new WorkerStatistics to set
+ * @param stateVersion the worker-side monotonic version of this state
+ * @param newState the reported WorkerState
*/
- def update(timeStamp: Long, state: WorkerState, stats: WorkerStatistics):
Unit = {
- if (this.lastUpdateTimeStamp < timeStamp) {
- this.stats = stats
- this.state = state
- this.lastUpdateTimeStamp = timeStamp
+ def updateState(stateVersion: Long, newState: WorkerState): Unit = {
+ if (isTerminal(this.state)) {
+ return
+ }
+ if (this.lastStateVersion < stateVersion) {
+ this.state = newState
+ this.lastStateVersion = stateVersion
}
}
/**
- * Updates only the worker state if the provided timestamp is newer than the
- * last recorded update timestamp.
- *
- * @param timeStamp the nanosecond-timestamp of this update
- * @param state the new WorkerState to set
+ * Forces the worker into TERMINATED, e.g. when the controller kills a
region. This
+ * still respects terminal-state absorption: a worker that already
COMPLETED on its
+ * own is left as COMPLETED. Uses the maximum version so it wins over any
in-flight
+ * non-terminal report for a worker that had not yet reached a terminal
state.
*/
- def update(timeStamp: Long, state: WorkerState): Unit = {
- if (this.lastUpdateTimeStamp < timeStamp) {
- this.state = state
- this.lastUpdateTimeStamp = timeStamp
- }
- }
+ def forceTerminate(): Unit = updateState(Long.MaxValue, TERMINATED)
Review Comment:
Agreed — keeping the two pieces as they are: the forceTerminate scaladoc
documents that it relies on terminal absorption, and the WorkerExecutionSpec
cases (forceTerminate on a COMPLETED vs. a non-terminal worker) pin both
behaviors.
##########
amber/src/main/protobuf/org/apache/texera/amber/engine/architecture/worker/statistics.proto:
##########
@@ -60,4 +60,7 @@ message WorkerStatistics {
message WorkerMetrics {
WorkerState worker_state = 1 [(scalapb.field).no_box = true];
WorkerStatistics worker_statistics = 2 [(scalapb.field).no_box = true];
+ // Monotonic per-worker version of worker_state, used by the controller to
order
+ // state reports causally instead of by wall-clock timestamp. See
WorkerStateManager.
+ int64 state_version = 3;
Review Comment:
Added in 0742501.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerTestSupport.scala:
##########
@@ -140,7 +140,8 @@ object RegionCoordinatorTestSupport {
case InitializeExecutor | OpenExecutor =>
Some(EmptyReturn())
case StartWorker =>
- Some(WorkerStateResponse(WorkerState.RUNNING))
+ // RUNNING is the worker's 2nd transition
(UNINITIALIZED→READY→RUNNING).
+ Some(WorkerStateResponse(WorkerState.RUNNING, stateVersion = 2L))
Review Comment:
Leaving as is per your call — the inline comment records where the 2 comes
from (UNINITIALIZED→READY→RUNNING).
##########
amber/src/test/python/core/runnables/test_main_loop.py:
##########
@@ -1099,13 +1102,19 @@ def send_pause(
output_queue,
):
input_queue.put(mock_pause)
- assert output_queue.get() == DCMElement(
+ elem = output_queue.get()
+ # version is the worker's logical state clock; read it from the actual
+ # report rather than pinning a brittle count (covered by StateManager
tests).
+ state_version =
elem.payload.return_invocation.return_value.worker_state_response.state_version
+ assert elem == DCMElement(
tag=mock_control_output_channel,
payload=DirectControlMessagePayloadV2(
return_invocation=ReturnInvocation(
command_id=command_sequence,
return_value=ControlReturn(
-
worker_state_response=WorkerStateResponse(WorkerState.PAUSED)
+ worker_state_response=WorkerStateResponse(
+ WorkerState.PAUSED, state_version=state_version
Review Comment:
Added in 0742501 — all three read-back sites now guard with `assert
state_version > 0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
