xref: /llvm-project/lldb/test/API/commands/trace/TestTraceSave.py (revision 1eeeab82c6eb185f5139e633a59c2dbcb15616e4) 
1import lldb
2import json
3from intelpt_testcase import *
4from lldbsuite.test.lldbtest import *
5from lldbsuite.test import lldbutil
6from lldbsuite.test.decorators import *
7
8
9def find(predicate, seq):
10    for item in seq:
11        if predicate(item):
12            return item
13
14
15class TestTraceSave(TraceIntelPTTestCaseBase):
16    def testErrorMessages(self):
17        # We first check the output when there are no targets
18        self.expect(
19            "trace save",
20            substrs=[
21                "error: invalid target, create a target using the 'target create' command"
22            ],
23            error=True,
24        )
25
26        # We now check the output when there's a non-running target
27        self.expect(
28            "target create "
29            + os.path.join(self.getSourceDir(), "intelpt-trace", "a.out")
30        )
31
32        self.expect(
33            "trace save",
34            substrs=["error: Command requires a current process."],
35            error=True,
36        )
37
38        # Now we check the output when there's a running target without a trace
39        self.expect("b main")
40        self.expect("run")
41
42        self.expect(
43            "trace save", substrs=["error: Process is not being traced"], error=True
44        )
45
46    def testSaveToInvalidDir(self):
47        self.expect(
48            "target create "
49            + os.path.join(self.getSourceDir(), "intelpt-trace", "a.out")
50        )
51        self.expect("b main")
52        self.expect("r")
53        self.expect("thread trace start")
54        self.expect("n")
55
56        # Check the output when saving without providing the directory argument
57        self.expect(
58            "trace save ",
59            substrs=[
60                "error: a single path to a directory where the trace bundle will be created is required"
61            ],
62            error=True,
63        )
64
65        # Check the output when saving to an invalid directory
66        self.expect(
67            "trace save /", substrs=["error: couldn't write to the file"], error=True
68        )
69
70    def testSaveWhenNotLiveTrace(self):
71        self.expect(
72            "trace load -v "
73            + os.path.join(self.getSourceDir(), "intelpt-trace", "trace.json"),
74            substrs=["intel-pt"],
75        )
76
77        # Check the output when not doing live tracing
78        self.expect(
79            "trace save "
80            + os.path.join(self.getBuildDir(), "intelpt-trace", "trace_not_live_dir")
81        )
82
83    def testSaveMultiCpuTrace(self):
84        """
85        This test starts a per-cpu tracing session, then saves the session to disk, and
86        finally it loads it again.
87        """
88        self.skipIfPerCpuTracingIsNotSupported()
89
90        self.expect(
91            "target create "
92            + os.path.join(self.getSourceDir(), "intelpt-trace", "a.out")
93        )
94        self.expect("b main")
95        self.expect("r")
96        self.expect("process trace start --per-cpu-tracing")
97        self.expect("b 7")
98
99        output_dir = os.path.join(self.getBuildDir(), "intelpt-trace", "trace_save")
100        self.expect("trace save " + output_dir)
101
102        def checkSessionBundle(session_file_path):
103            with open(session_file_path) as session_file:
104                session = json.load(session_file)
105                # We expect tsc conversion info
106                self.assertIn("tscPerfZeroConversion", session)
107                # We expect at least one cpu
108                self.assertGreater(len(session["cpus"]), 0)
109
110                # We expect the required trace files to be created
111                for cpu in session["cpus"]:
112                    cpu_files_prefix = os.path.join(output_dir, "cpus", str(cpu["id"]))
113                    self.assertTrue(os.path.exists(cpu_files_prefix + ".intelpt_trace"))
114                    self.assertTrue(
115                        os.path.exists(cpu_files_prefix + ".perf_context_switch_trace")
116                    )
117
118                # We expect at least one one process
119                self.assertGreater(len(session["processes"]), 0)
120                for process in session["processes"]:
121                    # We expect at least one thread
122                    self.assertGreater(len(process["threads"]), 0)
123                    # We don't expect thread traces
124                    for thread in process["threads"]:
125                        self.assertTrue(
126                            ("iptTrace" not in thread) or (thread["iptTrace"] is None)
127                        )
128
129        original_trace_session_file = os.path.join(output_dir, "trace.json")
130        checkSessionBundle(original_trace_session_file)
131
132        output_dir = os.path.join(self.getBuildDir(), "intelpt-trace", "trace_save")
133        self.expect("trace load " + os.path.join(output_dir, "trace.json"))
134        output_copy_dir = os.path.join(
135            self.getBuildDir(), "intelpt-trace", "copy_trace_save"
136        )
137        self.expect("trace save " + output_copy_dir)
138
139        # We now check that the new bundle is correct on its own
140        copied_trace_session_file = os.path.join(output_copy_dir, "trace.json")
141        checkSessionBundle(copied_trace_session_file)
142
143        # We finally check that the new bundle has the same information as the original one
144        with open(original_trace_session_file) as original_file:
145            original = json.load(original_file)
146            with open(copied_trace_session_file) as copy_file:
147                copy = json.load(copy_file)
148
149                self.assertEqual(len(original["processes"]), len(copy["processes"]))
150
151                for process in original["processes"]:
152                    copied_process = find(
153                        lambda proc: proc["pid"] == process["pid"], copy["processes"]
154                    )
155                    self.assertIsNotNone(copied_process)
156
157                    for thread in process["threads"]:
158                        copied_thread = find(
159                            lambda thr: thr["tid"] == thread["tid"],
160                            copied_process["threads"],
161                        )
162                        self.assertIsNotNone(copied_thread)
163
164                for cpu in original["cpus"]:
165                    copied_cpu = find(lambda cor: cor["id"] == cpu["id"], copy["cpus"])
166                    self.assertIsNotNone(copied_cpu)
167
168    def testSaveTrace(self):
169        self.expect(
170            "target create "
171            + os.path.join(self.getSourceDir(), "intelpt-trace", "a.out")
172        )
173        self.expect("b main")
174        self.expect("r")
175        self.expect("thread trace start")
176        self.expect("b 7")
177
178        ci = self.dbg.GetCommandInterpreter()
179        res = lldb.SBCommandReturnObject()
180
181        ci.HandleCommand("thread trace dump instructions -c 10 --forwards", res)
182        self.assertTrue(res.Succeeded())
183        first_ten_instructions = res.GetOutput()
184
185        ci.HandleCommand("thread trace dump instructions -c 10", res)
186        self.assertTrue(res.Succeeded())
187        last_ten_instructions = res.GetOutput()
188
189        # Now, save the trace to <trace_copy_dir>
190        self.expect(
191            "trace save "
192            + os.path.join(self.getBuildDir(), "intelpt-trace", "trace_copy_dir")
193        )
194
195        # Load the trace just saved
196        self.expect(
197            "trace load -v "
198            + os.path.join(
199                self.getBuildDir(), "intelpt-trace", "trace_copy_dir", "trace.json"
200            ),
201            substrs=["intel-pt"],
202        )
203
204        # Compare with instructions saved at the first time
205        ci.HandleCommand("thread trace dump instructions -c 10 --forwards", res)
206        self.assertTrue(res.Succeeded())
207        self.assertEqual(res.GetOutput(), first_ten_instructions)
208
209        ci.HandleCommand("thread trace dump instructions -c 10", res)
210        self.assertTrue(res.Succeeded())
211        self.assertEqual(res.GetOutput(), last_ten_instructions)
212
213    def testSaveKernelTrace(self):
214        original_trace_file = os.path.join(
215            self.getSourceDir(), "intelpt-kernel-trace", "trace.json"
216        )
217        copied_trace_dir = os.path.join(self.getBuildDir(), "intelpt-kernel-trace")
218        copied_trace_file = os.path.join(copied_trace_dir, "trace.json")
219
220        self.expect("trace load -v " + original_trace_file, substrs=["intel-pt"])
221        self.expect("trace save " + copied_trace_dir)
222
223        # We finally check that the new json has the same information as the original one
224        with open(original_trace_file) as original_file:
225            original_file = json.load(original_file)
226            with open(copied_trace_file) as copy_file:
227                copy_file = json.load(copy_file)
228                self.assertIn("kernel", copy_file)
229
230                self.assertEqual(
231                    os.path.basename(original_file["kernel"]["file"]),
232                    os.path.basename(copy_file["kernel"]["file"]),
233                )
234