Yicong-Huang commented on code in PR #7207: URL: https://github.com/apache/texera/pull/7207#discussion_r3753548921
########## common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala: ########## @@ -0,0 +1,490 @@ +/* + * 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 com.typesafe.scalalogging.LazyLogging +import org.apache.texera.amber.util.JSONUtils.objectMapper + +import java.io.{BufferedReader, BufferedWriter, InputStreamReader, OutputStreamWriter} +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path, StandardCopyOption} +import java.util.concurrent.{ + Callable, + ConcurrentHashMap, + ExecutionException, + ExecutorService, + Executors, + LinkedBlockingQueue, + TimeUnit, + TimeoutException +} +import java.util.concurrent.atomic.AtomicInteger +import scala.annotation.tailrec +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +/** + * Pools of persistent Python "worker" processes that eliminate the per-call + * interpreter-boot + import cost a test otherwise pays on every subprocess + * spawn. Testing operators one at a time does not scale when each one costs a + * spawn: a bare `-I -S` interpreter boots in ~25 ms, and once pandas and plotly + * are imported a spawn costs ~260-310 ms — ~96% of a job whose real work is Review Comment: Is the ~96% measuring something other than these two figures? A 260-310 ms spawn around ~4 ms of work makes the spawn ~99% of the job. I assume the 96% is #6975's decomposition and a share of something slightly different; naming what it is a share of would keep the sentence self-checking. As written, a reader who divides the two numbers here gets a different answer. ########## common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala: ########## @@ -0,0 +1,490 @@ +/* + * 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 com.typesafe.scalalogging.LazyLogging +import org.apache.texera.amber.util.JSONUtils.objectMapper + +import java.io.{BufferedReader, BufferedWriter, InputStreamReader, OutputStreamWriter} +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path, StandardCopyOption} +import java.util.concurrent.{ + Callable, + ConcurrentHashMap, + ExecutionException, + ExecutorService, + Executors, + LinkedBlockingQueue, + TimeUnit, + TimeoutException +} +import java.util.concurrent.atomic.AtomicInteger +import scala.annotation.tailrec +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +/** + * Pools of persistent Python "worker" processes that eliminate the per-call + * interpreter-boot + import cost a test otherwise pays on every subprocess + * spawn. Testing operators one at a time does not scale when each one costs a + * spawn: a bare `-I -S` interpreter boots in ~25 ms, and once pandas and plotly + * are imported a spawn costs ~260-310 ms — ~96% of a job whose real work is + * ~4 ms. A worker pays that once at startup, then serves many jobs over its + * lifetime, so N spawns become one. + * + * Lives in test scope here, rather than beside a single caller, because tests in + * several modules run generated operator code and would otherwise each + * hand-roll a driver, a stdout protocol and a timeout. Other modules reach it + * through a `test->test` dependency on this one. + * + * Generic over the worker script — the pool never interprets the payload — so + * one implementation serves a syntax check, template execution and DataFrame + * comparison alike. Each distinct (resource, interpreterArgs, launchArgs, + * python, env) combination gets its own sub-pool. + * + * Protocol (line-delimited JSON, shared by all worker scripts): + * startup worker -> pool: {"ready": true} + * request pool -> worker: <caller-supplied JSON object>\n + * response worker -> pool: {"exit": <int>, "stdout": "...", "stderr": "..."}\n + * + * Concurrency: callers submit from several threads at once — one test fanning + * its cases out, or suites running in parallel — so each sub-pool holds up to + * [[maxWorkers]] workers, each serving one job at a time (borrow -> run -> + * return). A worker script may chdir per job, so a worker must never run two + * jobs at once — the borrow/return discipline guarantees that. + * + * Robustness: an ordinary *job* failure comes back as an [[Outcome]] with + * `exit != 0` (worker keeps running). A hard interpreter crash ends a worker; + * the pool detects the EOF / broken pipe, discards it, and throws + * [[WorkerDiedException]] so the caller can fall back to a one-shot subprocess + * — behavior is then never worse than the pre-pool path. A worker that stays + * alive but stops answering ends the same way, on the [[Timeouts]] below. + */ +object PythonWorkerPool extends LazyLogging { + + /** Worker response: process-like exit code plus captured streams. */ + final case class Outcome(exit: Int, stdout: String, stderr: String) + + /** Thrown when a worker dies mid-job (hard crash / broken pipe). Callers + * catch this and fall back to a one-shot subprocess. + */ Review Comment: This doc did not move with the contract. `run`'s Scaladoc at `:144` now promises this exception for a worker that would not start or never signalled ready too, and `create` (`:266`) and `awaitReady` (`:393`) deliver on it. A caller reading only the exception's own doc would still write the narrower fallback the widening was meant to retire. ```suggestion /** Thrown for a worker the pool could not give out or keep: one that would not * start, one that never signalled ready, or one that died or fell silent * mid-job. Callers catch this and fall back to a one-shot subprocess. */ ``` ########## common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala: ########## @@ -49,8 +57,99 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite { private val MaxDepth: Int = 3 private val AcceptPackages: Seq[String] = Seq("org.apache.texera.amber.operator") + /** Budget for one whole fanned-out pass over every descriptor. Deliberately far + * above a real run (well under a second), so it only ever fires on a hang. + */ + private val PassTimeout: FiniteDuration = 10.minutes + + /** Runs the given work concurrently and returns the results in submission + * order, rethrowing the first failure so it fails the test. Sized to the pool + * so the threads match the workers available to serve them. + * + * Daemon threads: a task parked on a subprocess pipe answers no interrupt, so + * `shutdownNow` need not end it, and a non-daemon one left there would hold the + * JVM — and the build — open after [[PassTimeout]] has already failed the test. + */ + private def awaitAll[T](work: Seq[() => T]): Seq[T] = { + val threads = Executors.newFixedThreadPool( + PythonWorkerPool.maxWorkers, + (r: Runnable) => { + val t = new Thread(r, "py-compile-check") + t.setDaemon(true) + t + } + ) + try { + implicit val ec: ExecutionContext = ExecutionContext.fromExecutorService(threads) + Await.result(Future.sequence(work.map(w => Future(w()))), PassTimeout) + } finally threads.shutdownNow() + } + + /** Checks the pooled path could not serve. Not a failure — the spawn each one + * fell back to is the pre-pool behavior — but a number the summary has to + * carry, since a check passing says nothing about which path answered it. + */ + private val spawnFallbacks = new AtomicInteger(0) + + /** Syntax-checks one generated module, through a pooled worker when available. + * + * The worker is launched with the same `-I -S` isolation the one-shot path + * uses, so what the check accepts is unchanged; it just stops paying an + * interpreter boot — the whole cost of a check whose real work is under a + * millisecond — once per descriptor. A worker the pool cannot give out, or + * loses mid-job, falls back to the spawn, so behavior is never worse than + * before the pool. + */ + private def syntaxCheck( + pythonExecutable: String, + pythonSource: String, + descriptorName: String + ): Either[String, Unit] = { + def viaPool: Either[String, Unit] = { + val request = objectMapper.createObjectNode() + request.put("source", pythonSource) + request.put("name", s"$descriptorName.py") + val outcome = PythonWorkerPool.run( + resourcePath = "/python/py_compile_worker.py", + launchArgs = Seq.empty, + pythonExe = pythonExecutable, + request = request, + interpreterArgs = Seq("-I", "-S") + ) + if (outcome.exit == 0) Right(()) + else { + val output = if (outcome.stderr.trim.nonEmpty) outcome.stderr.trim else "(no output)" + Left( + s"py_compile failed (exit=${outcome.exit})\nOutput:\n" + + truncateBlock(output, maxLines = 40, maxChars = 8000) + ) + } + } + + if (PythonWorkerPool.enabled) { + try viaPool + catch { + // Anything the pooled path throws leaves the spawn as the answer, which is + // what makes it never worse than before: not only a worker that died + // mid-job, but equally an interpreter that could not be started at all — + // `ProcessBuilder.start` throws a bare IOException, which the pool passes + // through, and 4-way concurrency is a live way to reach it. Counted, so a + // run the pool served none of does not read as a green pooled run. Review Comment: The rationale no longer matches the code. As of this round the pool does not pass that `IOException` through: `PythonWorkerPool.scala:266-270` converts it, and the new case at `PythonWorkerPoolSpec.scala:192` asserts so. The wider catch is still right, for a different reason. `ensureScript()` (`PythonWorkerPool.scala:255`) runs before `create`'s wrap, so materializing the worker script is the one step that can still escape as something else. ```suggestion // mid-job, but equally one the pool could not hand out at all. Those // arrive as WorkerDiedException; NonFatal also covers the steps outside // that contract, such as materializing the worker script. Counted, so a // run the pool served none of does not read as a green pooled run. ``` ########## common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala: ########## @@ -0,0 +1,242 @@ +/* + * 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 def python(): String = { Review Comment: Being a `def`, this re-probes the interpreter on every call. `call` (`:100`) invokes it per pool call, so the suite spawns `python3 --version` about 19 times for an answer that cannot change mid-run — and in the cap case all five callers pay it concurrently before reaching the pool. A `lazy val` fixes it, at the cost of dropping the parens at the four call sites. Minor, but this is the cost the PR exists to remove. -- 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]
