path: root/registry/vulkan/scripts/cereal/api_log_decoder.py

import os
from typing import List, Set, Dict, Optional

from . import VulkanType, VulkanCompoundType
from .wrapperdefs import VulkanWrapperGenerator


class ApiLogDecoder(VulkanWrapperGenerator):
    """
    This class generates decoding logic for the graphics API logs captured by
    [GfxApiLogger](http://source/play-internal/battlestar/aosp/device/generic/vulkan-cereal/base/GfxApiLogger.h)

    This allows developers to see a pretty-printed version of the API log data when using
    print_gfx_logs.py
    """

    # List of Vulkan APIs that we will generate decoding logic for
    generated_apis = [
        "vkAcquireImageANDROID",
        "vkAllocateMemory",
        "vkBeginCommandBufferAsyncGOOGLE",
        "vkBindBufferMemory",
        "vkBindImageMemory",
        "vkCmdBeginRenderPass",
        "vkCmdBindDescriptorSets",
        "vkCmdBindIndexBuffer",
        "vkCmdBindPipeline",
        "vkCmdBindVertexBuffers",
        "vkCmdClearAttachments",
        "vkCmdClearColorImage",
        "vkCmdCopyBufferToImage",
        "vkCmdCopyImageToBuffer",
        "vkCmdDraw",
        "vkCmdDrawIndexed",
        "vkCmdEndRenderPass",
        "vkCmdPipelineBarrier",
        "vkCmdSetScissor",
        "vkCmdSetViewport",
        "vkCollectDescriptorPoolIdsGOOGLE",
        "vkCreateBufferWithRequirementsGOOGLE",
        "vkCreateDescriptorPool",
        "vkCreateDescriptorSetLayout",
        "vkCreateFence",
        "vkCreateFramebuffer",
        "vkCreateGraphicsPipelines",
        "vkCreateImageView",
        "vkCreateImageWithRequirementsGOOGLE",
        "vkCreatePipelineCache",
        "vkCreateRenderPass",
        "vkCreateSampler",
        "vkCreateSemaphore",
        "vkCreateShaderModule",
        "vkDestroyBuffer",
        "vkDestroyCommandPool",
        "vkDestroyDescriptorPool",
        "vkDestroyDescriptorSetLayout",
        "vkDestroyDevice",
        "vkDestroyFence",
        "vkDestroyFramebuffer",
        "vkDestroyImage",
        "vkDestroyImageView",
        "vkDestroyInstance",
        "vkDestroyPipeline",
        "vkDestroyPipelineCache",
        "vkDestroyPipelineLayout",
        "vkDestroyRenderPass",
        "vkDestroySemaphore",
        "vkDestroyShaderModule",
        "vkEndCommandBufferAsyncGOOGLE",
        "vkFreeCommandBuffers",
        "vkFreeMemory",
        "vkFreeMemorySyncGOOGLE",
        "vkGetFenceStatus",
        "vkGetMemoryHostAddressInfoGOOGLE",
        "vkGetBlobGOOGLE",
        "vkGetPhysicalDeviceFormatProperties",
        "vkGetPhysicalDeviceProperties2KHR",
        "vkGetPipelineCacheData",
        "vkGetSwapchainGrallocUsageANDROID",
        "vkQueueCommitDescriptorSetUpdatesGOOGLE",
        "vkQueueFlushCommandsGOOGLE",
        "vkQueueSignalReleaseImageANDROIDAsyncGOOGLE",
        "vkQueueSubmitAsyncGOOGLE",
        "vkQueueWaitIdle",
        "vkResetFences",
        "vkWaitForFences",
    ]

    def __init__(self, module, typeInfo):
        VulkanWrapperGenerator.__init__(self, module, typeInfo)
        self.typeInfo = typeInfo

        # Set of Vulkan structs that we need to write decoding logic for
        self.structs: Set[str] = set()

        # Maps enum group names to the list of enums in the group, for all enum groups in the spec
        # E.g.:  "VkResult": ["VK_SUCCESS", "VK_NOT_READY", "VK_TIMEOUT", etc...]
        self.all_enums: Dict[str, List[str]] = {}

        # Set of Vulkan enums that we need to write decoding logic for
        self.needed_enums: Set[str] = {"VkStructureType"}

    def onBegin(self):
        self.module.append("""
#####################################################################################################
# Pretty-printer functions for Vulkan data structures
# THIS FILE IS AUTO-GENERATED - DO NOT EDIT
#
# To re-generate this file, run generate-vulkan-sources.sh
#####################################################################################################

""".lstrip())

    def onGenGroup(self, groupinfo, groupName, alias=None):
        """Called for each enum group in the spec"""
        for enum in groupinfo.elem.findall("enum"):
            self.all_enums[groupName] = self.all_enums.get(groupName, []) + [enum.get('name')]

    def onEnd(self):
        for api_name in sorted(self.generated_apis):
            self.process_api(api_name)
        self.process_structs()
        self.process_enums()

    def process_api(self, api_name):
        """Main entry point to generate decoding logic for each Vulkan API"""
        api = self.typeInfo.apis[api_name]
        self.module.append('def OP_{}(printer, indent: int):\n'.format(api_name))

        # Decode the sequence number. All commands have sequence numbers, except those handled
        # by VkSubdecoder.cpp. The logic here is a bit of a hack since it's based on the command
        # name. Ideally, we would detect whether a particular command is part of a subdecode block
        # in the decoding script.
        if not api_name.startswith("vkCmd") and api_name != "vkBeginCommandBufferAsyncGOOGLE":
            self.module.append('    printer.write_int("seqno: ", 4, indent)\n')

        for param in api.parameters:
            # Add any structs that this API uses to the list of structs to write decoding logic for
            if self.typeInfo.isCompoundType(param.typeName):
                self.structs.add(param.typeName)

            # Don't try to print the pData field of vkQueueFlushCommandsGOOGLE, those are the
            # commands processed as part of the subdecode pass
            if api.name == "vkQueueFlushCommandsGOOGLE" and param.paramName == "pData":
                continue

            # Write out decoding logic for that parameter
            self.process_type(param)

        # Finally, add a return statement. This is needed in case the API has no parameters.
        self.module.append('    return\n\n')

    def process_structs(self):
        """Writes decoding logic for all the structs that we use"""

        # self.structs now contains all the structs used directly by the Vulkan APIs we use.
        # Recursively expand this set to add all the structs used by these structs.
        copy = self.structs.copy()
        self.structs.clear()
        for struct_name in copy:
            self.expand_needed_structs(struct_name)

        # Now we have the full list of structs that we need to write decoding logic for.
        # Write a decoder for each of them
        for struct_name in sorted(self.structs):
            struct = self.typeInfo.structs[struct_name]
            self.module.append('def struct_{}(printer, indent: int):\n'.format(struct_name))
            for member in self.get_members(struct):
                self.process_type(member)
            self.module.append('\n')

    def expand_needed_structs(self, struct_name: str):
        """
        Recursively adds all the structs used by a given struct to the list of structs to process
        """
        if struct_name in self.structs:
            return
        self.structs.add(struct_name)
        struct = self.typeInfo.structs[struct_name]
        for member in self.get_members(struct):
            if self.typeInfo.isCompoundType(member.typeName):
                self.expand_needed_structs(member.typeName)

    def get_members(self, struct: VulkanCompoundType):
        """
        Returns the members of a struct/union that we need to process.
        For structs, returns the list of all members
        For unions, returns a list with just the first member.
        """
        return struct.members[0:1] if struct.isUnion else struct.members

    def process_type(self, type: VulkanType):
        """
        Writes decoding logic for a single Vulkan type. This could be the parameter in a Vulkan API,
        or a struct member.
        """
        if type.typeName == "VkStructureType":
            self.module.append(
                '    printer.write_stype_and_pnext("{}", indent)\n'.format(
                    type.parent.structEnumExpr))
            return

        if type.isNextPointer():
            return

        if type.paramName == "commandBuffer":
            if type.parent.name != "vkQueueFlushCommandsGOOGLE":
                return

        # Enums
        if type.isEnum(self.typeInfo):
            self.needed_enums.add(type.typeName)
            self.module.append(
                '    printer.write_enum("{}", {}, indent)\n'.format(
                    type.paramName, type.typeName))
            return

        # Bitmasks
        if type.isBitmask(self.typeInfo):
            enum_type = self.typeInfo.bitmasks.get(type.typeName)
            if enum_type:
                self.needed_enums.add(enum_type)
                self.module.append(
                    '    printer.write_flags("{}", {}, indent)\n'.format(
                        type.paramName, enum_type))
                return
            # else, fall through and let the primitive type logic handle it

        # Structs or unions
        if self.typeInfo.isCompoundType(type.typeName):
            self.module.append(
                '    printer.write_struct("{name}", struct_{type}, {optional}, {count}, indent)\n'
                    .format(name=type.paramName,
                            type=type.typeName,
                            optional=type.isOptionalPointer(),
                            count=self.get_length_expression(type)))
            return

        # Null-terminated strings
        if type.isString():
            self.module.append('    printer.write_string("{}", None, indent)\n'.format(
                type.paramName))
            return

        # Arrays of primitive types
        if type.staticArrExpr and type.primitiveEncodingSize and type.primitiveEncodingSize <= 8:
            # Array sizes are specified either as a number, or as an enum value
            array_size = int(type.staticArrExpr) if type.staticArrExpr.isdigit() \
                else self.typeInfo.enumValues.get(type.staticArrExpr)
            assert array_size is not None, type.staticArrExpr

            if type.typeName == "char":
                self.module.append(
                    '    printer.write_string("{}", {}, indent)\n'.format(
                        type.paramName, array_size))
            elif type.typeName == "float":
                self.module.append(
                    '    printer.write_float("{}", indent, count={})\n'
                        .format(type.paramName, array_size))
            else:
                self.module.append(
                    '    printer.write_int("{name}", {int_size}, indent, signed={signed}, count={array_size})\n'
                        .format(name=type.paramName,
                                array_size=array_size,
                                int_size=type.primitiveEncodingSize,
                                signed=type.isSigned()))
            return

        # Pointers
        if type.pointerIndirectionLevels > 0:
            # Assume that all uint32* are always serialized directly rather than passed by pointers.
            # This is probably not always true (e.g. out params) - fix this as needed.
            size = 4 if type.primitiveEncodingSize == 4 else 8
            self.module.append(
                '    {name} = printer.write_int("{name}", {size}, indent, optional={opt}, count={count}, big_endian={big_endian})\n'
                    .format(name=type.paramName,
                            size=size,
                            opt=type.isOptionalPointer(),
                            count=self.get_length_expression(type),
                            big_endian=self.using_big_endian(type)))
            return

        # Primitive types (ints, floats)
        if type.isSimpleValueType(self.typeInfo) and type.primitiveEncodingSize:
            if type.typeName == "float":
                self.module.append(
                    '    printer.write_float("{name}", indent)\n'.format(name=type.paramName))
            else:
                self.module.append(
                    '    {name} = printer.write_int("{name}", {size}, indent, signed={signed}, big_endian={big_endian})\n'.format(
                        name=type.paramName,
                        size=type.primitiveEncodingSize,
                        signed=type.isSigned(),
                        big_endian=self.using_big_endian(type))
                )
            return

        raise NotImplementedError(
            "No decoding logic for {} {}".format(type.typeName, type.paramName))

    def using_big_endian(self, type: VulkanType):
        """For some reason gfxstream serializes some types as big endian"""
        return type.typeName == "size_t"

    def get_length_expression(self, type: VulkanType) -> Optional[str]:
        """Returns the length expression for a given type"""
        if type.lenExpr is None:
            return None

        if type.lenExpr.isalpha():
            return type.lenExpr

        # There are a couple of instances in the spec where we use a math expression to express the
        # length (e.g. VkPipelineMultisampleStateCreateInfo). CodeGen().generalLengthAccess() has
        # logic o parse these expressions correctly, but for now,we just use a simple lookup table.
        known_expressions = {
            r"latexmath:[\lceil{\mathit{rasterizationSamples} \over 32}\rceil]":
                "int(rasterizationSamples / 32)",
            r"latexmath:[\textrm{codeSize} \over 4]": "int(codeSize / 4)",
            r"null-terminated": None
        }
        if type.lenExpr in known_expressions:
            return known_expressions[type.lenExpr]

        raise NotImplementedError("Unknown length expression: " + type.lenExpr)

    def process_enums(self):
        """
        For each Vulkan enum that we use, write out a python dictionary mapping the enum values back
        to the enum name as a string
        """
        for enum_name in sorted(self.needed_enums):
            self.module.append('{} = {{\n'.format(enum_name))
            for identifier in self.all_enums[enum_name]:
                value = self.typeInfo.enumValues.get(identifier)
                if value is not None and isinstance(value, int):
                    self.module.append('    {}: "{}",\n'.format(value, identifier))
            self.module.append('}\n\n')