lunaix-os/scripts/qemu.py

   1 #!/usr/bin/env python
   2
   3 import subprocess, time, os, re, argparse, json
   4 from pathlib import PurePosixPath
   5
   6 g_lookup = {}
   7
   8 subsitution = re.compile(r"\$[A-Za-z_][A-Za-z0-9_]*")
   9
  10 def do_expand(val, path):
  11     global g_lookup
  12
  13     parts = []
  14     prev_pos = 0
  15     iters = subsitution.finditer(val)
  16     for _, m in enumerate(iters, start=1):
  17         start_pos = m.start()
  18         end_pos = m.end()
  19
  20         parts.append(val[prev_pos:start_pos])
  21
  22         key = val[start_pos + 1:end_pos]
  23         if key not in g_lookup:
  24             raise Exception(
  25                 f"reference key {key} (config: {path}) is not defined")
  26
  27         parts.append(g_lookup[key])
  28         prev_pos = end_pos
  29
  30     parts.append(val[prev_pos:])
  31     return "".join(parts)
  32
  33 def get_config(opt, path, default=None, required=False):
  34     _path = PurePosixPath(path)
  35     for p in _path.parts:
  36         if p in opt:
  37             opt = opt[p]
  38             continue
  39         if required:
  40             raise Exception(f"config: {path} is required")
  41         return default
  42
  43     if not isinstance(opt, str):
  44         return opt
  45
  46     return do_expand(opt, path)
  47
  48 def join_attrs(attrs):
  49     return ",".join(attrs)
  50
  51 def parse_protocol(opt):
  52     protocol = get_config(opt, "protocol", "telnet")
  53     addr     = get_config(opt, "addr", ":12345")
  54     logfile  = get_config(opt, "logfile")
  55
  56     return (f"{protocol}:{addr}", logfile)
  57
  58 class QEMUPeripherals:
  59     def __init__(self, name, opt) -> None:
  60         self.name = name
  61         self._opt = opt
  62
  63     def get_qemu_opts(self) -> list:
  64         pass
  65
  66 class BasicSerialDevice(QEMUPeripherals):
  67     def __init__(self, opt) -> None:
  68         super().__init__("serial", opt)
  69
  70     def get_qemu_opts(self):
  71         link, logfile = parse_protocol(self._opt)
  72
  73         cmds = [ link, "server", "nowait" ]
  74         if logfile:
  75             cmds.append(f"logfile={logfile}")
  76         return [ "-serial", join_attrs(cmds) ]
  77
  78 class PCISerialDevice(QEMUPeripherals):
  79     def __init__(self, opt) -> None:
  80         super().__init__("pci-serial", opt)
  81
  82     def get_qemu_opts(self):
  83         name = f"chrdev.{hex(self.__hash__())[2:]}"
  84         cmds = [ "pci-serial", f"chardev={name}" ]
  85         chrdev = [ "file", f"id={name}" ]
  86
  87         logfile = get_config(self._opt, "logfile", required=True)
  88         chrdev.append(f"path={logfile}")
  89         ()
  90         return [
  91             "-chardev", join_attrs(chrdev),
  92             "-device", join_attrs(cmds)
  93          ]
  94
  95 class AHCIBus(QEMUPeripherals):
  96     def __init__(self, opt) -> None:
  97         super().__init__("ahci", opt)
  98
  99     def get_qemu_opts(self):
 100         opt = self._opt
 101         name: str = get_config(opt, "name", required=True)
 102         name = name.strip().replace(" ", "_")
 103         cmds = [ "-device", f"ahci,id={name}" ]
 104
 105         for i, disk in enumerate(get_config(opt, "disks", default=[])):
 106             d_type = get_config(disk, "type",   default="ide-hd")
 107             d_img  = get_config(disk, "img",    required=True)
 108             d_ro   = get_config(disk, "ro",     default=False)
 109             d_fmt  = get_config(disk, "format", default="raw")
 110             d_id   = f"disk_{i}"
 111
 112             cmds += [
 113                 "-drive", join_attrs([
 114                     f"id={d_id},"
 115                     f"file={d_img}",
 116                     f"readonly={'on' if d_ro else 'off'}",
 117                     f"if=none",
 118                     f"format={d_fmt}"
 119                 ]),
 120                 "-device", join_attrs([
 121                     d_type,
 122                     f"drive={d_id}",
 123                     f"bus={name}.{i}"
 124                 ])
 125             ]
 126
 127         return cmds
 128
 129 class RTCDevice(QEMUPeripherals):
 130     def __init__(self, opt) -> None:
 131         super().__init__("rtc", opt)
 132
 133     def get_qemu_opts(self):
 134         opt = self._opt
 135         base = get_config(opt, "base", default="utc")
 136         return [ "-rtc", f"base={base}" ]
 137
 138 class QEMUMonitor(QEMUPeripherals):
 139     def __init__(self, opt) -> None:
 140         super().__init__("monitor", opt)
 141
 142     def get_qemu_opts(self):
 143         link, logfile = parse_protocol(self._opt)
 144
 145         return [
 146             "-monitor", join_attrs([
 147                 link,
 148                 "server",
 149                 "nowait",
 150                 f"logfile={logfile}"
 151             ])
 152         ]
 153
 154 class QEMUExec:
 155     devices = {
 156         "basic_serial": BasicSerialDevice,
 157         "ahci": AHCIBus,
 158         "rtc": RTCDevice,
 159         "hmp": QEMUMonitor,
 160         "pci-serial": PCISerialDevice
 161     }
 162
 163     def __init__(self, options) -> None:
 164         self._opt = options
 165         self._devices = []
 166
 167         for dev in get_config(options, "devices", default=[]):
 168             dev_class = get_config(dev, "class")
 169             if dev_class not in QEMUExec.devices:
 170                 raise Exception(f"device class: {dev_class} is not defined")
 171
 172             self._devices.append(QEMUExec.devices[dev_class](dev))
 173
 174     def get_qemu_exec_name(self):
 175         pass
 176
 177     def get_qemu_arch_opts(self):
 178         cmds = [
 179             "-machine", get_config(self._opt, "machine"),
 180             "-cpu", join_attrs([
 181                  get_config(self._opt, "cpu/type", required=True),
 182                 *get_config(self._opt, "cpu/features", default=[]),
 183              ])
 184         ]
 185
 186         return cmds
 187
 188     def get_qemu_debug_opts(self):
 189         cmds = [ "-no-reboot", "-no-shutdown" ]
 190         debug = get_config(self._opt, "debug")
 191         if not debug:
 192             return cmds
 193
 194         cmds.append("-S")
 195         cmds += [ "-gdb", f"tcp::{get_config(debug, 'gdb_port', default=1234)}" ]
 196
 197         trace_opts = get_config(debug, "traced", [])
 198         for trace in trace_opts:
 199             cmds += [ "-d", f"trace:{trace}"]
 200
 201         return cmds
 202
 203     def get_qemu_general_opts(self):
 204         return [
 205             "-m", get_config(self._opt, "memory", required=True),
 206             "-smp", get_config(self._opt, "ncpu", default=1)
 207         ]
 208
 209     def add_peripheral(self, peripheral):
 210         self._devices.append(peripheral)
 211
 212     def start(self, qemu_dir_override=""):
 213         qemu_path = self.get_qemu_exec_name()
 214         qemu_path = os.path.join(qemu_dir_override, qemu_path)
 215         cmds = [
 216             qemu_path,
 217             *self.get_qemu_arch_opts(),
 218             *self.get_qemu_debug_opts()
 219         ]
 220
 221         for dev in self._devices:
 222             cmds += dev.get_qemu_opts()
 223
 224         print(" ".join(cmds), "\n")
 225
 226         handle = subprocess.Popen(cmds)
 227
 228         while True:
 229             ret_code = handle.poll()
 230             if ret_code is not None:
 231                 return ret_code
 232             time.sleep(5)
 233
 234 class QEMUx86Exec(QEMUExec):
 235     def __init__(self, options) -> None:
 236         super().__init__(options)
 237
 238     def get_qemu_exec_name(self):
 239         if get_config(self._opt, "arch") in ["i386", "x86_32"]:
 240             return "qemu-system-i386"
 241         else:
 242             return "qemu-system-x86_64"
 243
 244 def main():
 245     global g_lookup
 246
 247     arg = argparse.ArgumentParser()
 248
 249     arg.add_argument("config_file")
 250     arg.add_argument("--qemu-dir", default="")
 251     arg.add_argument("-v", "--values", action='append', default=[])
 252
 253     arg_opt = arg.parse_args()
 254
 255     opts = {}
 256     with open(arg_opt.config_file, 'r') as f:
 257         opts.update(json.loads(f.read()))
 258
 259     for kv in arg_opt.values:
 260         [k, v] = kv.split('=')
 261         g_lookup[k] = v
 262
 263     arch = get_config(opts, "arch")
 264
 265     q = None
 266     if arch in ["i386", "x86_32", "x86_64"]:
 267         q = QEMUx86Exec(opts)
 268     else:
 269         raise Exception(f"undefined arch: {arch}")
 270
 271     q.start(arg_opt.qemu_dir)
 272
 273 if __name__ == "__main__":
 274     try:
 275         main()
 276     except Exception as e:
 277         print(e)