Extracted resoures to trait, usable on non biopet CommandLineFunctions

protected/biopet-gatk-extensions/src/main/scala/nl/lumc/sasc/biopet/extensions/gatk/broad/HaplotypeCaller.scala

@@ -41,7 +41,6 @@ class HaplotypeCaller(val root: Configurable) extends org.broadinstitute.gatk.qu
   override def beforeGraph() {
     super.beforeGraph()
     if (bamOutput != null && nct.getOrElse(1) > 1) {
-      threads = 1
       logger.warn("BamOutput is on, nct/threads is forced to set on 1, this option is only for debug")
     }
     nct = Some(getThreads)

public/basty/src/main/scala/nl/lumc/sasc/biopet/pipelines/basty/BastyTrait.scala

@@ -137,7 +137,6 @@ trait BastyTrait extends MultiSampleQScript {
       val numBoot = config("boot_runs", default = 100, submodule = "raxml").asInt
       val bootList = for (t <- 0 until numBoot) yield {
         val raxmlBoot = new Raxml(this)
-        raxmlBoot.threads = 1
         raxmlBoot.input = variants
         raxmlBoot.m = config("raxml_ml_model", default = "GTRGAMMAX")
         raxmlBoot.p = Some(seed)

public/biopet-core/src/main/scala/nl/lumc/sasc/biopet/core/BiopetCommandLineFunction.scala

@@ -19,8 +19,6 @@ import java.io.{ PrintWriter, File, FileInputStream }
 import java.security.MessageDigest
 
 import nl.lumc.sasc.biopet.utils.Logging
-import nl.lumc.sasc.biopet.utils.config.Configurable
-import org.broadinstitute.gatk.queue.function.CommandLineFunction
 import org.broadinstitute.gatk.utils.commandline.{ Output, Input }
 import org.broadinstitute.gatk.utils.runtime.ProcessSettings
 import org.ggf.drmaa.JobTemplate
@@ -32,7 +30,7 @@ import scala.util.matching.Regex
 import scala.collection.JavaConversions._
 
 /** Biopet command line trait to auto check executable and cluster values */
-trait BiopetCommandLineFunction extends CommandLineFunction with Configurable { biopetFunction =>
+trait BiopetCommandLineFunction extends CommandLineResources { biopetFunction =>
   analysisName = configName
 
   @Input(doc = "deps", required = false)
@@ -41,19 +39,6 @@ trait BiopetCommandLineFunction extends CommandLineFunction with Configurable {
   @Output
   var outputFiles: List[File] = Nil
 
-  var threads = 0
-  def defaultThreads = 1
-
-  var vmem: Option[String] = config("vmem")
-  def defaultCoreMemory: Double = 1.0
-  def defaultVmemFactor: Double = 1.4
-  var vmemFactor: Double = config("vmem_factor", default = defaultVmemFactor)
-
-  var residentFactor: Double = config("resident_factor", default = 1.2)
-
-  private var _coreMemory: Double = 2.0
-  def coreMemeory = _coreMemory
-
   var executable: String = _
 
   /** This is the default shell for drmaa jobs */
@@ -102,44 +87,10 @@ trait BiopetCommandLineFunction extends CommandLineFunction with Configurable {
 
   /** Set default output file, threads and vmem for current job */
   final def internalBeforeGraph(): Unit = {
-    val firstOutput = try {
-      this.firstOutput
-    } catch {
-      case e: NullPointerException => null
-    }
 
     pipesJobs.foreach(_.beforeGraph())
     pipesJobs.foreach(_.internalBeforeGraph())
 
-    if (jobOutputFile == null && firstOutput != null)
-      jobOutputFile = new File(firstOutput.getAbsoluteFile.getParent, "." + firstOutput.getName + "." + configName + ".out")
-
-    val ownThreads = getThreads(defaultThreads)
-    if (threads == 0) threads = ownThreads + pipesJobs.map(_.threads).map(i => if (i == 0) 1 else i).sum
-    if (threads > 1) nCoresRequest = Option(threads)
-
-    _coreMemory = config("core_memory", default = (defaultCoreMemory * (ownThreads.toDouble / this.threads.toDouble)) +
-      pipesJobs.map(job => job.coreMemeory * (job.threads.toDouble / this.threads.toDouble)).sum).asDouble +
-      (0.5 * retry)
-
-    if (config.contains("memory_limit")) memoryLimit = config("memory_limit")
-    else memoryLimit = Some(_coreMemory * threads)
-
-    if (config.contains("resident_limit")) residentLimit = config("resident_limit")
-    else residentLimit = Some((_coreMemory + (0.5 * retry)) * residentFactor)
-
-    if (!config.contains("vmem")) vmem = Some((_coreMemory * (vmemFactor + (0.5 * retry))) + "G")
-    jobName = configName + ":" + (if (firstOutput != null) firstOutput.getName else jobOutputFile)
-  }
-
-  var retry = 0
-
-  override def setupRetry(): Unit = {
-    super.setupRetry()
-    if (vmem.isDefined) jobResourceRequests = jobResourceRequests.filterNot(_.contains("h_vmem="))
-    logger.info("Auto raise memory on retry")
-    retry += 1
-    this.freeze()
   }
 
   /** can override this value is executable may not be converted to CanonicalPath */
@@ -251,33 +202,6 @@ trait BiopetCommandLineFunction extends CommandLineFunction with Configurable {
     BiopetCommandLineFunction.versionCache.get(versionCommand)
   }
 
-  def getThreads: Int = getThreads(defaultThreads)
-
-  /**
-   * Get threads from config
-   * @param default default when not found in config
-   * @return number of threads
-   */
-  def getThreads(default: Int): Int = {
-    val maxThreads: Int = config("maxthreads", default = 24)
-    val threads: Int = config("threads", default = default)
-    if (maxThreads > threads) threads
-    else maxThreads
-  }
-
-  /**
-   * Get threads from config
-   * @param default default when not found in config
-   * @param module Module when this is difrent from default
-   * @return number of threads
-   */
-  def getThreads(default: Int, module: String): Int = {
-    val maxThreads: Int = config("maxthreads", default = 24, submodule = module)
-    val threads: Int = config("threads", default = default, submodule = module)
-    if (maxThreads > threads) threads
-    else maxThreads
-  }
-
   private[core] var _inputAsStdin = false
   def inputAsStdin = _inputAsStdin
   private[core] var _outputAsStdout = false

public/biopet-core/src/main/scala/nl/lumc/sasc/biopet/core/CommandLineResources.scala

+package nl.lumc.sasc.biopet.core
+
+import java.io.File
+
+import nl.lumc.sasc.biopet.utils.config.Configurable
+import org.broadinstitute.gatk.queue.function.CommandLineFunction
+
+/**
+ * Created by pjvanthof on 01/10/15.
+ */
+trait CommandLineResources extends CommandLineFunction with Configurable {
+
+  def defaultThreads = 1
+  final def threads = nCoresRequest match {
+    case Some(i) => i
+    case _ => {
+      val t = getThreads
+      nCoresRequest = Some(t)
+      t
+    }
+  }
+
+  var vmem: Option[String] = config("vmem")
+  def defaultCoreMemory: Double = 1.0
+  def defaultVmemFactor: Double = 1.4
+  var vmemFactor: Double = config("vmem_factor", default = defaultVmemFactor)
+
+  var residentFactor: Double = config("resident_factor", default = 1.2)
+
+  private var _coreMemory: Double = 2.0
+  def coreMemeory = _coreMemory
+
+  var retry = 0
+
+  override def freezeFieldValues(): Unit = {
+    setResources()
+    super.freezeFieldValues()
+  }
+
+  def getThreads: Int = getThreads(defaultThreads)
+
+  /**
+   * Get threads from config
+   * @param default default when not found in config
+   * @return number of threads
+   */
+  private def getThreads(default: Int): Int = {
+    val maxThreads: Int = config("maxthreads", default = 24)
+    val threads: Int = config("threads", default = default)
+    if (maxThreads > threads) threads
+    else maxThreads
+  }
+
+  def setResources(): Unit = {
+    val firstOutput = try {
+      this.firstOutput
+    } catch {
+      case e: NullPointerException => null
+    }
+
+    if (jobOutputFile == null && firstOutput != null)
+      jobOutputFile = new File(firstOutput.getAbsoluteFile.getParent, "." + firstOutput.getName + "." + configName + ".out")
+
+    nCoresRequest = Option(threads)
+
+    _coreMemory = config("core_memory", default = defaultCoreMemory).asDouble +
+      (0.5 * retry)
+
+    if (config.contains("memory_limit")) memoryLimit = config("memory_limit")
+    else memoryLimit = Some(_coreMemory * threads)
+
+    if (config.contains("resident_limit")) residentLimit = config("resident_limit")
+    else residentLimit = Some((_coreMemory + (0.5 * retry)) * residentFactor)
+
+    if (!config.contains("vmem")) vmem = Some((_coreMemory * (vmemFactor + (0.5 * retry))) + "G")
+    jobName = configName + ":" + (if (firstOutput != null) firstOutput.getName else jobOutputFile)
+  }
+
+  override def setupRetry(): Unit = {
+    super.setupRetry()
+    if (vmem.isDefined) jobResourceRequests = jobResourceRequests.filterNot(_.contains("h_vmem="))
+    logger.info("Auto raise memory on retry")
+    retry += 1
+    this.freeze()
+  }
+
+  protected def combineResources(commands: List[CommandLineResources]): Unit = {
+    commands.foreach(_.setResources())
+    nCoresRequest = Some(commands.map(_.threads).sum)
+
+    _coreMemory = commands.map(cmd => cmd.coreMemeory * (cmd.threads.toDouble / threads.toDouble)).sum
+    memoryLimit = Some(_coreMemory * threads)
+    residentLimit = Some((_coreMemory + (0.5 * retry)) * residentFactor)
+    vmem = Some((_coreMemory * (vmemFactor + (0.5 * retry))) + "G")
+  }
+
+}

public/biopet-extentsions/src/main/scala/nl/lumc/sasc/biopet/extensions/Raxml.scala

@@ -74,7 +74,6 @@ class Raxml(val root: Configurable) extends BiopetCommandLineFunction {
   /** Sets correct output files to job */
   override def beforeGraph() {
     require(w != null)
-    if (threads == 0) threads = getThreads
     executable = if (threads > 1 && executableThreads.isDefined) executableThreads.get else executableNonThreads
     super.beforeGraph()
     out :::= List(Some(getInfoFile), getBestTreeFile, getBootstrapFile, getBipartitionsFile).flatten