fix(LinuxHelper): fix a bug in the LinuxHelper

The application crashed when the getProcesses method was called with a non-existing application

fix(PowerSpy): fix a problem with Process/Application targets

The idle power was added to the dynamic power for these targets, it was not correct

refactoring(targets): add missing toString method

Add a toString method to each target

refactoring(PowerChannel): modify the AggregatePowerReport

Given the fact the AggregatePowerReport aggregates data from several targets and sources, all these informations are kept in order to de displayed to the end-user.

Author: mcolmant

powerapi-core/src/main/scala/org/powerapi/PowerMeter.scala

@@ -239,5 +239,5 @@ trait PowerDisplay {
   /**
    * Displays data from power reports.
    */
-  def display(timestamp: Long, target: Target, device: String, power: Power)
+  def display(timestamp: Long, targets: Set[Target], devices: Set[String], power: Power)
 }

powerapi-core/src/main/scala/org/powerapi/core/MonitorActors.scala

@@ -28,13 +28,12 @@ import akka.event.LoggingReceive
 import java.util.UUID
 import org.powerapi.core.ClockChannel.ClockTick
 import org.powerapi.core.power._
-import org.powerapi.module.PowerChannel.PowerReport
 import org.powerapi.core.target.Target
 import org.powerapi.{PowerDisplay, PowerMonitoring}
 import org.powerapi.core.ClockChannel.{startClock, stopClock, subscribeClockTick, unsubscribeClockTick}
 import org.powerapi.core.MonitorChannel.{MonitorAggFunction, MonitorStart, MonitorStarted, MonitorStop, MonitorStopAll, formatMonitorChildName, subscribeMonitorsChannel}
 import org.powerapi.core.MonitorChannel.{publishMonitorTick, setAggFunction, stopMonitor}
-import org.powerapi.module.PowerChannel.{AggregateReport, render, subscribePowerReport, unsubscribePowerReport, subscribeAggPowerReport}
+import org.powerapi.module.PowerChannel.{AggregatePowerReport, RawPowerReport, render, subscribeRawPowerReport, unsubscribeRawPowerReport, subscribeAggPowerReport}
 import org.powerapi.module.SensorChannel.{monitorAllStopped, monitorStopped}
 import org.powerapi.reporter.ReporterComponent
 import scala.concurrent.duration.FiniteDuration
