Xiao-zhen-Liu commented on code in PR #6011:
URL: https://github.com/apache/texera/pull/6011#discussion_r3581132743
##########
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:
The file has no trailing newline. Worth adding one while you are touching
the last line.
##########
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:
Passing `Long.MaxValue` here is only safe because of the terminal check at
the top of `updateState`. Without it, `forceTerminate` would overwrite a worker
that already reached COMPLETED. It is tested, so this is just a note to keep
the two pieces linked for future readers.
##########
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:
Every report site reads the state and the version as two separate calls
(`getCurrentState` then `getStateVersion`), for example in `StartHandler` and
`QueryStatisticsHandler`. That is safe today because the worker is
single-threaded and nothing transitions in between. If you want the rule that
the version always matches the reported state to be impossible to break later,
consider a single accessor that returns both, e.g. `getStateWithVersion: (T,
Long)`, and use it at every report site.
##########
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:
Reading `state_version` back out of the produced element and feeding it into
the expected value makes the check on that field always pass, so a handler that
reported the wrong version, or dropped it, would not fail here. The
`StateManager` tests cover the version value itself, so this is acceptable, but
a lightweight `assert state_version > 0` would at least catch a missing version.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionCoordinatorTestSupport.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:
`stateVersion = 2L` hard-codes the transition count (UNINITIALIZED -> READY
-> RUNNING). The comment helps, but if the startup path ever gains or drops a
transition this silently goes stale. Minor, fine to leave.
--
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]
