tests/migration/guestperf/engine.py

   1 from __future__ import print_function
   2 #
   3 # Migration test main engine
   4 #
   5 # Copyright (c) 2016 Red Hat, Inc.
   6 #
   7 # This library is free software; you can redistribute it and/or
   8 # modify it under the terms of the GNU Lesser General Public
   9 # License as published by the Free Software Foundation; either
  10 # version 2 of the License, or (at your option) any later version.
  11 #
  12 # This library is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15 # Lesser General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU Lesser General Public
  18 # License along with this library; if not, see <http://www.gnu.org/licenses/>.
  19 #
  20
  21
  22 import os
  23 import re
  24 import sys
  25 import time
  26
  27 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', '..', 'scripts'))
  28 import qemu
  29 import qmp.qmp
  30 from guestperf.progress import Progress, ProgressStats
  31 from guestperf.report import Report
  32 from guestperf.timings import TimingRecord, Timings
  33
  34
  35 class Engine(object):
  36
  37     def __init__(self, binary, dst_host, kernel, initrd, transport="tcp",
  38                  sleep=15, verbose=False, debug=False):
  39
  40         self._binary = binary # Path to QEMU binary
  41         self._dst_host = dst_host # Hostname of target host
  42         self._kernel = kernel # Path to kernel image
  43         self._initrd = initrd # Path to stress initrd
  44         self._transport = transport # 'unix' or 'tcp' or 'rdma'
  45         self._sleep = sleep
  46         self._verbose = verbose
  47         self._debug = debug
  48
  49         if debug:
  50             self._verbose = debug
  51
  52     def _vcpu_timing(self, pid, tid_list):
  53         records = []
  54         now = time.time()
  55
  56         jiffies_per_sec = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
  57         for tid in tid_list:
  58             statfile = "/proc/%d/task/%d/stat" % (pid, tid)
  59             with open(statfile, "r") as fh:
  60                 stat = fh.readline()
  61                 fields = stat.split(" ")
  62                 stime = int(fields[13])
  63                 utime = int(fields[14])
  64                 records.append(TimingRecord(tid, now, 1000 * (stime + utime) / jiffies_per_sec))
  65         return records
  66
  67     def _cpu_timing(self, pid):
  68         records = []
  69         now = time.time()
  70
  71         jiffies_per_sec = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
  72         statfile = "/proc/%d/stat" % pid
  73         with open(statfile, "r") as fh:
  74             stat = fh.readline()
  75             fields = stat.split(" ")
  76             stime = int(fields[13])
  77             utime = int(fields[14])
  78             return TimingRecord(pid, now, 1000 * (stime + utime) / jiffies_per_sec)
  79
  80     def _migrate_progress(self, vm):
  81         info = vm.command("query-migrate")
  82
  83         if "ram" not in info:
  84             info["ram"] = {}
  85
  86         return Progress(
  87             info.get("status", "active"),
  88             ProgressStats(
  89                 info["ram"].get("transferred", 0),
  90                 info["ram"].get("remaining", 0),
  91                 info["ram"].get("total", 0),
  92                 info["ram"].get("duplicate", 0),
  93                 info["ram"].get("skipped", 0),
  94                 info["ram"].get("normal", 0),
  95                 info["ram"].get("normal-bytes", 0),
  96                 info["ram"].get("dirty-pages-rate", 0),
  97                 info["ram"].get("mbps", 0),
  98                 info["ram"].get("dirty-sync-count", 0)
  99             ),
 100             time.time(),
 101             info.get("total-time", 0),
 102             info.get("downtime", 0),
 103             info.get("expected-downtime", 0),
 104             info.get("setup-time", 0),
 105             info.get("x-cpu-throttle-percentage", 0),
 106         )
 107
 108     def _migrate(self, hardware, scenario, src, dst, connect_uri):
 109         src_qemu_time = []
 110         src_vcpu_time = []
 111         src_pid = src.get_pid()
 112
 113         vcpus = src.command("query-cpus")
 114         src_threads = []
 115         for vcpu in vcpus:
 116             src_threads.append(vcpu["thread_id"])
 117
 118         # XXX how to get dst timings on remote host ?
 119
 120         if self._verbose:
 121             print("Sleeping %d seconds for initial guest workload run" % self._sleep)
 122         sleep_secs = self._sleep
 123         while sleep_secs > 1:
 124             src_qemu_time.append(self._cpu_timing(src_pid))
 125             src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 126             time.sleep(1)
 127             sleep_secs -= 1
 128
 129         if self._verbose:
 130             print("Starting migration")
 131         if scenario._auto_converge:
 132             resp = src.command("migrate-set-capabilities",
 133                                capabilities = [
 134                                    { "capability": "auto-converge",
 135                                      "state": True }
 136                                ])
 137             resp = src.command("migrate-set-parameters",
 138                                x_cpu_throttle_increment=scenario._auto_converge_step)
 139
 140         if scenario._post_copy:
 141             resp = src.command("migrate-set-capabilities",
 142                                capabilities = [
 143                                    { "capability": "postcopy-ram",
 144                                      "state": True }
 145                                ])
 146             resp = dst.command("migrate-set-capabilities",
 147                                capabilities = [
 148                                    { "capability": "postcopy-ram",
 149                                      "state": True }
 150                                ])
 151
 152         resp = src.command("migrate_set_speed",
 153                            value=scenario._bandwidth * 1024 * 1024)
 154
 155         resp = src.command("migrate_set_downtime",
 156                            value=scenario._downtime / 1024.0)
 157
 158         if scenario._compression_mt:
 159             resp = src.command("migrate-set-capabilities",
 160                                capabilities = [
 161                                    { "capability": "compress",
 162                                      "state": True }
 163                                ])
 164             resp = src.command("migrate-set-parameters",
 165                                compress_threads=scenario._compression_mt_threads)
 166             resp = dst.command("migrate-set-capabilities",
 167                                capabilities = [
 168                                    { "capability": "compress",
 169                                      "state": True }
 170                                ])
 171             resp = dst.command("migrate-set-parameters",
 172                                decompress_threads=scenario._compression_mt_threads)
 173
 174         if scenario._compression_xbzrle:
 175             resp = src.command("migrate-set-capabilities",
 176                                capabilities = [
 177                                    { "capability": "xbzrle",
 178                                      "state": True }
 179                                ])
 180             resp = dst.command("migrate-set-capabilities",
 181                                capabilities = [
 182                                    { "capability": "xbzrle",
 183                                      "state": True }
 184                                ])
 185             resp = src.command("migrate-set-cache-size",
 186                                value=(hardware._mem * 1024 * 1024 * 1024 / 100 *
 187                                       scenario._compression_xbzrle_cache))
 188
 189         resp = src.command("migrate", uri=connect_uri)
 190
 191         post_copy = False
 192         paused = False
 193
 194         progress_history = []
 195
 196         start = time.time()
 197         loop = 0
 198         while True:
 199             loop = loop + 1
 200             time.sleep(0.05)
 201
 202             progress = self._migrate_progress(src)
 203             if (loop % 20) == 0:
 204                 src_qemu_time.append(self._cpu_timing(src_pid))
 205                 src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 206
 207             if (len(progress_history) == 0 or
 208                 (progress_history[-1]._ram._iterations <
 209                  progress._ram._iterations)):
 210                 progress_history.append(progress)
 211
 212             if progress._status in ("completed", "failed", "cancelled"):
 213                 if progress._status == "completed" and paused:
 214                     dst.command("cont")
 215                 if progress_history[-1] != progress:
 216                     progress_history.append(progress)
 217
 218                 if progress._status == "completed":
 219                     if self._verbose:
 220                         print("Sleeping %d seconds for final guest workload run" % self._sleep)
 221                     sleep_secs = self._sleep
 222                     while sleep_secs > 1:
 223                         time.sleep(1)
 224                         src_qemu_time.append(self._cpu_timing(src_pid))
 225                         src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 226                         sleep_secs -= 1
 227
 228                 return [progress_history, src_qemu_time, src_vcpu_time]
 229
 230             if self._verbose and (loop % 20) == 0:
 231                 print("Iter %d: remain %5dMB of %5dMB (total %5dMB @ %5dMb/sec)" % (
 232                     progress._ram._iterations,
 233                     progress._ram._remaining_bytes / (1024 * 1024),
 234                     progress._ram._total_bytes / (1024 * 1024),
 235                     progress._ram._transferred_bytes / (1024 * 1024),
 236                     progress._ram._transfer_rate_mbs,
 237                 ))
 238
 239             if progress._ram._iterations > scenario._max_iters:
 240                 if self._verbose:
 241                     print("No completion after %d iterations over RAM" % scenario._max_iters)
 242                 src.command("migrate_cancel")
 243                 continue
 244
 245             if time.time() > (start + scenario._max_time):
 246                 if self._verbose:
 247                     print("No completion after %d seconds" % scenario._max_time)
 248                 src.command("migrate_cancel")
 249                 continue
 250
 251             if (scenario._post_copy and
 252                 progress._ram._iterations >= scenario._post_copy_iters and
 253                 not post_copy):
 254                 if self._verbose:
 255                     print("Switching to post-copy after %d iterations" % scenario._post_copy_iters)
 256                 resp = src.command("migrate-start-postcopy")
 257                 post_copy = True
 258
 259             if (scenario._pause and
 260                 progress._ram._iterations >= scenario._pause_iters and
 261                 not paused):
 262                 if self._verbose:
 263                     print("Pausing VM after %d iterations" % scenario._pause_iters)
 264                 resp = src.command("stop")
 265                 paused = True
 266
 267     def _get_common_args(self, hardware, tunnelled=False):
 268         args = [
 269             "noapic",
 270             "edd=off",
 271             "printk.time=1",
 272             "noreplace-smp",
 273             "cgroup_disable=memory",
 274             "pci=noearly",
 275             "console=ttyS0",
 276         ]
 277         if self._debug:
 278             args.append("debug")
 279         else:
 280             args.append("quiet")
 281
 282         args.append("ramsize=%s" % hardware._mem)
 283
 284         cmdline = " ".join(args)
 285         if tunnelled:
 286             cmdline = "'" + cmdline + "'"
 287
 288         argv = [
 289             "-machine", "accel=kvm",
 290             "-cpu", "host",
 291             "-kernel", self._kernel,
 292             "-initrd", self._initrd,
 293             "-append", cmdline,
 294             "-chardev", "stdio,id=cdev0",
 295             "-device", "isa-serial,chardev=cdev0",
 296             "-m", str((hardware._mem * 1024) + 512),
 297             "-smp", str(hardware._cpus),
 298         ]
 299
 300         if self._debug:
 301             argv.extend(["-device", "sga"])
 302
 303         if hardware._prealloc_pages:
 304             argv_source += ["-mem-path", "/dev/shm",
 305                             "-mem-prealloc"]
 306         if hardware._locked_pages:
 307             argv_source += ["-realtime", "mlock=on"]
 308         if hardware._huge_pages:
 309             pass
 310
 311         return argv
 312
 313     def _get_src_args(self, hardware):
 314         return self._get_common_args(hardware)
 315
 316     def _get_dst_args(self, hardware, uri):
 317         tunnelled = False
 318         if self._dst_host != "localhost":
 319             tunnelled = True
 320         argv = self._get_common_args(hardware, tunnelled)
 321         return argv + ["-incoming", uri]
 322
 323     @staticmethod
 324     def _get_common_wrapper(cpu_bind, mem_bind):
 325         wrapper = []
 326         if len(cpu_bind) > 0 or len(mem_bind) > 0:
 327             wrapper.append("numactl")
 328             if cpu_bind:
 329                 wrapper.append("--physcpubind=%s" % ",".join(cpu_bind))
 330             if mem_bind:
 331                 wrapper.append("--membind=%s" % ",".join(mem_bind))
 332
 333         return wrapper
 334
 335     def _get_src_wrapper(self, hardware):
 336         return self._get_common_wrapper(hardware._src_cpu_bind, hardware._src_mem_bind)
 337
 338     def _get_dst_wrapper(self, hardware):
 339         wrapper = self._get_common_wrapper(hardware._dst_cpu_bind, hardware._dst_mem_bind)
 340         if self._dst_host != "localhost":
 341             return ["ssh",
 342                     "-R", "9001:localhost:9001",
 343                     self._dst_host] + wrapper
 344         else:
 345             return wrapper
 346
 347     def _get_timings(self, vm):
 348         log = vm.get_log()
 349         if not log:
 350             return []
 351         if self._debug:
 352             print(log)
 353
 354         regex = r"[^\s]+\s\((\d+)\):\sINFO:\s(\d+)ms\scopied\s\d+\sGB\sin\s(\d+)ms"
 355         matcher = re.compile(regex)
 356         records = []
 357         for line in log.split("\n"):
 358             match = matcher.match(line)
 359             if match:
 360                 records.append(TimingRecord(int(match.group(1)),
 361                                             int(match.group(2)) / 1000.0,
 362                                             int(match.group(3))))
 363         return records
 364
 365     def run(self, hardware, scenario, result_dir=os.getcwd()):
 366         abs_result_dir = os.path.join(result_dir, scenario._name)
 367
 368         if self._transport == "tcp":
 369             uri = "tcp:%s:9000" % self._dst_host
 370         elif self._transport == "rdma":
 371             uri = "rdma:%s:9000" % self._dst_host
 372         elif self._transport == "unix":
 373             if self._dst_host != "localhost":
 374                 raise Exception("Running use unix migration transport for non-local host")
 375             uri = "unix:/var/tmp/qemu-migrate-%d.migrate" % os.getpid()
 376             try:
 377                 os.remove(uri[5:])
 378                 os.remove(monaddr)
 379             except:
 380                 pass
 381
 382         if self._dst_host != "localhost":
 383             dstmonaddr = ("localhost", 9001)
 384         else:
 385             dstmonaddr = "/var/tmp/qemu-dst-%d-monitor.sock" % os.getpid()
 386         srcmonaddr = "/var/tmp/qemu-src-%d-monitor.sock" % os.getpid()
 387
 388         src = qemu.QEMUMachine(self._binary,
 389                                args=self._get_src_args(hardware),
 390                                wrapper=self._get_src_wrapper(hardware),
 391                                name="qemu-src-%d" % os.getpid(),
 392                                monitor_address=srcmonaddr)
 393
 394         dst = qemu.QEMUMachine(self._binary,
 395                                args=self._get_dst_args(hardware, uri),
 396                                wrapper=self._get_dst_wrapper(hardware),
 397                                name="qemu-dst-%d" % os.getpid(),
 398                                monitor_address=dstmonaddr)
 399
 400         try:
 401             src.launch()
 402             dst.launch()
 403
 404             ret = self._migrate(hardware, scenario, src, dst, uri)
 405             progress_history = ret[0]
 406             qemu_timings = ret[1]
 407             vcpu_timings = ret[2]
 408             if uri[0:5] == "unix:":
 409                 os.remove(uri[5:])
 410             if self._verbose:
 411                 print("Finished migration")
 412
 413             src.shutdown()
 414             dst.shutdown()
 415
 416             return Report(hardware, scenario, progress_history,
 417                           Timings(self._get_timings(src) + self._get_timings(dst)),
 418                           Timings(qemu_timings),
 419                           Timings(vcpu_timings),
 420                           self._binary, self._dst_host, self._kernel,
 421                           self._initrd, self._transport, self._sleep)
 422         except Exception as e:
 423             if self._debug:
 424                 print("Failed: %s" % str(e))
 425             try:
 426                 src.shutdown()
 427             except:
 428                 pass
 429             try:
 430                 dst.shutdown()
 431             except:
 432                 pass
 433
 434             if self._debug:
 435                 print(src.get_log())
 436                 print(dst.get_log())
 437             raise
 438