TIMEOUT = 0.5
+STACK_HEADER_STR = r'Stack (most recent call first):'
+
+# Regular expressions
+STACK_HEADER = re.escape(STACK_HEADER_STR)
+THREAD_NAME = r'( \[.*\])?'
+THREAD_ID = fr'Thread 0x[0-9a-f]+{THREAD_NAME}'
+THREAD_HEADER = fr'{THREAD_ID} \(most recent call first\):'
+CURRENT_THREAD_ID = fr'Current thread 0x[0-9a-f]+{THREAD_NAME}'
+CURRENT_THREAD_HEADER = fr'{CURRENT_THREAD_ID} \(most recent call first\):'
+
def expected_traceback(lineno1, lineno2, header, min_count=1):
regex = header
)
if all_threads and not all_threads_disabled:
if know_current_thread:
- header = 'Current thread 0x[0-9a-f]+'
+ header = CURRENT_THREAD_HEADER
else:
- header = 'Thread 0x[0-9a-f]+'
+ header = THREAD_HEADER
else:
- header = 'Stack'
+ header = STACK_HEADER
regex = [f'^{fatal_error}']
if py_fatal_error:
regex.append("Python runtime state: initialized")
regex.append('')
if all_threads_disabled and not py_fatal_error:
regex.append("<Cannot show all threads while the GIL is disabled>")
- regex.append(fr'{header} \(most recent call first\):')
+ regex.append(fr'{header}')
if support.Py_GIL_DISABLED and py_fatal_error and not know_current_thread:
regex.append(" <tstate is freed>")
else:
else:
lineno = 14
expected = [
- 'Stack (most recent call first):',
+ f'{STACK_HEADER_STR}',
' File "<string>", line %s in funcB' % lineno,
' File "<string>", line 17 in funcA',
' File "<string>", line 19 in <module>'
func_name=func_name,
)
expected = [
- 'Stack (most recent call first):',
+ f'{STACK_HEADER_STR}',
' File "<string>", line 4 in %s' % truncated,
' File "<string>", line 6 in <module>'
]
lineno = 10
# When the traceback is dumped, the waiter thread may be in the
# `self.running.set()` call or in `self.stop.wait()`.
- regex = r"""
- ^Thread 0x[0-9a-f]+ \(most recent call first\):
+ regex = fr"""
+ ^{THREAD_HEADER}
(?: File ".*threading.py", line [0-9]+ in [_a-z]+
){{1,3}} File "<string>", line (?:22|23) in run
File ".*threading.py", line [0-9]+ in _bootstrap_inner
File ".*threading.py", line [0-9]+ in _bootstrap
- Current thread 0x[0-9a-f]+ \(most recent call first\):
+ {CURRENT_THREAD_HEADER}
File "<string>", line {lineno} in dump
File "<string>", line 28 in <module>$
"""
- regex = dedent(regex.format(lineno=lineno)).strip()
+ regex = dedent(regex).strip()
self.assertRegex(output, regex)
self.assertEqual(exitcode, 0)
count = loops
if repeat:
count *= 2
- header = r'Timeout \(%s\)!\nThread 0x[0-9a-f]+ \(most recent call first\):\n' % timeout_str
+ header = (fr'Timeout \({timeout_str}\)!\n'
+ fr'{THREAD_HEADER}\n')
regex = expected_traceback(17, 26, header, min_count=count)
self.assertRegex(trace, regex)
else:
trace = '\n'.join(trace)
if not unregister:
if all_threads:
- regex = r'Current thread 0x[0-9a-f]+ \(most recent call first\):\n'
+ regex = fr'{CURRENT_THREAD_HEADER}\n'
else:
- regex = r'Stack \(most recent call first\):\n'
+ regex = fr'{STACK_HEADER}\n'
regex = expected_traceback(14, 32, regex)
self.assertRegex(trace, regex)
else:
#define OFF(x) offsetof(PyTracebackObject, x)
-#define PUTS(fd, str) (void)_Py_write_noraise(fd, str, (int)strlen(str))
+#define PUTS(fd, str) (void)_Py_write_noraise(fd, str, strlen(str))
#define MAX_STRING_LENGTH 500
#define MAX_FRAME_DEPTH 100
_Py_DumpHexadecimal(fd,
tstate->thread_id,
sizeof(unsigned long) * 2);
+
+ // Write the thread name
+#if defined(HAVE_PTHREAD_GETNAME_NP) || defined(HAVE_PTHREAD_GET_NAME_NP)
+ char name[100];
+ pthread_t thread = (pthread_t)tstate->thread_id;
+#ifdef HAVE_PTHREAD_GETNAME_NP
+ int rc = pthread_getname_np(thread, name, Py_ARRAY_LENGTH(name));
+#else /* defined(HAVE_PTHREAD_GET_NAME_NP) */
+ int rc = 0; /* pthread_get_name_np() returns void */
+ pthread_get_name_np(thread, name, Py_ARRAY_LENGTH(name));
+#endif
+ if (!rc) {
+ size_t len = strlen(name);
+ if (len) {
+ PUTS(fd, " [");
+ (void)_Py_write_noraise(fd, name, len);
+ PUTS(fd, "]");
+ }
+ }
+#endif
+
PUTS(fd, " (most recent call first):\n");
}