Simulate interactivity boost for simulated user input events

telemetry/telemetry/internal/platform/cros_platform_backend.py

 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 import logging
 
 from telemetry import decorators
 from telemetry.core import cros_interface
 from telemetry.core import platform
 from telemetry.core import util
 from telemetry.internal.forwarders import cros_forwarder
 from telemetry.internal.platform import cros_device
 from telemetry.internal.platform import linux_based_platform_backend
 from telemetry.internal.platform.power_monitor import cros_power_monitor
 from telemetry.internal.util import ps_util
+from telemetry.internal.util import repeating_timer
 
 
 class CrosPlatformBackend(
     linux_based_platform_backend.LinuxBasedPlatformBackend):
   def __init__(self, device=None):
     super(CrosPlatformBackend, self).__init__(device)
     if device and not device.is_local:
       self._cri = cros_interface.CrOSInterface(
           device.host_name, device.ssh_port, device.ssh_identity)
       self._cri.TryLogin()
     else:
       self._cri = cros_interface.CrOSInterface()
     self._powermonitor = cros_power_monitor.CrosPowerMonitor(self)
+    self._InitialiseInteractiveBoost()
 
   @classmethod
   def IsPlatformBackendForHost(cls):
     return util.IsRunningOnCrosDevice()
 
   @classmethod
   def SupportsDevice(cls, device):
     return isinstance(device, cros_device.CrOSDevice)
 
   @classmethod
@@ skipping 122 matching lines @@
       logging.warning(
           'PathExists: params timeout and retries are not support on CrOS.')
     return self._cri.FileExistsOnDevice(path)
 
   def CanTakeScreenshot(self):
     # crbug.com/609001: screenshots don't work on VMs.
     return not self.cri.IsRunningOnVM()
 
   def TakeScreenshot(self, file_path):
     return self._cri.TakeScreenshot(file_path)
+
+  def _InitialiseInteractiveBoost(self):
+    """On ChromeOS on ARM, the kernel boosts CPU frequency when an input (key
+    press, touch event, etc) is received. For simulated inputs, this doesn't
+    happen, so this functionality activates the same boost mechanism by
+    using the sysfs controls.
+    The boost should be turned on for specific interaction events, for the
+    duration of the interaction, to simulate normal system behaviour.
+    On systems which do not support frequency boost, this has no effect
+    (consistent with normal non-simulated behaviour).
+    """
+    self._boost_duration = None
+    self._boost_timer = None
+    governor = self.GetFileContents(
+        "/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor").strip()
+    if governor != "interactive":
+      logging.warning("no interactive cpufreq governor present, not setting"
+          " up interactivity boost")
+      return
+    core = 0
+    durations = {}
+    while True:
+      path = "/sys/devices/system/cpu/cpu" + str(core) + "/cpufreq/interactive"
+      if self.PathExists(path + "/boostpulse_duration") \
+        and self.PathExists(path + "/boostpulse"):
+        durations[core] = self.GetFileContents(path
+            + "/boostpulse_duration").strip()
+      else:
+        break
+      core = core + 1
+    if durations:
+      # Set timer to drive boost every 9 minutes. It lasts for 10 minutes
+      # so this ensures it will remain constantly on with minimal
+      # overhead, until turned off.
+      self._boost_timer = repeating_timer.RepeatingTimer(
+          lambda: self._WriteCPUFreqControl("boostpulse", "1"), 9 * 60)
+      self._boost_duration = durations
+      logging.info("interactivity boost set up")
+    else:
+      logging.info("kernel does not support frequency boost controls,"
+          "interactivity boost could not be set up")
+
+  def _WriteCPUFreqControl(self, leaf, content):
+    if self._boost_duration is not None:
+      cmds = []
+      for core in self._boost_duration:
+        path = "/sys/devices/system/cpu/cpu" + str(core) \
+            + "/cpufreq/interactive"
+        value = content if not isinstance(content, dict) else content[core]
+        if cmds:
+          cmds.append(';')
+        cmds.extend(['echo', value, '>', path + "/" + leaf])
+      self.RunCommand(cmds)
+
+  def SetInteractivityBoost(self, enable):
+    if self._boost_duration is not None \
+        and self._boost_timer.IsRunning() != enable:
+      logging.info('SetInteractivityBoost: %s boost on %d cores',
+          ("enabling" if enable else "disabling"), len(self._boost_duration))
+
+      # enable a timer to initiate a regular boost pulse
+      #
+      # the boost pulse duration and timer are such that boost will remain
+      # constantly enabled
+      if enable:
+        # set the boost duration to 10 minutes
+        self._WriteCPUFreqControl("boostpulse_duration", "600000000")
+        self._WriteCPUFreqControl("boostpulse", "1")
+        self._boost_timer.Start()
+      else:
+        # set the boost duration back to the default setting if disabling.
+        self._WriteCPUFreqControl("boostpulse_duration", self._boost_duration)
+        self._boost_timer.Stop()
+        # this ensures that the pulse continues for the default pulse duration
+        # (typically 80 ms) after the input event ends - as would normally
+        # happen with real user interaction.
+        self._WriteCPUFreqControl("boostpulse", "1")