parallel gap filling for same range code
package net.sf.mzmine.modules.peaklistmethods.gapfilling.samerange;
import java.util.logging.Logger;
+import java.util.List;
+import java.util.Collections;
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
import net.sf.mzmine.datamodel.DataPoint;
import net.sf.mzmine.datamodel.Feature;
@@ -55,6 +59,7 @@ class SameRangeTask extends AbstractTask {
private boolean removeOriginal;
private int processedRows, totalRows;
+ private AtomicInteger processedRowsAtomic;;
private ParameterSet parameters;
@@ -89,8 +94,22 @@ class SameRangeTask extends AbstractTask {
// Create new peak list
processedPeakList = new SimplePeakList(peakList + " " + suffix, columns);
- // Fill gaps in given column
+
+ /*************************************************************
+ Creating a stream to process the data in parallel
+ */
+
+ List<Integer> rowList = new ArrayList<Integer>();
for (int row = 0; row < totalRows; row++){
+ rowList.add(new Integer(row));
+ }
+
+ processedRowsAtomic = new AtomicInteger(0);
+
+ List outputList = Collections.synchronizedList(new ArrayList());
+
+ rowList.parallelStream().forEach(rowObj -> {
+ int row = rowObj.intValue();
// Canceled?
if (isCanceled())
@@ -102,6 +121,58 @@ class SameRangeTask extends AbstractTask {
// Copy comment
newRow.setComment(sourceRow.getComment());
+ // Copy identities
+ for (PeakIdentity ident : sourceRow.getPeakIdentities())
+ newRow.addPeakIdentity(ident, false);
+ if (sourceRow.getPreferredPeakIdentity() != null)
+ newRow.setPreferredPeakIdentity(sourceRow
+ .getPreferredPeakIdentity());
+
+ // Copy each peaks and fill gaps
+ for (RawDataFile column : columns) {
+
+ // Canceled?
+ if (isCanceled())
+ return;
+
+ // Get current peak
+ Feature currentPeak = sourceRow.getPeak(column);
+
+ // If there is a gap, try to fill it
+ if (currentPeak == null)
+ currentPeak = fillGap(sourceRow, column);
+
+ // If a peak was found or created, add it
+ if (currentPeak != null)
+ newRow.addPeak(column, currentPeak);
+ }
+
+ outputList.add(newRow);
+
+ processedRowsAtomic.getAndAdd(1);
+ //processedRows++;
+ });
+
+ outputList.stream().forEach(newRow -> {
+ processedPeakList.addRow((PeakListRow)newRow);
+ });
+
+ /* End Parallel Implementation */
+ /*******************************************************************************/
+
+
+ // Fill gaps in given column
+ /*for (int row = 0; row < totalRows; row++) {
+ // Canceled?
+ if (isCanceled())
+ return;
+
+ PeakListRow sourceRow = peakList.getRow(row);
+ PeakListRow newRow = new SimplePeakListRow(sourceRow.getID());
+
+ // Copy comment
+ newRow.setComment(sourceRow.getComment());
+
// Copy identities
for (PeakIdentity ident : sourceRow.getPeakIdentities())
newRow.addPeakIdentity(ident, false);
@@ -133,7 +204,7 @@ class SameRangeTask extends AbstractTask {
processedRows++;
- }
+ }*/
// Append processed peak list to the project
project.addPeakList(processedPeakList);
@@ -222,7 +293,7 @@ class SameRangeTask extends AbstractTask {
public double getFinishedPercentage() {
if (totalRows == 0)
return 0;
- return (double) processedRows / (double) totalRows;
+ return (double) processedRowsAtomic.get() / (double) totalRows;
}
