cpuset.py 8.5 KB
Newer Older
1 2 3 4
# coding: utf-8
import logging
import os
import os.path
5 6
import pwd
import time
7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43

from zope import interface as zope_interface
from slapos.manager import interface

logger = logging.getLogger(__name__)


class Manager(object):
  """Manage cgroup's cpuset in terms on initializing and runtime operations.

  CPUSET manager moves PIDs between CPU cores using Linux cgroup system.

  In order to use this feature put "cpuset" into "manager_list" into your slapos
  configuration file inside [slapos] section.

  TODO: there is no limit on number of reserved cores per user.
  """
  zope_interface.implements(interface.IManager)

  cpu_exclusive_file = ".slapos-cpu-exclusive"
  cpuset_path = "/sys/fs/cgroup/cpuset/"
  task_write_mode = "wt"
  config_power_user_option = "power_user_list"

  def __init__(self, config):
    """Retain access to dict-like configuration."""
    self.config = config

  def software(self, software):
    """We don't need to mingle with software."""
    pass

  def format(self, computer):
    """Create cgroup folder per-CPU with exclusive access to the CPU.

    - Those folders are "/sys/fs/cgroup/cpuset/cpu<N>".
    """
44
    if not os.path.exists(os.path.join(self.cpuset_path, "cpuset.cpus")):
45
      logger.warning("CPUSet Manager cannot format computer because cgroups do not exist.")
46
      return
47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86

    for cpu in self._cpu_id_list():
      cpu_path = self._prepare_folder(
        os.path.join(self.cpuset_path, "cpu" + str(cpu)))
      with open(cpu_path + "/cpuset.cpus", "wt") as fx:
        fx.write(str(cpu))  # this cgroup manages only this cpu
      with open(cpu_path + "/cpuset.cpu_exclusive", "wt") as fx:
        fx.write("1")  # manages it exclusively
      with open(cpu_path + "/cpuset.mems", "wt") as fx:
        fx.write("0")  # it doesn't work without that

  def instance(self, partition):
    """Control runtime state of the computer."""

    if not os.path.exists(os.path.join(self.cpuset_path, "cpu0")):
      # check whether the computer was formatted
      logger.warning("CGROUP's CPUSET Manager cannot update computer because it is not cpuset-formatted.")
      return

    request_file = os.path.join(partition.instance_path, self.cpu_exclusive_file)
    if not os.path.exists(request_file) or not read_file(request_file):
      # This instance does not ask for cpu exclusive access
      return

    # Gather list of users allowed to request exlusive cores
    power_user_list = self.config.get(self.config_power_user_option, "").split()
    uid, gid = partition.getUserGroupId()
    uname = pwd.getpwuid(uid).pw_name
    if uname not in power_user_list:
      logger.warning("User {} not allowed to modify cpuset! "
                     "Allowed users are in {} option in config file.".format(
                        uname, self.config_power_user_option))
      return

    # prepare paths to tasks file for all and per-cpu
    tasks_file = os.path.join(self.cpuset_path, "tasks")
    cpu_tasks_file_list = [os.path.join(cpu_folder, "tasks")
                           for cpu_folder in self._cpu_folder_list()]

    # Gather exclusive CPU usage map {username: set[cpu_id]}
87 88 89 90 91 92 93 94
    # We do not need to gather that since we have no limits yet
    #cpu_usage = defaultdict(set)
    #for cpu_id in self._cpu_id_list()[1:]:  # skip the first public CPU
    #  pids = [int(pid)
    #          for pid in read_file(cpu_tasks_file_list[cpu_id]).splitlines()]
    #  for pid in pids:
    #    process = psutil.Process(pid)
    #    cpu_usage[process.username()].add(cpu_id)
95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110

    # Move all PIDs from the pool of all CPUs onto the first exclusive CPU.
    running_list = sorted(list(map(int, read_file(tasks_file).split())), reverse=True)
    first_cpu = self._cpu_id_list()[0]
    success_set, refused_set = set(), set()
    for pid in running_list:
      try:
        self._move_task(pid, first_cpu)
        success_set.add(pid)
        time.sleep(0.01)
      except IOError as e:
        refused_set.add(pid)
    logger.debug("Refused to move {:d} PIDs: {!s}\n"
                 "Suceeded in moving {:d} PIDs {!s}\n".format(
                     len(refused_set), refused_set, len(success_set), success_set))

111
    cpu_folder_list = self._cpu_folder_list()
