import gdbremote_testcase
class GdbRemoteForkTestBase(gdbremote_testcase.GdbRemoteTestCaseBase):
fork_regex = (
"[$]T[0-9a-fA-F]{{2}}thread:p([0-9a-f]+)[.]([0-9a-f]+);.*"
"{}:p([0-9a-f]+)[.]([0-9a-f]+).*"
)
fork_regex_nonstop = (
"%Stop:T[0-9a-fA-F]{{2}}"
"thread:p([0-9a-f]+)[.]([0-9a-f]+);.*"
"{}:p([0-9a-f]+)[.]([0-9a-f]+).*"
)
fork_capture = {1: "parent_pid", 2: "parent_tid", 3: "child_pid", 4: "child_tid"}
stop_regex_base = "T[0-9a-fA-F]{{2}}thread:p{}.{};.*reason:signal.*"
stop_regex = "^[$]" + stop_regex_base
def start_fork_test(self, args, variant="fork", nonstop=False):
self.build()
self.prep_debug_monitor_and_inferior(inferior_args=args)
self.add_qSupported_packets(["multiprocess+", "{}-events+".format(variant)])
ret = self.expect_gdbremote_sequence()
self.assertIn("{}-events+".format(variant), ret["qSupported_response"])
self.reset_test_sequence()
# continue and expect fork
if nonstop:
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#00",
"send packet: $OK#00",
{
"direction": "send",
"regex": self.fork_regex_nonstop.format(variant),
"capture": self.fork_capture,
},
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
else:
self.test_sequence.add_log_lines(
[
"read packet: $c#00",
{
"direction": "send",
"regex": self.fork_regex.format(variant),
"capture": self.fork_capture,
},
],
True,
)
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
return tuple(
ret[x] for x in ("parent_pid", "parent_tid", "child_pid", "child_tid")
)
def fork_and_detach_test(self, variant, nonstop=False):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
[variant], variant, nonstop=nonstop
)
# detach the forked child
self.test_sequence.add_log_lines(
[
"read packet: $D;{}#00".format(child_pid),
"send packet: $OK#00",
# verify that the current process is correct
"read packet: $qC#00",
"send packet: $QCp{}.{}#00".format(parent_pid, parent_tid),
# verify that the correct processes are detached/available
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: $Eff#00",
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
self.reset_test_sequence()
return parent_pid, parent_tid
def fork_and_follow_test(self, variant, nonstop=False):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
[variant], variant, nonstop=nonstop
)
# switch to the forked child
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: $OK#00",
"read packet: $Hcp{}.{}#00".format(child_pid, child_tid),
"send packet: $OK#00",
# detach the parent
"read packet: $D;{}#00".format(parent_pid),
"send packet: $OK#00",
# verify that the correct processes are detached/available
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: $Eff#00",
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: $OK#00",
# then resume the child
"read packet: $c#00",
],
True,
)
if nonstop:
self.test_sequence.add_log_lines(
[
"send packet: $OK#00",
"send packet: %Stop:W00;process:{}#00".format(child_pid),
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
else:
self.test_sequence.add_log_lines(
[
"send packet: $W00;process:{}#00".format(child_pid),
],
True,
)
self.expect_gdbremote_sequence()
def detach_all_test(self, nonstop=False):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork"], nonstop=nonstop
)
self.test_sequence.add_log_lines(
[
# double-check our PIDs
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: $OK#00",
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: $OK#00",
# detach all processes
"read packet: $D#00",
"send packet: $OK#00",
# verify that both PIDs are invalid now
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: $Eff#00",
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: $Eff#00",
],
True,
)
self.expect_gdbremote_sequence()
def vkill_test(self, kill_parent=False, kill_child=False, nonstop=False):
assert kill_parent or kill_child
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork"], nonstop=nonstop
)
if kill_parent:
self.test_sequence.add_log_lines(
[
# kill the process
"read packet: $vKill;{}#00".format(parent_pid),
"send packet: $OK#00",
],
True,
)
if kill_child:
self.test_sequence.add_log_lines(
[
# kill the process
"read packet: $vKill;{}#00".format(child_pid),
"send packet: $OK#00",
],
True,
)
self.test_sequence.add_log_lines(
[
# check child PID/TID
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: ${}#00".format("Eff" if kill_child else "OK"),
# check parent PID/TID
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: ${}#00".format("Eff" if kill_parent else "OK"),
],
True,
)
self.expect_gdbremote_sequence()
def resume_one_test(self, run_order, use_vCont=False, nonstop=False):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "stop"], nonstop=nonstop
)
parent_expect = [
self.stop_regex_base.format(parent_pid, parent_tid),
"W00;process:{}#.*".format(parent_pid),
]
child_expect = [
self.stop_regex_base.format(child_pid, child_tid),
"W00;process:{}#.*".format(child_pid),
]
for x in run_order:
if x == "parent":
pidtid = (parent_pid, parent_tid)
expect = parent_expect.pop(0)
elif x == "child":
pidtid = (child_pid, child_tid)
expect = child_expect.pop(0)
else:
assert False, "unexpected x={}".format(x)
if use_vCont:
self.test_sequence.add_log_lines(
[
# continue the selected process
"read packet: $vCont;c:p{}.{}#00".format(*pidtid),
],
True,
)
else:
self.test_sequence.add_log_lines(
[
# continue the selected process
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
],
True,
)
if nonstop:
self.test_sequence.add_log_lines(
[
"send packet: $OK#00",
{"direction": "send", "regex": "%Stop:" + expect},
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
else:
self.test_sequence.add_log_lines(
[
{"direction": "send", "regex": "[$]" + expect},
],
True,
)
# if at least one process remained, check both PIDs
if parent_expect or child_expect:
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{}.{}#00".format(parent_pid, parent_tid),
"send packet: ${}#00".format("OK" if parent_expect else "Eff"),
"read packet: $Hgp{}.{}#00".format(child_pid, child_tid),
"send packet: ${}#00".format("OK" if child_expect else "Eff"),
],
True,
)
self.expect_gdbremote_sequence()
Codebase Browser
llvm