tzh
2024-08-22 c7d0944258c7d0943aa7b2211498fd612971ce27
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
#!/usr/bin/env python
 
# Copyright 2015 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
 
import unittest
import logging
 
from systrace import decorators
from systrace import run_systrace
from systrace.tracing_agents import ftrace_agent
 
 
SYSTRACE_HOST_CMD_DEFAULT = ['./systrace.py', '--target=linux']
FT_DIR = "/sys/kernel/debug/tracing/"
FT_EVENT_DIR = FT_DIR + "events/"
FT_TRACE_ON = FT_DIR + "tracing_on"
FT_TRACE = FT_DIR + "trace"
FT_BUFFER_SIZE = FT_DIR + "buffer_size_kb"
 
 
def make_test_io_interface(permitted_files):
  class TestIoImpl(object):
 
    @staticmethod
    def writeFile(path, data):
      permitted_files[path] = data
 
    @staticmethod
    def readFile(path):
      if path in permitted_files:
        return permitted_files[path]
      else:
        return ""
 
    @staticmethod
    def haveWritePermissions(path):
      return path in permitted_files
 
  return TestIoImpl
 
 
class FtraceAgentTest(unittest.TestCase):
 
  @decorators.HostOnlyTest
  def test_avail_categories(self):
    # sched only has required events
    permitted_files = {
      FT_EVENT_DIR + "sched/sched_switch/enable": "0",
      FT_EVENT_DIR + "sched/sched_wakeup/enable": "0"
    }
    io_interface = make_test_io_interface(permitted_files)
    agent = ftrace_agent.FtraceAgent(io_interface)
    self.assertEqual(['sched'], agent._avail_categories())
 
    # check for no available categories
    permitted_files = {}
    io_interface = make_test_io_interface(permitted_files)
    agent = ftrace_agent.FtraceAgent(io_interface)
    self.assertEqual([], agent._avail_categories())
 
    # block has some required, some optional events
    permitted_files = {
      FT_EVENT_DIR + "block/block_rq_complete/enable": "0",
      FT_EVENT_DIR + "block/block_rq_issue/enable": "0"
    }
    io_interface = make_test_io_interface(permitted_files)
    agent = ftrace_agent.FtraceAgent(io_interface)
    self.assertEqual(['disk'], agent._avail_categories())
 
  @decorators.HostOnlyTest
  def test_tracing_bootstrap(self):
    workq_event_path = FT_EVENT_DIR + "workqueue/enable"
    permitted_files = {
      workq_event_path: "0",
      FT_TRACE: "x"
    }
    io_interface = make_test_io_interface(permitted_files)
    systrace_cmd = SYSTRACE_HOST_CMD_DEFAULT + ["workq"]
    options, categories = run_systrace.parse_options(systrace_cmd)
    agent = ftrace_agent.FtraceAgent(io_interface)
    self.assertEqual(['workq'], agent._avail_categories())
 
    # confirm tracing is enabled, buffer is cleared
    agent.StartAgentTracing(options, categories)
    self.assertEqual(permitted_files[FT_TRACE_ON], "1")
    self.assertEqual(permitted_files[FT_TRACE], "")
 
    # fill in file with dummy contents
    dummy_trace = "trace_contents"
    permitted_files[FT_TRACE] = dummy_trace
 
    # confirm tracing is disabled
    agent.StopAgentTracing()
    agent.GetResults()
    self.assertEqual(permitted_files[FT_TRACE_ON], "0")
 
    # confirm trace is expected, and read from fs
    self.assertEqual(agent.GetResults().raw_data, dummy_trace)
 
    # confirm buffer size is reset to 1
    self.assertEqual(permitted_files[FT_BUFFER_SIZE], "1")
 
  @decorators.HostOnlyTest
  def test_tracing_event_enable_disable(self):
    # turn on irq tracing
    ipi_event_path = FT_EVENT_DIR + "ipi/enable"
    irq_event_path = FT_EVENT_DIR + "irq/enable"
    permitted_files = {
      ipi_event_path: "0",
      irq_event_path: "0"
    }
    io_interface = make_test_io_interface(permitted_files)
    systrace_cmd = SYSTRACE_HOST_CMD_DEFAULT + ["irq"]
    options, categories = run_systrace.parse_options(systrace_cmd)
    options.ftrace_categories = categories
    agent = ftrace_agent.FtraceAgent(io_interface)
    self.assertEqual(['irq'], agent._avail_categories())
 
    # confirm all the event nodes are turned on during tracing
    agent.StartAgentTracing(options)
    self.assertEqual(permitted_files[irq_event_path], "1")
    self.assertEqual(permitted_files[ipi_event_path], "1")
 
    # and then turned off when completed.
    agent.StopAgentTracing()
    agent.GetResults()
    self.assertEqual(permitted_files[irq_event_path], "0")
    self.assertEqual(permitted_files[ipi_event_path], "0")
 
  @decorators.HostOnlyTest
  def test_buffer_size(self):
    systrace_cmd = SYSTRACE_HOST_CMD_DEFAULT + ['-b', '16000']
    options, categories = run_systrace.parse_options(systrace_cmd)
    agent = ftrace_agent.FtraceAgent()
    agent._config = options
    agent._config.atrace_categories = categories
    self.assertEqual(agent._get_trace_buffer_size(), 16000)
 
if __name__ == "__main__":
  logging.getLogger().setLevel(logging.DEBUG)
  unittest.main(verbosity=2)