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
144
145
146
147
148
149
150
151
152
153
154
|
#!/usr/bin/env python3
#
# SPDX-FileCopyrightText: 2024 Kienan Stewart <kstewart@efficios.com>
# SPDX-License-Identifier: GPL-2.0-only
#
"""
Tests that the lttng command-line client emits a warning when the
the a shared memory path for a session is smaller than an estimate
of the minimum memory allocation required based on the number of
sub-buffers, the sub-buffer size, the number of CPUs, and if the
session is in snapshot mode or not.
"""
import math
import os
import pathlib
import subprocess
import sys
import time
test_utils_import_path = pathlib.Path(__file__).absolute().parents[3] / "utils"
sys.path.append(str(test_utils_import_path))
import lttngtest
def test_shm_warning(tap, test_env, tests):
if os.getuid() != 0:
tap.skip_all_remaining("This test requires root to make a temporary shm mount")
return
# Create a 64M shm mount. Many containers default to a shm of this size.
# @see https://github.com/moby/moby/blob/a95a6788b59885056512c837897db20684433780/daemon/config/config.go#L39
# @see https://docs.podman.io/en/v5.3.0/markdown/podman-run.1.html#shm-size-number-unit
shm_path = lttngtest.TemporaryDirectory("tmp")
p = subprocess.Popen(
["mount", "-t", "tmpfs", "-o", "size=64M", "tmpfs", str(shm_path.path)]
)
p.wait()
if p.returncode != 0:
tap.skip_all_remaining("Couldn't create tmpfs for testing alternate shm path")
return
shm_path.add_cleanup_callback(
lambda path: subprocess.Popen(["umount", path]).wait(), str(shm_path.path)
)
# This may not be the CPUs available on the system, as it could be limited
# by `-X cpu-count` or `PYTHON_CPU_COUNT` as of Python 3.13
ncpus = os.cpu_count()
if ncpus is None:
tap.skip_all_remaining("Cannot determine CPU count")
return
client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
for test in tests:
subbuf_count = 2 if test["snapshot"] else 1
subbuf_size = (
test["target_usage_mib"] / ncpus / test["nchannels"] / subbuf_count
)
if not math.log(subbuf_size, 2).is_integer():
tag.diagnostic(subbuf_size)
tap.skip("Sub-buffer size {} is not a power of 2".format(subbuf_size))
continue
subbuf_size = int(subbuf_size)
tap.diagnostic(
"Case: nCPUs {} with {} channels of {} sub-buffer(s) of size {}M [{}] {} warn".format(
ncpus,
test["nchannels"],
subbuf_count,
subbuf_size,
"snapshot enabled" if test["snapshot"] else "snapshot disabled",
"should" if test["warning_expected"] else "should not",
)
)
session = client.create_session(
output=None, shm_path=shm_path.path, snapshot=test["snapshot"]
)
channels = []
for channel in range(test["nchannels"]):
channel = session.add_channel(
lttngtest.lttngctl.TracingDomain.User,
lttngtest.lttngctl.BufferSharingPolicy.PerUID,
subbuf_size="{}M".format(subbuf_size),
subbuf_count=subbuf_count,
)
channel.add_recording_rule(
lttngtest.lttngctl.UserTracepointEventRule("tp:tptest")
)
output = client._run_cmd("start '{}'".format(session.name))
tap.diagnostic("\n{}\n".format(output))
session.destroy()
tap.test(
("Warning" in output and test["warning_expected"])
or ("Warning" not in output and not test["warning_expected"]),
"Warning {} in lttng client output when starting session".format(
"present" if test["warning_expected"] else "not present"
),
)
if __name__ == "__main__":
tests = [
{
"warning_expected": False,
"target_usage_mib": 32,
"snapshot": False,
"nchannels": 1,
},
{
"warning_expected": True,
"target_usage_mib": 64,
"snapshot": False,
"nchannels": 1,
},
{
"warning_expected": True,
"target_usage_mib": 128,
"snapshot": False,
"nchannels": 1,
},
{
"warning_expected": False,
"target_usage_mib": 32,
"snapshot": True,
"nchannels": 1,
},
{
"warning_expected": True,
"target_usage_mib": 64,
"snapshot": True,
"nchannels": 1,
},
{
"warning_expected": False,
"target_usage_mib": 32,
"snapshot": False,
"nchannels": 2,
},
{
"warning_expected": True,
"target_usage_mib": 64,
"snapshot": False,
"nchannels": 2,
},
]
tap = lttngtest.TapGenerator(len(tests))
with lttngtest.test_environment(log=tap.diagnostic, with_sessiond=True) as test_env:
test_shm_warning(tap, test_env, tests)
|