@@ -58,10 +57,10 @@ class MonitorChild(eventBus: MessageBus,
   /**
    * Running state.
    */
-  def running(aggR: AggregateReport, aggFunction: Seq[Power] => Power): Actor.Receive = LoggingReceive {
+  def running(aggR: AggregatePowerReport, aggFunction: Seq[Power] => Power): Actor.Receive = LoggingReceive {
     case tick: ClockTick => produceMessages(tick)
     case MonitorAggFunction(_, id, aggF) if muid == id => setAggregatingFunction(aggR, aggF)
-    case powerReport: PowerReport => aggregate(aggR, powerReport, aggFunction)
+    case powerReport: RawPowerReport => aggregate(aggR, powerReport, aggFunction)
     case MonitorStop(_, id) if muid == id => stop()
     case _: MonitorStopAll => stop()
   } orElse default
@@ -73,15 +72,15 @@ class MonitorChild(eventBus: MessageBus,
   def start(): Unit = {
     startClock(frequency)(eventBus)
     subscribeClockTick(frequency)(eventBus)(self)
-    subscribePowerReport(muid)(eventBus)(self)
+    subscribeRawPowerReport(muid)(eventBus)(self)
     log.info("monitor is started, muid: {}", muid)
-    context.become(running(AggregateReport(muid, SUM), SUM))
+    context.become(running(AggregatePowerReport(muid, SUM), SUM))
   }
 
   /**
    * Change the aggregating function for this monitor
    */
-  def setAggregatingFunction(aggR: AggregateReport, aggF: Seq[Power] => Power): Unit = {
+  def setAggregatingFunction(aggR: AggregatePowerReport, aggF: Seq[Power] => Power): Unit = {
     log.info("aggregating function is changed")
     context.become(running(aggR, aggF))
   }
@@ -97,11 +96,11 @@ class MonitorChild(eventBus: MessageBus,
    * Wait to retrieve power reports of all targets from a same monitor to aggregate them
    * into once power report.
    */
-  def aggregate(aggR: AggregateReport, powerReport: PowerReport, aggF: Seq[Power] => Power): Unit = {
+  def aggregate(aggR: AggregatePowerReport, powerReport: RawPowerReport, aggF: Seq[Power] => Power): Unit = {
     aggR += powerReport
     if (aggR.size >= targets.size) {
       render(aggR)(eventBus)
-      context.become(running(AggregateReport(muid, aggF), aggF))
+      context.become(running(AggregatePowerReport(muid, aggF), aggF))
     }
     else
       context.become(running(aggR, aggF))
@@ -114,7 +113,7 @@ class MonitorChild(eventBus: MessageBus,
   def stop(): Unit = {
     stopClock(frequency)(eventBus)
     unsubscribeClockTick(frequency)(eventBus)(self)
-    unsubscribePowerReport(muid)(eventBus)(self)
+    unsubscribeRawPowerReport(muid)(eventBus)(self)
     log.info("monitor is stopped, muid: {}", muid)
     self ! PoisonPill
   }

powerapi-core/src/main/scala/org/powerapi/core/OSHelper.scala

@@ -66,21 +66,21 @@ trait OSHelper {
   /**
    * Get the list of frequencies available on the CPU.
    */
-  def getCPUFrequencies(topology: Map[Int, Iterable[Int]]): Iterable[Long]
+  def getCPUFrequencies: Set[Long]
 
   /**
    * Get the list of processes behind an Application.
    *
    * @param application: targeted application.
    */
-  def getProcesses(application: Application): Iterable[Process]
+  def getProcesses(application: Application): Set[Process]
 
   /**
    * Get the list of thread behind a Process.
    *
    * @param process: targeted process.
    */
-  def getThreads(process: Process): Iterable[Thread]
+  def getThreads(process: Process): Set[Thread]
 
   /**
    * Get the process execution time on the cpu.
@@ -128,7 +128,7 @@ trait OSHelper {
 class LinuxHelper extends OSHelper with Configuration {
   private val log = LogManager.getLogger
 
-  private val PSFormat = """^\s*(\d+)""".r
+  private val PSFormat = """^\s*(\d+)\s*""".r
   private val GlobalStatFormat = """cpu\s+([\d\s]+)""".r
   private val TimeInStateFormat = """(\d+)\s+(\d+)""".r
 
@@ -177,44 +177,44 @@ class LinuxHelper extends OSHelper with Configuration {
   /**
    * CPU's topology.
    */
-  lazy val topology: Map[Int, Iterable[Int]] = load { conf =>
+  lazy val topology: Map[Int, Set[Int]] = load { conf =>
     (for (item: Config <- conf.getConfigList("powerapi.cpu.topology"))
-      yield (item.getInt("core"), item.getDoubleList("indexes").map(_.toInt).toList)).toMap
+      yield (item.getInt("core"), item.getDoubleList("indexes").map(_.toInt).toSet)).toMap
   } match {
     case ConfigValue(values) => values
     case _ => Map()
   }
 
-  def getCPUFrequencies(topology: Map[Int, Iterable[Int]]): Iterable[Long] = {
+  def getCPUFrequencies: Set[Long] = {
     (for(index <- topology.values.flatten) yield {
       try {
         using(frequenciesPath.replace("%?core", s"$index"))(source => {
           log.debug("using {} as a frequencies path", frequenciesPath)
-          source.getLines.toIndexedSeq(0).split("\\s").toList.map(_.toLong)
+          source.getLines.toIndexedSeq(0).split("\\s").map(_.toLong).toSet
         })
       }
       catch {
-        case ioe: IOException => log.warn("i/o exception: {}", ioe.getMessage); List()
+        case ioe: IOException => log.warn("i/o exception: {}", ioe.getMessage); Set[Long]()
       }
     }).flatten.toSet
   }
 
-  def getProcesses(application: Application): Iterable[Process] = {
-    Seq("ps", "-C", application.name, "-o", "pid", "--no-headers").!!.split("\n").map {
+  def getProcesses(application: Application): Set[Process] = {
+    Seq("ps", "-C", application.name, "-o", "pid", "--no-headers").lineStream_!.map {
       case PSFormat(pid) => Process(pid.toInt)
-    }
+    }.toSet
   }
 
-  def getThreads(process: Process): Iterable[Thread] = {
+  def getThreads(process: Process): Set[Thread] = {
     val pidDirectory = new File(taskPath.replace("%?pid", s"${process.pid}"))
 
     if (pidDirectory.exists && pidDirectory.isDirectory) {
       /**
        * The pid is removed because it corresponds to the main thread.
        */
-      pidDirectory.listFiles.filter(dir => dir.isDirectory && dir.getName != s"${process.pid}").map(dir => Thread(dir.getName.toInt))
+      pidDirectory.listFiles.filter(dir => dir.isDirectory && dir.getName != s"${process.pid}").map(dir => Thread(dir.getName.toInt)).toSet
     }
-    else Iterable[Thread]()
+    else Set[Thread]()
   }
 
   def getProcessCpuTime(process: Process): Option[Long] = {

powerapi-core/src/main/scala/org/powerapi/core/target/Target.scala

@@ -38,7 +38,9 @@ trait Target
  * @author <a href="mailto:romain.rouvoy@univ-lille1.fr">Romain Rouvoy</a>
  * @author <a href="mailto:maxime.colmant@gmail.com">Maxime Colmant</a>
  */
-case class Process(pid: Int) extends Target
+case class Process(pid: Int) extends Target {
+  override def toString(): String = s"$pid"
+}
 
 /**
  * Monitoring target for a specific application.
@@ -48,7 +50,9 @@ case class Process(pid: Int) extends Target
  * @author <a href="mailto:romain.rouvoy@univ-lille1.fr">Romain Rouvoy</a>
  * @author <a href="mailto:maxime.colmant@gmail.com">Maxime Colmant</a>
  */
-case class Application(name: String) extends Target
+case class Application(name: String) extends Target {
+  override def toString(): String = name
+}
 
 /**
  * Target usage ratio.

powerapi-core/src/main/scala/org/powerapi/module/PowerChannel.scala

@@ -27,7 +27,7 @@ import java.util.UUID
 import org.powerapi.core.{Message, MessageBus, Channel}
 import org.powerapi.core.ClockChannel.ClockTick
 import org.powerapi.core.power._
-import org.powerapi.core.target.{intToProcess, Target}
+import org.powerapi.core.target.Target
 import scala.concurrent.duration.DurationInt
 
 /**
@@ -41,18 +41,12 @@ object PowerChannel extends Channel {
 
   type M = PowerReport
 
-  case object PowerReport
-
   /**
    * Base trait for each power report
    */
   trait PowerReport extends Message {
     def muid: UUID
-    def target: Target
-    def power: Power
-    def device: String
     def tick: ClockTick
-    override def toString() = s"timestamp=${tick.timestamp};target=$target;device=$device;value=$power"
   }
 
   /**
@@ -73,46 +67,37 @@ object PowerChannel extends Channel {
                             tick: ClockTick) extends PowerReport
 
   /**
-   * Used to represent an aggregated PowerReport.
-   *
-   * @param muid: monitor unique identifier (MUID), which is at the origin of the report flow.
-   * @param aggFunction: aggregate power estimation for a specific sample of power reports.
+   * AggregatePowerReport is represented as a dedicated type of message.
    */
-  case class AggregateReport(muid: UUID, aggFunction: Seq[Power] => Power) extends PowerReport {
-    private val values: collection.mutable.Buffer[Power] = collection.mutable.Buffer.empty
-    private lazy val agg = aggFunction(values.seq)
-    private var lastPowerReport: PowerReport = RawPowerReport(aggPowerReportTopic(muid),
-                                                              muid,
-                                                              -1,
-                                                              0.W,
-                                                              "none",
-                                                              ClockTick("none", 0.milliseconds))
+  case class AggregatePowerReport(muid: UUID, aggFunction: Seq[Power] => Power) extends PowerReport {
+    private val reports = collection.mutable.Buffer[RawPowerReport]()
+    private lazy val agg = aggFunction(for(report <- reports) yield report.power)
 
-    def size: Int = values.size
-    def +=(value: PowerReport):AggregateReport = {
-      values append value.power
-      lastPowerReport = value
+    def size: Int = reports.size
+
+    def +=(report: RawPowerReport): AggregatePowerReport = {
+      reports += report
       this
     }
 
     val topic: String = aggPowerReportTopic(muid)
-    def target: Target = lastPowerReport.target
+    def targets: Set[Target] = (for(report <- reports) yield report.target).toSet
     def power: Power = agg
-    def device: String = lastPowerReport.device
-    def tick: ClockTick = lastPowerReport.tick
+    def devices: Set[String] = (for(report <- reports) yield report.device).toSet
+    def tick: ClockTick = if(reports.nonEmpty) reports.last.tick else ClockTick("", 0.seconds)
   }
 
   /**
-   * Publish a power report in the event bus.
+   * Publish a raw power report in the event bus.
    */
-  def publishPowerReport(muid: UUID, target: Target, power: Power, device: String, tick: ClockTick): MessageBus => Unit = {
-    publish(RawPowerReport(powerReportMuid(muid), muid, target, power, device, tick))
+  def publishRawPowerReport(muid: UUID, target: Target, power: Power, device: String, tick: ClockTick): MessageBus => Unit = {
+    publish(RawPowerReport(rawPowerReportMuid(muid), muid, target, power, device, tick))
   }
 
   /**
    * Publish an aggregated power report in the event bus.
    */
-  def render(aggR: AggregateReport): MessageBus => Unit = {
+  def render(aggR: AggregatePowerReport): MessageBus => Unit = {
     publish(aggR)
   }
 
@@ -130,18 +115,18 @@ object PowerChannel extends Channel {
   /**
    * External method used by the MonitorChild actors for interacting with the bus.
    */
-  def subscribePowerReport(muid: UUID): MessageBus => ActorRef => Unit = {
-    subscribe(powerReportMuid(muid))
+  def subscribeRawPowerReport(muid: UUID): MessageBus => ActorRef => Unit = {
+    subscribe(rawPowerReportMuid(muid))
   }
 
-  def unsubscribePowerReport(muid: UUID): MessageBus => ActorRef => Unit = {
-    unsubscribe(powerReportMuid(muid))
+  def unsubscribeRawPowerReport(muid: UUID): MessageBus => ActorRef => Unit = {
+    unsubscribe(rawPowerReportMuid(muid))
   }
 
   /**
    * Use to format a MUID to an associated topic.
    */
-  private def powerReportMuid(muid: UUID): String  = {
+  private def rawPowerReportMuid(muid: UUID): String  = {
     s"power:$muid"
   }
 

powerapi-core/src/main/scala/org/powerapi/module/cpu/dvfs/CpuFormula.scala

@@ -25,7 +25,7 @@ package org.powerapi.module.cpu.dvfs
 import org.powerapi.core.MessageBus
 import org.powerapi.core.power._
 import org.powerapi.module.cpu.UsageMetricsChannel.{subscribeDvfsUsageReport, UsageReport}
-import org.powerapi.module.PowerChannel.publishPowerReport
+import org.powerapi.module.PowerChannel.publishRawPowerReport
 
 /**
  * CPU formula component giving CPU energy of a given process in computing the ratio between
@@ -71,6 +71,6 @@ class CpuFormula(eventBus: MessageBus, tdp: Double, tdpFactor: Double, frequenci
       case _ => 0d.W
     }
 
-    publishPowerReport(sensorReport.muid, sensorReport.target, p, "cpu", sensorReport.tick)(eventBus)
+    publishRawPowerReport(sensorReport.muid, sensorReport.target, p, "cpu", sensorReport.tick)(eventBus)
   }
 }

powerapi-core/src/main/scala/org/powerapi/module/cpu/simple/CpuFormula.scala

@@ -26,7 +26,7 @@ import org.powerapi.core.{Configuration, MessageBus}
 import org.powerapi.core.power._
 import org.powerapi.module.FormulaComponent
 import org.powerapi.module.cpu.UsageMetricsChannel.{subscribeSimpleUsageReport, UsageReport}
-import org.powerapi.module.PowerChannel.publishPowerReport
+import org.powerapi.module.PowerChannel.publishRawPowerReport
 
 
 /**
@@ -45,6 +45,6 @@ class CpuFormula(eventBus: MessageBus, tdp: Double, tdpFactor: Double) extends F
 
   def compute(sensorReport: UsageReport): Unit = {
     lazy val power = ((tdp * tdpFactor) * sensorReport.targetRatio.ratio).W
-    publishPowerReport(sensorReport.muid, sensorReport.target, power, "cpu", sensorReport.tick)(eventBus)
+    publishRawPowerReport(sensorReport.muid, sensorReport.target, power, "cpu", sensorReport.tick)(eventBus)
   }
 }

powerapi-core/src/main/scala/org/powerapi/module/libpfm/LibpfmCoreModule.scala

@@ -28,7 +28,7 @@ import org.powerapi.module.libpfm.cycles.{LibpfmCoreCyclesFormulaConfiguration,
 import scala.collection.BitSet
 import scala.concurrent.duration.FiniteDuration
 
-class LibpfmCoreModule(timeout: Timeout, topology: Map[Int, Iterable[Int]], configuration: BitSet, events: List[String],
+class LibpfmCoreModule(timeout: Timeout, topology: Map[Int, Set[Int]], configuration: BitSet, events: Set[String],
                        cyclesThreadName: String, cyclesRefName: String, formulae: Map[Double, List[Double]], samplingInterval: FiniteDuration) extends PowerModule {
 
   lazy val underlyingSensorsClasses  = Seq((classOf[LibpfmCoreSensor], Seq(timeout, topology, configuration, events)))
@@ -37,11 +37,11 @@ class LibpfmCoreModule(timeout: Timeout, topology: Map[Int, Iterable[Int]], conf
 
 object LibpfmCoreModule extends LibpfmCoreSensorConfiguration with LibpfmCoreCyclesFormulaConfiguration {
   def apply(): LibpfmCoreModule = {
-    new LibpfmCoreModule(timeout, topology, configuration, events, cyclesThreadName, cyclesThreadName, formulae, samplingInterval)
+    new LibpfmCoreModule(timeout, topology, configuration, events, cyclesThreadName, cyclesRefName, formulae, samplingInterval)
   }
 }
 
-class LibpfmCoreSensorModule(timeout: Timeout, topology: Map[Int, Iterable[Int]], configuration: BitSet, events: List[String]) extends PowerModule {
+class LibpfmCoreSensorModule(timeout: Timeout, topology: Map[Int, Set[Int]], configuration: BitSet, events: Set[String]) extends PowerModule {
   lazy val underlyingSensorsClasses  = Seq((classOf[LibpfmCoreSensor], Seq(timeout, topology, configuration, events)))
   lazy val underlyingFormulaeClasses = Seq()
 }
@@ -51,7 +51,7 @@ object LibpfmCoreSensorModule extends LibpfmCoreSensorConfiguration {
     new LibpfmCoreSensorModule(timeout, topology, configuration, events)
   }
 
-  def apply(events: List[String]): LibpfmCoreSensorModule = {
+  def apply(events: Set[String]): LibpfmCoreSensorModule = {
     new LibpfmCoreSensorModule(timeout, topology, configuration, events)
   }
 }

powerapi-core/src/main/scala/org/powerapi/module/libpfm/LibpfmCoreProcessModule.scala

@@ -29,7 +29,7 @@ import org.powerapi.module.libpfm.cycles.{LibpfmCoreCyclesFormulaConfiguration,
 import scala.collection.BitSet
 import scala.concurrent.duration.FiniteDuration
 
-class LibpfmCoreProcessModule(timeout: Timeout, topology: Map[Int, Iterable[Int]], configuration: BitSet, events: List[String], inDepth: Boolean,
+class LibpfmCoreProcessModule(timeout: Timeout, topology: Map[Int, Set[Int]], configuration: BitSet, events: Set[String], inDepth: Boolean,
                               cyclesThreadName: String, cyclesRefName: String, formulae: Map[Double, List[Double]], samplingInterval: FiniteDuration) extends PowerModule {
 
   lazy val underlyingSensorsClasses  = Seq((classOf[LibpfmCoreProcessSensor], Seq(new LinuxHelper, timeout, topology, configuration, events, inDepth)))

powerapi-core/src/main/scala/org/powerapi/module/libpfm/LibpfmCoreProcessSensor.scala

@@ -41,7 +41,7 @@ import scala.reflect.ClassTag
  *
  * @author <a href="mailto:maxime.colmant@gmail.com">Maxime Colmant</a>
  */
-class LibpfmCoreProcessSensor(eventBus: MessageBus, osHelper: OSHelper, timeout: Timeout, topology: Map[Int, Iterable[Int]], configuration: BitSet, events: List[String], inDepth: Boolean) extends APIComponent {
+class LibpfmCoreProcessSensor(eventBus: MessageBus, osHelper: OSHelper, timeout: Timeout, topology: Map[Int, Set[Int]], configuration: BitSet, events: Set[String], inDepth: Boolean) extends APIComponent {
   val processClaz = implicitly[ClassTag[Process]].runtimeClass
   val appClaz = implicitly[ClassTag[Application]].runtimeClass