diff --git a/gortools/src/main/java/gorsat/BatchedPipeStepIteratorAdaptor.java b/gortools/src/main/java/gorsat/BatchedPipeStepIteratorAdaptor.java index d543d3993..3198379a4 100644 --- a/gortools/src/main/java/gorsat/BatchedPipeStepIteratorAdaptor.java +++ b/gortools/src/main/java/gorsat/BatchedPipeStepIteratorAdaptor.java @@ -23,6 +23,7 @@ package gorsat; import gorsat.Commands.Analysis; +import org.gorpipe.exceptions.GorCancelledException; import org.gorpipe.exceptions.GorException; import org.gorpipe.exceptions.GorSystemException; import org.gorpipe.gor.model.GenomicIterator; @@ -218,8 +219,9 @@ public boolean hasNext() { } return ret; } catch (InterruptedException e) { + // Must not look like end of stream: callers would treat the truncated output as complete (ENGKNOW-3979). Thread.currentThread().interrupt(); - return false; + throw new GorCancelledException("Query interrupted", e); } } diff --git a/gortools/src/main/java/gorsat/process/ParallelExecutor.java b/gortools/src/main/java/gorsat/process/ParallelExecutor.java index 2e2de416a..de69a5101 100644 --- a/gortools/src/main/java/gorsat/process/ParallelExecutor.java +++ b/gortools/src/main/java/gorsat/process/ParallelExecutor.java @@ -27,6 +27,7 @@ import java.util.Arrays; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; /** * This class encapsulates a general execution in parallel of the pgor command when a @@ -37,11 +38,21 @@ public class ParallelExecutor { private Throwable firstException; private final Thread[] threads; private final Function0[] commands; + private final AtomicBoolean cancelled; public ParallelExecutor(int workers, Function0[] commands) { + this(workers, commands, new AtomicBoolean(false)); + } + + /** + * @param cancelled set when any command fails, before the other workers are interrupted. Commands can check + * it to avoid committing partial results, since the interrupt flag alone is easily lost. + */ + public ParallelExecutor(int workers, Function0[] commands, AtomicBoolean cancelled) { this.commands = commands; this.threads = new Thread[workers]; this.firstException = null; + this.cancelled = cancelled; } @SuppressWarnings("squid:S00112") // We need to handle Throwable here, sorry @@ -50,7 +61,7 @@ public void parallelExecute() throws Throwable { for( int i = 0; i < threads.length; i++ ) { Thread t = new Thread(() -> { Function0 func = clq.poll(); - while( func != null ) { + while( func != null && !cancelled.get() ) { func.apply(); func = clq.poll(); } @@ -70,6 +81,7 @@ public void parallelExecute() throws Throwable { private synchronized void parallelExcecuteUncaughtExceptionHandler(Thread thread, Throwable throwable) { if (firstException == null) { firstException = throwable; + cancelled.set(true); for (Thread t : threads) { if (t != thread) { t.interrupt(); diff --git a/gortools/src/main/scala/gorsat/QueryHandlers/GeneralQueryHandler.scala b/gortools/src/main/scala/gorsat/QueryHandlers/GeneralQueryHandler.scala index 258e39ae0..340419c4b 100644 --- a/gortools/src/main/scala/gorsat/QueryHandlers/GeneralQueryHandler.scala +++ b/gortools/src/main/scala/gorsat/QueryHandlers/GeneralQueryHandler.scala @@ -35,7 +35,7 @@ import gorsat.Utilities.{AnalysisUtilities, MacroUtilities} import gorsat.process.{GorJavaUtilities, ParallelExecutor} import org.apache.commons.io.FilenameUtils import org.gorpipe.client.FileCache -import org.gorpipe.exceptions.{GorException, GorSystemException, GorUserException} +import org.gorpipe.exceptions.{GorCancelledException, GorException, GorSystemException, GorUserException} import org.gorpipe.gor.binsearch.GorIndexType import org.gorpipe.gor.driver.meta.DataType import org.gorpipe.gor.model.{DriverBackedFileReader, FileReader, GorMeta, GorOptions, GorParallelQueryHandler} @@ -48,6 +48,8 @@ import org.gorpipe.gor.util.DataUtil import org.slf4j.LoggerFactory import java.util.Optional +import java.util.concurrent.atomic.AtomicBoolean +import java.util.function.BooleanSupplier import scala.jdk.CollectionConverters.IteratorHasAsScala class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParallelQueryHandler { @@ -67,7 +69,8 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal (linkCacheFilePath, extension) } - def runAndStoreLinkFileInCache(nested: GorContext, writeLocationPath: String, fileCache: FileCache, useMd5: Boolean): String = { + def runAndStoreLinkFileInCache(nested: GorContext, writeLocationPath: String, fileCache: FileCache, useMd5: Boolean, + isCancelled: BooleanSupplier = GeneralQueryHandler.NotCancelled): String = { val startTime = System.currentTimeMillis val fileReader = nested.getSession.getProjectContext.getFileReader val commandToExecute = nested.getCommand @@ -76,7 +79,7 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal val noDict = commandToExecute.toLowerCase.contains(" -nodict ") val writeGord = isGord && !noDict var cacheRes = writeLocationPath - val resultFileName = runCommand(nested, commandToExecute, if (isGord) writeLocationPath else null, useMd5, theTheDict = true) + val resultFileName = runCommand(nested, commandToExecute, if (isGord) writeLocationPath else null, useMd5, theTheDict = true, isCancelled) val isCacheDir = fileReader.resolveUrl(writeLocationPath,true).isDirectory() if(fileCache != null && (!isCacheDir || writeGord)) { @@ -89,12 +92,13 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal cacheRes } - def runAndStoreInCache(nested: GorContext, fileCache: FileCache, useMd5: Boolean): String = { + def runAndStoreInCache(nested: GorContext, fileCache: FileCache, useMd5: Boolean, + isCancelled: BooleanSupplier = GeneralQueryHandler.NotCancelled): String = { val startTime = System.currentTimeMillis val commandToExecute = nested.getCommand val commandSignature = nested.getSignature var cacheFile = findCacheFile(commandSignature, commandToExecute, header, fileCache, AnalysisUtilities.theCacheDirectory(context.getSession)) - val resultFileName = runCommand(nested, commandToExecute, cacheFile, useMd5, theTheDict = false) + val resultFileName = runCommand(nested, commandToExecute, cacheFile, useMd5, theTheDict = false, isCancelled) if (fileCache != null) { val extension = CommandParseUtilities.getExtensionForQuery(commandToExecute, header) val overheadTime = findOverheadTime(commandToExecute) @@ -135,6 +139,9 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal val fileReader = context.getSession.getProjectContext.getFileReader var commandList: List[() => Unit] = Nil val useMd5 = System.getProperty("gor.caching.md5.enabled", "false").toBoolean + // Set by ParallelExecutor when any part fails, so the other parts never commit partial output (ENGKNOW-3979) + val cancelled = new AtomicBoolean(false) + val isCancelled: BooleanSupplier = () => cancelled.get() for (i <- commandSignatures.indices) { val executeFunction = block2Function { @@ -149,9 +156,9 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal fileNames(i) = if (cacheFile == null || !fileReader.exists(cacheFile)) { val writeLocationPath = cacheFiles(i) if (writeLocationPath != null) { - runAndStoreLinkFileInCache(nested, writeLocationPath, fileCache, useMd5) + runAndStoreLinkFileInCache(nested, writeLocationPath, fileCache, useMd5, isCancelled) } else { - runAndStoreInCache(nested, fileCache, useMd5) + runAndStoreInCache(nested, fileCache, useMd5, isCancelled) } } else { generateDictionaryFile(commandToExecute, fileReader, useMd5, cacheFile) @@ -168,13 +175,13 @@ class GeneralQueryHandler(context: GorContext, header: Boolean) extends GorParal commandList ::= executeFunction } - if (commandList != Nil) parallelExecution(commandList.reverse.toArray) + if (commandList != Nil) parallelExecution(commandList.reverse.toArray, cancelled) fileNames } - def parallelExecution(commands: Array[() => Unit]): Unit = { - val pe = new ParallelExecutor(context.getSession.getSystemContext.getWorkers, commands) + def parallelExecution(commands: Array[() => Unit], cancelled: AtomicBoolean = new AtomicBoolean(false)): Unit = { + val pe = new ParallelExecutor(context.getSession.getSystemContext.getWorkers, commands, cancelled) try pe.parallelExecute() catch { @@ -210,24 +217,50 @@ object GeneralQueryHandler { CommandParseUtilities.getExtensionForQuery(commandToExecute, header)) } - def runCommand(context: GorContext, commandToExecute: String, outfile: String, useMd5: Boolean, theTheDict: Boolean): String = { + private val NotCancelled: BooleanSupplier = () => false + + /** + * True if the output must not be committed: the run was cancelled (e.g. a sibling parallel part failed) or the + * thread was interrupted. A cancelled source may end early without an exception, so its output is partial. + */ + private def mustNotCommit(isCancelled: BooleanSupplier): Boolean = + isCancelled.getAsBoolean || Thread.currentThread().isInterrupted + + def runCommand(context: GorContext, commandToExecute: String, outfile: String, useMd5: Boolean, theTheDict: Boolean, + isCancelled: BooleanSupplier = NotCancelled): String = { context.start(outfile) // We are using absolute paths here val fileReader = context.getSession.getProjectContext.getSystemFileReader val result = if (commandToExecute.toUpperCase().startsWith(CommandParseUtilities.GOR_DICTIONARY_PART) || commandToExecute.toUpperCase().startsWith(CommandParseUtilities.GOR_DICTIONARY_FOLDER_PART)) { - writeOutGorDictionaryPart(commandToExecute, fileReader, outfile, theTheDict) + checkDictionaryCommit(writeOutGorDictionaryPart(commandToExecute, fileReader, outfile, theTheDict), fileReader, commandToExecute, isCancelled) } else if (commandToExecute.toUpperCase().startsWith(CommandParseUtilities.GOR_DICTIONARY)) { - writeOutGorDictionary(commandToExecute, fileReader, outfile, theTheDict) + checkDictionaryCommit(writeOutGorDictionary(commandToExecute, fileReader, outfile, theTheDict), fileReader, commandToExecute, isCancelled) } else if (commandToExecute.toUpperCase().startsWith(CommandParseUtilities.NOR_DICTIONARY)) { - writeOutNorDictionaryPart(commandToExecute, fileReader, outfile) + checkDictionaryCommit(writeOutNorDictionaryPart(commandToExecute, fileReader, outfile), fileReader, commandToExecute, isCancelled) } else { - runCommandInternal(context, commandToExecute, outfile, useMd5) + runCommandInternal(context, commandToExecute, outfile, useMd5, isCancelled) } context.end() result } - private def runCommandInternal(context: GorContext, commandToExecute: String, outfile: String, useMd5: Boolean): String = { + /** + * Dictionaries are written in place, so on cancel remove a possibly partial dictionary file to keep it from + * being picked up from the cache (ENGKNOW-3979). + */ + private def checkDictionaryCommit(outfile: String, fileReader: FileReader, commandToExecute: String, isCancelled: BooleanSupplier): String = { + if (mustNotCommit(isCancelled)) { + try { + if (outfile != null && fileReader.exists(outfile) && !fileReader.isDirectory(outfile)) fileReader.delete(outfile) + } catch { + case _: Exception => /* do nothing */ + } + throw new GorCancelledException(s"Query cancelled, result not stored: $commandToExecute", null) + } + outfile + } + + private def runCommandInternal(context: GorContext, commandToExecute: String, outfile: String, useMd5: Boolean, isCancelled: BooleanSupplier): String = { val theSource = new DynamicRowSource(commandToExecute, context) val theHeader = theSource.getHeader @@ -294,6 +327,12 @@ object GeneralQueryHandler { } } + // A cancelled run (e.g. a sibling parallel part failed) may have ended its source early without an + // exception. Never commit that partial output to the cache; the catch below removes the temp file (ENGKNOW-3979). + if (mustNotCommit(isCancelled)) { + throw new GorCancelledException(s"Query cancelled, result not stored: $commandToExecute", null) + } + if(oldName!=null && fileReader.exists(oldName) && !oldName.equals(newName)) { fileReader.move(oldName, newName) val oldMetaName = DataUtil.toFile(oldName, DataType.META) diff --git a/gortools/src/test/java/gorsat/QueryHandlers/UTestGeneralQueryHandler.java b/gortools/src/test/java/gorsat/QueryHandlers/UTestGeneralQueryHandler.java new file mode 100644 index 000000000..dab3abfff --- /dev/null +++ b/gortools/src/test/java/gorsat/QueryHandlers/UTestGeneralQueryHandler.java @@ -0,0 +1,92 @@ +package gorsat.QueryHandlers; + +import gorsat.DynIterator; +import gorsat.process.GorInputSources; +import gorsat.process.GorPipeCommands; +import gorsat.process.PipeInstance; +import gorsat.process.PipeOptions; +import gorsat.process.TestSessionFactory; +import org.gorpipe.exceptions.GorCancelledException; +import org.gorpipe.gor.session.GorSession; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.stream.Stream; + +/** + * ENGKNOW-3979: output of a cancelled parallel part must never be committed, whether or not the part's + * thread still has its interrupt flag set. + */ +public class UTestGeneralQueryHandler { + + @Rule + public TemporaryFolder workDir = new TemporaryFolder(); + + private Path workPath; + private GorSession session; + + @Before + public void setUp() { + GorPipeCommands.register(); + GorInputSources.register(); + DynIterator.createGorIterator_$eq(PipeInstance::createGorIterator); + workPath = workDir.getRoot().toPath(); + var options = new PipeOptions(); + options.gorRoot_$eq(workPath.toString()); + options.cacheDir_$eq(workPath.resolve("cache").toString()); + options.requestId_$eq("test"); + session = new TestSessionFactory(options, null, false, null, null).create(); + } + + @After + public void tearDown() { + if (session != null) session.close(); + } + + @Test + public void cancelledQueryIsNotCommitted() throws IOException { + var outfile = workPath.resolve("out.gor"); + + Assert.assertThrows(GorCancelledException.class, () -> GeneralQueryHandler.runCommand( + session.getGorContext(), "gorrows -p chr1:1-100", outfile.toString(), false, false, () -> true)); + + Assert.assertFalse("Cancelled output must not be committed", Files.exists(outfile)); + assertNoFilesLeft(); + } + + @Test + public void cancelledDictionaryIsNotCommitted() throws IOException { + Files.writeString(workPath.resolve("a.tsv"), "#col\n1\n"); + Files.writeString(workPath.resolve("b.tsv"), "#col\n2\n"); + var outfile = workPath.resolve("out.nord"); + + Assert.assertThrows(GorCancelledException.class, () -> GeneralQueryHandler.runCommand( + session.getGorContext(), "NORDICT a.tsv a b.tsv b", outfile.toString(), false, false, () -> true)); + + Assert.assertFalse("Cancelled dictionary must not be committed", Files.exists(outfile)); + } + + @Test + public void notCancelledQueryIsCommitted() { + var outfile = workPath.resolve("out.gor"); + + GeneralQueryHandler.runCommand( + session.getGorContext(), "gorrows -p chr1:1-100", outfile.toString(), false, false, () -> false); + + Assert.assertTrue(Files.exists(outfile)); + } + + private void assertNoFilesLeft() throws IOException { + try (Stream files = Files.list(workPath)) { + var left = files.filter(p -> p.getFileName().toString().startsWith("out")).toList(); + Assert.assertTrue("Temp output must be removed, found: " + left, left.isEmpty()); + } + } +} diff --git a/gortools/src/test/java/gorsat/Script/UTestSignature.java b/gortools/src/test/java/gorsat/Script/UTestSignature.java index 2355e622f..ab7fd90f6 100644 --- a/gortools/src/test/java/gorsat/Script/UTestSignature.java +++ b/gortools/src/test/java/gorsat/Script/UTestSignature.java @@ -228,6 +228,7 @@ public void testSignatureWithVersionedLinkFile() throws IOException, Interrupted } Files.writeString(dataPath1, "chr1\t2\n", StandardOpenOption.APPEND); + Files.setLastModifiedTime(dataPath1, FileTime.fromMillis(Files.getLastModifiedTime(dataPath1).toMillis() + 1000)); try (var session = factory.create()) { var engine = ScriptEngineFactory.create(session.getGorContext()); diff --git a/gortools/src/test/java/gorsat/UTestParallel.java b/gortools/src/test/java/gorsat/UTestParallel.java index 558bc26c5..dd37b339b 100644 --- a/gortools/src/test/java/gorsat/UTestParallel.java +++ b/gortools/src/test/java/gorsat/UTestParallel.java @@ -22,12 +22,23 @@ package gorsat; +import org.gorpipe.exceptions.GorException; import org.gorpipe.exceptions.GorParsingException; import org.junit.Assert; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; public class UTestParallel { + @Rule + public TemporaryFolder workDir = new TemporaryFolder(); + @Test public void testNorParallelQuery() { String query = "create pnlist = norrows 100 -offset 100 | calc pn 'PN_'+rownum | signature -timeres 1; parallel -parts [pnlist] <(norrows 100 -offset #{col:rownum} | calc pn '#{col:pn}')"; @@ -79,4 +90,32 @@ public void testGorParallelQueryExceedingLimit() { String query = "create pnlist = norrows 100 -offset 100 | calc pn 'PN_'+rownum | signature -timeres 1; parallel -parts [pnlist] -limit 10 <(gorrows -p chr1:0-#{col:rownum} | calc pn '#{col:pn}')"; TestUtils.runGorPipeCount(query); } + + /** + * ENGKNOW-3979: when one part of a parallel create fails, a sibling part that is still running gets + * interrupted. Its partial (header-only) output must not be committed to the result cache, otherwise + * a rerun after the failure is fixed reuses the empty part and silently drops its rows. + */ + @Test + public void testFailedPartDoesNotLeaveSiblingPartInCache() throws IOException { + Path root = workDir.getRoot().toPath(); + Path cacheDir = Files.createDirectory(root.resolve("result_cache")); + Path failingPart = root.resolve("a.gor"); + Files.writeString(failingPart, "chrom\tpos\tval\tdelay\nchr1\t1\tbad\t0\n"); + Files.writeString(root.resolve("b.gor"), + "chrom\tpos\tval\tdelay\nchr2\t1\tok\t500\nchr2\t2\tok\t500\nchr2\t3\tok\t500\n"); + Files.writeString(root.resolve("parts.tsv"), "#name\na\nb\n"); + + String query = "create xx = parallel -parts parts.tsv <(gor #{col:name}.gor | calc s sleep(delay) | throwif val = 'bad'); gor [xx]"; + + Assert.assertThrows(GorException.class, + () -> TestUtils.runGorPipe(query, root.toString(), cacheDir.toString(), false, null, null)); + + Files.writeString(failingPart, "chrom\tpos\tval\tdelay\nchr1\t1\tok\t0\n"); + Files.setLastModifiedTime(failingPart, FileTime.fromMillis(System.currentTimeMillis() + 10000)); + + String result = TestUtils.runGorPipe(query, root.toString(), cacheDir.toString(), false, null, null); + Assert.assertEquals("All rows from all parts expected on rerun, got:\n" + result, + 5, result.split("\n").length); + } } diff --git a/gortools/src/test/java/gorsat/process/UTestParallelExecutor.java b/gortools/src/test/java/gorsat/process/UTestParallelExecutor.java new file mode 100644 index 000000000..b0dce55dc --- /dev/null +++ b/gortools/src/test/java/gorsat/process/UTestParallelExecutor.java @@ -0,0 +1,58 @@ +package gorsat.process; + +import org.junit.Assert; +import org.junit.Test; +import scala.Function0; +import scala.Unit; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +public class UTestParallelExecutor { + + /** + * ENGKNOW-3979: a failed command must cancel the run through an explicit flag, not only by interrupting + * siblings. A sibling that loses its interrupt flag (e.g. code that calls Thread.interrupted() or swallows + * InterruptedException) must still see the run as cancelled, and queued commands must not be started. + */ + @Test + @SuppressWarnings("unchecked") + public void failedCommandCancelsRunEvenIfSiblingClearsInterrupt() { + var cancelled = new AtomicBoolean(false); + var siblingStarted = new CountDownLatch(1); + var siblingSawCancel = new AtomicBoolean(false); + var queuedRan = new AtomicBoolean(false); + + Function0 sibling = () -> { + siblingStarted.countDown(); + long deadline = System.currentTimeMillis() + 10000; + // Thread.interrupted() clears the flag, like code that swallows the interrupt + while (!Thread.interrupted() && System.currentTimeMillis() < deadline) { + Thread.onSpinWait(); + } + siblingSawCancel.set(cancelled.get()); + return null; + }; + Function0 failing = () -> { + try { + siblingStarted.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + throw new IllegalStateException("part failed"); + }; + Function0 queued = () -> { + queuedRan.set(true); + return null; + }; + + var pe = new ParallelExecutor(2, new Function0[]{sibling, failing, queued}, cancelled); + + var ex = Assert.assertThrows(IllegalStateException.class, pe::parallelExecute); + Assert.assertEquals("part failed", ex.getMessage()); + Assert.assertTrue("Run must be marked cancelled", cancelled.get()); + Assert.assertTrue("Sibling must see cancel flag after its interrupt flag is cleared", siblingSawCancel.get()); + Assert.assertFalse("Queued command must not start after a failure", queuedRan.get()); + } +}