acrn-hypervisor/misc/acrn-config/library/scenario_cfg_lib.py

# Copyright (C) 2019 Intel Corporation. All rights reserved.
#
# SPDX-License-Identifier: BSD-3-Clause
#

import common
import board_cfg_lib

HEADER_LICENSE = common.open_license()

LOAD_ORDER_TYPE = ['PRE_LAUNCHED_VM', 'SOS_VM', 'POST_LAUNCHED_VM']
START_HPA_LIST = ['0', '0x100000000', '0x120000000']

KERN_TYPE_LIST = ['KERNEL_BZIMAGE', 'KERNEL_ZEPHYR']
KERN_BOOT_ADDR_LIST = ['0x100000']

VM_SEVERITY = ['SEVERITY_SAFETY_VM', 'SEVERITY_RTVM', 'SEVERITY_SOS', 'SEVERITY_STANDARD_VM']
VUART_TYPE = ['VUART_LEGACY_PIO', 'VUART_PCI']
VUART_BASE = ['SOS_COM1_BASE', 'SOS_COM2_BASE', 'COM1_BASE',
              'COM2_BASE', 'COM3_BASE', 'COM4_BASE', 'INVALID_COM_BASE']

AVALIBLE_COM1_BASE = ['INVALID_COM_BASE', 'COM1_BASE']
AVALIBLE_COM2_BASE = ['INVALID_COM_BASE', 'COM2_BASE']

VUART_IRQ = ['SOS_COM1_IRQ', 'SOS_COM2_IRQ', 'COM1_IRQ', 'COM2_IRQ', 'COM3_IRQ',
             'COM4_IRQ', 'CONFIG_COM_IRQ', '3', '4', '6', '7']

PCI_DEV_NUM_LIST = ['SOS_EMULATED_PCI_DEV_NUM', 'VM0_CONFIG_PCI_DEV_NUM', 'VM1_CONFIG_PCI_DEV_NUM']
PCI_DEVS_LIST = ['sos_pci_devs', 'vm0_pci_devs', 'vm1_pci_devs']
# Support 512M, 1G, 2G
# pre launch less then 2G, sos vm less than 24G
START_HPA_SIZE_LIST = ['0x20000000', '0x40000000', '0x80000000', 'CONFIG_SOS_RAM_SIZE']

COMMUNICATE_VM_ID = []

ERR_LIST = {}

KATA_VM_COUNT = 0


def check_board_private_info():

    if 'SOS_VM' not in common.VM_TYPES.values():
        return
    branch_tag = "board_private"
    private_info = {}
    dev_private_tags = ['rootfs']
    for tag_str in dev_private_tags:
        dev_setting = common.get_leaf_tag_map(common.SCENARIO_INFO_FILE, branch_tag, tag_str)
        private_info[tag_str] = dev_setting

    if not private_info['rootfs'] and err_dic:
        ERR_LIST['vm:id=0,boot_private,rootfs'] = "The board have to chose one rootfs partition"
        ERR_LIST.update(err_dic)


def vm_name_check(vm_names, item):
    """
    Check vm name
    :param vm_names: dictionary of vm name
    :param item: vm name item in xml
    :return: None
    """
    for name_i, name_str in vm_names.items():
        name_len = len(name_str)
        if name_len > 32 or name_len == 0:
            key = "vm:id={},{}".format(name_i, item)
            ERR_LIST[key] = "VM name length should be in range [1,32] bytes"


def load_order_check(load_orders, item):
    """
    Check load order type
    :param load_orders: dictionary of vm load_order
    :param item: vm name item in xml
    :return: None
    """
    sos_vm_ids = []
    pre_vm_ids = []
    post_vm_ids = []
    for order_i, load_str in load_orders.items():
        if load_str == "SOS_VM":
            sos_vm_ids.append(order_i)

        if load_str == "PRE_LAUNCHED_VM":
            pre_vm_ids.append(order_i)

        if load_str == "POST_LAUNCHED_VM":
            post_vm_ids.append(order_i)

        if not load_str:
            key = "vm:id={},{}".format(order_i, item)
            ERR_LIST[key] = "VM load should not empty"
            return

        if load_str not in LOAD_ORDER_TYPE:
            key = "vm:id={},{}".format(order_i, item)
            ERR_LIST[key] = "VM load order unknown"

    if len(sos_vm_ids) >= 2:
        key = "vm:id={},{}".format(sos_vm_ids[0], item)
        ERR_LIST[key] = "SOS_VM number should not be greater than 1"

    if post_vm_ids and sos_vm_ids:
        if post_vm_ids[0] < sos_vm_ids[-1]:
            key = "vm:id={},{}".format(post_vm_ids[0], item)
            ERR_LIST[key] = "POST_LAUNCHED_VM should be configured after SOS_VM"

    if pre_vm_ids and sos_vm_ids:
        if sos_vm_ids[-1] < pre_vm_ids[-1]:
            key = "vm:id={},{}".format(sos_vm_ids[0], item)
            ERR_LIST[key] = "PRE_LAUNCHED_VM should be configured before SOS_VM"


