diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 161affb5..9ad070eb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -79,6 +79,9 @@ jobs: - name: Perl tests (prove) run: devenv shell -- bash -c 'prove tests/t' + - name: Tokenizer worker protocol tests + run: devenv shell -- bash tests/test_token_worker.sh + - name: Pipeline tests (shell) run: devenv shell -- bash -c 'tests/tokenize_gate.sh && tests/pipeline_workdir_guard.sh' diff --git a/blobExec/src/main/scala/cregit/blobexec/BlobExec.scala b/blobExec/src/main/scala/cregit/blobexec/BlobExec.scala index b71e7f2e..3690dbf6 100644 --- a/blobExec/src/main/scala/cregit/blobexec/BlobExec.scala +++ b/blobExec/src/main/scala/cregit/blobexec/BlobExec.scala @@ -46,11 +46,16 @@ object BlobExec { inserter: ObjectInserter, timeoutSeconds: Int = DefaultTimeoutSeconds, onTimeout: () => Unit = () => (), - onParserCrash: () => Unit = () => () + onParserCrash: () => Unit = () => (), + workerPool: Option[TokenizerWorkerPool] = None ): Outcome = { val env = Seq("BFG_BLOB" -> origSha, "BFG_FILENAME" -> filename, "BFG_PATH" -> fullPath) - new ChildRunner(timeoutSeconds).run(command, bytes, env) match { + val ran = workerPool match { + case Some(pool) => pool.invoke(bytes, origSha, filename, fullPath, timeoutSeconds) + case None => new ChildRunner(timeoutSeconds).run(command, bytes, env) + } + ran match { case ChildRunner.Outcome.Killed(why) => System.err.println( s"Warning: command [$command] on blob $origSha at path [$fullPath] gave no usable " + diff --git a/blobExec/src/main/scala/cregit/blobexec/Main.scala b/blobExec/src/main/scala/cregit/blobexec/Main.scala index c608e035..9d872643 100644 --- a/blobExec/src/main/scala/cregit/blobexec/Main.scala +++ b/blobExec/src/main/scala/cregit/blobexec/Main.scala @@ -71,10 +71,16 @@ object Main { // `raw` (not `s`): the mask example below contains a regex backslash, which a // processed-escape interpolator rejects. `$$` therefore renders a literal `$`. private val Usage = - raw"""Usage: blobExec [--abort-on-error] [--pipeline | --pipeline-trees | --shard=K/N] [--warm=] [--blob-timeout=] [--stall-timeout=] + raw"""Usage: blobExec [--abort-on-error] [--pipeline | --pipeline-trees | --shard=K/N] [--warm=] [--tokenizer-worker=] [--blob-timeout=] [--stall-timeout=] | | --abort-on-error exit immediately (status 2) on the first non-zero | exit from , instead of skipping that blob + | --tokenizer-worker= + | keep one protocol-speaking tokenizer process alive per + | available processor and route blob requests through it. + | The worker inherits this process's environment. Needs + | --pipeline or --pipeline-trees; without this flag + | is run once per blob. | --blob-timeout= | budget for one invocation (default ${BlobExec.DefaultTimeoutSeconds}). | A child that exceeds it is killed with everything @@ -142,6 +148,7 @@ object Main { var pipelineTrees = false var shard: Option[(Int, Int)] = None var warmPath: Option[java.nio.file.Path] = None + var tokenizerWorkerPath: Option[java.nio.file.Path] = None var blobTimeoutSeconds = BlobExec.DefaultTimeoutSeconds var stallTimeoutSeconds = Walker.DefaultStallTimeoutSeconds var stallExplicit = false @@ -171,6 +178,8 @@ object Main { sys.exit(1) } warmPath = Some(p) + case w if w.startsWith("--tokenizer-worker=") => + tokenizerWorkerPath = Some(Paths.get(w.stripPrefix("--tokenizer-worker="))) case t if t.startsWith("--blob-timeout=") => val spec = t.stripPrefix("--blob-timeout=") parsePositiveSeconds(spec) match { @@ -238,6 +247,16 @@ object Main { System.err.println(s"Error: command [$command] does not exist") sys.exit(1) } + tokenizerWorkerPath.foreach { path => + if (!Files.isRegularFile(path) || !Files.isExecutable(path)) { + System.err.println(s"Error: tokenizer worker [$path] is not an executable file") + sys.exit(1) + } + if (!(pipeline || pipelineTrees)) { + System.err.println("Error: --tokenizer-worker needs --pipeline or --pipeline-trees") + sys.exit(1) + } + } if (mask.isEmpty) { System.err.println("Error: fileMaskRegex must be non-empty") sys.exit(1) @@ -258,7 +277,8 @@ object Main { s"abortOnError=$abortOnError pipeline=$pipeline pipelineTrees=$pipelineTrees " + s"shard=$shardStr warm=$warmStr blobTimeout=${blobTimeoutSeconds}s " + s"stallTimeout=${stallTimeoutSeconds}s incremental=$incremental " + - s"denylistEntries=${denylist.size}" + s"denylistEntries=${denylist.size} " + + s"tokenizerWorker=${tokenizerWorkerPath.map(_.toString).getOrElse("off")}" ) val src: FileRepository = openSrc(srcPath) @@ -272,14 +292,19 @@ object Main { // The cumulative figure is forensic only: gating on it would make one past // timeout permanent. + var workerPool: Option[TokenizerWorkerPool] = None val (stats, timedOutEver) = try { val parallelism = math.max(1, Runtime.getRuntime.availableProcessors) + workerPool = tokenizerWorkerPath.map { path => + new TokenizerWorkerPool(Seq(path.toString), parallelism) + } val walker = new Walker( src, dst, mapping, mask.r, command, abortOnError, parallelism, pipeline, pipelineTrees, shard, destinationMayContainObjects = incremental, blobTimeoutSeconds = blobTimeoutSeconds, stallTimeoutSeconds = stallTimeoutSeconds, + workerPool = workerPool, denylist = denylist ) val s = walker.run() @@ -288,6 +313,7 @@ object Main { if (s.blobsTimedOut > 0) mapping.setMeta(BlobsTimedOutMetaKey, total.toString) (s, total) } finally { + workerPool.foreach(_.close()) mapping.close() dst.close() src.close() diff --git a/blobExec/src/main/scala/cregit/blobexec/TokenizerWorkerPool.scala b/blobExec/src/main/scala/cregit/blobexec/TokenizerWorkerPool.scala new file mode 100644 index 00000000..a81025b0 --- /dev/null +++ b/blobExec/src/main/scala/cregit/blobexec/TokenizerWorkerPool.scala @@ -0,0 +1,258 @@ +package cregit.blobexec + +import java.io.{BufferedInputStream, BufferedOutputStream, BufferedReader, EOFException, IOException, InputStream, InputStreamReader} +import java.nio.charset.StandardCharsets.UTF_8 +import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference} +import java.util.concurrent.{ArrayBlockingQueue, ConcurrentHashMap, CountDownLatch, TimeUnit} +import scala.jdk.CollectionConverters._ + +final class TokenizerWorkerPool( + workerCommand: Seq[String], + size: Int +) extends AutoCloseable { + + require(workerCommand.nonEmpty, "workerCommand must not be empty") + require(size > 0, "worker pool size must be positive") + + private val WorkerTimeoutExitCode = 124 + private val ReadySeconds = 30L + + private final class Response(val exitCode: Int, val stdout: Array[Byte], val stderr: String) + + private final class Worker( + val index: Int, + val process: Process, + val stdin: BufferedOutputStream, + val stdout: BufferedInputStream + ) { + private val diagnostics = new StringBuilder + + def appendDiagnostic(line: String): Unit = diagnostics.synchronized { + diagnostics.append(line).append('\n') + } + + def takeDiagnostics(): String = diagnostics.synchronized { + try diagnostics.toString finally diagnostics.clear() + } + } + + private val idle = new ArrayBlockingQueue[Worker](size) + private val workers = ConcurrentHashMap.newKeySet[Worker]() + private val closed = new AtomicBoolean(false) + + try { + (0 until size).foreach { index => + val worker = startWorker(index) + workers.add(worker) + idle.put(worker) + } + } catch { + case failure: Throwable => + close() + throw failure + } + + def invoke( + bytes: Array[Byte], + origSha: String, + filename: String, + fullPath: String, + timeoutSeconds: Int + ): ChildRunner.Outcome = { + require(timeoutSeconds > 0, "worker timeout must be positive") + if (closed.get()) throw new IllegalStateException("tokenizer worker pool is closed") + + val worker = idle.take() + worker.takeDiagnostics() + try writeRequest(worker, bytes, origSha, filename, fullPath, timeoutSeconds) + catch { case failure: IOException => return failedWorker(worker, failure) } + + val backstopSeconds = ChildRunner.maxLifetimeSeconds(timeoutSeconds) + readWithin(s"token-worker-response-${worker.index}", backstopSeconds)(readResponse(worker.stdout)) match { + case None => + replaceWorker(worker) + ChildRunner.Outcome.Killed(s"no response within ${backstopSeconds}s; worker replaced") + case Some(Left(failure)) => + failedWorker(worker, failure) + case Some(Right(result)) => + returnWorker(worker) + if (result.exitCode == WorkerTimeoutExitCode) + ChildRunner.Outcome.Killed(s"no exit within ${timeoutSeconds}s") + else ChildRunner.Outcome.Exited(result.exitCode, result.stdout, result.stderr) + } + } + + override def close(): Unit = { + if (closed.compareAndSet(false, true)) { + val snapshot = workers.asScala.toVector + snapshot.foreach(worker => try worker.stdin.close() catch { case _: IOException => () }) + val deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2) + snapshot.foreach { worker => + val remaining = deadline - System.nanoTime() + if (remaining > 0) worker.process.waitFor(remaining, TimeUnit.NANOSECONDS) + } + snapshot.filter(_.process.isAlive).foreach(_.process.destroyForcibly()) + snapshot.foreach(_.process.waitFor(2, TimeUnit.SECONDS)) + workers.clear() + idle.clear() + } + } + + private def startWorker(index: Int): Worker = { + val builder = new ProcessBuilder(workerCommand: _*) + val process = builder.start() + val worker = new Worker( + index, + process, + new BufferedOutputStream(process.getOutputStream), + new BufferedInputStream(process.getInputStream) + ) + startStderrForwarder(worker) + + readWithin(s"token-worker-ready-$index", ReadySeconds)(readHeader(worker.stdout)) match { + case Some(Right("READY")) => worker + case notReady => + process.destroyForcibly() + throw startFailure(index, notReady) + } + } + + private def startFailure(index: Int, ready: Option[Either[Throwable, String]]): IllegalStateException = + ready match { + case None => + new IllegalStateException(s"tokenizer worker $index did not print READY within $ReadySeconds seconds") + case Some(Right(line)) => + new IllegalStateException(s"tokenizer worker $index printed [$line] instead of READY") + case Some(Left(failure)) => + new IllegalStateException(s"tokenizer worker $index failed before READY", failure) + } + + private def readWithin[A](threadName: String, seconds: Long)(read: => A): Option[Either[Throwable, A]] = { + val result = new AtomicReference[Either[Throwable, A]]() + val done = new CountDownLatch(1) + val reader = new Thread( + () => { + try result.set(Right(read)) + catch { case failure: Throwable => result.set(Left(failure)) } + finally done.countDown() + }, + threadName + ) + reader.setDaemon(true) + reader.start() + if (done.await(seconds, TimeUnit.SECONDS)) Some(result.get()) else None + } + + private def startStderrForwarder(worker: Worker): Unit = { + def forward(line: String): Unit = { + worker.appendDiagnostic(line) + System.err.println(s"tokenWorker[${worker.index}]: $line") + } + val thread = new Thread( + () => { + val reader = new BufferedReader(new InputStreamReader(worker.process.getErrorStream, UTF_8)) + try Iterator.continually(reader.readLine()).takeWhile(_ != null).foreach(forward) + catch { + case failure: IOException if worker.process.isAlive => + forward(s"stderr forwarding failed: ${failure.getMessage}") + case _: IOException => () + } finally try reader.close() catch { case _: IOException => () } + }, + s"token-worker-stderr-${worker.index}" + ) + thread.setDaemon(true) + thread.start() + } + + private def writeRequest( + worker: Worker, + bytes: Array[Byte], + origSha: String, + filename: String, + fullPath: String, + timeoutSeconds: Int + ): Unit = { + val filenameBytes = filename.getBytes(UTF_8) + val pathBytes = fullPath.getBytes(UTF_8) + val header = s"REQ $origSha ${bytes.length} ${filenameBytes.length} ${pathBytes.length} $timeoutSeconds\n" + .getBytes(UTF_8) + worker.stdin.write(header) + worker.stdin.write(filenameBytes) + worker.stdin.write(pathBytes) + worker.stdin.write(bytes) + worker.stdin.flush() + } + + private def readResponse(input: InputStream): Response = { + val fields = readHeader(input).split(" ", -1) + if (fields.length != 4 || fields(0) != "RES") + throw new IOException(s"malformed tokenizer worker response header [${fields.mkString(" ")}]") + + val exitCode = parseInt(fields(1), "exit") + val stdoutLength = parseLength(fields(2), "outLen") + val stderrLength = parseLength(fields(3), "errLen") + val stdout = readExactly(input, stdoutLength) + val stderr = new String(readExactly(input, stderrLength), UTF_8) + new Response(exitCode, stdout, stderr) + } + + private def readHeader(input: InputStream): String = { + val bytes = new java.io.ByteArrayOutputStream + var next = input.read() + while (next != -1 && next != '\n') { + if (bytes.size() >= 8192) throw new IOException("tokenizer worker response header exceeds 8192 bytes") + bytes.write(next) + next = input.read() + } + if (next == -1) throw new EOFException("tokenizer worker closed stdout") + new String(bytes.toByteArray, UTF_8) + } + + private def readExactly(input: InputStream, length: Int): Array[Byte] = { + val bytes = new Array[Byte](length) + var offset = 0 + while (offset < length) { + val read = input.read(bytes, offset, length - offset) + if (read == -1) throw new EOFException(s"tokenizer worker closed stdout after $offset of $length bytes") + offset += read + } + bytes + } + + private def parseInt(value: String, name: String): Int = + value.toIntOption.getOrElse(throw new IOException(s"invalid $name [$value] in tokenizer response")) + + private def parseLength(value: String, name: String): Int = { + val length = parseInt(value, name) + if (length < 0) throw new IOException(s"negative $name [$value] in tokenizer response") + length + } + + private def failedWorker(worker: Worker, failure: Throwable): ChildRunner.Outcome = { + val workerExit = if (worker.process.waitFor(100, TimeUnit.MILLISECONDS)) Some(worker.process.exitValue()) else None + val diagnostics = worker.takeDiagnostics() + replaceWorker(worker) + if (workerExit.contains(WorkerTimeoutExitCode)) + ChildRunner.Outcome.Killed(s"worker exited $WorkerTimeoutExitCode (timeout)") + else { + val stderr = if (diagnostics.nonEmpty) diagnostics else Option(failure.getMessage).getOrElse(failure.toString) + ChildRunner.Outcome.Exited(workerExit.filter(_ != 0).getOrElse(1), Array.emptyByteArray, stderr) + } + } + + private def returnWorker(worker: Worker): Unit = { + if (worker.process.isAlive && !closed.get()) idle.put(worker) + else replaceWorker(worker) + } + + private def replaceWorker(worker: Worker): Unit = { + workers.remove(worker) + try worker.stdin.close() catch { case _: IOException => () } + if (worker.process.isAlive) ChildRunner.killTree(worker.process) + if (!closed.get()) { + val replacement = startWorker(worker.index) + workers.add(replacement) + idle.put(replacement) + } + } +} diff --git a/blobExec/src/main/scala/cregit/blobexec/Walker.scala b/blobExec/src/main/scala/cregit/blobexec/Walker.scala index a9c9f474..46e01538 100644 --- a/blobExec/src/main/scala/cregit/blobexec/Walker.scala +++ b/blobExec/src/main/scala/cregit/blobexec/Walker.scala @@ -57,6 +57,7 @@ final class Walker( destinationMayContainObjects: Boolean = true, blobTimeoutSeconds: Int = BlobExec.DefaultTimeoutSeconds, stallTimeoutSeconds: Int = Walker.DefaultStallTimeoutSeconds, + workerPool: Option[TokenizerWorkerPool] = None, denylist: BlobDenylist = BlobDenylist.shipped ) { import Walker._ @@ -852,7 +853,8 @@ final class Walker( inserter = workerInserter, timeoutSeconds = blobTimeoutSeconds, onTimeout = () => { blobsTimedOut.increment(); unusable.set(true) }, - onParserCrash = () => { blobsParserCrashed.increment(); unusable.set(true) } + onParserCrash = () => { blobsParserCrashed.increment(); unusable.set(true) }, + workerPool = workerPool ) val res = outcome match { case BlobExec.Outcome.Skip if unusable.get() => diff --git a/blobExec/src/test/scala/cregit/blobexec/TokenizerWorkerPoolSpec.scala b/blobExec/src/test/scala/cregit/blobexec/TokenizerWorkerPoolSpec.scala new file mode 100644 index 00000000..c14a1913 --- /dev/null +++ b/blobExec/src/test/scala/cregit/blobexec/TokenizerWorkerPoolSpec.scala @@ -0,0 +1,209 @@ +package cregit.blobexec + +import org.scalatest.BeforeAndAfterAll +import org.scalatest.funsuite.AnyFunSuite +import org.scalatest.matchers.should.Matchers + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Path, Paths} +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContext, Future} +import scala.jdk.CollectionConverters._ + +class TokenizerWorkerPoolSpec extends AnyFunSuite with Matchers with BeforeAndAfterAll { + + import ChildRunner.Outcome.{Exited, Killed} + + private def exited(outcome: ChildRunner.Outcome): (Int, Array[Byte], String) = outcome match { + case Exited(status, stdout, stderr) => (status, stdout, stderr) + case other => fail(s"expected Exited, got $other") + } + + private def request( + pool: TokenizerWorkerPool, + filename: String, + body: String, + sha: String, + timeoutSeconds: Int = 2 + ): ChildRunner.Outcome = + pool.invoke(body.getBytes(UTF_8), sha, filename, s"src/$filename", timeoutSeconds) + + private val tmpDir = Files.createTempDirectory("tokenizer-worker-pool-") + private val worker = tmpDir.resolve("fake-worker.pl") + + override def beforeAll(): Unit = { + Files.writeString( + worker, + """|#!/usr/bin/env perl + |use strict; + |use warnings; + | + |binmode STDIN; + |binmode STDOUT; + |$| = 1; + | + |sub read_exact { + | my ($length) = @_; + | my $value = ''; + | while (length($value) < $length) { + | my $read = read(STDIN, my $chunk, $length - length($value)); + | exit 2 unless defined($read) && $read > 0; + | $value .= $chunk; + | } + | return $value; + |} + | + |print "READY\n"; + |while (defined(my $header = )) { + | chomp $header; + | my ($tag, $sha, $body_len, $fn_len, $path_len, $timeout) = split / /, $header; + | exit 2 unless $tag eq 'REQ'; + | my $filename = read_exact($fn_len); + | my $path = read_exact($path_len); + | my $body = read_exact($body_len); + | + | if ($filename eq 'hang.c') { + | my $child = fork() // exit 3; + | exec 'sleep', '300' if $child == 0; + | open(my $pidFile, '>', "$0.grandchild") or exit 3; + | print {$pidFile} $child; + | close($pidFile); + | sleep 300 while 1; + | } elsif ($filename eq 'slow.c') { + | sleep 2; + | my $out = uc($body); + | print "RES 0 ", length($out), " 0\n", $out; + | } elsif ($filename eq 'crash.c') { + | print "RES 33 0 5\ncrash"; + | } elsif ($filename eq 'timeout.c') { + | print "RES 124 0 7\ntimeout"; + | } elsif ($filename eq 'budget.c') { + | print "RES 0 ", length($timeout), " 0\n", $timeout; + | } elsif ($filename eq 'die.c') { + | print STDERR "worker died\n"; + | exit 7; + | } else { + | my $out = uc($body); + | print "RES 0 ", length($out), " 0\n", $out; + | } + |} + |""".stripMargin + ) + worker.toFile.setExecutable(true) + } + + override def afterAll(): Unit = deleteRecursive(tmpDir) + + test("round-trips bytes and preserves parser crash status") { + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 1) + try { + val (exit, stdout, stderr) = exited(request(pool, "hello.c", "hello", "a" * 40)) + exit shouldEqual 0 + stdout shouldEqual "HELLO".getBytes(UTF_8) + stderr shouldBe empty + + val (crashExit, crashOut, crashErr) = + exited(request(pool, "crash.c", "bad", "b" * 40)) + crashExit shouldEqual BlobExec.ParserCrashExitCode + crashOut shouldBe empty + crashErr shouldEqual "crash" + } finally pool.close() + } + + test("sends each request's own budget to the worker") { + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 1) + try { + exited(request(pool, "budget.c", "x", "9" * 40, timeoutSeconds = 7))._2 shouldEqual "7".getBytes(UTF_8) + exited(request(pool, "budget.c", "x", "9" * 40, timeoutSeconds = 21))._2 shouldEqual "21".getBytes(UTF_8) + } finally pool.close() + } + + test("times out a wedged worker, kills its children, replaces it, and serves the next request") { + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 1) + try { + val started = System.nanoTime() + val hung = request(pool, "hang.c", "hang", "c" * 40, timeoutSeconds = 1) + val elapsed = (System.nanoTime() - started).nanos + + hung shouldBe a[Killed] + elapsed should be < 15.seconds + val grandchild = Files.readString(Paths.get(s"$worker.grandchild")).trim.toLong + eventuallyStopped(Seq(grandchild)) shouldBe true + val (nextExit, nextOut, nextErr) = + exited(request(pool, "next.c", "next", "d" * 40, timeoutSeconds = 1)) + nextExit shouldEqual 0 + nextOut shouldEqual "NEXT".getBytes(UTF_8) + nextErr shouldBe empty + } finally pool.close() + } + + test("serves pool-sized requests in parallel") { + implicit val ec: ExecutionContext = ExecutionContext.global + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 3) + try { + val started = System.nanoTime() + val requests = (1 to 3).map { n => + Future(exited(request(pool, "slow.c", s"body$n", f"$n%040x", timeoutSeconds = 5))) + } + val results = Await.result(Future.sequence(requests), 10.seconds) + val elapsed = (System.nanoTime() - started).nanos + + results.map(_._1) should contain only 0 + results.map(r => new String(r._2, UTF_8)).toSet shouldEqual Set("BODY1", "BODY2", "BODY3") + elapsed should be < 4.seconds + } finally pool.close() + } + + test("maps worker timeout status and respawns after mid-request death") { + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 1) + try { + val timedOut = request(pool, "timeout.c", "slow", "e" * 40) + timedOut shouldBe a[Killed] + + val died = exited(request(pool, "die.c", "die", "f" * 40)) + died._1 should not equal 0 + died._2 shouldBe empty + died._3 should include("worker died") + + val recovered = exited(request(pool, "ok.c", "ok", "1" * 40)) + recovered._1 shouldEqual 0 + recovered._2 shouldEqual "OK".getBytes(UTF_8) + } finally pool.close() + } + + test("close terminates every worker process") { + val pool = new TokenizerWorkerPool(Seq(worker.toString), size = 2) + val pids = workerPids() + pids should have size 2 + + pool.close() + + eventuallyStopped(pids) shouldBe true + } + + private def workerPids(): Seq[Long] = + ProcessHandle.current().children().iterator().asScala + .filter(_.info().commandLine().orElse("").contains(worker.toString)) + .map(_.pid()) + .toVector + + private def eventuallyStopped(pids: Seq[Long]): Boolean = { + val deadline = System.nanoTime() + 5.seconds.toNanos + while (System.nanoTime() < deadline && pids.exists(isAlive)) Thread.sleep(25) + pids.forall(pid => !isAlive(pid)) + } + + private def isAlive(pid: Long): Boolean = { + val handle = ProcessHandle.of(pid) + handle.isPresent && handle.get().isAlive + } + + private def deleteRecursive(path: Path): Unit = { + if (Files.isDirectory(path)) { + val entries = Files.list(path) + try entries.forEach(deleteRecursive) finally entries.close() + } + Files.deleteIfExists(path) + () + } +} diff --git a/run_pipeline_process.sh b/run_pipeline_process.sh index b7ed4ed0..e7201c01 100755 --- a/run_pipeline_process.sh +++ b/run_pipeline_process.sh @@ -82,6 +82,11 @@ Output: with millions of loose objects. A failed pack never aborts the run: the Parquet dataset is the product, and packing only makes later steps faster. + --tokenizer-worker + keep one tokenizeByBlobId/tokenWorker.pl process alive per + blobExec worker slot instead of spawning the full wrapper + chain for every blob (default: off). Needs --mode + pipeline or pipeline-trees. --memory-limit SIZE cap the DuckDB heap in the dataset generator (step 10). Omit to accept that script's own default of 8GB. Takes an @@ -198,6 +203,7 @@ WORK="../cregit-files" MEMO_DIR="" SKIP_HTML=0 GC_MODE="plain" +TOKENIZER_WORKER="" MEMORY_LIMIT="" DUCKDB_THREADS="" # Empty means "do not pass the flag", as above: step 10 then emits the @@ -367,6 +373,7 @@ while [ $# -gt 0 ]; do --blob-timeout) need_val "$@"; BLOB_TIMEOUT="$2"; shift 2 ;; --stall-timeout) need_val "$@"; STALL_TIMEOUT="$2"; shift 2 ;; --gc) need_val "$@"; GC_MODE="$2"; shift 2 ;; + --tokenizer-worker) TOKENIZER_WORKER=1; shift ;; --memory-limit) need_val "$@"; MEMORY_LIMIT="$2"; shift 2 ;; --duckdb-threads) need_val "$@"; DUCKDB_THREADS="$2"; shift 2 ;; --project-meta) need_val "$@"; PROJECT_META="$2"; shift 2 ;; @@ -486,6 +493,9 @@ case "$MODE" in serial|pipeline|pipeline-trees|sharded) ;; *) echo "invalid --mode: $MODE (serial|pipeline|pipeline-trees|sharded)" >&2; exit 2 ;; esac +if [ -n "$TOKENIZER_WORKER" ] && [ "$MODE" != pipeline ] && [ "$MODE" != pipeline-trees ]; then + echo "--tokenizer-worker needs --mode pipeline or pipeline-trees" >&2; exit 2 +fi if [ "$MODE" = "sharded" ]; then [ "$SHARDS" -ge 1 ] 2>/dev/null || { echo "--shards must be a positive integer" >&2; exit 2; } fi @@ -744,6 +754,7 @@ else [ "$MODE" = "pipeline" ] && MODE_FLAG="--pipeline" [ "$MODE" = "pipeline-trees" ] && MODE_FLAG="--pipeline-trees" java -jar "$BFG" $MODE_FLAG \ + ${TOKENIZER_WORKER:+"--tokenizer-worker=${CREGIT}/tokenizeByBlobId/tokenWorker.pl"} \ ${BLOB_TIMEOUT:+--blob-timeout=$BLOB_TIMEOUT} \ ${STALL_TIMEOUT:+--stall-timeout=$STALL_TIMEOUT} \ "$REPO_PATH_ORIGINAL_BARE" \ diff --git a/tests/fixtures/Tiny.java b/tests/fixtures/Tiny.java new file mode 100644 index 00000000..c9b995f9 --- /dev/null +++ b/tests/fixtures/Tiny.java @@ -0,0 +1,4 @@ +package p; +public class Tiny { + int f(int a) { return a; } +} diff --git a/tests/fixtures/empty.c b/tests/fixtures/empty.c new file mode 100644 index 00000000..e69de29b diff --git a/tests/fixtures/quote'name.c b/tests/fixtures/quote'name.c new file mode 100644 index 00000000..041c166d --- /dev/null +++ b/tests/fixtures/quote'name.c @@ -0,0 +1 @@ +static int quoted_name(void) { return 7; } diff --git a/tests/fixtures/tiny.c b/tests/fixtures/tiny.c new file mode 100644 index 00000000..78f2de10 --- /dev/null +++ b/tests/fixtures/tiny.c @@ -0,0 +1 @@ +int main(void) { return 0; } diff --git a/tests/fixtures/tiny.cpp b/tests/fixtures/tiny.cpp new file mode 100644 index 00000000..88e4ae71 --- /dev/null +++ b/tests/fixtures/tiny.cpp @@ -0,0 +1,2 @@ +struct S { int m(int a) const { return a; } }; +template T id(T v) { return v; } diff --git a/tests/fixtures/upper.H b/tests/fixtures/upper.H new file mode 100644 index 00000000..e039d96f --- /dev/null +++ b/tests/fixtures/upper.H @@ -0,0 +1,6 @@ +namespace n { +class A { +public: + int x; +}; +} diff --git a/tests/test_token_worker.sh b/tests/test_token_worker.sh new file mode 100755 index 00000000..4751b571 --- /dev/null +++ b/tests/test_token_worker.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +# Run inside `devenv shell` (needs perl, python3, srcml, ctags on PATH). +set -euo pipefail + +ROOT=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) + +SRCML2TOKEN="$ROOT/tokenize/srcMLtoken/srcml2token" +[[ -x "$SRCML2TOKEN" ]] || make -C "$ROOT/tokenize/srcMLtoken" + +SRCML=$(command -v srcml) +CTAGS=$(command -v ctags) +TOKENIZE_COMMAND="$ROOT/tokenize/tokenize.pl --srcml2token=$SRCML2TOKEN --srcml=$SRCML --ctags=$CTAGS" +TEMP_ROOT=$(mktemp -d "${TMPDIR:-/tmp}/cregit-token-worker.XXXXXX") +trap 'rm -rf "$TEMP_ROOT"' EXIT + +python3 "$ROOT/tests/token_worker_driver.py" \ + --worker "$ROOT/tokenizeByBlobId/tokenWorker.pl" \ + --token-by-sha "$ROOT/tokenizeByBlobId/tokenBySha.pl" \ + --tokenize-command "$TOKENIZE_COMMAND" \ + --fixture "$ROOT/tests/fixtures/tiny.c" \ + --fixture "$ROOT/tests/fixtures/quote'name.c" \ + --fixture "$ROOT/tests/fixtures/tiny.cpp" \ + --fixture "$ROOT/tests/fixtures/Tiny.java" \ + --fixture "$ROOT/tests/fixtures/empty.c" \ + --fixture "$ROOT/tests/fixtures/upper.H" \ + --temp-root "$TEMP_ROOT" diff --git a/tests/token_worker_driver.py b/tests/token_worker_driver.py new file mode 100755 index 00000000..c083a7c3 --- /dev/null +++ b/tests/token_worker_driver.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 + +import argparse +import hashlib +import os +import pathlib +import shutil +import subprocess +import tempfile +import time + + +class Worker: + def __init__(self, executable, memo_dir, tokenize_command): + env = os.environ.copy() + env["BFG_MEMO_DIR"] = str(memo_dir) + env["BFG_TOKENIZE_CMD"] = tokenize_command + self.process = subprocess.Popen( + [str(executable)], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + ) + ready = self.process.stdout.readline() + if ready != b"READY\n": + raise AssertionError(f"worker startup failed: {ready!r}") + + def request(self, filename, full_path, contents, timeout_secs): + filename_bytes = os.fsencode(filename) + path_bytes = os.fsencode(full_path) + orig_sha = hashlib.sha1(contents).hexdigest() + header = ( + f"REQ {orig_sha} {len(contents)} {len(filename_bytes)} " + f"{len(path_bytes)} {timeout_secs}\n" + ).encode("ascii") + self.process.stdin.write(header + filename_bytes + path_bytes + contents) + self.process.stdin.flush() + + response = self.process.stdout.readline() + parts = response.decode("ascii").rstrip("\n").split(" ") + if len(parts) != 4 or parts[0] != "RES": + raise AssertionError(f"invalid response header: {response!r}") + exit_code, out_len, err_len = map(int, parts[1:]) + stdout = self._read_exact(out_len) + stderr = self._read_exact(err_len) + return exit_code, stdout, stderr + + def close(self): + self.process.stdin.close() + exit_code = self.process.wait(timeout=5) + stderr = self.process.stderr.read() + if exit_code != 0: + raise AssertionError( + f"worker exited {exit_code} on EOF: {stderr.decode(errors='replace')}" + ) + + def _read_exact(self, length): + value = b"" + while len(value) < length: + chunk = self.process.stdout.read(length - len(value)) + if not chunk: + raise AssertionError("unexpected EOF from worker") + value += chunk + return value + + +def reference(token_by_sha, memo_dir, tokenize_command, fixture): + contents = fixture.read_bytes() + env = os.environ.copy() + env.update( + { + "BFG_BLOB": hashlib.sha1(contents).hexdigest(), + "BFG_FILENAME": fixture.name, + "BFG_PATH": f"fixtures/{fixture.name}", + "BFG_MEMO_DIR": str(memo_dir), + "BFG_TOKENIZE_CMD": tokenize_command, + } + ) + return subprocess.run( + [str(token_by_sha)], + input=contents, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + check=False, + ) + + +def memo_file(memo_dir, contents): + digest = hashlib.sha1(contents).hexdigest() + return memo_dir / digest[:2] / digest[2:4] / digest + + +def assert_no_timeout_process(marker): + time.sleep(0.2) + result = subprocess.run( + ["pgrep", "-f", marker], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + if result.returncode != 1: + raise AssertionError( + f"timeout process still running: {result.stdout.decode().strip()}" + ) + + +def write_script(path, body): + path.write_text("#!/usr/bin/env bash\n" + body, encoding="utf-8") + path.chmod(0o755) + return path + + +def in_process_command(args, marker): + fake_srcml = write_script( + args.temp_root / "fake-srcml.sh", + 'src="${@: -3:1}"\n' + f'grep -q SLEEP_MARKER "$src" && exec perl -e "sleep 100" {marker}\n' + 'grep -q ABORT_MARKER "$src" && kill -ABRT $$\n' + f'exec {shutil.which("srcml")} "$@"\n', + ) + fake_ctags = write_script( + args.temp_root / "fake-ctags.sh", + "echo HELPER_STDERR_MARKER >&2\n" + f'exec {shutil.which("ctags")} "$@"\n', + ) + options = [o for o in args.tokenize_command.split() if not o.startswith(("--srcml=", "--ctags="))] + return " ".join(options + [f"--srcml={fake_srcml}", f"--ctags={fake_ctags}"]) + + +def in_process_checks(args, marker, fixture, expected): + memo = args.temp_root / "inproc-memo" + memo.mkdir() + worker = Worker(args.worker, memo, in_process_command(args, marker)) + + started = time.monotonic() + timed_out = worker.request("timeout.c", "fixtures/timeout.c", b"int SLEEP_MARKER;\n", 1) + elapsed = time.monotonic() - started + if timed_out[0] != 124 or elapsed > 3.0: + raise AssertionError(f"in-process timeout response was {timed_out!r} after {elapsed:.2f}s") + assert_no_timeout_process(marker) + print(f"PASS in-process timeout exits 124 in {elapsed:.2f}s with no child left") + + crash = worker.request("crash.c", "fixtures/crash.c", b"int ABORT_MARKER;\n", 20) + if crash[0] != 33 or b"killed by signal 6" not in crash[2]: + raise AssertionError(f"srcml death was not reported as a parser crash: {crash!r}") + if b"HELPER_STDERR_MARKER" not in crash[2]: + raise AssertionError(f"helper stderr did not reach the request's stderr: {crash!r}") + print("PASS in-process srcml crash exits 33 without output, with helper stderr") + + if any(memo.rglob("*")): + raise AssertionError("failed in-process requests were memoized") + recovered = request_fixture(worker, fixture) + if recovered != (0, expected, b""): + raise AssertionError(f"in-process worker did not recover: {recovered!r}") + worker.close() + print("PASS in-process worker recovers after timeout and crash") + + +def request_fixture(worker, fixture): + return worker.request(fixture.name, f"fixtures/{fixture.name}", fixture.read_bytes(), 20) + + +def byte_identity_checks(args, worker, worker_memo): + expected_outputs = [] + for index, fixture in enumerate(args.fixture): + reference_memo = args.temp_root / f"reference-memo-{index}" + reference_memo.mkdir() + expected = reference( + args.token_by_sha, reference_memo, args.tokenize_command, fixture + ) + if expected.returncode != 0: + raise AssertionError( + f"reference failed for {fixture}: " + f"{expected.stderr.decode(errors='replace')}" + ) + actual = request_fixture(worker, fixture) + if actual[0] != 0 or actual[1] != expected.stdout: + raise AssertionError(f"worker output differs for {fixture}") + if memo_file(worker_memo, fixture.read_bytes()).read_bytes() != expected.stdout: + raise AssertionError(f"worker memo differs for {fixture}") + expected_outputs.append(expected.stdout) + print(f"PASS byte-identical: {fixture.name}") + return expected_outputs + + +def memo_and_recovery_checks(worker, fixture, expected): + if request_fixture(worker, fixture) != (0, expected, b""): + raise AssertionError("memo-hit response changed bytes") + print("PASS repeated request uses identical memo bytes") + + unknown = worker.request("unknown.xyzzy", "fixtures/unknown.xyzzy", b"x", 20) + if unknown[0] == 0 or b"unknown file extension" not in unknown[2]: + raise AssertionError(f"unknown extension was not rejected: {unknown!r}") + recovered = request_fixture(worker, fixture) + if recovered[0] != 0 or recovered[1] != expected: + raise AssertionError("worker did not recover after unknown extension") + print("PASS unknown extension is non-fatal to worker") + + +def timeout_checks(args, marker): + timeout_script = write_script( + args.temp_root / "timeout-parser.sh", f"exec perl -e 'sleep 100' {marker}\n" + ) + timeout_memo = args.temp_root / "timeout-memo" + timeout_memo.mkdir() + timeout_worker = Worker(args.worker, timeout_memo, str(timeout_script)) + started = time.monotonic() + timed_out = timeout_worker.request("timeout.c", "fixtures/timeout.c", b"int x;\n", 1) + elapsed = time.monotonic() - started + if timed_out[0] != 124 or elapsed > 3.0: + raise AssertionError(f"timeout response was {timed_out[0]} after {elapsed:.2f}s") + if any(timeout_memo.rglob("*")): + raise AssertionError("timeout response was memoized") + assert_no_timeout_process(marker) + timeout_worker.close() + print(f"PASS timeout exits 124 in {elapsed:.2f}s with no child left") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--worker", required=True, type=pathlib.Path) + parser.add_argument("--token-by-sha", required=True, type=pathlib.Path) + parser.add_argument("--tokenize-command", required=True) + parser.add_argument("--fixture", action="append", required=True, type=pathlib.Path) + parser.add_argument("--temp-root", required=True, type=pathlib.Path) + args = parser.parse_args() + + worker_memo = args.temp_root / "worker-memo" + worker_memo.mkdir() + worker = Worker(args.worker, worker_memo, args.tokenize_command) + expected_outputs = byte_identity_checks(args, worker, worker_memo) + memo_and_recovery_checks(worker, args.fixture[0], expected_outputs[0]) + worker.close() + + marker = f"cregit-token-worker-timeout-{os.getpid()}" + timeout_checks(args, marker) + in_process_checks(args, marker, args.fixture[0], expected_outputs[0]) + + print("ALL TOKEN WORKER TESTS PASSED") + + +if __name__ == "__main__": + main() diff --git a/tokenize/CregitSrcMl.pm b/tokenize/CregitSrcMl.pm new file mode 100644 index 00000000..bddb8b15 --- /dev/null +++ b/tokenize/CregitSrcMl.pm @@ -0,0 +1,383 @@ +package CregitSrcMl; + +# In-process replacement for tokenizeSrcMl.pl: one persistent ctags per language +# and one persistent `srcml2token --server`; only srcml is spawned per file. + +use strict; +use warnings; +use bytes; +use Errno qw(EINTR); +use File::Temp (); +use POSIX qw(_exit); + +our $PARSER_CRASH_EXIT = 33; + +sub new { + my ($class, %options) = @_; + my $self = { + srcml => $options{srcml} // "srcml", + srcml2token => $options{srcml2token} // "srcml2token", + ctags => $options{ctags} // "ctags-universal", + position => $options{position} // 0, + ctagsProc => {}, + s2t => undef, + child => undef, + timedOut => 0, + }; + return bless $self, $class; +} + +sub tokenize { + my ($self, $language, $inputName, $workDir, $timeoutSecs) = @_; + + $self->{timedOut} = 0; + local $SIG{ALRM} = sub { + $self->{timedOut} = 1; + $self->kill_all(); + }; + alarm($timeoutSecs) if $timeoutSecs; + + my @result; + my $ok = eval { + @result = $self->run($language, $inputName, $workDir); + 1; + }; + my $error = $@; + alarm(0); + my $helperErr = $self->take_helper_stderr(); + + if ($self->{timedOut}) { + $self->shutdown(); + return (124, "", $helperErr . "tokenize command timed out after [$timeoutSecs] seconds\n"); + } + return (255, "", $helperErr . ($error || "srcml tokenization failed\n")) if not $ok; + $result[2] = $helperErr . $result[2]; + return @result; +} + +sub run { + my ($self, $language, $inputName, $workDir) = @_; + my $stderr = ""; + + my ($declarations, $listDeclarations) = + $self->read_declarations($language, $inputName, "$workDir/$inputName"); + + my $xmlPath = "$workDir/srcml-output.xml"; + my ($srcmlStatus, $srcmlErr) = $self->run_srcml($language, $inputName, $workDir, $xmlPath); + $stderr .= $srcmlErr; + if ($srcmlStatus & 127) { + my $signal = $srcmlStatus & 127; + return parser_crash($inputName, $stderr, + "srcml was killed by signal $signal (shell status " . (128 + $signal) . "). " + . "srcML 1.1.0 dies on a signal on some C/C++ inputs when --position " + . "is given; --position cannot be dropped because the token format " + . "depends on it."); + } + if ($srcmlStatus != 0) { + return parser_crash($inputName, $stderr, "srcml exited " . ($srcmlStatus >> 8) . "."); + } + + my ($tokenStatus, $tokenLines, $tokenErr) = $self->parse_xml($xmlPath); + unlink $xmlPath; + $stderr .= $tokenErr; + if (not defined $tokenStatus) { + return parser_crash($inputName, $stderr, "srcml2token died before answering."); + } + if ($tokenStatus != 0) { + return parser_crash($inputName, $stderr, "srcml2token exited $tokenStatus."); + } + if (not @$tokenLines) { + return parser_crash($inputName, $stderr, + "the token stream was empty. A successful srcML parse always emits at " + . "least the begin_unit/end_unit wrapper, even for an empty file, so an " + . "empty stream is a failed parse and never a legitimately empty result."); + } + + my $output = $self->merge($tokenLines, $declarations, $listDeclarations); + return (0, $output, $stderr); +} + +sub merge { + my ($self, $tokenLines, $declarations, $listDeclarations) = @_; + my $position = $self->{position}; + my $lastLine = -1; + my $output = ""; + + for my $tokenLine (@$tokenLines) { + die "unable to parse srcml line [$tokenLine]" + unless $tokenLine =~ /^([0-9]+|-):([0-9]+|-)\s+(.+)$/; + my ($line, $col, $token) = ($1, $2, $3); + my $lineNum = $line eq '-' ? 0 : $line; + $output .= declaration_lines($line, $declarations, $listDeclarations, $position) + if $lineNum != $lastLine; + $lastLine = $lineNum; + $output .= "$line:$col|" if $position; + $output .= "$token\n"; + if ($token =~ /^end_/) { + $output .= "-:-|" if $position; + $output .= "\n"; + } + } + return $output; +} + +sub declaration_lines { + my ($line, $declarations, $listDeclarations, $position) = @_; + my $output = ""; + for my $name (@{ $listDeclarations->{$line} || [] }) { + my $decl = $declarations->{$line}{$name}; + die "Illegal value in get declaration [$line][$name]" unless defined $decl; + $output .= "$line:-|" if $position; + $output .= "DECL|$decl->{type}|$decl->{name}\n"; + } + return $output; +} + +sub parser_crash { + my ($filename, $stderr, $why) = @_; + $stderr .= "cregit: tokenization of [$filename] FAILED: $why\n"; + $stderr .= "cregit: refusing to emit a tokenization for [$filename]; " + . "exiting $PARSER_CRASH_EXIT so the caller can count this blob.\n"; + return ($PARSER_CRASH_EXIT, "", $stderr); +} + +sub ctags_for { + my ($self, $language) = @_; + my $proc = $self->{ctagsProc}{$language}; + return $proc if $proc; + + $proc = spawn_pipe([ $self->{ctags}, "--_interactive", "--language-force=$language", + "--fields=+nKS", "--sort=no" ]); + my $hello = readline($proc->{out}); + die "ctags did not start: " . ($hello // "EOF") . "\n" + unless defined $hello and $hello =~ /"_type": "program"/; + $self->{ctagsProc}{$language} = $proc; + return $proc; +} + +sub read_declarations { + my ($self, $language, $inputName, $path) = @_; + my (%declarations, %listDeclarations); + + my $source = do { + open(my $fh, '<:raw', $path) or die "unable to read [$path]: $!\n"; + local $/; + <$fh>; + }; + $source = "" unless defined $source; + + my $proc = $self->ctags_for($language); + my $in = $proc->{in}; + print {$in} qq({"command":"generate-tags","filename":"$inputName","size":) + . length($source) . "}\n", $source; + $in->flush; + + while (1) { + my $line = readline($proc->{out}); + if (not defined $line) { + $self->drop_ctags($language); + last; + } + last if $line =~ /"_type": "completed"/; + if ($line =~ /"_type": "error"/) { + $self->drop_ctags($language); + die "ctags rejected the request: $line"; + } + my $decl = declaration_from_tag($line) or next; + $declarations{$decl->{line}}{$decl->{name}} = $decl; + push @{ $listDeclarations{$decl->{line}} }, $decl->{name}; + } + return (\%declarations, \%listDeclarations); +} + +sub declaration_from_tag { + my ($line) = @_; + return undef unless $line =~ /"_type": "tag"/; + + my ($n) = $line =~ /"line": ([0-9]+)/; + my $name = json_field($line, "name"); + my $kind = json_field($line, "kind"); + my $sig = json_field($line, "signature"); + return undef unless defined $n and defined $name and defined $kind; + $sig = "-" unless defined $sig; + for ($name, $kind, $sig) { s/\\/\\\\/g } + + my $xref = "$n $name @@@ $kind @@@ $sig"; + die "unable to parse output [$xref]" unless $xref =~ /^([0-9]+) (.+) @@@ (.*) @@@ (.*)$/; + my %decl = (line => $1, name => $2, type => $3, sig => $4); + $decl{name} .= " " . $decl{sig} if $decl{sig} ne "-"; + return \%decl; +} + +sub drop_ctags { + my ($self, $language) = @_; + $self->retire(delete $self->{ctagsProc}{$language}); +} + +my %JSON_ESCAPE = ( + '"' => '"', '\\' => '\\', '/' => '/', + b => "\b", f => "\f", n => "\n", r => "\r", t => "\t", +); + +sub json_str { + my ($s) = @_; + $s =~ s{\\(["\\/bfnrt]|u[0-9a-fA-F]{4})}{$JSON_ESCAPE{$1} // chr(hex(substr($1, 1)))}ge; + return $s; +} + +sub json_field { + my ($line, $key) = @_; + return undef unless $line =~ /"\Q$key\E": "((?:[^"\\]|\\.)*)"/; + return json_str($1); +} + +sub run_srcml { + my ($self, $language, $inputName, $workDir, $xmlPath) = @_; + pipe(my $errRead, my $errWrite) or die "unable to create stderr pipe: $!\n"; + + my $pid = fork(); + die "unable to fork srcml: $!\n" if not defined $pid; + if ($pid == 0) { + close($errRead); + open(STDIN, '<', '/dev/null'); + open(STDOUT, '>', '/dev/null'); + open(STDERR, '>&', $errWrite) or _exit(255); + close($errWrite); + setpgrp(0, 0); + chdir($workDir) or do { print STDERR "unable to enter [$workDir]: $!\n"; _exit(255) }; + { no warnings 'exec'; exec $self->{srcml}, "-l", $language, "--position", $inputName, "-o", $xmlPath; } + print STDERR "unable to execute srcml [$self->{srcml}]: $!\n"; + _exit(255); + } + close($errWrite); + $self->{child} = $pid; + + my $stderr = ""; + while (1) { + my $read = sysread($errRead, $stderr, 65536, length($stderr)); + if (not defined $read) { + next if $! == EINTR; + last; + } + last if $read == 0; + } + close($errRead); + waitpid($pid, 0); + my $status = $?; + $self->{child} = undef; + return ($status, $stderr); +} + +sub s2t { + my ($self) = @_; + return $self->{s2t} if $self->{s2t}; + $self->{s2t} = spawn_pipe([ $self->{srcml2token}, "--server" ]); + return $self->{s2t}; +} + +sub parse_xml { + my ($self, $xmlPath) = @_; + my $proc = $self->s2t(); + my $in = $proc->{in}; + print {$in} "PARSE $xmlPath\n"; + $in->flush; + + my (@tokens, $status); + my $stderr = ""; + while (1) { + my $line = readline($proc->{out}); + if (not defined $line) { + $self->retire(delete $self->{s2t}); + return (undef, \@tokens, $stderr); + } + if ($line =~ /\A\x01END ([0-9]+)\n\z/) { + $status = $1; + last; + } + if ($line =~ /\A\x01ERR (.*)\n\z/s) { + $stderr .= "$1\n"; + next; + } + chomp $line; + push @tokens, $line; + } + return ($status, \@tokens, $stderr); +} + +sub spawn_pipe { + my ($command) = @_; + pipe(my $childIn, my $parentOut) or die "unable to create pipe: $!\n"; + pipe(my $parentIn, my $childOut) or die "unable to create pipe: $!\n"; + my $err = File::Temp->new(TEMPLATE => "cregit-helper-err-XXXXX", TMPDIR => 1); + + my $pid = fork(); + die "unable to fork [$command->[0]]: $!\n" if not defined $pid; + if ($pid == 0) { + close($parentOut); + close($parentIn); + open(STDIN, '<&', $childIn) or _exit(255); + open(STDOUT, '>&', $childOut) or _exit(255); + open(STDERR, '>>', $err->filename) or _exit(255); + close($childIn); + close($childOut); + { no warnings 'exec'; exec @$command; } + print STDERR "unable to execute [$command->[0]]: $!\n"; + _exit(255); + } + close($childIn); + close($childOut); + binmode $parentOut; + binmode $parentIn; + return { pid => $pid, in => $parentOut, out => $parentIn, err => $err }; +} + +# A helper's stderr goes to a file, not a pipe: a full pipe would block the +# helper while the worker waits for its stdout. +sub take_helper_stderr { + my ($self) = @_; + my @procs = grep { $_ } values(%{ $self->{ctagsProc} }), $self->{s2t}; + my $text = ""; + for my $err (@{ delete $self->{retiredErr} // [] }, map { $_->{err} } @procs) { + seek($err, 0, 0); + local $/; + $text .= <$err> // ""; + truncate($err, 0); + } + return $text; +} + +sub retire { + my ($self, $proc) = @_; + return unless $proc; + push @{ $self->{retiredErr} }, $proc->{err}; + reap_pipe($proc); +} + +sub reap_pipe { + my ($proc) = @_; + return unless $proc; + close($proc->{in}) if defined fileno($proc->{in}); + close($proc->{out}) if defined fileno($proc->{out}); + kill 'TERM', $proc->{pid}; + waitpid($proc->{pid}, 0); +} + +sub kill_all { + my ($self) = @_; + if (my $pid = $self->{child}) { + kill 'TERM', -$pid; + select(undef, undef, undef, 0.1); + kill 'KILL', -$pid; + } + for my $proc (values %{ $self->{ctagsProc} }, $self->{s2t}) { + kill 'KILL', $proc->{pid} if $proc; + } +} + +sub shutdown { + my ($self) = @_; + reap_pipe(delete $self->{ctagsProc}{$_}) for keys %{ $self->{ctagsProc} }; + reap_pipe(delete $self->{s2t}); +} + +1; diff --git a/tokenize/srcMLtoken/srcml2token.cpp b/tokenize/srcMLtoken/srcml2token.cpp index a91f110e..60702dc6 100644 --- a/tokenize/srcMLtoken/srcml2token.cpp +++ b/tokenize/srcMLtoken/srcml2token.cpp @@ -22,6 +22,8 @@ #include #include "srcml2token.hpp" #include +#include +#include // --------------------------------------------------------------------------- // Local helper methods @@ -29,11 +31,71 @@ void usage() { XERCES_STD_QUALIFIER cout << "\nUsage:\n" - " srcml2token \n\n" + " srcml2token \n" + " srcml2token --server\n\n" "This program converts the output of srcML into a simplified tokenized version\n" + "--server reads `PARSE ` lines on stdin and answers each with the tokens,\n" + "then `\\x01ERR ` lines and a final `\\x01END ` line.\n" << XERCES_STD_QUALIFIER endl; } +static int parseOne(SAX2XMLReader* parser, const char* path) +{ + srcml2tokenResetState(); + srcml2tokenHandlers handler; + parser->setContentHandler(&handler); + parser->setErrorHandler(&handler); + try + { + if (path == NULL) { + StdInInputSource src; + parser->parse(src); + } else { + parser->parse(path); + } + } + catch (const OutOfMemoryException&) + { + *srcml2tokenDiag << "OutOfMemoryException" << XERCES_STD_QUALIFIER endl; + return 4; + } + catch (const XMLException& e) + { + *srcml2tokenDiag << "\nError during parsing: \n" + << StrX(e.getMessage()) + << "\n" << XERCES_STD_QUALIFIER endl; + return 4; + } + catch (...) + { + *srcml2tokenDiag << "Unexpected exception during parsing" << XERCES_STD_QUALIFIER endl; + return 4; + } + return 0; +} + +static int serverLoop(SAX2XMLReader* parser) +{ + std::string line; + while (std::getline(std::cin, line)) { + if (line.compare(0, 6, "PARSE ") != 0) { + std::cout << "\x01" "ERR malformed request [" << line << "]\n" "\x01" "END 2\n" << std::flush; + continue; + } + std::ostringstream diag; + srcml2tokenDiag = &diag; + int status = parseOne(parser, line.c_str() + 6); + srcml2tokenDiag = &std::cerr; + + std::istringstream lines(diag.str()); + std::string d; + while (std::getline(lines, d)) { + std::cout << "\x01" "ERR " << d << "\n"; + } + std::cout << "\x01" "END " << status << "\n" << std::flush; + } + return 0; +} // --------------------------------------------------------------------------- // Program entry point @@ -56,42 +118,11 @@ int main(int argC, char* argV[]) SAX2XMLReader* parser = XMLReaderFactory::createXMLReader(); - srcml2tokenHandlers handler; - parser->setContentHandler(&handler); - parser->setErrorHandler(&handler); - - int errorCount = 0; - // create a faux scope so that 'src' destructor is called before - // XMLPlatformUtils::Terminate - { - // - // Kick off the parse and catch any exceptions. Create a standard - // input input source and tell the parser to parse from that. - // - try - { - if (argC < 2) { - StdInInputSource src; - parser->parse(src); - } else { - parser->parse(argV[1]); - } - } - catch (const OutOfMemoryException&) - { - XERCES_STD_QUALIFIER cerr << "OutOfMemoryException" << XERCES_STD_QUALIFIER endl; - errorCount = 2; - return 4; - } - catch (const XMLException& e) - { - XERCES_STD_QUALIFIER cerr << "\nError during parsing: \n" - << StrX(e.getMessage()) - << "\n" << XERCES_STD_QUALIFIER endl; - errorCount = 1; - return 4; - } - + int status; + if (argC >= 2 && strcmp(argV[1], "--server") == 0) { + status = serverLoop(parser); + } else { + status = parseOne(parser, argC < 2 ? NULL : argV[1]); } // @@ -101,10 +132,5 @@ int main(int argC, char* argV[]) XMLPlatformUtils::Terminate(); - if (errorCount > 0) - return 4; - else - return 0; + return status; } - - diff --git a/tokenize/srcMLtoken/srcml2tokenHandlers.cpp b/tokenize/srcMLtoken/srcml2tokenHandlers.cpp index a6ba5884..153e1713 100644 --- a/tokenize/srcMLtoken/srcml2tokenHandlers.cpp +++ b/tokenize/srcMLtoken/srcml2tokenHandlers.cpp @@ -36,14 +36,25 @@ std::vector lineMarkers {0}; std::vector lineNumbers {1}; +std::ostream* srcml2tokenDiag = &std::cerr; + +void srcml2tokenResetState() { + lineMarkers.assign(1, 0); + lineNumbers.assign(1, 1); +} + +static std::string transcoded(const XMLCh* s) { + char* local = XMLString::transcode(s); + std::string result = local == NULL ? "" : local; + XMLString::release(&local); + return result; +} std::string get_attribute_value(const Attributes& attrs, std::string name) { XMLCh* revName= XMLString::transcode(name.c_str()); - char* st = XMLString::transcode(attrs.getValue(revName)); - std::string result = ""; - if (st != NULL) - result = st; + std::string result = transcoded(attrs.getValue(revName)); + XMLString::release(&revName); return result; } @@ -152,7 +163,7 @@ void srcml2tokenHandlers::startElement(const XMLCh* const //uri , const XMLCh* const qname , const Attributes& attrs) { - std::string tagLocal = XMLString::transcode(localname); + std::string tagLocal = transcoded(localname); // char *tagName = XMLString::transcode(qname); // std::string savePos = position(attrs); std::string savePos = newGetPosition(); @@ -205,7 +216,7 @@ void srcml2tokenHandlers::endElement (const XMLCh *const /*uri*/, const XMLCh *const localname, const XMLCh *const /*qname*/) { - char *tagName = XMLString::transcode(localname); + std::string tagName = transcoded(localname); // No escapes are legal here if (currentContent.length() > 0 ) { @@ -230,7 +241,7 @@ void srcml2tokenHandlers::endElement (const XMLCh *const /*uri*/, void srcml2tokenHandlers::characters( const XMLCh* const chars , const XMLSize_t length) { - std::string original = XMLString::transcode(chars); + std::string original = transcoded(chars); std::string st = mytrim(original); std::string node = mystack.top(); @@ -265,7 +276,7 @@ void srcml2tokenHandlers::startDocument() // --------------------------------------------------------------------------- void srcml2tokenHandlers::error(const SAXParseException& e) { - XERCES_STD_QUALIFIER cerr << "\nError at file " << StrX(e.getSystemId()) + *srcml2tokenDiag << "\nError at file " << StrX(e.getSystemId()) << ", line " << e.getLineNumber() << ", char " << e.getColumnNumber() << "\n Message: " << StrX(e.getMessage()) << XERCES_STD_QUALIFIER endl; @@ -273,7 +284,7 @@ void srcml2tokenHandlers::error(const SAXParseException& e) void srcml2tokenHandlers::fatalError(const SAXParseException& e) { - XERCES_STD_QUALIFIER cerr << "\nFatal Error at file " << StrX(e.getSystemId()) + *srcml2tokenDiag << "\nFatal Error at file " << StrX(e.getSystemId()) << ", line " << e.getLineNumber() << ", char " << e.getColumnNumber() << "\n Message: " << StrX(e.getMessage()) << XERCES_STD_QUALIFIER endl; @@ -281,7 +292,7 @@ void srcml2tokenHandlers::fatalError(const SAXParseException& e) void srcml2tokenHandlers::warning(const SAXParseException& e) { - XERCES_STD_QUALIFIER cerr << "\nWarning at file " << StrX(e.getSystemId()) + *srcml2tokenDiag << "\nWarning at file " << StrX(e.getSystemId()) << ", line " << e.getLineNumber() << ", char " << e.getColumnNumber() << "\n Message: " << StrX(e.getMessage()) << XERCES_STD_QUALIFIER endl; diff --git a/tokenize/srcMLtoken/srcml2tokenHandlers.hpp b/tokenize/srcMLtoken/srcml2tokenHandlers.hpp index 256f5178..70649942 100644 --- a/tokenize/srcMLtoken/srcml2tokenHandlers.hpp +++ b/tokenize/srcMLtoken/srcml2tokenHandlers.hpp @@ -20,6 +20,7 @@ // --------------------------------------------------------------------------- // Includes // --------------------------------------------------------------------------- +#include #include #include #include @@ -27,6 +28,9 @@ XERCES_CPP_NAMESPACE_USE +extern std::ostream* srcml2tokenDiag; +void srcml2tokenResetState(); + class srcml2tokenHandlers : public DefaultHandler { public: diff --git a/tokenize/tokenize.pl b/tokenize/tokenize.pl old mode 100644 new mode 100755 diff --git a/tokenizeByBlobId/WORKER_PROTOCOL.md b/tokenizeByBlobId/WORKER_PROTOCOL.md new file mode 100644 index 00000000..ddcfc70a --- /dev/null +++ b/tokenizeByBlobId/WORKER_PROTOCOL.md @@ -0,0 +1,76 @@ +# Persistent tokenizer worker protocol + +`tokenizeByBlobId/tokenWorker.pl` replaces the per-blob chain +`tokenBySha.pl → sh → tokenize.pl → tokenizeSrcMl.pl` with one long-lived perl +process per pool slot. With `--tokenizer-worker=` (pipelined modes only), +blobExec starts `parallelism` workers and talks to each over its stdin/stdout. +One request in flight per worker. + +The output contract is byte-identical to `tokenBySha.pl`: same tokens on +stdout, same memo file written at `$BFG_MEMO_DIR/xx/yy/`, same +exit-code meaning. A tokenized blob's id is a pure function of content, so the +worker must never change bytes. + +## Startup + +``` +tokenWorker.pl # no args; inherits blobExec's environment, see below +``` + +Worker writes `READY\n` to stdout once, then waits for requests. stderr is for +diagnostics only; blobExec forwards it. + +## Request (JVM → worker stdin) + +``` +REQ \n + +``` + +- `origSha`: hex id of the original blob (was `BFG_BLOB`). +- `filename`: basename with extension (was `BFG_FILENAME`). +- `fullPath`: repo-relative path (was `BFG_PATH`). +- Lengths are decimal byte counts. Strings are raw bytes, no escaping. +- `timeoutSecs`: per-request parser budget the worker must enforce itself. + +## Response (worker stdout → JVM) + +``` +RES \n + +``` + +- `exit`: what `tokenBySha.pl` would have exited with. `0` = tokens follow; + `33` = parser crash (`$PARSER_CRASH_EXIT`); `124` = the worker killed the + parser on timeout; any other non-zero = tokenizer error, stdout empty. +- On a non-zero exit nothing is memoized, exactly as `tokenBySha.pl` does. +- stderr holds the request's diagnostics, including what the persistent + helpers wrote during it. blobExec logs it only when the blob fails. + +## Timeout + +The worker runs the parser chain in its own process group (`setpgrp` in the +child) and enforces `timeoutSecs` with `alarm`; on expiry it `kill -TERM` then +`-KILL`s the group and answers `RES 124 0 `. blobExec keeps a backstop: +if no `RES` header arrives within `ChildRunner.maxLifetimeSeconds(timeoutSecs)` +(`timeoutSecs + 10`), it kills the worker and its children, starts a new one, +and reports the blob as a timeout (`Skip`, never `Replace`). + +## Lifecycle + +- EOF on stdin → worker exits 0. +- A malformed header → worker writes the reason to stderr and exits 2. blobExec + respawns and reports that blob as a Skip. +- Each worker keeps one `ctags --_interactive` per language and one + `srcml2token --server` alive between requests (`tokenize/CregitSrcMl.pm`), + and starts them again after a timeout or a crash. Only `srcml` is spawned per + blob. Request order is irrelevant. +- Languages that `tokenize.pl` does not send to `tokenizeSrcMl.pl`, and any + `BFG_TOKENIZE_CMD` the worker does not recognise, still spawn + `BFG_TOKENIZE_CMD` per blob. + +## Environment + +- `BFG_MEMO_DIR`, `BFG_TOKENIZE_CMD`: as for `tokenBySha.pl`. +- `BFG_WORKER_INPROC=0`: spawn `BFG_TOKENIZE_CMD` for each blob instead of + the in-process srcML chain (A/B measurement only). diff --git a/tokenizeByBlobId/tokenBySha.pl b/tokenizeByBlobId/tokenBySha.pl old mode 100644 new mode 100755 diff --git a/tokenizeByBlobId/tokenWorker.pl b/tokenizeByBlobId/tokenWorker.pl new file mode 100755 index 00000000..22aa4dfc --- /dev/null +++ b/tokenizeByBlobId/tokenWorker.pl @@ -0,0 +1,275 @@ +#!/usr/bin/env perl + +use strict; +use warnings; +use bytes; +use Digest::SHA qw(sha1_hex); +use Errno qw(EINTR); +use File::Copy qw(move); +use File::Path qw(make_path); +use File::Temp qw(tempfile); +use FindBin qw($RealBin); +use IO::Select; +use POSIX qw(_exit); +use File::Basename qw(basename dirname); +use lib "$RealBin/../tokenize"; +use CregitLanguages; +use CregitSrcMl; + +binmode STDIN; +binmode STDOUT; +$| = 1; +$SIG{PIPE} = 'IGNORE'; + +my $memoDir = $ENV{BFG_MEMO_DIR}; +die "You must define BFG_MEMO_DIR\n" if not defined($memoDir) or $memoDir eq ""; +die "Sha dir [$memoDir] does not exist\n" if not -d $memoDir; + +my $tokenizeCmd = $ENV{BFG_TOKENIZE_CMD}; +die "Tokenize command not defined. Use BFG_TOKENIZE_CMD\n" + if not defined($tokenizeCmd) or $tokenizeCmd eq ""; + +my @tokenizeCommand = grep { length($_) } split /\s+/, $tokenizeCmd; +my $buildDir = "$RealBin/build"; +make_path($buildDir) if not -d $buildDir; + +my $inProcess = in_process_tokenizer(@tokenizeCommand); + +print STDOUT "READY\n"; + +while (defined(my $header = )) { + if ($header !~ /\AREQ ([[:xdigit:]]+) ([0-9]+) ([0-9]+) ([0-9]+) ([0-9]+)\n\z/) { + protocol_error("malformed request header"); + } + + my ($origSha, $bodyLen, $fnLen, $pathLen, $timeoutSecs) = ($1, $2, $3, $4, $5); + my $filename = read_exact($fnLen, "filename"); + my $fullPath = read_exact($pathLen, "full path"); + my $contents = read_exact($bodyLen, "body"); + + my ($exitCode, $stdout, $stderr); + my $ok = eval { + ($exitCode, $stdout, $stderr) = process_request( + $origSha, $filename, $fullPath, $contents, $timeoutSecs + ); + 1; + }; + if (not $ok) { + $exitCode = 255; + $stdout = ""; + $stderr = $@ || "tokenizer request failed\n"; + } + + print STDOUT "RES $exitCode " . length($stdout) . " " . length($stderr) . "\n"; + print STDOUT $stdout, $stderr; +} + +$inProcess->shutdown() if $inProcess; +exit 0; + +sub in_process_tokenizer { + my ($program, @options) = @_; + return undef if ($ENV{BFG_WORKER_INPROC} // "1") eq "0"; + return undef unless defined $program; + my $name = basename($program); + return undef unless $name eq "tokenize.pl" or $name eq $CregitLanguages::SRCML_PARSER_REL; + + my %config = ( + srcml => "srcml", + srcml2token => dirname($program) . "/srcMLtoken/srcml2token", + ctags => "ctags-universal", + position => 0, + ); + for my $option (@options) { + if ($option =~ /\A--srcml=(.+)\z/) { $config{srcml} = $1 } + elsif ($option =~ /\A--srcml2token=(.+)\z/) { $config{srcml2token} = $1 } + elsif ($option =~ /\A--ctags=(.+)\z/) { $config{ctags} = $1 } + elsif ($option eq "--position") { $config{position} = 1 } + else { return undef } + } + return CregitSrcMl->new(%config); +} + +sub srcml_language { + my ($language) = @_; + my $parser = $CregitLanguages::LANG_PARSER_REL{$language}; + return (defined $parser and $parser eq $CregitLanguages::SRCML_PARSER_REL); +} + +sub read_exact { + my ($length, $label) = @_; + my $value = ""; + + while (length($value) < $length) { + my $read = read(STDIN, $value, $length - length($value), length($value)); + protocol_error("unable to read $label: $!") if not defined($read); + protocol_error("unexpected EOF while reading $label") if $read == 0; + } + + return $value; +} + +sub protocol_error { + my ($message) = @_; + print STDERR "$message\n"; + exit 2; +} + +sub process_request { + my ($origSha, $filename, $fullPath, $contents, $timeoutSecs) = @_; + + my ($ext) = $filename =~ /\.([^.]+)\z/; + my $language = defined($ext) ? CregitLanguages::language_for_ext($ext) : undef; + return (255, "", "unknown file extension [" . lc($ext // "") . "]\n") if not defined($language); + + my $sha1 = sha1_hex($contents); + my $memoFile = "$memoDir/" . substr($sha1, 0, 2) . "/" . substr($sha1, 2, 2) . "/$sha1"; + return (0, read_memo($memoFile), "") if -f $memoFile; + + my $tempDir = File::Temp->newdir( + "tokdir-XXXXX", + DIR => $buildDir, + CLEANUP => 1, + ); + my $inputName = "input." . lc($ext); + write_input("$tempDir/$inputName", $contents); + + my ($exitCode, $output, $error) = tokenize_input($language, $inputName, "$tempDir", $timeoutSecs); + return ($exitCode, "", $error) if $exitCode != 0; + + write_memo($memoFile, $output); + return (0, $output, $error); +} + +sub tokenize_input { + my ($language, $inputName, $workDir, $timeoutSecs) = @_; + return $inProcess->tokenize($language, $inputName, $workDir, $timeoutSecs) + if $inProcess and srcml_language($language); + return run_command($workDir, $timeoutSecs, @tokenizeCommand, "--language=$language", $inputName); +} + +sub read_memo { + my ($memoFile) = @_; + open(my $memo, '<:raw', $memoFile) + or die "unable to open memoized file [$memoFile]: $!\n"; + local $/; + my $output = <$memo>; + close($memo) or die "unable to close memoized file [$memoFile]: $!\n"; + return $output; +} + +sub write_input { + my ($inputFile, $contents) = @_; + open(my $input, '>:raw', $inputFile) + or die "unable to write temp input [$inputFile]: $!\n"; + print {$input} $contents; + close($input) or die "unable to close temp input [$inputFile]: $!\n"; +} + +sub write_memo { + my ($memoFile, $output) = @_; + my $dir = dirname($memoFile); + make_path($dir) if not -d $dir; + my ($memoOut, $tempOutput) = tempfile( + "tmpfile-out-XXXXX", + DIR => $buildDir, + UNLINK => 0, + ); + binmode $memoOut; + print {$memoOut} $output; + close($memoOut) or die "unable to close tokenizer output [$tempOutput]: $!\n"; + move($tempOutput, $memoFile) + or die "unable to move tokenizer output to [$memoFile]: $!\n"; +} + +sub run_command { + my ($workingDir, $timeoutSecs, @command) = @_; + pipe(my $stdoutRead, my $stdoutWrite) or die "unable to create stdout pipe: $!\n"; + pipe(my $stderrRead, my $stderrWrite) or die "unable to create stderr pipe: $!\n"; + + my $pid = fork(); + die "unable to fork tokenizer: $!\n" if not defined($pid); + + if ($pid == 0) { + close($stdoutRead); + close($stderrRead); + exec_in_group($workingDir, $stdoutWrite, $stderrWrite, @command); + } + + close($stdoutWrite); + close($stderrWrite); + + my $timedOut = 0; + local $SIG{ALRM} = sub { + $timedOut = 1; + kill 'TERM', -$pid; + select(undef, undef, undef, 0.1); + kill 'KILL', -$pid; + }; + alarm($timeoutSecs); + my ($stdout, $stderr) = read_streams($stdoutRead, $stderrRead); + waitpid($pid, 0); + my $status = $?; + alarm(0); + + if ($timedOut) { + $stderr .= "tokenize command timed out after [$timeoutSecs] seconds\n"; + return (124, "", $stderr); + } + + if ($status == -1) { + return (255, "", $stderr . "unable to wait for tokenizer: $!\n"); + } + my $signal = $status & 127; + if ($signal) { + return (128 + $signal, "", $stderr . "tokenizer killed by signal [$signal]\n"); + } + + return ($status >> 8, $stdout, $stderr); +} + +sub exec_in_group { + my ($workingDir, $stdoutWrite, $stderrWrite, @command) = @_; + open(STDOUT, '>&', $stdoutWrite) or _exit(255); + open(STDERR, '>&', $stderrWrite) or _exit(255); + binmode STDOUT; + binmode STDERR; + close($stdoutWrite); + close($stderrWrite); + setpgrp(0, 0) or do { + print STDERR "unable to create tokenizer process group: $!\n"; + _exit(255); + }; + chdir($workingDir) or do { + print STDERR "unable to enter tokenizer directory [$workingDir]: $!\n"; + _exit(255); + }; + { no warnings 'exec'; exec {$command[0]} @command; } + print STDERR "unable to execute tokenizer [$command[0]]: $!\n"; + _exit(255); +} + +sub read_streams { + my ($stdoutRead, $stderrRead) = @_; + binmode $stdoutRead; + binmode $stderrRead; + my %stream = (fileno($stdoutRead) => "", fileno($stderrRead) => ""); + my @order = (fileno($stdoutRead), fileno($stderrRead)); + my $selector = IO::Select->new($stdoutRead, $stderrRead); + while ($selector->count()) { + for my $handle ($selector->can_read()) { + my $read = sysread($handle, my $chunk, 65536); + if (not defined($read)) { + next if $! == EINTR; + die "unable to read tokenizer output: $!\n"; + } + if ($read == 0) { + $selector->remove($handle); + close($handle); + next; + } + $stream{fileno($handle)} .= $chunk; + } + } + return @stream{@order}; +}