Yicong-Huang commented on code in PR #7207: URL: https://github.com/apache/texera/pull/7207#discussion_r3754680059
########## 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 — mostly the imports — 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, so callers on distinct worker + * scripts add up rather than share this bound. Defaults to 4, override via + * `TEXERA_TEST_PYTHON_WORKERS`. Public so a caller fanning out jobs within one + * test can size that fan-out to the workers it will get. + */ + val maxWorkers: Int = + sys.env + .get("TEXERA_TEST_PYTHON_WORKERS") + .flatMap(s => scala.util.Try(s.trim.toInt).toOption) + .filter(_ > 0) + .getOrElse(4) + + /** How long a caller waits on a worker before the pool kills and discards it. + * A read on a process pipe cannot be interrupted — a suite or executor timeout + * leaves the reading thread stuck on it — so a worker that stays alive without + * answering has to be bounded here. `response` keeps the 30 seconds the + * one-shot spawn this pool replaced allowed a job; `startup` is longer because + * a worker imports its libraries before it reports ready, and a loaded CI + * machine makes that slow. Override in seconds via + * `TEXERA_TEST_PYTHON_WORKER_TIMEOUT` / `TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT`. + */ + final case class Timeouts(responseMillis: Long, startupMillis: Long) + + object Timeouts { + private def envSeconds(name: String, default: Long): Long = + sys.env + .get(name) + .flatMap(s => scala.util.Try(s.trim.toLong).toOption) + .filter(_ > 0) + .getOrElse(default) * 1000 + + val Default: Timeouts = Timeouts( + responseMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_TIMEOUT", 30), + startupMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT", 60) + ) + } + + /** + * Run one job through a pooled worker for `resourcePath`, launched as + * `pythonExe <interpreterArgs> <script> <launchArgs>` with extra environment + * `env`. `request` is the worker-specific JSON payload (the pool does not + * interpret it). Throws [[WorkerDiedException]] for every worker the pool could + * not give out or keep — one that would not start, would not report ready, or + * died mid-job — so one `catch` covers a caller's whole fallback. + * + * `interpreterArgs` are the flags that must precede the script — a syntax + * checker wants `-I -S` so it validates under the same isolation a one-shot + * `python -I -S -m py_compile` gave it. `launchArgs` are the script's own + * (e.g. `--serve`), and `env` carries what a flag cannot (e.g. PYTHONPATH). Review Comment: `-I` implies `-E`, so a worker launched with the `-I -S` this paragraph recommends three lines up ignores PYTHONPATH and every other `PYTHON*` variable. Checked it: `PYTHONPATH=<dir> python3 -I -S -c "import <mod>"` raises `ModuleNotFoundError` and `sys.path` shows no such entry; drop `-I` and it imports. The parameter itself is fine — non-`PYTHON*` variables do survive `-I`. It is the example that misleads, and #7149's template runner is the caller that will reach for PYTHONPATH. ```suggestion * (e.g. `--serve`), and `env` carries what a flag cannot — though not * PYTHONPATH or any other PYTHON* var, which the `-I` above makes CPython ignore. ``` ########## project/TestFilters.scala: ########## @@ -0,0 +1,46 @@ +/* + * 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. + */ + +import sbt._ + +/** + * Selects a module's tagged tests for the fast-unit job or the integration job: + * skip-integration excludes them, integration-only runs only them, unset runs + * everything. Shared because the mapping is identical in every module, while the + * env var and the tag are not — the tag annotation has to live somewhere the + * module's own Test config can see. + */ +object TestFilters { + + /** @param envVar the variable the two CI jobs set to opposite values; it has to + * be the one the workflow already sets on the step that invokes + * this module's tests, or neither subset is selected. + * @param tag fully-qualified name of the tag annotation, as + * `classOf[...].getName` gives it at the test site — ScalaTest + * matches these by string, so a rename that misses one side + * silently stops filtering. + */ + def integrationSplit(envVar: String, tag: String): Seq[TestOption] = + sys.env.get(envVar) match { + case Some("skip-integration") => + Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", tag)) + case Some("integration-only") => + Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", tag)) + case _ => Nil Review Comment: A value neither branch recognizes falls through to "run everything". So an `AMBER_TEST_FILTER` typo in either job runs the full suite in `amber-integration`, and the pandas probe then reads `provisioned = false` and cancels. That is the non-result this PR exists to remove, arriving with nothing red. The inline code did the same, but it was one module's. This is now the single helper both jobs route through, which makes it the place to fail loudly instead. ```suggestion case Some(other) => sys.error(s"unrecognized $envVar value: $other") case None => Nil ``` ########## common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java: ########## @@ -0,0 +1,51 @@ +/* + * 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.operator.tags; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.scalatest.TagAnnotation; + +/** + * Marks a test in this module as needing more than a bare Python interpreter — + * pandas or plotly, which the {@code amber} job does not install. See the + * AMBER_TEST_FILTER block in {@code common/workflow-operator/build.sbt} for how + * it routes to {@code amber-integration}. + * + * <p>Apply it to a whole spec as an annotation, or to a single case — {@code + * test(name, Tag(classOf[IntegrationTest].getName))} in a FunSuite, + * {@code taggedAs} in a FlatSpec — so that a spec's cheaper assertions stay in + * the unit job and its coverage report. Review Comment: The trailing clause reads as the purpose of both options, but annotating a whole spec takes its cheap assertions out of the unit job too. That gain belongs only to the per-case form. ```suggestion * <p>Apply it to a whole spec as an annotation, or — so that a spec's cheaper * assertions stay in the unit job and its coverage report — to a single case: * {@code test(name, Tag(classOf[IntegrationTest].getName))} in a FunSuite, * {@code taggedAs} in a FlatSpec. ``` ########## .github/workflows/build.yml: ########## @@ -284,9 +284,13 @@ jobs: # Test config (sbt's `test` task does not transit dependsOn), # so common modules' tests are listed explicitly here. # - # AMBER_TEST_FILTER=skip-integration tells amber/build.sbt to - # exclude @org.apache.texera.amber.tags.IntegrationTest specs; - # those run in the amber-integration job below. + # AMBER_TEST_FILTER=skip-integration excludes the integration-tagged + # specs of two modules, each under its own tag: amber/build.sbt drops Review Comment: No spec is excluded on the workflow-operator side — the tag sits on one case inside `PythonCodeRawInvalidTextSpec`, whose other tests still run in this job. That per-case granularity is the departure from amber worth not hiding here. ```suggestion # AMBER_TEST_FILTER=skip-integration excludes the integration-tagged # tests of two modules, each under its own tag: amber/build.sbt drops ``` ########## 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 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")) + } + + 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: + // those holding one time out and are discarded, which frees a slot without + // handing anything back, and the caller waiting at the cap has to notice that Review Comment: "those holding one time out and are discarded" makes the callers the thing discarded; it is their workers. This sentence carries the mechanism the case exists to test, so the mismatch costs a re-read at exactly the wrong place. ```suggestion // 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 ``` ########## 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 — mostly the imports — 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, so callers on distinct worker + * scripts add up rather than share this bound. Defaults to 4, override via + * `TEXERA_TEST_PYTHON_WORKERS`. Public so a caller fanning out jobs within one + * test can size that fan-out to the workers it will get. + */ + val maxWorkers: Int = + sys.env + .get("TEXERA_TEST_PYTHON_WORKERS") + .flatMap(s => scala.util.Try(s.trim.toInt).toOption) + .filter(_ > 0) + .getOrElse(4) + + /** How long a caller waits on a worker before the pool kills and discards it. + * A read on a process pipe cannot be interrupted — a suite or executor timeout + * leaves the reading thread stuck on it — so a worker that stays alive without + * answering has to be bounded here. `response` keeps the 30 seconds the + * one-shot spawn this pool replaced allowed a job; `startup` is longer because Review Comment: `response` and `startup` are not the field names — the case class below declares `responseMillis` and `startupMillis`, which is also what a caller constructing a `Timeouts` has to type. ```suggestion * answering has to be bounded here. `responseMillis` keeps the 30 seconds the * one-shot spawn allowed a job; `startupMillis` is longer because ``` -- 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]
