Merge pull request #395 from AdamaJava/qsnp_unfiltered_read_bug

fix(qsnp): OOM caused by large number of failed filter reads at same location

Author: holmeso

qsnp/src/org/qcmg/snp/Pipeline.java

@@ -28,7 +28,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
@@ -450,7 +449,7 @@ private void checkHeadersSortOrder(List<SAMFileHeader> headers, boolean isNormal
 	}
 
 	List<SAMSequenceDictionary> getSequenceDictionaries(List<SAMFileHeader> headers) {
-		final List<SAMSequenceDictionary> seqDictionaries = new ArrayList<SAMSequenceDictionary>();
+		final List<SAMSequenceDictionary> seqDictionaries = new ArrayList<>();
 		for (final SAMFileHeader header : headers) {
 			seqDictionaries.add(header.getSequenceDictionary());
 		}
@@ -485,7 +484,7 @@ private void checkSequenceDictionaries(List<SAMFileHeader> normalHeaders, List<S
 
 
 		// pick the first normal sequence to check against fasta
-		final List<SAMSequenceRecord> normalSequences = normalSeqDictionaries.get(0).getSequences();
+		final List<SAMSequenceRecord> normalSequences = normalSeqDictionaries.getFirst().getSequences();
 
 		// now check against the supplied reference file
 		final FastaSequenceFile ref = new FastaSequenceFile(new File(referenceFile), true);
@@ -633,7 +632,7 @@ void walkBams() throws Exception {
 	/**
 	 * Sets up 2 Producer threads, 2 Consumer threads and a Cleaner thread, along with the concurrent collections, queues, and barriers used by them all
 	 * 
-	 * @param ignoreDuplicates indicates whether duplicate records should be discarded out right. Not useful for torrent mode
+	 * @param includeDups indicates whether duplicate records should be discarded out right. Not useful for torrent mode
 	 * @throws Exception
 	 */
 	void walkBams(boolean includeDups) throws Exception {
@@ -736,7 +735,10 @@ public class Producer implements Runnable {
 		private XXHash64 xxhash64;
 		private final static int seed = 0x9747b28c; // used to initialize the hash value, use whatever value you want, but always the same
 		private final static int ONE_MILLION = 1_000_000;
-		
+		private int failedFilterStartPosition = 0;
+		private int failedFilterCountAtStartPosition = 0;
+		private static final int MAX_FAILED_RECORDS_PER_POSITION = 1000;
+
 		public Producer(final String[] bamFiles, final CountDownLatch latch, final boolean isNormal, 
 				final Queue<SAMRecordFilterWrapper> samQueue, final Thread mainThread, final String query, 
 				final CyclicBarrier barrier, boolean includeDups, Accumulator [] accum) throws Exception {
@@ -874,7 +876,24 @@ private void processRecord(SAMRecord record) throws Exception {
 				/*
 				 * we now want to keep track of reads that don't pass the filter for test as well as control
 				 */
-				addRecordToQueue(record, passesFilter);
+				if ( ! passesFilter) {
+					/*
+					There are instances in the genome where there are a large number of poor quality reads mapping to the same location.
+					This is causing memory issues when trying to process these reads, and so will ignore these reads once we have seen 100 reads at the same start position that fail the filter.
+					 */
+					int start = record.getAlignmentStart();
+					if (start == failedFilterStartPosition) {
+						failedFilterCountAtStartPosition++;
+					} else {
+						failedFilterCountAtStartPosition = 1;
+						failedFilterStartPosition = start;
+					}
+					if (failedFilterCountAtStartPosition <= MAX_FAILED_RECORDS_PER_POSITION) {
+						addRecordToQueue(record, false);
+					}
+				} else {
+					addRecordToQueue(record, true);
+				}
 			} else {
 				// didn't have any filtering defined - add all
 				addRecordToQueue(record, true);
@@ -986,7 +1005,7 @@ public void processSAMRecord(final SAMRecordFilterWrapper record) {
 		 * @param length
 		 * @param referenceOffset
 		 * @param passesFilter
-		 * @param readStartPosition start position of the read - depends on strand as to whether this is the alignemtnEnd or alignmentStart
+		 * @param readStartPosition start position of the read - depends on strand as to whether this is the alignmentEnd or alignmentStart
 		 */
 		public void updateMapWithAccums(int startPosition, final byte[] bases, final byte[] qualities,
 				boolean forwardStrand, int offset, int length, int referenceOffset, final boolean passesFilter, final int readEndPosition, long readNameHash) {
@@ -1275,10 +1294,10 @@ private void interrogateAccumulations(Accumulator control, Accumulator test) {
 		 */
 		if (null != control && null != test) {
 			if (control.getPosition() != test.getPosition()) {
-				throw new IllegalArgumentException("Control and test accumulator positions do not match!!! control: " + control.toString() + ", and test: " + test.toString());
+				throw new IllegalArgumentException("Control and test accumulator positions do not match!!! control: " + control + ", and test: " + test);
 			}
 		}
-		final int position = control != null ? control.getPosition() : test.getPosition();
+		final int position = control != null ? control.getPosition() : test != null ? test.getPosition() : Integer.MAX_VALUE;
 
 		// if we are over the length of this particular sequence - return
 		if (position - 1 >= referenceBasesLength) return;
@@ -1377,7 +1396,7 @@ private void interrogateAccumulations(Accumulator control, Accumulator test) {
 					/*
 					 * populate adjacentAccumulators so that compound snp decision can be made
 					 */
-					adjacentAccumulators.put(v, new Pair<Accumulator, Accumulator>(control, test));
+					adjacentAccumulators.put(v, new Pair<>(control, test));
 				}
 			}
 		}

qsnp/test/org/qcmg/snp/PipelineTest.java

@@ -1,12 +1,13 @@
 package org.qcmg.snp;
 
-import static org.junit.Assert.assertEquals;
-
 import java.io.IOException;
 import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -21,21 +22,22 @@
 import htsjdk.samtools.SAMFileHeader;
 import htsjdk.samtools.SAMRecord;
 
+import static org.junit.Assert.*;
+
 public class PipelineTest {
 
 	@Rule
 	public final TemporaryFolder testFolder = new TemporaryFolder();
-	
+
 	@Test
 	public void getHeader() {
 		VcfHeader h = Pipeline.getHeaderForQSnp("abc", "123", "456", "qsnp", null, null, "xyz");
-		assertEquals(true, null != h.getFormatRecord(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND));
-		assertEquals(true, h.getFormatRecord(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND).toString().contains(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND_DESC));
+        assertNotNull(h.getFormatRecord(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND));
+        assertTrue(h.getFormatRecord(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND).toString().contains(VcfHeaderUtils.FORMAT_OBSERVED_ALLELES_BY_STRAND_DESC));
 	}
-	
-	
+
 	@Test
-	public void getAccumulatorsFromReads() throws IOException {
+	public void getAccumulatorsFromReads() {
 		final Pipeline pipeline = new TestPipeline();
 
 		CountDownLatch consumerLatch = new CountDownLatch(1);

qsnp/test/org/qcmg/snp/StandardPipelineTest.java

@@ -12,19 +12,14 @@
 import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.qcmg.common.commandline.Executor;
-import org.qcmg.common.model.Accumulator;
 
 import htsjdk.samtools.SAMFileHeader;
 import htsjdk.samtools.SAMFileHeader.SortOrder;
@@ -96,9 +91,9 @@ public void testRunStandardMode() throws Exception{
 		final String command = "-log " + logFile.getAbsolutePath() + " -i " + ini.getAbsolutePath();
 		final Executor exec = new Executor(command, "org.qcmg.snp.Main");
 		assertEquals(0, exec.getErrCode());
-		List<String> vcfs = Files.lines(Paths.get(vcf.getPath())).filter(s -> ! s.startsWith("#")).collect(Collectors.toList());
+		List<String> vcfs = Files.lines(Paths.get(vcf.getPath())).filter(s -> ! s.startsWith("#")).toList();
 
-		vcfs.stream().forEach(System.out::println);
+		vcfs.forEach(System.out::println);
 
 		assertEquals(60, vcfs.size());
 		AtomicInteger ai = new AtomicInteger(0);
@@ -147,42 +142,18 @@ public void testNumbers() throws IOException, InterruptedException {
 		final String command = "-log " + logFile.getAbsolutePath() + " -i " + ini.getAbsolutePath();
 		final Executor exec = new Executor(command, "org.qcmg.snp.Main");
 		assertEquals(0, exec.getErrCode());
-		List<String> vcfs = Files.lines(Paths.get(vcf.getPath())).filter(s -> ! s.startsWith("#")).collect(Collectors.toList());
+		List<String> vcfs = Files.lines(Paths.get(vcf.getPath())).filter(s -> ! s.startsWith("#")).toList();
 
-		vcfs.stream().forEach(System.out::println);
+		vcfs.forEach(System.out::println);
 
 		assertEquals(4, vcfs.size());
 //		assertEquals("GL000208.1	53	.	T	G	.	.	FLANK=AAGTGGTTCAA	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	0/1:90,6:96:T1[]5[]:C1;G51;T89:.:.:6:G1[41]5[40.2];T9[35.22]81[39.68]	0/1:90,6:96:T1[]5[]:C1;G51;T89:.:.:6:G1[41]5[40.2];T9[35.22]81[39.68]", vcfs.get(0));
-		assertEquals("GL000208.1	53	.	T	G	.	.	FLANK=AAGTGGTTCAA	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	0/1:115,10:125:T1[]7[]:C1;G51;T89:.:.:9:G2[41]8[40.5];T9[35.22]106[39.88]	0/1:115,10:125:T1[]7[]:C1;G51;T89:.:.:9:G2[41]8[40.5];T9[35.22]106[39.88]", vcfs.get(0));
+		assertEquals("GL000208.1	53	.	T	G	.	.	FLANK=AAGTGGTTCAA	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	0/1:115,10:125:T1[]7[]:C1;G51;T89:.:.:9:G2[41]8[40.5];T9[35.22]106[39.88]	0/1:115,10:125:T1[]7[]:C1;G51;T89:.:.:9:G2[41]8[40.5];T9[35.22]106[39.88]", vcfs.getFirst());
 //		assertEquals("GL000208.1	77	.	T	C	.	.	FLANK=AAAGACGTATT	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	1/1:4,94:98:C0[]3[];T0[]1[]:C171;T11:.:.:64:C9[35.78]85[39.94];T1[41]3[39.67]	1/1:4,94:98:C0[]3[];T0[]1[]:C171;T11:.:.:64:C9[35.78]85[39.94];T1[41]3[39.67]", vcfs.get(1));
 //		assertEquals("GL000208.1	84	.	C	A	.	.	FLANK=TATTCAACTCA	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	0/1:22,71:93:A2[]6[]:A178;C6:.:.:50:A8[38.62]63[38.92];C2[36.5]20[40.2]	0/1:22,71:93:A2[]6[]:A178;C6:.:.:50:A8[38.62]63[38.92];C2[36.5]20[40.2]", vcfs.get(2));
 //		assertEquals("GL000208.1	98	.	C	A	.	.	FLANK=ACTTTAATGCA	GT:AD:DP:EOR:FF:FT:INF:NNS:OABS	1/1:0,83:83:A0[]8[]:A188;G1:.:.:59:A8[37]75[38.63]	1/1:0,83:83:A0[]8[]:A188;G1:.:.:59:A8[37]75[38.63]", vcfs.get(3));
 	}
 
-	@Ignore
-	public void testContainsAndRemove() {
-		final ConcurrentMap<Integer, Accumulator> map = new ConcurrentHashMap<Integer, Accumulator>();
-		final int noOfLoops = 100000;
-		long time = System.currentTimeMillis();
-		for (int i = 0 ; i < noOfLoops ; i++) {
-			map.remove(i);
-		}
-		System.out.println("EMPTY: remove time: " + (System.currentTimeMillis() - time));
-		
-		time = System.currentTimeMillis();
-		for (int i = 0 ; i < noOfLoops ; i++) {
-			if (map.containsKey(i))
-				map.remove(i);
-		}
-		System.out.println("EMPTY: contains and remove time: " + (System.currentTimeMillis() - time));
-		
-		time = System.currentTimeMillis();
-		for (int i = 0 ; i < noOfLoops ; i++) {
-			map.remove(i);
-		}
-		System.out.println("EMPTY: remove time: " + (System.currentTimeMillis() - time));
-	}
-	
 	public void checkBam(File f) {
 		SamReader reader = SamReaderFactory.makeDefault().open(f);
 		assertEquals(SortOrder.coordinate, reader.getFileHeader().getSortOrder());