WIP

bammetrics/src/main/scala/nl/lumc/sasc/biopet/pipelines/bammetrics/BammetricsReport.scala

@@ -160,7 +160,7 @@ object BammetricsReport extends ReportBuilder {
     plot.output = pngFile
     plot.ylabel = Some("Reads")
     plot.width = Some(200 + (results.size * 10))
-    plot.title = Some("Aligned reads")
+    plot.title = Some("Aligned_reads")
     plot.runLocal()
   }
 

biopet-core/src/main/scala/nl/lumc/sasc/biopet/core/report/MultisampleReportBuilder.scala

@@ -16,7 +16,6 @@ package nl.lumc.sasc.biopet.core.report
 
 import scala.concurrent.{ Await, Future }
 import scala.concurrent.duration.Duration
-import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * This trait will generate a report with added function to generate sample and library pages for those existing in the summary.

biopet-core/src/main/scala/nl/lumc/sasc/biopet/core/report/ReportBuilder.scala

@@ -17,14 +17,14 @@ package nl.lumc.sasc.biopet.core.report
 import java.io._
 
 import nl.lumc.sasc.biopet.core.ToolCommandFunction
-import nl.lumc.sasc.biopet.utils.summary.db.Schema.{ Library, Module, Pipeline, Sample }
+import nl.lumc.sasc.biopet.utils.summary.db.Schema.{Library, Module, Pipeline, Sample}
 import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb
-import nl.lumc.sasc.biopet.utils.{ IoUtils, Logging, ToolCommand }
+import nl.lumc.sasc.biopet.utils.{IoUtils, Logging, ToolCommand}
 import org.broadinstitute.gatk.utils.commandline.Input
 import org.fusesource.scalate.TemplateEngine
 
 import scala.collection.mutable
-import scala.concurrent.{ Await, ExecutionContextExecutor, Future }
+import scala.concurrent._
 import scala.concurrent.duration.Duration
 import scala.language.postfixOps
 import scala.language.implicitConversions
