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
|
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2025 Arm Limited
"""Class to run blocking apps in the background.
The class won't automatically start the app. The start-up is done as part of the
:meth:`BlockingApp.wait_until_ready` method, which will return execution to the caller only
when the desired stdout has been returned by the app. Usually this is used to detect when the app
has been loaded and ready to be used.
This module also provides the class :class:`BlockingDPDKApp` useful to run any DPDK app from the
DPDK build dir.
Example:
..code:: python
pdump = BlockingDPDKApp(
PurePath("app/dpdk-pdump"),
app_params="--pdump 'port=0,queue=*,rx-dev=/tmp/rx-dev.pcap'"
)
pdump.wait_until_ready("65535") # start app
# pdump is now ready to capture
pdump.close() # stop/close app
"""
from pathlib import PurePath
from typing import Generic, TypeVar, cast
from typing_extensions import Self
from framework.context import get_ctx
from framework.params import Params
from framework.params.eal import EalParams
from framework.remote_session.dpdk_shell import compute_eal_params
from framework.remote_session.interactive_shell import InteractiveShell
from framework.testbed_model.node import Node
P = TypeVar("P", bound=Params)
class BlockingApp(InteractiveShell, Generic[P]):
"""Class to manage generic blocking apps."""
_app_params: P
def __init__(
self,
node: Node,
path: str | PurePath,
name: str | None = None,
privileged: bool = False,
app_params: P | str = "",
add_to_shell_pool: bool = True,
) -> None:
"""Constructor.
Args:
node: The node to run the app on.
path: Path to the application on the node.s
name: Name to identify this application.
privileged: Run as privileged user.
app_params: The application parameters. Can be of any type inheriting :class:`Params` or
a plain string.
add_to_shell_pool: If :data:`True`, the blocking app's shell will be added to the
shell pool.
"""
if isinstance(app_params, str):
params = Params()
params.append_str(app_params)
app_params = cast(P, params)
self._path = path
self._add_to_shell_pool = add_to_shell_pool
super().__init__(node, name, privileged, app_params)
@property
def path(self) -> str | PurePath:
"""The path of the DPDK app relative to the DPDK build folder."""
return self._path
def wait_until_ready(self, end_token: str) -> Self:
"""Start app and wait until ready.
Args:
end_token: The string at the end of a line that indicates the app is ready.
Returns:
Itself.
"""
self.start_application(end_token, self._add_to_shell_pool)
return self
def close(self) -> None:
"""Close the application.
Sends a SIGINT to close the application.
"""
self.send_command("\x03")
super().close()
PE = TypeVar("PE", bound=EalParams)
class BlockingDPDKApp(BlockingApp, Generic[PE]):
"""Class to manage blocking DPDK apps on the SUT."""
_app_params: PE
def __init__(
self,
path: PurePath,
name: str | None = None,
privileged: bool = True,
app_params: PE | str = "",
) -> None:
"""Constructor.
Args:
path: Path relative to the DPDK build to the executable.
name: Name to identify this application.
privileged: Run as privileged user.
app_params: The application parameters. If a string or an incomplete :class:`EalParams`
object are passed, the EAL params are computed based on the current context.
"""
if isinstance(app_params, str):
eal_params = compute_eal_params()
eal_params.append_str(app_params)
app_params = cast(PE, eal_params)
else:
app_params = cast(PE, compute_eal_params(app_params))
node = get_ctx().sut_node
path = PurePath(get_ctx().dpdk_build.remote_dpdk_build_dir).joinpath(self.path)
super().__init__(node, path, name, privileged, app_params)
|
