| """ | |
| Test SBDebugger.InterruptRequested and SBCommandInterpreter.WasInterrupted. | |
| """ | |
| import lldb | |
| import lldbsuite.test.lldbutil as lldbutil | |
| from lldbsuite.test.lldbtest import * | |
| import threading | |
| import os | |
| class TestDebuggerInterruption(TestBase): | |
| """This test runs a command that starts up, rendevous with the test thread | |
| using threading barriers, then checks whether it has been interrupted. | |
| The command's first argument is either 'interp' or 'debugger', to test | |
| InterruptRequested and WasInterrupted respectively. | |
| The command has two modes, interrupt and check, the former is the one that | |
| waits for an interrupt. Then latter just returns whether an interrupt was | |
| requested. We use the latter to make sure we took down the flag correctly.""" | |
| NO_DEBUG_INFO_TESTCASE = True | |
| class CommandRunner(threading.Thread): | |
| """This class is for running a command, and for making a thread to run the command on. | |
| It gets passed the test it is working on behalf of, and most of the important | |
| objects come from the test.""" | |
| def __init__(self, test): | |
| super().__init__() | |
| self.test = test | |
| def rendevous(self): | |
| # We smuggle out barriers and event to the runner thread using thread local data: | |
| import interruptible | |
| interruptible.local_data = interruptible.BarrierContainer( | |
| self.test.before_interrupt_barrier, | |
| self.test.after_interrupt_barrier, | |
| self.test.event, | |
| ) | |
| class DirectCommandRunner(CommandRunner): | |
| """ "This version runs a single command using HandleCommand.""" | |
| def __init__(self, test, command): | |
| super().__init__(test) | |
| self.command = command | |
| def run(self): | |
| self.rendevous() | |
| result = self.test.dbg.GetCommandInterpreter().HandleCommand( | |
| self.command, self.test.result | |
| ) | |
| if self.test.result_barrier: | |
| self.test.result_barrier.wait() | |
| class CommandInterpreterRunner(CommandRunner): | |
| """This version runs the CommandInterpreter and feeds the command to it.""" | |
| def __init__(self, test): | |
| super().__init__(test) | |
| def run(self): | |
| self.rendevous() | |
| test = self.test | |
| # We will use files for debugger input and output: | |
| # First write down the command: | |
| with open(test.getBuildArtifact(test.in_filename), "w") as f: | |
| f.write(f"{test.command}\n") | |
| # Now set the debugger's stdout & stdin to our files, and run | |
| # the CommandInterpreter: | |
| with open(test.out_filename, "w") as outf, open( | |
| test.in_filename, "r" | |
| ) as inf: | |
| outsbf = lldb.SBFile(outf.fileno(), "w", False) | |
| orig_outf = test.dbg.GetOutputFile() | |
| error = test.dbg.SetOutputFile(outsbf) | |
| test.assertSuccess(error, "Could not set outfile") | |
| insbf = lldb.SBFile(inf.fileno(), "r", False) | |
| orig_inf = test.dbg.GetOutputFile() | |
| error = test.dbg.SetInputFile(insbf) | |
| test.assertSuccess(error, "Could not set infile") | |
| options = lldb.SBCommandInterpreterRunOptions() | |
| options.SetPrintResults(True) | |
| options.SetEchoCommands(False) | |
| test.dbg.RunCommandInterpreter(True, False, options, 0, False, False) | |
| test.dbg.GetOutputFile().Flush() | |
| error = test.dbg.SetOutputFile(orig_outf) | |
| test.assertSuccess(error, "Restored outfile") | |
| test.dbg.SetInputFile(orig_inf) | |
| test.assertSuccess(error, "Restored infile") | |
| def command_setup(self, args): | |
| """Insert our command, if needed. Then set up event and barriers if needed. | |
| Then return the command to run.""" | |
| self.interp = self.dbg.GetCommandInterpreter() | |
| self.command_name = "interruptible_command" | |
| self.cmd_result = lldb.SBCommandReturnObject() | |
| if not "check" in args: | |
| self.event = threading.Event() | |
| self.result_barrier = threading.Barrier(2, timeout=10) | |
| self.before_interrupt_barrier = threading.Barrier(2, timeout=10) | |
| self.after_interrupt_barrier = threading.Barrier(2, timeout=10) | |
| else: | |
| self.event = None | |
| self.result_barrier = None | |
| self.before_interrupt_barrier = None | |
| self.after_interrupt_barrier = None | |
| if not self.interp.UserCommandExists(self.command_name): | |
| # Make the command we're going to use - it spins calling WasInterrupted: | |
| cmd_filename = "interruptible" | |
| cmd_filename = os.path.join(self.getSourceDir(), "interruptible.py") | |
| self.runCmd(f"command script import {cmd_filename}") | |
| cmd_string = f"command script add {self.command_name} --class interruptible.WelcomeCommand" | |
| self.runCmd(cmd_string) | |
| if len(args) == 0: | |
| command = self.command_name | |
| else: | |
| command = self.command_name + " " + args | |
| return command | |
| def run_single_command(self, command): | |
| # Now start up a thread to run the command: | |
| self.result.Clear() | |
| self.runner = TestDebuggerInterruption.DirectCommandRunner(self, command) | |
| self.runner.start() | |
| def start_command_interp(self): | |
| self.runner = TestDebuggerInterruption.CommandInterpreterRunner(self) | |
| self.runner.start() | |
| def check_text(self, result_text, interrupted): | |
| if interrupted: | |
| self.assertIn( | |
| "Command was interrupted", result_text, "Got the interrupted message" | |
| ) | |
| else: | |
| self.assertIn( | |
| "Command was not interrupted", | |
| result_text, | |
| "Got the not interrupted message", | |
| ) | |
| def gather_output(self): | |
| # Now wait for the interrupt to interrupt the command: | |
| self.runner.join(10.0) | |
| finished = not self.runner.is_alive() | |
| # Don't leave the runner thread stranded if the interrupt didn't work. | |
| if not finished: | |
| self.event.set() | |
| self.runner.join(10.0) | |
| self.assertTrue(finished, "We did finish the command") | |
| def check_result(self, interrupted=True): | |
| self.gather_output() | |
| self.check_text(self.result.GetOutput(), interrupted) | |
| def check_result_output(self, interrupted=True): | |
| self.gather_output() | |
| buffer = "" | |
| # Okay, now open the file for reading, and read. | |
| with open(self.out_filename, "r") as f: | |
| buffer = f.read() | |
| self.assertNotEqual(len(buffer), 0, "No command data") | |
| self.check_text(buffer, interrupted) | |
| def debugger_interrupt_test(self, use_interrupt_requested): | |
| """Test that debugger interruption interrupts a command | |
| running directly through HandleCommand. | |
| If use_interrupt_requested is true, we'll check that API, | |
| otherwise we'll check WasInterrupted. They should both do | |
| the same thing.""" | |
| if use_interrupt_requested: | |
| command = self.command_setup("debugger") | |
| else: | |
| command = self.command_setup("interp") | |
| self.result = lldb.SBCommandReturnObject() | |
| self.run_single_command(command) | |
| # Okay now wait till the command has gotten started to issue the interrupt: | |
| self.before_interrupt_barrier.wait() | |
| # I'm going to do it twice here to test that it works as a counter: | |
| self.dbg.RequestInterrupt() | |
| self.dbg.RequestInterrupt() | |
| def cleanup(): | |
| self.dbg.CancelInterruptRequest() | |
| self.addTearDownHook(cleanup) | |
| # Okay, now set both sides going: | |
| self.after_interrupt_barrier.wait() | |
| # Check that the command was indeed interrupted. First rendevous | |
| # after the runner thread had a chance to execute the command: | |
| self.result_barrier.wait() | |
| self.assertTrue(self.result.Succeeded(), "Our command succeeded") | |
| result_output = self.result.GetOutput() | |
| self.check_result(True) | |
| # Do it again to make sure that the counter is counting: | |
| self.dbg.CancelInterruptRequest() | |
| command = self.command_setup("debugger") | |
| self.run_single_command(command) | |
| # This time we won't even get to run the command, since HandleCommand | |
| # checks for the interrupt state on entry, so we don't wait on the command | |
| # barriers. | |
| self.result_barrier.wait() | |
| # Again check that we were | |
| self.assertFalse(self.result.Succeeded(), "Our command was not allowed to run") | |
| error_output = self.result.GetError() | |
| self.assertIn( | |
| "... Interrupted", error_output, "Command was cut short by interrupt" | |
| ) | |
| # Now take down the flag, and make sure that we aren't interrupted: | |
| self.dbg.CancelInterruptRequest() | |
| # Now make sure that we really did take down the flag: | |
| command = self.command_setup("debugger check") | |
| self.run_single_command(command) | |
| result_output = self.result.GetOutput() | |
| self.check_result(False) | |
| def test_debugger_interrupt_use_dbg(self): | |
| self.debugger_interrupt_test(True) | |
| def test_debugger_interrupt_use_interp(self): | |
| self.debugger_interrupt_test(False) | |
| def test_interp_doesnt_interrupt_debugger(self): | |
| """Test that interpreter interruption does not interrupt a command | |
| running directly through HandleCommand. | |
| If use_interrupt_requested is true, we'll check that API, | |
| otherwise we'll check WasInterrupted. They should both do | |
| the same thing.""" | |
| command = self.command_setup("debugger poll") | |
| self.result = lldb.SBCommandReturnObject() | |
| self.run_single_command(command) | |
| # Now raise the debugger interrupt flag. It will also interrupt the command: | |
| self.before_interrupt_barrier.wait() | |
| self.dbg.GetCommandInterpreter().InterruptCommand() | |
| self.after_interrupt_barrier.wait() | |
| # Check that the command was indeed interrupted: | |
| self.result_barrier.wait() | |
| self.assertTrue(self.result.Succeeded(), "Our command succeeded") | |
| result_output = self.result.GetOutput() | |
| self.check_result(False) | |
| def interruptible_command_test(self, use_interrupt_requested): | |
| """Test that interpreter interruption interrupts a command | |
| running in the RunCommandInterpreter loop. | |
| If use_interrupt_requested is true, we'll check that API, | |
| otherwise we'll check WasInterrupted. They should both do | |
| the same thing.""" | |
| self.out_filename = self.getBuildArtifact("output") | |
| self.in_filename = self.getBuildArtifact("input") | |
| # We're going to overwrite the input file, but we | |
| # don't want data accumulating in the output file. | |
| if os.path.exists(self.out_filename): | |
| os.unlink(self.out_filename) | |
| # You should be able to use either check method interchangeably: | |
| if use_interrupt_requested: | |
| self.command = self.command_setup("debugger") + "\n" | |
| else: | |
| self.command = self.command_setup("interp") + "\n" | |
| self.start_command_interp() | |
| # Now give the interpreter a chance to run this command up | |
| # to the first barrier | |
| self.before_interrupt_barrier.wait() | |
| # Then issue the interrupt: | |
| sent_interrupt = self.dbg.GetCommandInterpreter().InterruptCommand() | |
| self.assertTrue(sent_interrupt, "Did send command interrupt.") | |
| # Now give the command a chance to finish: | |
| self.after_interrupt_barrier.wait() | |
| self.check_result_output(True) | |
| os.unlink(self.out_filename) | |
| # Now send the check command, and make sure the flag is now down. | |
| self.command = self.command_setup("interp check") + "\n" | |
| self.start_command_interp() | |
| self.check_result_output(False) | |
| def test_interruptible_command_check_dbg(self): | |
| self.interruptible_command_test(True) | |
| def test_interruptible_command_check_interp(self): | |
| self.interruptible_command_test(False) | |
| def test_debugger_doesnt_interrupt_command(self): | |
| """Test that debugger interruption doesn't interrupt a command | |
| running in the RunCommandInterpreter loop.""" | |
| self.out_filename = self.getBuildArtifact("output") | |
| self.in_filename = self.getBuildArtifact("input") | |
| # We're going to overwrite the input file, but we | |
| # don't want data accumulating in the output file. | |
| if os.path.exists(self.out_filename): | |
| os.unlink(self.out_filename) | |
| self.command = self.command_setup("interp poll") + "\n" | |
| self.start_command_interp() | |
| self.before_interrupt_barrier.wait() | |
| self.dbg.RequestInterrupt() | |
| def cleanup(): | |
| self.dbg.CancelInterruptRequest() | |
| self.addTearDownHook(cleanup) | |
| self.after_interrupt_barrier.wait() | |
| self.check_result_output(False) | |
| os.unlink(self.out_filename) |