[ceph.git] / ceph / src / seastar / scripts / addr2line.py

#!/usr/bin/env python3
#
# This file is open source software, licensed to you under the terms
# of the Apache License, Version 2.0 (the "License").  See the NOTICE file
# distributed with this work for additional information regarding copyright
# ownership.  You may not use this file except in compliance with the License.
#
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
# Copyright (C) 2017 ScyllaDB

import argparse
import collections
import re
import sys
import subprocess
from enum import Enum

class Addr2Line:
    def __init__(self, binary, concise=False):
        self._binary = binary

        # Print warning if binary has no debug info according to `file`.
        # Note: no message is printed for system errors as they will be
        # printed also by addr2line later on.
        output = subprocess.check_output(["file", self._binary])
        s = output.decode("utf-8")
        if s.find('ELF') >= 0 and s.find('debug_info', len(self._binary)) < 0:
            print('{}'.format(s))

        options = f"-{'C' if not concise else ''}fpia"
        self._input = subprocess.Popen(["addr2line", options, "-e", self._binary], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
        if concise:
            self._output = subprocess.Popen(["c++filt", "-p"], stdin=self._input.stdout, stdout=subprocess.PIPE, universal_newlines=True)
        else:
            self._output = self._input

        # If a library doesn't exist in a particular path, addr2line
        # will just exit.  We need to be robust against that.  We
        # can't just wait on self._addr2line since there is no
        # guarantee on what timeout is sufficient.
        self._input.stdin.write('\n')
        self._input.stdin.flush()
        res = self._output.stdout.readline()
        self._missing = res == ''

    def _read_resolved_address(self):
        res = self._output.stdout.readline()
        # remove the address
        res = res.split(': ', 1)[1]
        dummy = '0x0000000000000000: ?? ??:0\n'
        line = ''
        while line != dummy:
            res += line
            line = self._output.stdout.readline()
        return res

    def __call__(self, address):
        if self._missing:
            return " ".join([self._binary, address, '\n'])
        # print two lines to force addr2line to output a dummy
        # line which we can look for in _read_address
        self._input.stdin.write(address + '\n\n')
        self._input.stdin.flush()
        return self._read_resolved_address()

class BacktraceResolver(object):

    class BacktraceParser(object):
        class Type(Enum):
            ADDRESS = 1
            SEPARATOR = 2

        def __init__(self):
            addr = "0x[0-9a-f]+"
            path = "\S+"
            token = f"(?:{path}\+)?{addr}"
            full_addr_match = f"(?:({path})\+)?({addr})"
            self.oneline_re = re.compile(f"^((?:.*(?:(?:at|backtrace):?|:))?(?:\s+))?({token}(?:\s+{token})*)(?:\).*|\s*)$", flags=re.IGNORECASE)
            self.address_re = re.compile(full_addr_match, flags=re.IGNORECASE)
            self.asan_re = re.compile(f"^(?:.*\s+)\(({full_addr_match})\)\s*$", flags=re.IGNORECASE)
            self.separator_re = re.compile('^\W*-+\W*$')

        def __call__(self, line):
            def get_prefix(s):
                if s is not None:
                    s = s.strip()
                return s or None

            m = re.match(self.oneline_re, line)
            if m:
                #print(f">>> '{line}': oneline {m.groups()}")
                ret = {'type': self.Type.ADDRESS}
                ret['prefix'] = get_prefix(m.group(1))
                addresses = []
                for obj in m.group(2).split():
                    m = re.match(self.address_re, obj)
                    #print(f"  >>> '{obj}': address {m.groups()}")
                    addresses.append({'path': m.group(1), 'addr': m.group(2)})
                ret['addresses'] = addresses
                return ret

            m = re.match(self.asan_re, line)
            if m:
                #print(f">>> '{line}': asan {m.groups()}")
                ret = {'type': self.Type.ADDRESS}
                ret['prefix'] = None
                ret['addresses'] = [{'path': m.group(2), 'addr': m.group(3)}]
                return ret

            match = re.match(self.separator_re, line)
            if match:
                return {'type': self.Type.SEPARATOR}

            #print(f">>> '{line}': None")
            return None

    def __init__(self, executable, before_lines=1, context_re='', verbose=False, concise=False):
        self._executable = executable
        self._current_backtrace = []
        self._prefix = None
        self._before_lines = before_lines
        self._before_lines_queue = collections.deque(maxlen=before_lines)
        self._i = 0
        self._known_backtraces = {}
        if context_re is not None:
            self._context_re = re.compile(context_re)
        else:
            self._context_re = None
        self._verbose = verbose
        self._concise = concise
        self._known_modules = {self._executable: Addr2Line(self._executable, concise)}
        self.parser = self.BacktraceParser()

    def _get_resolver_for_module(self, module):
        if not module in self._known_modules:
            self._known_modules[module] = Addr2Line(module, self._concise)
        return self._known_modules[module]

    def __enter__(self):
        return self

    def __exit__(self, type, value, tb):
        self._print_current_backtrace()

    def resolve_address(self, address, module=None, verbose=None):
        if module is None:
            module = self._executable
        if verbose is None:
            verbose = self._verbose
        resolved_address = self._get_resolver_for_module(module)(address)
        if verbose:
            resolved_address = '{{{}}} {}: {}'.format(module, address, resolved_address)
        return resolved_address

    def _print_resolved_address(self, module, address):
        sys.stdout.write(self.resolve_address(address, module))

    def _backtrace_context_matches(self):
        if self._context_re is None:
            return True

        if any(map(lambda x: self._context_re.search(x) is not None, self._before_lines_queue)):
            return True

        if (not self._prefix is None) and self._context_re.search(self._prefix):
            return True

        return False

    def _print_current_backtrace(self):
        if len(self._current_backtrace) == 0:
            return

        if not self._backtrace_context_matches():
            self._current_backtrace = []
            return

        for line in self._before_lines_queue:
            sys.stdout.write(line)

        if not self._prefix is None:
            print(self._prefix)
            self._prefix = None

        backtrace = "".join(map(str, self._current_backtrace))
        if backtrace in self._known_backtraces:
            print("[Backtrace #{}] Already seen, not resolving again.".format(self._known_backtraces[backtrace]))
            print("") # To separate traces with an empty line
            self._current_backtrace = []
            return

        self._known_backtraces[backtrace] = self._i

        print("[Backtrace #{}]".format(self._i))

        for module, addr in self._current_backtrace:
            self._print_resolved_address(module, addr)

        print("") # To separate traces with an empty line

        self._current_backtrace = []
        self._i += 1

    def __call__(self, line):
        res = self.parser(line)

        if not res:
            self._print_current_backtrace()
            if self._before_lines > 0:
                self._before_lines_queue.append(line)
            elif self._before_lines < 0:
                sys.stdout.write(line) # line already has a trailing newline
            else:
                pass # when == 0 no non-backtrace lines are printed
        elif res['type'] == self.BacktraceParser.Type.SEPARATOR:
            pass
        elif res['type'] == self.BacktraceParser.Type.ADDRESS:
            addresses = res['addresses']
            if len(addresses) > 1:
                self._print_current_backtrace()
            if len(self._current_backtrace) == 0:
                self._prefix = res['prefix']
            for r in addresses:
                if r['path']:
                    self._current_backtrace.append((r['path'], r['addr']))
                else:
                    self._current_backtrace.append((self._executable, r['addr']))
            if len(addresses) > 1:
                self._print_current_backtrace()
        else:
            print(f"Unknown '{line}': {res}")
            raise RuntimeError("Unknown result type {res}")
Commit	Line	Data
20effc67 TL	1	#!/usr/bin/env python3
	2	#
	3	# This file is open source software, licensed to you under the terms
	4	# of the Apache License, Version 2.0 (the "License"). See the NOTICE file
	5	# distributed with this work for additional information regarding copyright
	6	# ownership. You may not use this file except in compliance with the License.
	7	#
	8	# You may obtain a copy of the License at
	9	#
	10	# http://www.apache.org/licenses/LICENSE-2.0
	11	#
	12	# Unless required by applicable law or agreed to in writing,
	13	# software distributed under the License is distributed on an
	14	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	15	# KIND, either express or implied. See the License for the
	16	# specific language governing permissions and limitations
	17	# under the License.
	18	#
	19	# Copyright (C) 2017 ScyllaDB
	20
	21	import argparse
	22	import collections
	23	import re
	24	import sys
	25	import subprocess
	26	from enum import Enum
	27
	28	class Addr2Line:
	29	def __init__(self, binary, concise=False):
	30	self._binary = binary
	31
	32	# Print warning if binary has no debug info according to `file`.
	33	# Note: no message is printed for system errors as they will be
	34	# printed also by addr2line later on.
	35	output = subprocess.check_output(["file", self._binary])
	36	s = output.decode("utf-8")
	37	if s.find('ELF') >= 0 and s.find('debug_info', len(self._binary)) < 0:
	38	print('{}'.format(s))
	39
	40	options = f"-{'C' if not concise else ''}fpia"
	41	self._input = subprocess.Popen(["addr2line", options, "-e", self._binary], stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
	42	if concise:
	43	self._output = subprocess.Popen(["c++filt", "-p"], stdin=self._input.stdout, stdout=subprocess.PIPE, universal_newlines=True)
	44	else:
	45	self._output = self._input
	46
	47	# If a library doesn't exist in a particular path, addr2line
	48	# will just exit. We need to be robust against that. We
	49	# can't just wait on self._addr2line since there is no
	50	# guarantee on what timeout is sufficient.
	51	self._input.stdin.write('\n')
	52	self._input.stdin.flush()
	53	res = self._output.stdout.readline()
	54	self._missing = res == ''
	55
	56	def _read_resolved_address(self):
	57	res = self._output.stdout.readline()
	58	# remove the address
	59	res = res.split(': ', 1)[1]
	60	dummy = '0x0000000000000000: ?? ??:0\n'
	61	line = ''
	62	while line != dummy:
	63	res += line
	64	line = self._output.stdout.readline()
65	return res
66
67	def __call__(self, address):
68	if self._missing:
69	return " ".join([self._binary, address, '\n'])
70	# print two lines to force addr2line to output a dummy
71	# line which we can look for in _read_address
72	self._input.stdin.write(address + '\n\n')
73	self._input.stdin.flush()
74	return self._read_resolved_address()
75
76	class BacktraceResolver(object):
77
78	class BacktraceParser(object):
79	class Type(Enum):
80	ADDRESS = 1
81	SEPARATOR = 2
82
83	def __init__(self):
84	addr = "0x[0-9a-f]+"
85	path = "\S+"
86	token = f"(?:{path}\+)?{addr}"
87	full_addr_match = f"(?:({path})\+)?({addr})"
88	self.oneline_re = re.compile(f"^((?:.(?:(?:at\|backtrace):?\|:))?(?:\s+))?({token}(?:\s+{token}))(?:\).\|\s)$", flags=re.IGNORECASE)
89	self.address_re = re.compile(full_addr_match, flags=re.IGNORECASE)
90	self.asan_re = re.compile(f"^(?:.\s+)\(({full_addr_match})\)\s$", flags=re.IGNORECASE)
91	self.separator_re = re.compile('^\W-+\W$')
92
93	def __call__(self, line):
94	def get_prefix(s):
95	if s is not None:
96	s = s.strip()
97	return s or None
98
99	m = re.match(self.oneline_re, line)
100	if m:
101	#print(f">>> '{line}': oneline {m.groups()}")
102	ret = {'type': self.Type.ADDRESS}
103	ret['prefix'] = get_prefix(m.group(1))
104	addresses = []
105	for obj in m.group(2).split():
106	m = re.match(self.address_re, obj)
107	#print(f" >>> '{obj}': address {m.groups()}")
108	addresses.append({'path': m.group(1), 'addr': m.group(2)})
109	ret['addresses'] = addresses
110	return ret
111
112	m = re.match(self.asan_re, line)
113	if m:
114	#print(f">>> '{line}': asan {m.groups()}")
115	ret = {'type': self.Type.ADDRESS}
116	ret['prefix'] = None
117	ret['addresses'] = [{'path': m.group(2), 'addr': m.group(3)}]
118	return ret
119
120	match = re.match(self.separator_re, line)
121	if match:
122	return {'type': self.Type.SEPARATOR}
123
124	#print(f">>> '{line}': None")
125	return None
126
127	def __init__(self, executable, before_lines=1, context_re='', verbose=False, concise=False):
128	self._executable = executable
129	self._current_backtrace = []
130	self._prefix = None
131	self._before_lines = before_lines
132	self._before_lines_queue = collections.deque(maxlen=before_lines)
133	self._i = 0
134	self._known_backtraces = {}
135	if context_re is not None:
136	self._context_re = re.compile(context_re)
137	else:
138	self._context_re = None
139	self._verbose = verbose
140	self._concise = concise
141	self._known_modules = {self._executable: Addr2Line(self._executable, concise)}
142	self.parser = self.BacktraceParser()
143
144	def _get_resolver_for_module(self, module):
145	if not module in self._known_modules:
146	self._known_modules[module] = Addr2Line(module, self._concise)
147	return self._known_modules[module]
148
149	def __enter__(self):
150	return self
151
152	def __exit__(self, type, value, tb):
153	self._print_current_backtrace()
154
155	def resolve_address(self, address, module=None, verbose=None):
156	if module is None:
157	module = self._executable
158	if verbose is None:
159	verbose = self._verbose
160	resolved_address = self._get_resolver_for_module(module)(address)
161	if verbose:
162	resolved_address = '{{{}}} {}: {}'.format(module, address, resolved_address)
163	return resolved_address
164
165	def _print_resolved_address(self, module, address):
166	sys.stdout.write(self.resolve_address(address, module))
167
168	def _backtrace_context_matches(self):
169	if self._context_re is None:
170	return True
171
172	if any(map(lambda x: self._context_re.search(x) is not None, self._before_lines_queue)):
173	return True
174
175	if (not self._prefix is None) and self._context_re.search(self._prefix):
176	return True
177
178	return False
179
180	def _print_current_backtrace(self):
181	if len(self._current_backtrace) == 0:
182	return
183
184	if not self._backtrace_context_matches():
185	self._current_backtrace = []
186	return
187
188	for line in self._before_lines_queue:
189	sys.stdout.write(line)
190
191	if not self._prefix is None:
192	print(self._prefix)
193	self._prefix = None
194
195	backtrace = "".join(map(str, self._current_backtrace))
196	if backtrace in self._known_backtraces:
197	print("[Backtrace #{}] Already seen, not resolving again.".format(self._known_backtraces[backtrace]))
198	print("") # To separate traces with an empty line
199	self._current_backtrace = []
200	return
201
202	self._known_backtraces[backtrace] = self._i
203
204	print("[Backtrace #{}]".format(self._i))
205
206	for module, addr in self._current_backtrace:
207	self._print_resolved_address(module, addr)
208
209	print("") # To separate traces with an empty line
210
211	self._current_backtrace = []
212	self._i += 1
213
214	def __call__(self, line):
215	res = self.parser(line)
216
217	if not res:
218	self._print_current_backtrace()
219	if self._before_lines > 0:
220	self._before_lines_queue.append(line)
221	elif self._before_lines < 0:
222	sys.stdout.write(line) # line already has a trailing newline
223	else:
224	pass # when == 0 no non-backtrace lines are printed
225	elif res['type'] == self.BacktraceParser.Type.SEPARATOR:
226	pass
227	elif res['type'] == self.BacktraceParser.Type.ADDRESS:
228	addresses = res['addresses']
229	if len(addresses) > 1:
230	self._print_current_backtrace()
231	if len(self._current_backtrace) == 0:
232	self._prefix = res['prefix']
233	for r in addresses:
234	if r['path']:
235	self._current_backtrace.append((r['path'], r['addr']))
236	else:
237	self._current_backtrace.append((self._executable, r['addr']))
238	if len(addresses) > 1:
239	self._print_current_backtrace()
240	else:
241	print(f"Unknown '{line}': {res}")
242	raise RuntimeError("Unknown result type {res}")
243