Copilot commented on code in PR #7870:
URL: https://github.com/apache/texera/pull/7870#discussion_r3837909389
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DPThreadSpec.scala:
##########
@@ -354,4 +406,343 @@ class DPThreadSpec extends AnyFlatSpec with MockFactory {
}
}
+ "DP Thread" should "treat stop() before start() as a no-op" in {
+ // stop() is also reached on the teardown path of a worker that never
started
+ // (e.g. an actor that fails during initialization). Neither the executor
+ // service nor the thread future exists yet, so both guards must
short-circuit
+ // rather than dereference a null.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = new DataProcessor(workerId, x => {}, inputMessageQueue =
inputQueue)
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+
+ // The entire contract on this path is that the call does not blow up.
Asserting
+ // that the two fields are null would assert nothing: stop() only ever
reads them,
+ // so `dpThread == null` is the field initializer talking, not production
code, and
+ // it would hold even if stop()'s body were deleted. Flipping either guard
to
+ // `== null` makes this call NPE, and that is what this test detects.
+ noException should be thrownBy dpThread.stop()
+ }
+
+ "DP Thread" should "not return from stop() until the DP thread has left the
main loop" in {
+ // stop() must be a synchronous join: WorkflowWorker tears down the output
gateway
+ // and the statistics manager right after it returns, so the DP thread
must already
+ // be out of run() by then. The join is `endFuture.get()`, and nothing
else in the
+ // suite constrains it. We park the DP thread in a NON-interruptible spin
(an
+ // interruptible wait would be swallowed by
DataProcessor.handleExecutorException
+ // and let stop() legitimately return early) and check that stop() blocks.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = newDataProcessor(inputQueue)
+ val release = new AtomicBoolean(false)
+ val inSpin = new CountDownLatch(1)
+ dp.executor = new OperatorExecutor {
+ override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike]
= {
+ inSpin.countDown()
+ while (!release.get()) {}
+ Iterator.empty
Review Comment:
Busy-wait spin in the executor (`while (!release.get()) {}`) will peg a CPU
core during the test run and can slow/flake CI under load. You can keep the
loop non-interruptible while being more scheduler-friendly by using
`Thread.onSpinWait()` inside the loop body.
This issue also appears on line 689 of the same file.
##########
amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DPThreadSpec.scala:
##########
@@ -354,4 +406,343 @@ class DPThreadSpec extends AnyFlatSpec with MockFactory {
}
}
+ "DP Thread" should "treat stop() before start() as a no-op" in {
+ // stop() is also reached on the teardown path of a worker that never
started
+ // (e.g. an actor that fails during initialization). Neither the executor
+ // service nor the thread future exists yet, so both guards must
short-circuit
+ // rather than dereference a null.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = new DataProcessor(workerId, x => {}, inputMessageQueue =
inputQueue)
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+
+ // The entire contract on this path is that the call does not blow up.
Asserting
+ // that the two fields are null would assert nothing: stop() only ever
reads them,
+ // so `dpThread == null` is the field initializer talking, not production
code, and
+ // it would hold even if stop()'s body were deleted. Flipping either guard
to
+ // `== null` makes this call NPE, and that is what this test detects.
+ noException should be thrownBy dpThread.stop()
+ }
+
+ "DP Thread" should "not return from stop() until the DP thread has left the
main loop" in {
+ // stop() must be a synchronous join: WorkflowWorker tears down the output
gateway
+ // and the statistics manager right after it returns, so the DP thread
must already
+ // be out of run() by then. The join is `endFuture.get()`, and nothing
else in the
+ // suite constrains it. We park the DP thread in a NON-interruptible spin
(an
+ // interruptible wait would be swallowed by
DataProcessor.handleExecutorException
+ // and let stop() legitimately return early) and check that stop() blocks.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = newDataProcessor(inputQueue)
+ val release = new AtomicBoolean(false)
+ val inSpin = new CountDownLatch(1)
+ dp.executor = new OperatorExecutor {
+ override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike]
= {
+ inSpin.countDown()
+ while (!release.get()) {}
+ Iterator.empty
+ }
+ }
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+ val stopReturned = new CountDownLatch(1)
+ try {
+ dpThread.start()
+ inputQueue.put(dataFrameOf(4))
+ assert(inSpin.await(5, TimeUnit.SECONDS), "the DP thread must be inside
the executor")
+ val stopper = new Thread(() => {
+ try {
+ dpThread.stop()
+ } finally {
+ stopReturned.countDown()
+ }
+ })
+ stopper.setDaemon(true)
+ stopper.start()
+ assert(
+ !stopReturned.await(500, TimeUnit.MILLISECONDS),
+ "stop() must not return while the DP thread is still inside run()"
+ )
+ release.set(true)
+ assert(
+ stopReturned.await(10, TimeUnit.SECONDS),
+ "stop() must return once the DP thread exits"
+ )
+ } finally {
+ release.set(true)
+ dpThread.stop()
+ }
+ }
+
+ "DP Thread" should "park when idle and exit quietly when stop() interrupts
it there" in {
+ // Two properties of the same moment. (1) An idle DP thread must PARK on
+ // internalQueue.take rather than spin: input selection sets
`waitingForInput = true`
+ // when no channel can be picked, and that flag is the only thing that
sends the loop
+ // back into the blocking take. Thread.State is the only way to tell
parking from
+ // spinning from outside. (2) stop() then interrupts the DP thread out of
that take,
+ // and run() must treat the InterruptedException as an ordinary teardown
-- log it and
+ // nothing more. The other catch arm delegates the error back to the
worker actor, and
+ // with the usual swallowing output handler the two are indistinguishable,
so record
+ // what the handler receives instead of dropping it.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val delegates = new ConcurrentLinkedQueue[MainThreadDelegateMessage]()
+ val dp = newDataProcessor(
+ inputQueue,
+ {
+ case Left(m) => delegates.add(m)
+ case _ => ()
+ }
+ )
+ // processTuple runs ON the DP thread, so this is how we get hold of that
exact
+ // thread. Every DPThread names its thread "DP-thread" and earlier tests
in this
+ // suite leave theirs alive, so matching by name would be ambiguous.
+ val dpWorkerThread = new AtomicReference[Thread]()
+ dp.executor = new OperatorExecutor {
+ override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike]
= {
+ dpWorkerThread.set(Thread.currentThread())
+ Iterator.empty
+ }
+ }
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+ val idleState = () => Option(dpWorkerThread.get()).map(_.getState)
+ try {
+ dpThread.start()
+ inputQueue.put(dataFrameOf(1))
+ assert(
+ awaitCond(idleState().contains(Thread.State.WAITING)),
+ s"an idle DP thread must park on take, was ${idleState()}"
+ )
+ // stop() joins the DP thread, so run() has returned by the time this
returns.
+ dpThread.stop()
+ assert(
+ delegates.isEmpty,
+ s"an interrupted teardown must not delegate an error, got
${delegates.peek()}"
+ )
+ } finally {
+ dpThread.stop()
+ }
+ }
+
+ "DP Thread" should "hold back data intake while a queued Backpressure
command is in effect" in {
+ // Backpressure delivered through the internal queue as an
ActorCommandElement,
+ // which is how the worker actor actually sends it. While it is on, the DP
+ // thread must stop taking data, so a data frame that arrives afterwards
stays
+ // queued in its channel and no tuple is processed until backpressure is
lifted.
+ val inputQueue = new LinkedBlockingQueue[DPInputQueueElement]()
+ val dp = newDataProcessor(inputQueue)
+ val processed = new AtomicInteger(0)
+ dp.executor = new OperatorExecutor {
+ override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike]
= {
+ processed.incrementAndGet()
+ Iterator.empty
+ }
+ }
+ val dpThread = new DPThread(workerId, dp, logManager, inputQueue)
+ try {
+ dpThread.start()
+ inputQueue.put(ActorCommandElement(Backpressure(enableBackpressure =
true)))
+ // `backpressureStatus` is a plain (non-volatile) var written by the DP
thread, so
+ // reading it from here needs a happens-before edge. CreditUpdate is
inert -- it
+ // falls through handleActorCommand's `case _ => // no op` arm -- and
observing the
+ // queue drained is a read of LinkedBlockingQueue's AtomicInteger count,
which the
+ // DP thread decremented after it had already handled the Backpressure
element.
+ inputQueue.put(ActorCommandElement(CreditUpdate()))
+ assert(awaitCond(inputQueue.isEmpty), "the queued commands must be
consumed")
+ assert(dpThread.backpressureStatus, "backpressure must be picked up")
+
+ inputQueue.put(dataFrameOf(200))
+ Thread.sleep(500)
+ assert(processed.get() == 0, "no data may be taken while backpressured")
Review Comment:
`Thread.sleep(500)` followed by `processed.get() == 0` is a weak negative
assertion: it can miss a bug that processes tuples while backpressured but only
after the fixed 500ms window. Since you already have `awaitCond`, you can
assert that the counter stays at 0 for the whole budget (i.e., it never becomes
non-zero) up until backpressure is lifted.
--
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]