112 113 114 115 116 117 118 119 120 121
    generic_cpu_path = cpu_folder_list[0]
    exclusive_cpu_path_list = cpu_folder_list[1:]

    # Gather all running PIDs for filtering out stale PIDs
    running_pid_set = set(running_list)
    running_pid_set.update(map(int, read_file(cpu_tasks_file_list[0]).split()))

    # gather already exclusively running PIDs
    exclusive_pid_set = set()
    for cpu_tasks_file in cpu_tasks_file_list[1:]:
122
      exclusive_pid_set.update(map(int, read_file(cpu_tasks_file).split()))
123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140

    # Move processes to their demanded exclusive CPUs
    with open(request_file, "rt") as fi:
      # take such PIDs which are either really running or are not already exclusive
      request_pid_list = [int(pid) for pid in fi.read().split()
                          if int(pid) in running_pid_set or int(pid) not in exclusive_pid_set]
    with open(request_file, "wt") as fo:
      fo.write("")  # empty file (we will write back only PIDs which weren't moved)
    for request_pid in request_pid_list:
      assigned_cpu = self._move_to_exclusive_cpu(request_pid)
      if assigned_cpu < 0:
        # if no exclusive CPU was assigned - write the PID back and try other time
        with open(request_file, "at") as fo:
          fo.write(str(request_pid) + "\n")

  def _cpu_folder_list(self):
    """Return list of folders for exclusive cpu cores."""
    return [os.path.join(self.cpuset_path, "cpu" + str(cpu_id))
141
            for cpu_id in self._cpu_id_list()]
142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220

  def _cpu_id_list(self):
    """Extract IDs of available CPUs and return them as a list.

    The first one will be always used for all non-exclusive processes.
    :return: list[int]
    """
    cpu_list = []  # types: list[int]
    with open(self.cpuset_path + "cpuset.cpus", "rt") as cpu_def:
      for cpu_def_split in cpu_def.read().strip().split(","):
        # IDs can be in form "0-4" or "0,1,2,3,4"
        if "-" in cpu_def_split:
          a, b = map(int, cpu_def_split.split("-"))
          cpu_list.extend(range(a, b + 1)) # because cgroup's range is inclusive
          continue
        cpu_list.append(int(cpu_def_split))
    return cpu_list

  def _move_to_exclusive_cpu(self, pid):
    """Try all exclusive CPUs and place the ``pid`` to the first available one.

    :return: int, cpu_id of used CPU, -1 if placement was not possible
    """
    exclusive_cpu_list = self._cpu_id_list()[1:]
    for exclusive_cpu in exclusive_cpu_list:
      # gather tasks assigned to current exclusive CPU
      task_path = os.path.join(self.cpuset_path, "cpu" + str(exclusive_cpu), "tasks")
      with open(task_path, "rt") as fi:
        task_list = fi.read().split()
      if len(task_list) > 0:
        continue  # skip occupied CPUs
      return self._move_task(pid, exclusive_cpu)[1]
    return -1

  def _move_task(self, pid, cpu_id, cpu_mode="performance"):
    """Move ``pid`` to ``cpu_id``.

    cpu_mode can be "performance" or "powersave"
    """
    known_cpu_mode_list = ("performance", "powersave")
    with open(os.path.join(self.cpuset_path, "cpu" + str(cpu_id), "tasks"), self.task_write_mode) as fo:
      fo.write(str(pid) + "\n")
    # set the core to `cpu_mode`
    scaling_governor_file = "/sys/devices/system/cpu/cpu{:d}/cpufreq/scaling_governor".format(cpu_id)
    if os.path.exists(scaling_governor_file):
      if cpu_mode not in known_cpu_mode_list:
        logger.warning("Cannot set CPU to mode \"{}\"! Known modes {!s}".format(
          cpu_mode, known_cpu_mode_list))
      else:
        try:
          with open(scaling_governor_file, self.task_write_mode) as fo:
            fo.write(cpu_mode + "\n")  # default is "powersave"
        except IOError as e:
          # handle permission error
          logger.error("Failed to write \"{}\" to {}".format(cpu_mode, scaling_governor_file))
    return pid, cpu_id

  def _prepare_folder(self, folder):
    """If-Create folder and set group write permission."""
    if not os.path.exists(folder):
      os.mkdir(folder)
      # make your life and testing easier and create mandatory files if they don't exist
      mandatory_file_list = ("tasks", "cpuset.cpus")
      for mandatory_file in mandatory_file_list:
        file_path = os.path.join(folder, mandatory_file)
        if not os.path.exists(file_path):
          with open(file_path, "wb"):
            pass  # touche
    return folder


def read_file(path, mode="rt"):
  with open(path, mode) as fi:
    return fi.read()


def write_file(content, path, mode="wt"):
  with open(path, mode) as fo:
    fo.write(content)