def get_load_order_cnt(load_orders, type_name):
    """
    Get load order type count
    :param load_orders: dictionary of vm load_order
    :param type_name: load order type for vm
    :return: number for this load order type name
    """
    type_cnt = 0
    for load_str in load_orders.values():
        if type_name == load_str:
            type_cnt += 1

    return type_cnt


def guest_flag_check(guest_flags, branch_tag, leaf_tag):

    for vm_i, flags in guest_flags.items():
        for guest_flag in flags:
            if guest_flag and guest_flag not in common.GUEST_FLAG:
                key = "vm:id={},{},{}".format(vm_i, branch_tag, leaf_tag)
                ERR_LIST[key] = "Unknow guest flag"


def uuid_format_check(uuid_dic, item):
    """
    Check uuid
    :param uuid_dic: dictionary of vm uuid
    :param item: vm uuid item in xml
    :return: None
    """
    uuid_len = 36

    for uuid_i, uuid_str in uuid_dic.items():

        if not uuid_str:
            key = "vm:id={},{}".format(uuid_i, item)
            ERR_LIST[key] = "VM uuid should not empty"
            return

        uuid_str_list = list(uuid_str)
        key = "vm:id={},{}".format(uuid_i, item)

        if len(uuid_str) != uuid_len:
            ERR_LIST[key] = "VM uuid length should be 36 bytes"

        if uuid_str_list[8] != '-':
            ERR_LIST[key] = "VM uuid format unknown"


def cpus_per_vm_check(id_cpus_per_vm_dic, item):
    """
    Check cpu number of per vm
    :param item: vm pcpu_id item in xml
    :return: None
    """
    for vm_i, vm_type in common.VM_TYPES.items():
        if vm_i not in id_cpus_per_vm_dic.keys() and vm_type == "SOS_VM":
            continue
        elif vm_i not in id_cpus_per_vm_dic.keys() and vm_type in ("PRE_LAUNCHED_VM", "POST_LAUNCHED"):
            key = "vm:id={},{}".format(vm_i, item)
            ERR_LIST[key] = "Pre launched_vm and Post launched vm should have cpus assignment"
            return

        cpus_vm_i = id_cpus_per_vm_dic[vm_i]
        if not cpus_vm_i:
            key = "vm:id={},{}".format(vm_i, item)
            ERR_LIST[key] = "{} VM have no assignment cpus".format(vm_type)
            return


def mem_start_hpa_check(id_start_hpa_dic, prime_item, item):
    """
    Check host physical address
    :param prime_item: the prime item in xml file
    :param item: vm start_hpa item in xml
    :return: None
    """

    for id_key, hpa_str in id_start_hpa_dic.items():
        hpa_strip_ul = hpa_str.strip('UL')
        hpa_strip_u = hpa_str.strip('U')

        if not hpa_str:
            key = "vm:id={},{}".format(id_key, item)
            ERR_LIST[key] = "VM start host physical memory address should not empty"
            return

        if hpa_strip_ul not in START_HPA_LIST and hpa_strip_u not in START_HPA_LIST:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            if '0x' not in hpa_str and '0X' not in hpa_str:
                ERR_LIST[key] = "Address should be Hex format"


def mem_size_check(id_hpa_size_dic, prime_item, item):
    """
    Check host physical size
    :param prime_item: the prime item in xml file
    :param item: vm size item in xml
    :return: None
    """

    for id_key, hpa_size in id_hpa_size_dic.items():
        hpa_sz_strip_ul = hpa_size.strip('UL')
        hpa_sz_strip_u = hpa_size.strip('U')

        if not hpa_size:
            key = "vm:id={},{}".format(id_key, item)
            ERR_LIST[key] = "VM start host physical memory size should not empty"
            return

        if hpa_sz_strip_ul not in START_HPA_SIZE_LIST and hpa_sz_strip_u not in \
                START_HPA_SIZE_LIST:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            if '0x' not in hpa_size and '0X' not in hpa_size:
                ERR_LIST[key] = "Mem size should be Hex format"


