| # |
| # Copyright (C) 2022 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import io |
| from datetime import datetime |
| import sys |
| import textwrap |
| from typing import Dict, Optional |
| from . import vulkan_printer |
| from . import opcodes |
| import struct |
| |
| |
| class CommandPrinter: |
| """This class is responsible for printing the commands found in the minidump file to the terminal.""" |
| |
| def __init__(self, opcode: int, original_size: int, data: bytes, timestamp: int, |
| stream_idx: int, cmd_idx: int, out=sys.stdout): |
| self.opcode = opcode |
| self.original_size = original_size |
| self.data = io.BytesIO(data) |
| self.timestamp = timestamp |
| self.stream_idx = stream_idx |
| self.cmd_idx = cmd_idx |
| self.out = out |
| |
| def print_cmd(self): |
| """ |
| Tries to decode and pretty print the command to the terminal. |
| Falls back to printing hex data if the command doesn't have a printer. |
| """ |
| |
| # Print out the command name |
| print("\n{}.{} - {}: ({} bytes)".format(self.stream_idx, self.cmd_idx, self.cmd_name(), |
| self.original_size - 8), file=self.out) |
| self.write_timestamp(4) |
| |
| if len(self.data.getbuffer()) == 0: |
| return |
| |
| pretty_printer = getattr(vulkan_printer, self.cmd_name(), None) |
| if not pretty_printer: |
| self.print_raw() |
| return |
| |
| try: |
| pretty_printer(self, indent=4) |
| self.check_no_more_bytes() |
| except Exception as ex: |
| print("Error while processing {}: {}".format(self.cmd_name(), repr(ex)), file=self.out) |
| print("Command raw data:", file=self.out) |
| self.print_raw() |
| raise ex |
| |
| def check_no_more_bytes(self): |
| """ |
| Checks that we processed all the bytes, otherwise there's probably a bug in the decoding |
| logic |
| """ |
| if self.data.tell() != len(self.data.getbuffer()): |
| raise BufferError( |
| "Not all data was decoded. Decoded {} bytes but command had {}".format( |
| self.data.tell(), len(self.data.getbuffer()))) |
| |
| def cmd_name(self) -> str: |
| """Returns the command name (e.g.: "OP_vkBeginCommandBuffer", or the opcode as a string if unknown""" |
| return opcodes.opcodes.get(self.opcode, str(self.opcode)) |
| |
| def print_raw(self): |
| """Prints the command data as a hex bytes, as a fallback if we don't know how to decode it""" |
| truncated = self.original_size > len(self.data.getbuffer()) + 8 |
| indent = 8 |
| hex = ' '.join(["{:02x}".format(x) for x in self.data.getbuffer()]) |
| if truncated: |
| hex += " [...]" |
| lines = textwrap.wrap(hex, width=16 * 3 + indent, initial_indent=' ' * indent, |
| subsequent_indent=' ' * indent) |
| for l in lines: |
| print(l, file=self.out) |
| |
| def read_bytes(self, size: int): |
| buf = self.data.read(size) |
| if len(buf) != size: |
| raise EOFError("Unexpectedly reached the end of the buffer") |
| return buf |
| |
| def read_int(self, size: int, signed: bool = False, big_endian: bool = False) -> int: |
| assert size in [1, 2, 4, 8], "Invalid size to read: " + str(size) |
| buf = self.read_bytes(size) |
| byte_order = 'big' if big_endian else 'little' |
| return int.from_bytes(buf, byteorder=byte_order, signed=signed) |
| |
| def read_float(self) -> float: |
| buf = self.read_bytes(4) |
| return struct.unpack('f', buf)[0] |
| |
| def write(self, msg: str, indent: int): |
| """Prints a string at a given indentation level""" |
| assert type(msg) == str |
| assert type(indent) == int and indent >= 0 |
| print(" " * indent + msg, end='', file=self.out) |
| |
| def write_int(self, |
| field_name: str, |
| size: int, |
| indent: int, |
| signed: bool = False, |
| big_endian: bool = False, |
| optional: bool = False, |
| count: Optional[int] = None) -> Optional[int]: |
| """ |
| Reads and prints integers from the data stream. |
| |
| When reading a single int (ie: when count=None), returns the int that was read, otherwise |
| returns None. |
| |
| size: size of the integer in bytes |
| indent: indentation level that we should write at |
| signed: whether to treat it as a signed or unsigned int |
| big_endian: whether to treat it as little endian or big endian |
| optional: if True, we will first read an 8 byte boolean value. If that value is false, we |
| will return without further reading. |
| count: how many integers to read, for repeated values. |
| """ |
| if optional and self.check_null(field_name, indent): |
| return |
| |
| # Write the field name |
| self.write("{name}: ".format(name=field_name), indent) |
| |
| if count is not None: |
| values = ["0x{:x}".format(self.read_int(size, signed, big_endian)) for i in |
| range(0, count)] |
| self.write("[{}]\n".format(", ".join(values)), indent=0) |
| else: |
| value = self.read_int(size, signed, big_endian) |
| # Print small values as decimal only, otherwise hex + decimal |
| format_str = ("{val}\n" if value < 10 else "0x{val:x} ({val})\n") |
| self.write(format_str.format(val=value), indent=0) |
| return value |
| |
| def write_float(self, field_name: str, indent: int, count: Optional[int] = None): |
| if count is not None: |
| values = [str(self.read_float()) for i in range(0, count)] |
| self.write("{}: [{}]\n".format(field_name, ", ".join(values)), indent) |
| else: |
| self.write("{}: {}\n".format(field_name, self.read_float()), indent) |
| |
| def write_enum(self, field_name: str, enum: Dict[int, str], indent: int) -> int: |
| """Reads the next 32-byte int from the data stream, prints it as an enum, and return it""" |
| value = self.read_int(4) |
| self.write("{}: {} ({})\n".format(field_name, enum.get(value, ""), value), indent) |
| return value |
| |
| def write_flags(self, field_name: str, enum: Dict[int, str], indent: int): |
| """Reads and prints Vulkan flags (byte masks)""" |
| remaining_flags = flags = self.read_int(4) |
| flags_list = [] |
| if remaining_flags == 0xffffffff: |
| # When the value is set to all flags, don't bother listing them all |
| flags_list.append("(all flags)") |
| else: |
| for (value, flag) in enum.items(): |
| if value & remaining_flags: |
| remaining_flags ^= value |
| flags_list.append(flag) |
| if remaining_flags != 0: |
| flags_list.insert(0, "0x{:x}".format(remaining_flags)) |
| self.write("{}: {} (0x{:x})\n".format(field_name, " | ".join(flags_list), flags), indent) |
| |
| def write_stype_and_pnext(self, expected_stype: str, indent: int): |
| """Reads and prints the sType and pNext fields found in many Vulkan structs, while also sanity checking them""" |
| stype = self.write_enum("sType", vulkan_printer.VkStructureType, indent) |
| stype_str = vulkan_printer.VkStructureType.get(stype) |
| if stype_str != expected_stype: |
| raise ValueError("Wrong structure type. Expected: {}, got {} ({}) instead".format( |
| expected_stype, stype, stype_str)) |
| |
| pnext_size = self.write_int("pNext_size", 4, indent, big_endian=True) |
| if pnext_size != 0: |
| self.write_enum("ext type", vulkan_printer.VkStructureType, indent + 1) |
| raise NotImplementedError("Decoding structs with extensions is not supported") |
| |
| def write_timestamp(self, indent): |
| if self.timestamp != 0: |
| self.write( |
| "Recorded at: {}\n".format(datetime.fromtimestamp(self.timestamp / 1000000.0)), |
| indent) |
| |
| def check_null(self, field_name: str, indent) -> bool: |
| is_null = self.read_int(8, big_endian=True) == 0 |
| if is_null: |
| self.write("{}: (null)\n".format(field_name), indent) |
| return is_null |
| |
| def write_struct(self, field_name: str, struct_fn, optional: bool, count: Optional[int], |
| indent: int): |
| """ |
| Reads and prints a struct, calling `struct_fn` to pretty-print it |
| optional: whether this is an optional structure. In this case, we will read an int64 first |
| and skip the struct if the result is null. |
| count: how many times this is repeated. Pass None for non-repeated fields. |
| """ |
| if optional and self.check_null(field_name, indent): |
| return |
| |
| is_repeated = count is not None |
| for i in range(0, count if is_repeated else 1): |
| suffix = " #{}".format(i) if is_repeated else "" |
| self.write("{}{}:\n".format(field_name, suffix), indent) |
| struct_fn(self, indent + 1) |
| |
| def write_string(self, field_name: str, size: Optional[int], indent: int): |
| """ |
| Reads a null-terminated string from the stream. |
| size: if specified, reads up to this many characters |
| """ |
| buf = bytearray() |
| if size is not None: |
| buf = self.read_bytes(size) |
| buf = buf.rstrip(b'\x00') |
| else: |
| # Reads from the string one char at a time, until we find a null |
| # Not the most efficient way of doing this, but whatever |
| while True: |
| c = self.read_int(1) |
| if c == 0: |
| break |
| buf.append(c) |
| |
| self.write("{}: \"{}\"\n".format(field_name, buf.decode('utf-8')), indent) |
