1"""This module defines the `GdbController` class
2which runs gdb as a subprocess and can write to it and read from it to get
3structured output.
4"""
5
6import logging
7import subprocess
8from distutils.spawn import find_executable
9from typing import Union, List, Optional
10from pygdbmi.IoManager import IoManager
11from pygdbmi.constants import (
12    DEFAULT_GDB_TIMEOUT_SEC,
13    DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
14)
15
16DEFAULT_GDB_LAUNCH_COMMAND = ["gdb", "--nx", "--quiet", "--interpreter=mi3"]
17logger = logging.getLogger(__name__)
18
19
20class GdbController:
21    def __init__(
22        self,
23        command: Optional[List[str]] = None,
24        time_to_check_for_additional_output_sec=DEFAULT_TIME_TO_CHECK_FOR_ADDITIONAL_OUTPUT_SEC,
25    ):
26        """
27        Run gdb as a subprocess. Send commands and receive structured output.
28        Create new object, along with a gdb subprocess
29
30        Args:
31            command: Command to run in shell to spawn new gdb subprocess
32            time_to_check_for_additional_output_sec: When parsing responses, wait this amout of time before exiting (exits before timeout is reached to save time). If <= 0, full timeout time is used.
33        Returns:
34            New GdbController object
35        """
36
37        if command is None:
38            command = DEFAULT_GDB_LAUNCH_COMMAND
39
40        if not any([("--interpreter=mi" in c) for c in command]):
41            logger.warning(
42                "Adding `--interpreter=mi3` (or similar) is recommended to get structured output. "
43                + "See https://sourceware.org/gdb/onlinedocs/gdb/Mode-Options.html#Mode-Options."
44            )
45        self.abs_gdb_path = None  # abs path to gdb executable
46        self.command = command  # type: List[str]
47        self.time_to_check_for_additional_output_sec = (
48            time_to_check_for_additional_output_sec
49        )
50        self.gdb_process = None
51        self._allow_overwrite_timeout_times = (
52            self.time_to_check_for_additional_output_sec > 0
53        )
54        gdb_path = command[0]
55        if not gdb_path:
56            raise ValueError("a valid path to gdb must be specified")
57
58        else:
59            abs_gdb_path = find_executable(gdb_path)
60            if abs_gdb_path is None:
61                raise ValueError(
62                    'gdb executable could not be resolved from "%s"' % gdb_path
63                )
64
65            else:
66                self.abs_gdb_path = abs_gdb_path
67
68        self.spawn_new_gdb_subprocess()
69
70    def spawn_new_gdb_subprocess(self):
71        """Spawn a new gdb subprocess with the arguments supplied to the object
72        during initialization. If gdb subprocess already exists, terminate it before
73        spanwing a new one.
74        Return int: gdb process id
75        """
76        if self.gdb_process:
77            logger.debug(
78                "Killing current gdb subprocess (pid %d)" % self.gdb_process.pid
79            )
80            self.exit()
81
82        logger.debug(f'Launching gdb: {" ".join(self.command)}')
83
84        # Use pipes to the standard streams
85        self.gdb_process = subprocess.Popen(
86            self.command,
87            shell=False,
88            stdout=subprocess.PIPE,
89            stdin=subprocess.PIPE,
90            stderr=subprocess.PIPE,
91            bufsize=0,
92        )
93
94        self.io_manager = IoManager(
95            self.gdb_process.stdin,
96            self.gdb_process.stdout,
97            self.gdb_process.stderr,
98            self.time_to_check_for_additional_output_sec,
99        )
100        return self.gdb_process.pid
101
102    def get_gdb_response(
103        self, timeout_sec: float = DEFAULT_GDB_TIMEOUT_SEC, raise_error_on_timeout=True
104    ):
105        """Get gdb response. See IoManager.get_gdb_response() for details"""
106        return self.io_manager.get_gdb_response(timeout_sec, raise_error_on_timeout)
107
108    def write(
109        self,
110        mi_cmd_to_write: Union[str, List[str]],
111        timeout_sec=DEFAULT_GDB_TIMEOUT_SEC,
112        raise_error_on_timeout: bool = True,
113        read_response: bool = True,
114    ):
115        """Write command to gdb. See IoManager.write() for details"""
116        return self.io_manager.write(
117            mi_cmd_to_write, timeout_sec, raise_error_on_timeout, read_response
118        )
119
120    def exit(self) -> None:
121        """Terminate gdb process"""
122        if self.gdb_process:
123            self.gdb_process.terminate()
124            self.gdb_process.communicate()
125        self.gdb_process = None
126        return None
127