def os_kern_name_check(id_kern_name_dic, prime_item, item):
    """
    Check os kernel name
    :param prime_item: the prime item in xml file
    :param item: vm os config name item in xml
    :return: None
    """

    for id_key, kern_name in id_kern_name_dic.items():
        if len(kern_name) > 32 or len(kern_name) == 0:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel name length should be in range [1,32] bytes"


def os_kern_type_check(id_kern_type_dic, prime_item, item):
    """
    Check os kernel type
    :param prime_item: the prime item in xml file
    :param item: vm os config type item in xml
    :return: None
    """

    for id_key, kern_type in id_kern_type_dic.items():

        if not kern_type:
            key = "vm:id={},{}".format(id_key, item)
            ERR_LIST[key] = "VM os config kernel type should not empty"
            return

        if kern_type not in KERN_TYPE_LIST:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel type unknown"


def os_kern_mod_check(id_kern_mod_dic, prime_item, item):
    """
    Check os kernel mod
    :param prime_item: the prime item in xml file
    :param item: vm os config mod item in xml
    :return: None
    """

    for id_key, kern_mod in id_kern_mod_dic.items():
        if len(kern_mod) > 32 or len(kern_mod) == 0:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel mod tag should be in range [1,32] bytes"


def os_kern_args_check(id_kern_args_dic, prime_item, item):
    """
    Check os kernel args
    :param prime_item: the prime item in xml file
    :param item: vm os config args item in xml
    :return: None
    """

    for vm_i,vm_type in common.VM_TYPES.items():
        if vm_i not in id_kern_args_dic.keys():
            continue
        kern_args = id_kern_args_dic[vm_i]
        if vm_type == "SOS_VM" and kern_args != "SOS_VM_BOOTARGS":
            key = "vm:id={},{},{}".format(vm_i, prime_item, item)
            ERR_LIST[key] = "VM os config kernel service os should be SOS_VM_BOOTARGS"


def os_kern_console_check(tty_console, prime_item, item):
    """
    Check os kernel console
    :param prime_item: the prime item in xml file
    :param item: vm os config console item in xml
    :return: None
    """

    if tty_console and "ttyS" not in tty_console:
        key = "hv:{},{}".format(prime_item, item)
        ERR_LIST[key] = "VM os config kernel console should be ttyS[0..3]"


def os_kern_load_addr_check(id_kern_load_addr_dic, prime_item, item):
    """
    Check os kernel load address
    :param prime_item: the prime item in xml file
    :param item: vm os config load address item in xml
    :return: None
    """

    for id_key, kern_load_addr in id_kern_load_addr_dic.items():

        if not kern_load_addr:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel load address should not empty"
            return

        if '0x' not in kern_load_addr and '0X' not in kern_load_addr:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel load address should Hex format"


def os_kern_entry_addr_check(id_kern_entry_addr_dic, prime_item, item):
    """
    Check os kernel entry address
    :param prime_item: the prime item in xml file
    :param item: vm os config entry address item in xml
    :return: None
    """

    for id_key, kern_entry_addr in id_kern_entry_addr_dic.items():

        if not kern_entry_addr:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel entry address should not empty"
            return

        if '0x' not in kern_entry_addr and '0X' not in kern_entry_addr:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel entry address should Hex format"


def os_kern_root_dev_check(id_kern_rootdev_dic, prime_item, item):
    """
    Check os kernel rootfs partitions
    :param prime_item: the prime item in xml file
    :param item: vm os config rootdev item in xml
    :return: None
    """

    for id_key, kern_rootdev in id_kern_rootdev_dic.items():
        if not kern_rootdev:
            key = "vm:id={},{},{}".format(id_key, prime_item, item)
            ERR_LIST[key] = "VM os config kernel root device should not empty"


def pci_dev_num_check(id_dev_num_dic, item):
    """
    Check vm pci device number
    :param item: vm pci_dev_num item in xml
    :return: None
    """

    for vm_i,vm_type in common.VM_TYPES.items():
        if vm_i not in id_dev_num_dic.keys():
            continue
        pci_dev_num = id_dev_num_dic[vm_i]
        if vm_type != "POST_LAUNCHED_VM" and pci_dev_num:
            if pci_dev_num not in PCI_DEV_NUM_LIST:
                key = "vm:id={},{}".format(vm_i, item)
                ERR_LIST[key] = "VM pci device number  shoud be one of {}".format(PCI_DEV_NUM_LIST)

