kz930 commented on code in PR #7207: URL: https://github.com/apache/texera/pull/7207#discussion_r3770685252
########## common/workflow-operator/src/test/resources/python/py_compile_worker.py: ########## @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Persistent worker that syntax-checks generated operator code, replacing one +`python -I -S -B -m py_compile <file>` spawn per operator descriptor. + +`compile(source, path, "exec")` is what `py_compile` does before writing a +`.pyc` and raises the same SyntaxError; skipping that write is why neither `-B` +nor a temp file is needed here. + +Protocol (line-delimited JSON, both directions): + + startup worker -> parent: {"ready": true} + request parent -> worker: {"source": "<code>", "name": "<label>"}\n + response worker -> parent: {"exit": 0, "stdout": "...", "stderr": "..."}\n + +`exit` is 1 when the source does not compile, with the SyntaxError in `stderr`, +mirroring the spawn it replaces so the parent's reporting is unchanged. That +does not end the worker; only a hard interpreter crash does. +""" +from __future__ import annotations + +import json +import sys +import traceback + + +def _compile_one(source: str, name: str) -> "dict[str, object]": + """Compile one generated module. `name` is the filename the traceback shows, + so a report names the descriptor rather than a temp path. + """ + try: + compile(source, name, "exec") Review Comment: Fixed in 8dd2ce3f2. The flag is right, but I could not reproduce the direction: on 3.12 and 3.14 the worker rejects `def f(x: (yield 1)): pass` too, only the message differs. The divergence I did find runs the other way. `def f(x: (y := 1)): pass` compiles under the spawn on 3.12 and raised here, so the old check was stricter, not more permissive. That is the case the docstring now names. ########## common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala: ########## @@ -0,0 +1,268 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.util.python + +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.funsuite.AnyFunSuite + +import java.util.concurrent.{Executors, TimeUnit} +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future} +import scala.util.Try + +/** + * What the pool owes a caller when a worker misbehaves. An ordinary job failure + * is the other suites' business; this one is about a worker that stays alive and + * stops taking part, which is the case that does not end by itself: a crash + * closes the pipe and the pending read returns, while silence would hold the + * caller forever — neither a read nor a write on a process pipe answers an + * interrupt or a deadline, so no suite-level timeout can release one. + * + * Every wait here is bounded and runs on daemon threads, so a regression fails + * these tests instead of wedging the run: a lost non-daemon thread parked on a + * pipe would keep the JVM, and the build, alive. + * + * The fixture worker is stdlib-only and runs under `-I -S`, so this needs an + * interpreter but none of the operator packages. + */ +final class PythonWorkerPoolSpec extends AnyFunSuite { + + private val HangingWorker = "/python/hanging_worker.py" + + /** Short enough to keep the suite quick, far enough above process startup to + * not be mistaken for one: a fixture asked to hang never answers at all, so + * these deadlines cannot race it. + */ + private val ShortTimeouts: PythonWorkerPool.Timeouts = + PythonWorkerPool.Timeouts(responseMillis = 1500, startupMillis = 1500) + + /** For the case that interrupts its own callers: far enough out that they are + * certainly still waiting on the worker, not already past their deadline. + */ + private val PatientTimeouts: PythonWorkerPool.Timeouts = + PythonWorkerPool.Timeouts(responseMillis = 60000, startupMillis = 60000) + + /** Ceiling on a whole case, well above the deadlines under test. Reaching it + * means something never gave up. + */ + private val Bound: FiniteDuration = 25.seconds + + /** Any interpreter serves — the fixture imports only `json` and `time` — so + * this deliberately skips the configured `python.path` the suites that need + * pandas resolve. A machine without one cancels rather than fails. + */ + private lazy val python: String = { + def isRunnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (p.waitFor(5, TimeUnit.SECONDS)) p.exitValue() == 0 else { p.destroyForcibly(); false } + } + + List("python3", "python", "py").find(isRunnable).getOrElse(cancel("no runnable python")) + } + + /** Resolves [[python]] here so that the cancel lands on the test thread. Forced + * inside a case, it would be forced inside a `Future`, and by the time it + * reached `intercept` a cancellation is a RuntimeException like any other: the + * cases would fail on a machine without an interpreter instead of cancelling, + * and the one at the cap would record it as the failure it asserts on and pass + * without ever reaching the cap. + */ + override def withFixture(test: NoArgTest): org.scalatest.Outcome = { + val _ = python + super.withFixture(test) + } + + private def onDaemonThreads[T](threads: Int)(body: ExecutionContext => T): T = { + val pool = Executors.newFixedThreadPool( + threads, + (r: Runnable) => { + val t = new Thread(r, "pool-spec-caller") + t.setDaemon(true) + t + } + ) + val ec: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(pool) + try body(ec) + finally pool.shutdownNow() + } + + private def call( + launchArgs: Seq[String], + request: ObjectNode, + timeouts: PythonWorkerPool.Timeouts + ): PythonWorkerPool.Outcome = + PythonWorkerPool.run( + resourcePath = HangingWorker, + launchArgs = launchArgs, + pythonExe = python, + request = request, + interpreterArgs = Seq("-I", "-S"), + timeouts = timeouts + ) + + /** A job the fixture takes and never answers. `hang` travels in the request + * rather than in `launchArgs` so that [[healthyCall]] lands in the same + * sub-pool: `Key` covers the script, its arguments and the environment, and a + * job routed elsewhere would be served by a pool whose queue this suite never + * touched. + */ + private def hangingCall( + launchArgs: Seq[String], + request: ObjectNode = objectMapper.createObjectNode(), + timeouts: PythonWorkerPool.Timeouts = ShortTimeouts + ): PythonWorkerPool.Outcome = + call(launchArgs, request.deepCopy().put("hang", true), timeouts) + + /** A job in that same sub-pool which the fixture does answer. */ + private def healthyCall(): PythonWorkerPool.Outcome = + call(Seq.empty, objectMapper.createObjectNode(), ShortTimeouts) + + /** The call, on a daemon thread and under [[Bound]], expected to give up. */ + private def interceptBounded(call: => Any): PythonWorkerPool.WorkerDiedException = + intercept[PythonWorkerPool.WorkerDiedException] { + onDaemonThreads(1)(ec => Await.result(Future(call)(ec), Bound)) + } + + test("a worker that takes the job and stops answering is killed and reported") { + val startedAt = System.nanoTime() + val thrown = interceptBounded(hangingCall(Seq.empty)) + val elapsedMillis = (System.nanoTime() - startedAt) / 1000000 + + assert(thrown.getMessage.contains("did not answer")) + assert(thrown.getMessage.contains("killed it")) + // Well under the default response budget: what fired is the timeout passed in, + // not a wait that happened to end. + assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.responseMillis / 2) + } + + test("a worker that never signals ready is killed and reported") { + val startedAt = System.nanoTime() + val thrown = interceptBounded(hangingCall(Seq("--hang-before-ready"))) + val elapsedMillis = (System.nanoTime() - startedAt) / 1000000 + + assert(thrown.getMessage.contains("did not signal ready")) + assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.startupMillis / 2) + } + + test("a worker that never reads its request is killed and reported") { + val request = objectMapper.createObjectNode() + // Past any pipe buffer, so the write cannot simply be handed to the kernel and + // left there: it is the blocked write itself that has to be given up on. + request.put("source", "x" * (4 * 1024 * 1024)) + + val thrown = interceptBounded(hangingCall(Seq("--deaf"), request)) + + assert(thrown.getMessage.contains("did not read its request")) + assert(thrown.getMessage.contains("killed it")) + } + + test("a caller waiting at the worker cap is not stranded by a discarded worker") { + // One caller more than there are workers, all onto workers that go quiet: + // the callers holding one time out and their workers are discarded, which + // frees a slot without handing anything back, and the caller waiting at the + // cap has to notice that rather than wait for a hand-back that never comes. + // The discards it is waiting on happen on workers other than the one ahead + // of it. + val callers = PythonWorkerPool.maxWorkers + 1 + + val outcomes = onDaemonThreads(callers) { implicit ec => + Await.result(Future.sequence(Seq.fill(callers)(Future(Try(hangingCall(Seq.empty))))), Bound) + } + + assert(outcomes.length == callers) + assert(outcomes.forall(_.isFailure)) + } + + test("the pool still serves jobs after it has discarded a timed-out worker") { + interceptBounded(hangingCall(Seq.empty)) + + // Deliberately the sub-pool the discard happened in — see [[hangingCall]] — + // so what is asserted is that a pool short one worker starts a replacement, + // not that an untouched pool works. + assert(healthyCall().exit == 0) + } + + test("an interpreter that cannot be started reaches the caller as a worker death") { + // Not the IOException ProcessBuilder raises: a caller's fallback is written + // against WorkerDiedException, and a pool that cannot hand out a worker at all + // is the case that fallback exists for. + val thrown = intercept[PythonWorkerPool.WorkerDiedException] { + PythonWorkerPool.run( + resourcePath = HangingWorker, + launchArgs = Seq.empty, + pythonExe = "no-such-python-on-this-machine", + request = objectMapper.createObjectNode(), + interpreterArgs = Seq("-I", "-S"), + timeouts = ShortTimeouts + ) + } + + assert(thrown.getMessage.contains("could not start python worker")) + } + + test("a caller interrupted mid-job does not cost the pool a worker") { + val workers = PythonWorkerPool.maxWorkers + + // Every slot taken by a job nobody will answer, and then the callers are + // interrupted — `shutdownNow` is what an executor does to a fan-out whose test + // has already failed. An interrupt is not a WorkerDiedException, so a worker + // left neither returned nor discarded would cost this sub-pool that slot for + // the rest of the JVM. + onDaemonThreads(workers) { implicit ec => + Seq.fill(workers)(Future(Try(hangingCall(Seq.empty, timeouts = PatientTimeouts)))) + // Enough for the callers to be waiting on a worker rather than starting one; + // the slot has to come back wherever the interrupt lands, so this only + // decides which of the two paths the case exercises. + Thread.sleep(500) + } + + // Serving `workers` jobs again is the whole assertion: a leaked slot leaves + // these at the cap, rechecking it until [[Bound]] runs out. + val outcomes = onDaemonThreads(workers) { implicit ec => + Await.result(Future.sequence(Seq.fill(workers)(Future(healthyCall()))), Bound) + } + + assert(outcomes.forall(_.exit == 0)) + } + + test("a worker that answers without an exit code is reported as the worker") { + // Not as a job that failed: a default would name the caller's own request as + // what went wrong, with an empty stdout as the evidence. + val request = objectMapper.createObjectNode().put("drop-exit", true) + + val thrown = intercept[PythonWorkerPool.WorkerDiedException] { + call(Seq.empty, request, ShortTimeouts) + } Review Comment: Fixed in 3e69a7867, taken as suggested. -- 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]
