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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
//
// Copyright 2017 Ettus Research, National Instruments Company
//
// SPDX-License-Identifier: GPL-3.0-or-later
//
#include "mpmd_xport_ctrl_udp.hpp"
#include "mpmd_impl.hpp"
#include "mpmd_xport_mgr.hpp"
#include <uhd/transport/udp_constants.hpp>
#include <uhd/transport/udp_simple.hpp>
#include <uhd/transport/udp_zero_copy.hpp>
using namespace uhd;
using namespace uhd::mpmd::xport;
namespace {
#if defined(UHD_PLATFORM_MACOS) || defined(UHD_PLATFORM_BSD)
//! Size of the host-side socket buffer for RX
const size_t MPMD_RX_SW_BUFF_SIZE_ETH = 0x100000; // 1Mib
#elif defined(UHD_PLATFORM_LINUX) || defined(UHD_PLATFORM_WIN32)
//! Size of the host-side socket buffer for RX
// For an ~8k frame size any size >32MiB is just wasted buffer space
const size_t MPMD_RX_SW_BUFF_SIZE_ETH = 0x2000000; // 32 MiB
#endif
//! Maximum CHDR packet size in bytes
const size_t MPMD_10GE_DATA_FRAME_MAX_SIZE = 4000;
//! Maximum CHDR packet size in bytes
const size_t MPMD_10GE_ASYNCMSG_FRAME_MAX_SIZE = 1472;
//! Number of send/recv frames
const size_t MPMD_ETH_NUM_FRAMES = 32;
//!
const double MPMD_BUFFER_FILL_RATE = 20.0e-3; // s
//! For MTU discovery, the time we wait for a packet before calling it
// oversized (seconds).
const double MPMD_MTU_DISCOVERY_TIMEOUT = 0.02;
// TODO: move these to appropriate header file for all other devices
const size_t MAX_RATE_1GIGE = 1e9 / 8; // byte/s
const size_t MAX_RATE_10GIGE = 10e9 / 8; // byte/s
std::vector<std::string> get_addrs_from_mb_args(const uhd::device_addr_t& mb_args)
{
// mb_args must always include addr
if (not mb_args.has_key(FIRST_ADDR_KEY)) {
throw uhd::runtime_error(
"The " + FIRST_ADDR_KEY
+ " key must be specified in "
"device args to create an Ethernet transport to an RFNoC block");
}
std::vector<std::string> addrs{mb_args[FIRST_ADDR_KEY]};
if (mb_args.has_key(SECOND_ADDR_KEY)) {
addrs.push_back(mb_args[SECOND_ADDR_KEY]);
}
return addrs;
}
/*! Do a binary search to discover MTU
*
* Uses the MPM echo service to figure out MTU. We simply send a bunch of
* packets and see if they come back until we converged on the path MTU.
* The end result must lie between \p min_frame_size and \p max_frame_size.
*
* \param address IP address
* \param port UDP port (yeah it's a string!)
* \param min_frame_size Minimum frame size, initialize algorithm to start
* with this value
* \param max_frame_size Maximum frame size, initialize algorithm to start
* with this value
* \param echo_timeout Timeout value in seconds. For frame sizes that
* exceed the MTU, we don't expect a response, and this
* is the amount of time we'll wait before we assume
* the frame size exceeds the MTU.
*/
size_t discover_mtu(const std::string& address,
const std::string& port,
size_t min_frame_size,
size_t max_frame_size,
const double echo_timeout = 0.020)
{
const size_t echo_prefix_offset = uhd::mpmd::mpmd_impl::MPM_ECHO_CMD.size();
const size_t mtu_hdr_len = echo_prefix_offset + 10;
UHD_ASSERT_THROW(min_frame_size < max_frame_size);
UHD_ASSERT_THROW(min_frame_size % 4 == 0);
UHD_ASSERT_THROW(max_frame_size % 4 == 0);
UHD_ASSERT_THROW(min_frame_size >= echo_prefix_offset + mtu_hdr_len);
using namespace uhd::transport;
// The return port will probably differ from the discovery port, so we
// need a "broadcast" UDP connection; using make_connected() would
// drop packets
udp_simple::sptr udp = udp_simple::make_broadcast(address, port);
std::string send_buf(uhd::mpmd::mpmd_impl::MPM_ECHO_CMD);
send_buf.resize(max_frame_size, '#');
UHD_ASSERT_THROW(send_buf.size() == max_frame_size);
std::vector<uint8_t> recv_buf;
recv_buf.resize(max_frame_size, ' ');
// Little helper to check returned packets match the sent ones
auto require_bufs_match = [&recv_buf, &send_buf, mtu_hdr_len](const size_t len) {
if (len < mtu_hdr_len
or std::memcmp((void*)&recv_buf[0], (void*)&send_buf[0], mtu_hdr_len) != 0) {
throw uhd::runtime_error("Unexpected content of MTU "
"discovery return packet!");
}
};
UHD_LOG_TRACE("MPMD", "Determining UDP MTU... ");
size_t seq_no = 0;
while (min_frame_size < max_frame_size) {
// Only test multiples of 4 bytes!
const size_t test_frame_size = (max_frame_size / 2 + min_frame_size / 2 + 3)
& ~size_t(3);
// Encode sequence number and current size in the string, makes it
// easy to debug in code or Wireshark. Is also used for identifying
// response packets.
std::sprintf(
&send_buf[echo_prefix_offset], ";%04lu,%04lu", seq_no++, test_frame_size);
UHD_LOG_TRACE("MPMD", "Testing frame size " << test_frame_size);
udp->send(boost::asio::buffer(&send_buf[0], test_frame_size));
const size_t len = udp->recv(boost::asio::buffer(recv_buf), echo_timeout);
if (len == 0) {
// Nothing received, so this is probably too big
max_frame_size = test_frame_size - 4;
} else if (len >= test_frame_size) {
// Size went through, so bump the minimum
require_bufs_match(len);
min_frame_size = test_frame_size;
} else if (len < test_frame_size) {
// This is an odd case. Something must have snipped the packet
// on the way back. Still, we'll just back off and try
// something smaller.
UHD_LOG_DEBUG("MPMD", "Unexpected packet truncation during MTU discovery.");
require_bufs_match(len);
max_frame_size = len;
}
}
UHD_LOG_DEBUG("MPMD", "Path MTU for address " << address << ": " << min_frame_size);
return min_frame_size;
}
} // namespace
mpmd_xport_ctrl_udp::mpmd_xport_ctrl_udp(const uhd::device_addr_t& mb_args)
: _mb_args(mb_args)
, _recv_args(filter_args(mb_args, "recv"))
, _send_args(filter_args(mb_args, "send"))
, _available_addrs(get_addrs_from_mb_args(mb_args))
, _mtu(MPMD_10GE_DATA_FRAME_MAX_SIZE)
{
const std::string mpm_discovery_port = _mb_args.get(
mpmd_impl::MPM_DISCOVERY_PORT_KEY, std::to_string(mpmd_impl::MPM_DISCOVERY_PORT));
auto discover_mtu_for_ip = [mpm_discovery_port](const std::string& ip_addr) {
return discover_mtu(ip_addr,
mpm_discovery_port,
IP_PROTOCOL_MIN_MTU_SIZE - IP_PROTOCOL_UDP_PLUS_IP_HEADER,
MPMD_10GE_DATA_FRAME_MAX_SIZE,
MPMD_MTU_DISCOVERY_TIMEOUT);
};
for (const auto& ip_addr : _available_addrs) {
_mtu = std::min(_mtu, discover_mtu_for_ip(ip_addr));
}
}
uhd::both_xports_t mpmd_xport_ctrl_udp::make_transport(
mpmd_xport_mgr::xport_info_t& xport_info,
const usrp::device3_impl::xport_type_t xport_type,
const uhd::device_addr_t& xport_args_)
{
auto xport_args = xport_args_;
if (xport_type == usrp::device3_impl::RX_DATA
and not xport_args.has_key("recv_buff_size")) {
xport_args["recv_buff_size"] = std::to_string(MPMD_RX_SW_BUFF_SIZE_ETH);
}
size_t link_speed = MAX_RATE_1GIGE;
if (xport_info.count("link_speed") == 0) {
UHD_LOG_WARNING("MPMD",
"Could not determine link speed; using 1GibE max speed of "
<< MAX_RATE_1GIGE);
} else {
link_speed = xport_info.at("link_speed") == "10000" ? MAX_RATE_10GIGE
: MAX_RATE_1GIGE;
}
transport::zero_copy_xport_params default_buff_args;
// Create actual UDP transport
default_buff_args.recv_frame_size =
xport_args.cast<size_t>("recv_frame_size", get_mtu(uhd::RX_DIRECTION));
default_buff_args.recv_buff_size = link_speed * MPMD_BUFFER_FILL_RATE;
default_buff_args.send_buff_size = link_speed * MPMD_BUFFER_FILL_RATE;
if (xport_type == usrp::device3_impl::ASYNC_MSG) {
default_buff_args.send_frame_size = MPMD_10GE_ASYNCMSG_FRAME_MAX_SIZE;
} else {
default_buff_args.send_frame_size =
xport_args.cast<size_t>("send_frame_size", get_mtu(uhd::TX_DIRECTION));
}
transport::udp_zero_copy::buff_params buff_params;
auto recv = transport::udp_zero_copy::make(xport_info["ipv4"],
xport_info["port"],
default_buff_args,
buff_params,
xport_args);
const uint16_t port = recv->get_local_port();
const std::string src_ip_addr = recv->get_local_addr();
xport_info["src_port"] = std::to_string(port);
xport_info["src_ipv4"] = src_ip_addr;
// Create both_xports_t object and finish:
both_xports_t xports;
xports.endianness = uhd::ENDIANNESS_BIG;
xports.send_sid = sid_t(xport_info["send_sid"]);
xports.recv_sid = xports.send_sid.reversed();
xports.recv_buff_size = buff_params.recv_buff_size;
xports.send_buff_size = buff_params.send_buff_size;
xports.recv = recv; // Note: This is a type cast!
xports.send = recv; // This too
return xports;
}
bool mpmd_xport_ctrl_udp::is_valid(const mpmd_xport_mgr::xport_info_t& xport_info) const
{
return std::find(
_available_addrs.cbegin(), _available_addrs.cend(), xport_info.at("ipv4"))
!= _available_addrs.cend();
}
size_t mpmd_xport_ctrl_udp::get_mtu(const uhd::direction_t /*dir*/) const
{
return _mtu;
}
|
