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