def pci_devs_check(id_devs_dic, item):
    """
    Check vm pci devices
    :param item: vm pci_devs item in xml
    :return: None
    """

    for vm_i,vm_type in common.VM_TYPES.items():
        if vm_i not in id_devs_dic.keys():
            continue
        pci_dev = id_devs_dic[vm_i]
        if vm_type != "POST_LAUNCHED_VM" and pci_dev:
            if pci_dev not in PCI_DEVS_LIST:
                key = "vm:id={},{}".format(vm_i, item)
                ERR_LIST[key] = "VM pci device shoud be one of {}".format(PCI_DEVS_LIST)


def get_vuart1_vmid(vm_vuart1):
    """
    Get vmid:vuart1 dictionary
    :param vm_vuart1: vm_vuart1 setting from xml
    :return: dictionary of vmid:vuart1
    """
    vm_id_dic = {}
    for i in list(common.VM_TYPES.keys()):
        for key in vm_vuart1[i].keys():
            if key == "target_vm_id":
                vm_id_dic[i] = vm_vuart1[i][key]

    # remove the unavailable vimid:target_vmid from dictonary
    vmid_list = list(vm_id_dic.keys())

    for vmid in vmid_list:
        new_vmid = vm_id_dic[vmid]
        if int(new_vmid) in vmid_list and vmid == int(vm_id_dic[int(new_vmid)]):
            continue
        else:
            vm_id_dic.pop(vmid)

    return vm_id_dic


def cpus_assignment(cpus_per_vm, index):
    """
    Get cpu id assignment for vm by vm index
    :param cpus_per_vm: a dictionary by vmid:cpus
    :param index: vm index
    :return: cpu assignment string
    """
    vm_cpu_bmp = {}

    for i in range(len(cpus_per_vm[index])):
        if i == 0:
            if len(cpus_per_vm[index]) == 1:
                cpu_str = "{{AFFINITY_CPU({0}U)}}".format(cpus_per_vm[index][0])
            else:
                cpu_str = "{{AFFINITY_CPU({0}U)".format(cpus_per_vm[index][0])
        else:
            if i == len(cpus_per_vm[index]) - 1:
                cpu_str = cpu_str + ", AFFINITY_CPU({0}U)}}".format(cpus_per_vm[index][i])
            else:
                cpu_str = cpu_str + ", AFFINITY_CPU({0}U)".format(cpus_per_vm[index][i])

    vm_cpu_bmp['cpu_map'] = cpu_str
    vm_cpu_bmp['cpu_num'] = len(cpus_per_vm[index])
    return vm_cpu_bmp

def clos_assignment(clos_per_vm, index):
    """
    Get clos id assignment for vm by vm index
    :param clos_per_vm: a dictionary by vmid:cpus
    :param index: vm index
    :return: clos assignment string
    """
    vm_clos_bmp = {}

    for i in range(len(clos_per_vm[index])):
        if i == 0:
            if len(clos_per_vm[index]) == 1:
                clos_str = "{{{0}U}}".format(clos_per_vm[index][0])
            else:
                clos_str = "{{{0}U".format(clos_per_vm[index][0])
        else:
            if i == len(clos_per_vm[index]) - 1:
                clos_str = clos_str + ", {0}U}}".format(clos_per_vm[index][i])
            else:
                clos_str = clos_str + ", {0}U".format(clos_per_vm[index][i])

    vm_clos_bmp['clos_map'] = clos_str
    return vm_clos_bmp


def avl_vuart_ui_select(scenario_info):
    tmp_vuart = {}
    for vm_i,vm_type in common.VM_TYPES.items():

        if vm_type == "SOS_VM":
            key = "vm={},vuart=0,base".format(vm_i)
            tmp_vuart[key] = ['SOS_COM1_BASE', 'INVALID_COM_BASE']
            key = "vm={},vuart=1,base".format(vm_i)
            tmp_vuart[key] = ['SOS_COM2_BASE', 'INVALID_COM_BASE']
        else:
            key = "vm={},vuart=0,base".format(vm_i)
            tmp_vuart[key] = ['INVALID_COM_BASE', 'COM1_BASE']
            key = "vm={},vuart=1,base".format(vm_i)
            tmp_vuart[key] = ['INVALID_COM_BASE', 'COM2_BASE']

    return tmp_vuart


def get_first_post_vm():

    i = 0
    for vm_i,vm_type in common.VM_TYPES.items():
        if vm_type == "POST_LAUNCHED_VM":
            i = vm_i
            break

    return (err_dic, i)