Yicong-Huang commented on code in PR #7207:
URL: https://github.com/apache/texera/pull/7207#discussion_r3754197809
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala:
##########
@@ -249,18 +350,50 @@ final class PythonCodeRawInvalidTextSpec extends
AnyFunSuite {
val findings = checkResult.findings ++ pyCompileFindings
if (findings.isEmpty && checkResult.code.nonEmpty) {
- ok += 1
- println(s"[py-compile OK $ok/$total | checked $checked/$total]
${descriptorClass.getName}")
+ println(
+ s"[py-compile OK ${ok.incrementAndGet()}/$total | " +
+ s"checked ${checked.get()}/$total] ${descriptorClass.getName}"
+ )
}
findings
- }
+ }).flatten
- println(s"[py-compile SUMMARY] ok=$ok/$total")
+ println(
+ s"[py-compile SUMMARY] ok=${ok.get()}/$total, spawn
fallbacks=${spawnFallbacks.get()}"
+ )
if (allFindings.nonEmpty) {
fail(PythonReflectionUtils.renderReport(allFindings, total = total))
}
}
+ /** py_compile above only parses the emitted code; running it needs the
packages
+ * it imports. Tagged, so only amber-integration — the job that installs
them —
+ * runs this. There a missing package is a defect; elsewhere it is a
local-setup
+ * fact, so cancel rather than fail.
+ */
+ test(
+ "the Python interpreter operator templates run in should import pandas and
plotly",
+ Tag(classOf[IntegrationTest].getName)
+ ) {
+ val provisioned =
sys.env.get("AMBER_TEST_FILTER").contains("integration-only")
Review Comment:
Keying fail-vs-cancel on `AMBER_TEST_FILTER` ties this test's severity to
*how* the integration subset gets selected. It is right today. But say that
selector is later replaced or renamed — which extracting
`TestFilters.integrationSplit` makes easy, and which its own `@param tag` doc
warns about. The test still runs in `amber-integration`, silently back to
cancelling: the non-result this PR removes, with nothing going red.
`TestFilters` is build-scope, so the string cannot be imported here. Worth a
line at each site pointing at the other?
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala:
##########
@@ -225,21 +324,23 @@ final class PythonCodeRawInvalidTextSpec extends
AnyFunSuite {
}
val total = descriptorCandidates.size
- var ok = 0
- var checked = 0
-
- val allFindings = descriptorCandidates.flatMap { descriptorClass =>
- checked += 1
+ val ok = new AtomicInteger(0)
+ val checked = new AtomicInteger(0)
+ // Checked concurrently: the fan-out is what turns the pool's workers into
+ // parallel interpreters rather than a queue in front of one. The executor
is
+ // sized to maxWorkers, so nothing is submitted past the cap.
Review Comment:
My wording from round 3 was wrong here: `awaitAll` builds all 117 futures up
front, so every task *is* submitted — the fixed-size executor caps how many run
at once.
```suggestion
// sized to maxWorkers, so nothing runs past the cap.
```
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala:
##########
@@ -0,0 +1,491 @@
+/*
+ * 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 — the imports ~96% of that —
dwarfing
+ * the ~4 ms of real work a job does. 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 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.
+ */
+ final class WorkerDiedException(message: String, cause: Throwable = null)
+ extends RuntimeException(message, cause)
+
+ /** Feature toggle. `TEXERA_TEST_PYTHON_WORKER=0` (or `false`/`off`) forces
the
+ * one-subprocess-per-call paths everywhere — an escape hatch for debugging
a
+ * suspected isolation leak. Default on.
+ */
+ val enabled: Boolean =
+ !sys.env
+ .get("TEXERA_TEST_PYTHON_WORKER")
+ .map(_.trim.toLowerCase)
+ .exists(Set("0", "false", "off"))
+
+ /** Max live workers per sub-pool — per sub-pool, so callers on distinct
worker
Review Comment:
`per sub-pool` twice reads as a stutter; the following line already says
what the bound is not shared with.
```suggestion
/** Max live workers per sub-pool, so callers on distinct worker
```
##########
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
Review Comment:
`Checks` is the noun here, but every other Scaladoc in this file opens with
a verb, so it first reads as "Checks that ..." — the next clause is what
disambiguates it.
```suggestion
/** Count of checks the pooled path could not serve. Not a failure — the
spawn each one
```
##########
.github/workflows/build.yml:
##########
@@ -631,6 +631,8 @@ jobs:
# specs. The Java @TagAnnotation makes the marker visible to
# ScalaTest's reflection, so `-n TAG` correctly narrows the
# run.
+ # WorkflowOperator/test is here for the same reason, with its own tag:
Review Comment:
This note landed here, but the `amber` job's counterpart at `:287-289` still
reads as though the env value only tells `amber/build.sbt` to exclude amber's
tag. It now also drives `common/workflow-operator/build.sbt:42-44` and a
differently-named tag — and that exclusion is what keeps the pandas probe out
of a job installing neither.
--
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]