@@ -73,7 +73,7 @@ trait ReportBuilderExtension extends ToolCommandFunction {
 
 trait ReportBuilder extends ToolCommand {
 
-  implicit lazy val global: ExecutionContextExecutor = scala.concurrent.ExecutionContext.global
+  implicit lazy val ec = ReportBuilder.ec
   implicit def toOption[T](x: T): Option[T] = Option(x)
   implicit def autoWait[T](x: Future[T]): T = Await.result(x, Duration.Inf)
 
@@ -278,6 +278,8 @@ trait ReportBuilder extends ToolCommand {
 
 object ReportBuilder {
 
+  implicit lazy val ec = ExecutionContext.global
+
   /** Single template render engine, this will have a cache for all compile templates */
   protected val engine = new TemplateEngine()
   engine.allowReload = false

biopet-core/src/main/scala/nl/lumc/sasc/biopet/core/summary/SummaryQScript.scala

@@ -26,6 +26,7 @@ import nl.lumc.sasc.biopet.LastCommitHash
 import scala.collection.mutable
 import scala.concurrent.Await
 import scala.concurrent.duration.Duration
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.io.Source
 
 /**

biopet-core/src/test/scala/nl/lumc/sasc/biopet/core/report/MultisampleReportBuilderTest.scala

@@ -25,7 +25,6 @@ import org.testng.annotations.Test
 
 import scala.concurrent.{ Await, Future }
 import scala.concurrent.duration.Duration
-import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * Created by pjvanthof on 24/02/16.
@@ -46,6 +45,9 @@ class MultisampleReportBuilderTest extends TestNGSuite with Matchers {
 
       def libraryPage(sampleId: Int, libraryId: Int, args: Map[String, Any]) = Future(ReportPage(Nil, Nil, Map()))
     }
+
+    import scala.concurrent.ExecutionContext.Implicits.global
+
     val dbFile = File.createTempFile("summary.", ".db")
     dbFile.deleteOnExit()
     val db = SummaryDb.openSqliteSummary(dbFile)

biopet-core/src/test/scala/nl/lumc/sasc/biopet/core/report/ReportBuilderTest.scala

@@ -26,8 +26,6 @@ import org.testng.annotations.{ DataProvider, Test }
 import scala.concurrent.{ Await, Future }
 import scala.concurrent.duration.Duration
 
-import scala.concurrent.ExecutionContext.Implicits.global
-
 /**
  * Created by pjvanthof on 24/02/16.
  */
@@ -53,6 +51,8 @@ class ReportBuilderTest extends TestNGSuite with Matchers {
         (if (nested) "p1" -> Future(ReportPage(Nil, Nil, Map())) :: Nil else Nil), Nil, Map()))
     }
 
+    import scala.concurrent.ExecutionContext.Implicits.global
+
     val dbFile = File.createTempFile("summary.", ".db")
     dbFile.deleteOnExit()
     val db = SummaryDb.openSqliteSummary(dbFile)

biopet-core/src/test/scala/nl/lumc/sasc/biopet/core/summary/WriteSummaryTest.scala

@@ -32,6 +32,7 @@ import org.testng.annotations.{ AfterClass, Test }
 
 import scala.concurrent.Await
 import scala.concurrent.duration.Duration
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.util.matching.Regex
 
 /**

biopet-tools/src/main/scala/nl/lumc/sasc/biopet/tools/RegionAfCount.scala

@@ -14,16 +14,18 @@
  */
 package nl.lumc.sasc.biopet.tools
 
-import java.io.{ PrintWriter, InputStream, File }
+import java.io.{File, PrintWriter}
 import java.util
 
 import htsjdk.variant.vcf.VCFFileReader
 import nl.lumc.sasc.biopet.utils.ToolCommand
 import nl.lumc.sasc.biopet.utils.rscript.ScatterPlot
-import nl.lumc.sasc.biopet.utils.intervals.{ BedRecord, BedRecordList }
+import nl.lumc.sasc.biopet.utils.intervals.{BedRecord, BedRecordList}
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable
+import scala.concurrent.ExecutionContext
+import scala.concurrent.ExecutionContext.Implicits.global
 
 object RegionAfCount extends ToolCommand {
   case class Args(bedFile: File = null,
@@ -123,7 +125,7 @@ object RegionAfCount extends ToolCommand {
       if (cmdArgs.scatterpPlot) generatePlot(tsvFile)
     }
 
-    def generatePlot(tsvFile: File): Unit = {
+    def generatePlot(tsvFile: File)(implicit ec: ExecutionContext): Unit = {
       logger.info(s"Generate plot for $tsvFile")
 
       val scatterPlot = new ScatterPlot(null)

biopet-tools/src/main/scala/nl/lumc/sasc/biopet/tools/bamstats/Histogram.scala

@@ -14,12 +14,14 @@
  */
 package nl.lumc.sasc.biopet.tools.bamstats
 
-import java.io.{ File, IOException, PrintWriter }
+import java.io.{File, IOException, PrintWriter}
 
 import nl.lumc.sasc.biopet.utils.rscript.LinePlot
-import nl.lumc.sasc.biopet.utils.{ Logging, sortAnyAny }
+import nl.lumc.sasc.biopet.utils.{Logging, sortAnyAny}
 
 import scala.collection.mutable
+import scala.concurrent.ExecutionContext
+import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * Created by pjvanthof on 05/07/16.
@@ -91,7 +93,7 @@ class Histogram[T](_counts: Map[T, Long] = Map[T, Long]())(implicit ord: Numeric
     writer.close()
   }
 
-  def writeFilesAndPlot(outputDir: File, prefix: String, xlabel: String, ylabel: String, title: String): Unit = {
+  def writeFilesAndPlot(outputDir: File, prefix: String, xlabel: String, ylabel: String, title: String)(implicit ec: ExecutionContext): Unit = {
     writeHistogramToTsv(new File(outputDir, prefix + ".histogram.tsv"))
     writeAggregateToTsv(new File(outputDir, prefix + ".stats.tsv"))
     val plot = new LinePlot(null)

biopet-utils/src/main/scala/nl/lumc/sasc/biopet/utils/process/AsyncProcess.scala

@@ -17,6 +17,8 @@ DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
 
+import nl.lumc.sasc.biopet.utils.Logging
+
 import scala.concurrent.{ ExecutionContext, Future, Promise }
 import scala.sys.process.{ Process, ProcessLogger }
 import scala.util.Try
@@ -129,12 +131,16 @@ trait Sys {
     val proc = Process(cmd).run(ProcessLogger(stdout.appendLine, stderr.appendLine))
     p.tryCompleteWith(Future(proc.exitValue).map(c => (c, stdout.get, stderr.get)))
 
-    val cancel = {
-      p.tryFailure(new ExecutionCanceled(s"Process: '${cmd.mkString(" ")}' canceled"))
+    val cancel = () => {
+      p.tryFailure {
+        Logging.logger.error("stdout: " + stdout.get)
+        Logging.logger.error("stderr: " + stderr.get)
+        new ExecutionCanceled(s"Process: '${cmd.mkString(" ")}' canceled")
+      }
       proc.destroy()
     }
 
-    (p.future, () => cancel)
+    (p.future, cancel)
   }
 
   class OutputSlurper {

biopet-utils/src/main/scala/nl/lumc/sasc/biopet/utils/rscript/Rscript.scala

@@ -14,12 +14,16 @@
  */
 package nl.lumc.sasc.biopet.utils.rscript
 
-import java.io.{ File, FileOutputStream }
+import java.io.{File, FileOutputStream}
 
 import nl.lumc.sasc.biopet.utils.Logging
 import nl.lumc.sasc.biopet.utils.config.Configurable
 import nl.lumc.sasc.biopet.utils.process.Sys
+import nl.lumc.sasc.biopet.utils.process.Sys.AsyncExecResult
 
+import scala.collection.parallel.mutable.ParSeq
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, ExecutionContext}
 import scala.sys.process.ProcessLogger
 
 /**
@@ -62,16 +66,26 @@ trait Rscript extends Configurable {
     }
   }
 
+  private val cache: ParSeq[AsyncExecResult] = Nil
+
   /**
    * Execute rscript on local system
    * @param logger How to handle stdout and stderr
    */
-  def runLocal(logger: ProcessLogger): Unit = {
+  def runLocal(logger: ProcessLogger)(implicit ec: ExecutionContext): Unit = {
     checkScript()
 
     Logging.logger.info("Running: " + cmd.mkString(" "))
 
-    val (exitcode, stdout, stderr) = Sys.exec(cmd)
+    while (cache.size > 5) for (c <- cache) Thread.
+
+    val results = Sys.execAsync(cmd)
+
+    while(results.isRunning) {
+      Thread.sleep(1000)
+    }
+
+    val (exitcode, stdout, stderr) = Await.result(results.map(x => (x._1, x._2, x._3)), Duration.Inf)
 
     Logging.logger.info("stdout:\n" + stdout + "\n")
     Logging.logger.info("stderr:\n" + stderr)
@@ -83,7 +97,7 @@ trait Rscript extends Configurable {
    * Execute rscript on local system
    * Stdout and stderr will go to biopet logger
    */
-  def runLocal(): Unit = {
+  def runLocal()(implicit ec: ExecutionContext): Unit = {
     runLocal(ProcessLogger(Logging.logger.info(_)))
   }
 }

biopet-utils/src/main/scala/nl/lumc/sasc/biopet/utils/summary/db/SummaryDb.scala

@@ -4,7 +4,6 @@ import nl.lumc.sasc.biopet.utils.ConfigUtils
 import nl.lumc.sasc.biopet.utils.summary.db.Schema._
 import slick.driver.H2Driver.api._
 
-import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.Duration
 import scala.concurrent.{ Await, ExecutionContext, Future }
 import java.io.{ Closeable, File }
@@ -22,6 +21,8 @@ import scala.language.implicitConversions
  */
 trait SummaryDb extends Closeable {
 
+  implicit val ec: ExecutionContext
+
   def db: Database
 
   def close(): Unit = db.close()
@@ -431,9 +432,9 @@ trait SummaryDb extends Closeable {
 
 }
 
-class SummaryDbReadOnly(val db: Database) extends SummaryDb
+class SummaryDbReadOnly(val db: Database)(implicit val ec: ExecutionContext) extends SummaryDb
 
-class SummaryDbWrite(val db: Database) extends SummaryDb {
+class SummaryDbWrite(val db: Database)(implicit val ec: ExecutionContext) extends SummaryDb {
   /** This method will create all tables */
   def createTables(): Unit = {
     try {
@@ -688,7 +689,7 @@ object SummaryDb {
   }
 
   /** This will open a sqlite database and create tables when the database did not exist yet */
-  def openSqliteSummary(file: File): SummaryDbWrite = {
+  def openSqliteSummary(file: File)(implicit ec: ExecutionContext): SummaryDbWrite = {
     if (!summaryConnections.contains(file)) {
       val config: org.sqlite.SQLiteConfig = new org.sqlite.SQLiteConfig()
       config.enforceForeignKeys(true)
@@ -703,7 +704,7 @@ object SummaryDb {
     summaryConnections(file)
   }
 
-  def openReadOnlySqliteSummary(file: File): SummaryDbReadOnly = {
+  def openReadOnlySqliteSummary(file: File)(implicit ec: ExecutionContext): SummaryDbReadOnly = {
     require(file.exists(), s"File does not exist: $file")
     val config: org.sqlite.SQLiteConfig = new org.sqlite.SQLiteConfig()
     config.enforceForeignKeys(true)
@@ -712,7 +713,7 @@ object SummaryDb {
     config.setReadOnly(true)
 
     val asyncExecutor = new AsyncExecutor {
-      override def executionContext: ExecutionContext = global
+      override def executionContext: ExecutionContext = ec
       override def close(): Unit = {}
     }
 

biopet-utils/src/test/scala/nl/lumc/sasc/biopet/utils/summary/db/SummaryDbTest.scala

@@ -9,6 +9,8 @@ import org.testng.annotations.Test
 
 import scala.concurrent.Await
 import scala.concurrent.duration.Duration
+import scala.concurrent.ExecutionContext.Implicits.global
+
 import SummaryDb.Implicts._
 import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb.{ NoLibrary, NoModule, NoSample }
 

gears/src/main/scala/nl/lumc/sasc/biopet/pipelines/gears/GearsReport.scala

@@ -24,7 +24,6 @@ import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb.{ NoLibrary, NoModule, Sam
 
 import scala.concurrent.{ Await, Future }
 import scala.concurrent.duration.Duration
-import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * Report for Gears

mapping/src/main/scala/nl/lumc/sasc/biopet/pipelines/mapping/MultisampleMappingReport.scala

@@ -23,7 +23,6 @@ import nl.lumc.sasc.biopet.utils.summary.db.SummaryDb._
 
 import scala.concurrent.{ Await, Future }
 import scala.concurrent.duration.Duration
-import scala.concurrent.ExecutionContext.Implicits.global
 
 /**
  * Created by pjvanthof on 11/01/16.