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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
|
//
// Copyright 2017 Ettus Research, National Instruments Company
// Copyright 2019 Ettus Research, National Instruments Brand
//
// SPDX-License-Identifier: GPL-3.0-or-later
//
#include "mpmd_link_if_ctrl_udp.hpp"
#include "mpmd_impl.hpp"
#include "mpmd_link_if_mgr.hpp"
#include <uhd/rfnoc/constants.hpp>
#include <uhd/transport/udp_constants.hpp>
#include <uhd/transport/udp_simple.hpp>
#include <uhd/transport/udp_zero_copy.hpp>
#include <uhdlib/rfnoc/rfnoc_common.hpp>
#include <uhdlib/transport/udp_boost_asio_link.hpp>
#include <uhdlib/transport/udp_common.hpp>
#include <uhdlib/utils/narrow.hpp>
#include <string>
#ifdef HAVE_DPDK
//# include <uhdlib/transport/dpdk_simple.hpp>
# include <uhdlib/transport/udp_dpdk_link.hpp>
#endif
using namespace uhd;
using namespace uhd::transport;
using namespace uhd::mpmd::xport;
const uhd::rfnoc::chdr::chdr_packet_factory mpmd_link_if_ctrl_udp::_pkt_factory(
uhd::rfnoc::CHDR_W_64, ENDIANNESS_LITTLE);
namespace {
//! Maximum CHDR packet size in bytes
const size_t MPMD_10GE_DATA_FRAME_MAX_SIZE = 8000;
const size_t MPMD_1GE_DATA_FRAME_MAX_SIZE = 1472;
const size_t MPMD_1GE_ASYNCMSG_FRAME_MAX_SIZE = 1472;
//! Number of send/recv frames
const size_t MPMD_ETH_NUM_FRAMES = 32;
//!
const double MPMD_BUFFER_DEPTH = 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
mpmd_link_if_ctrl_udp::udp_link_info_map get_udp_info_from_xport_info(
const mpmd_link_if_mgr::xport_info_list_t& link_info_list)
{
mpmd_link_if_ctrl_udp::udp_link_info_map result;
for (const auto& link_info : link_info_list) {
if (!link_info.count("ipv4")) {
UHD_LOG_ERROR("MPMD::XPORT::UDP",
"Invalid response from get_chdr_link_options()! No `ipv4' key!");
throw uhd::runtime_error(
"Invalid response from get_chdr_link_options()! No `ipv4' key!");
}
if (!link_info.count("port")) {
UHD_LOG_ERROR("MPMD::XPORT::UDP",
"Invalid response from get_chdr_link_options()! No `port' key!");
throw uhd::runtime_error(
"Invalid response from get_chdr_link_options()! No `port' key!");
}
const std::string udp_port = link_info.at("port");
const size_t link_rate = link_info.count("link_rate")
? std::stoul(link_info.at("link_rate"))
: MAX_RATE_1GIGE;
result.emplace(link_info.at("ipv4"),
mpmd_link_if_ctrl_udp::udp_link_info_t{udp_port, link_rate});
}
return result;
}
std::vector<std::string> get_addrs_from_mb_args(const uhd::device_addr_t& mb_args,
const mpmd_link_if_ctrl_udp::udp_link_info_map& link_info_list)
{
// mb_args must always include addr
if (not mb_args.has_key(FIRST_ADDR_KEY)) {
UHD_LOG_WARNING("MPMD::XPORT::UDP",
"The `" << FIRST_ADDR_KEY
<< "' key must be specified in "
"device args to create an Ethernet transport to an RFNoC block");
return {};
}
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]);
}
// This is where in UHD we encode the knowledge about what
// get_chdr_link_options() returns to us.
for (const auto& ip_addr : addrs) {
if (link_info_list.count(ip_addr)) {
continue;
}
UHD_LOG_WARNING("MPMD::XPORT::UDP",
"Cannot create UDP link to device: The IP address `"
<< ip_addr << "' is requested, but not reachable.");
return {};
}
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,
const bool use_dpdk)
{
if (use_dpdk) {
#ifdef HAVE_DPDK
// FIXME
UHD_LOG_WARNING("MPMD", "Using hard-coded MTU of 8000 for DPDK");
return 8000;
#else
UHD_LOG_WARNING("MPMD",
"DPDK was requested but is not available, falling back to regular UDP");
#endif
}
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;
}
// DPDK version
//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 auto& ctx = uhd::transport::uhd_dpdk_ctx::get();
//const size_t echo_prefix_offset = uhd::mpmd::mpmd_impl::MPM_ECHO_CMD.size();
//const size_t mtu_hdr_len = echo_prefix_offset + 10;
//const int port_id = ctx.get_route(address);
//UHD_ASSERT_THROW(port_id >= 0);
//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;
//uhd::transport::zero_copy_xport_params buff_args;
//buff_args.recv_frame_size = max_frame_size;
//buff_args.send_frame_size = max_frame_size;
//buff_args.num_send_frames = 1;
//buff_args.num_recv_frames = 1;
//auto dev_addr = uhd::device_addr_t();
//dpdk_zero_copy::sptr sock = dpdk_zero_copy::make(
//ctx, (unsigned int)port_id, address, port, "0", buff_args, dev_addr);
//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);
//// Little helper to check returned packets match the sent ones
//auto require_bufs_match = [&send_buf, mtu_hdr_len](
//const uint8_t* recv_buf, 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) {
//managed_send_buffer::sptr msbuf = sock->get_send_buff(0);
//UHD_ASSERT_THROW(msbuf.get() != nullptr);
//max_frame_size = std::min(msbuf->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);
//// Copy to real buffer
//UHD_LOG_TRACE("MPMD", "Testing frame size " << test_frame_size);
//auto* tx_buf = msbuf->cast<uint8_t*>();
//std::memcpy(tx_buf, &send_buf[0], test_frame_size);
//msbuf->commit(test_frame_size);
//msbuf.reset();
//managed_recv_buffer::sptr mrbuf = sock->get_recv_buff(echo_timeout);
//if (mrbuf.get() == nullptr || mrbuf->size() == 0) {
//// Nothing received, so this is probably too big
//max_frame_size = test_frame_size - 4;
//} else if (mrbuf->size() >= test_frame_size) {
//// Size went through, so bump the minimum
//require_bufs_match(mrbuf->cast<uint8_t*>(), mrbuf->size());
//min_frame_size = test_frame_size;
//} else if (mrbuf->size() < 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(mrbuf->cast<uint8_t*>(), mrbuf->size());
//max_frame_size = mrbuf->size();
//}
//mrbuf.reset();
//}
//UHD_LOG_DEBUG("MPMD", "Path MTU for address " << address << ": " << min_frame_size);
//return min_frame_size;
//}
} // namespace
/******************************************************************************
* Structors
*****************************************************************************/
mpmd_link_if_ctrl_udp::mpmd_link_if_ctrl_udp(const uhd::device_addr_t& mb_args,
const mpmd_link_if_mgr::xport_info_list_t& xport_info)
: _mb_args(mb_args)
, _udp_info(get_udp_info_from_xport_info(xport_info))
, _mtu(MPMD_10GE_DATA_FRAME_MAX_SIZE)
{
const bool use_dpdk = mb_args.has_key("use_dpdk"); // FIXME use constrained_device_args
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, use_dpdk](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,
use_dpdk);
};
const std::vector<std::string> requested_addrs(
get_addrs_from_mb_args(mb_args, _udp_info));
for (const auto& ip_addr : requested_addrs) {
try {
// If MTU discovery fails, we gracefully recover, but declare that
// link invalid.
_mtu = std::min(_mtu, discover_mtu_for_ip(ip_addr));
_available_addrs.push_back(ip_addr);
} catch (const uhd::exception& ex) {
UHD_LOG_WARNING("MPMD::XPORT::UDP",
"Error during MTU discovery on address " << ip_addr << ": " << ex.what());
}
}
}
/******************************************************************************
* API
*****************************************************************************/
uhd::transport::both_links_t mpmd_link_if_ctrl_udp::get_link(const size_t link_idx,
const uhd::transport::link_type_t link_type,
const uhd::device_addr_t& link_args)
{
UHD_ASSERT_THROW(link_idx < _available_addrs.size());
const std::string ip_addr = _available_addrs.at(link_idx);
const std::string udp_port = _udp_info.at(ip_addr).udp_port;
const size_t link_rate = get_link_rate(link_idx);
link_params_t default_link_params;
default_link_params.num_send_frames = MPMD_ETH_NUM_FRAMES;
default_link_params.num_recv_frames = MPMD_ETH_NUM_FRAMES;
default_link_params.send_frame_size = (link_rate == MAX_RATE_10GIGE)
? MPMD_10GE_DATA_FRAME_MAX_SIZE
: (link_rate == MAX_RATE_1GIGE)
? MPMD_1GE_DATA_FRAME_MAX_SIZE
: get_mtu(uhd::TX_DIRECTION);
default_link_params.recv_frame_size = (link_rate == MAX_RATE_10GIGE)
? MPMD_10GE_DATA_FRAME_MAX_SIZE
: (link_rate == MAX_RATE_1GIGE)
? MPMD_1GE_DATA_FRAME_MAX_SIZE
: get_mtu(uhd::RX_DIRECTION);
default_link_params.send_buff_size = get_link_rate(link_idx) * MPMD_BUFFER_DEPTH;
default_link_params.recv_buff_size = get_link_rate(link_idx) * MPMD_BUFFER_DEPTH;
link_params_t link_params = calculate_udp_link_params(link_type,
get_mtu(uhd::TX_DIRECTION),
get_mtu(uhd::RX_DIRECTION),
default_link_params,
_mb_args,
link_args);
// Enforce a minimum bound of the number of receive and send frames.
link_params.num_send_frames =
std::max(uhd::rfnoc::MIN_NUM_FRAMES, link_params.num_send_frames);
link_params.num_recv_frames =
std::max(uhd::rfnoc::MIN_NUM_FRAMES, link_params.num_recv_frames);
if (_mb_args.has_key("use_dpdk")) { // FIXME use constrained device args
#ifdef HAVE_DPDK
auto link = uhd::transport::udp_dpdk_link::make(ip_addr, udp_port, link_params);
return std::make_tuple(
link, link_params.send_buff_size, link, link_params.recv_buff_size, true);
#else
UHD_LOG_WARNING("X300", "Cannot create DPDK transport, falling back to UDP");
#endif
}
auto link = uhd::transport::udp_boost_asio_link::make(ip_addr,
udp_port,
link_params,
link_params.recv_buff_size,
link_params.send_buff_size);
return std::make_tuple(
link, link_params.send_buff_size, link, link_params.recv_buff_size, true);
}
size_t mpmd_link_if_ctrl_udp::get_num_links() const
{
return _available_addrs.size();
}
//! Return the rate of the underlying link in bytes/sec
double mpmd_link_if_ctrl_udp::get_link_rate(const size_t link_idx) const
{
UHD_ASSERT_THROW(link_idx < get_num_links());
return _udp_info.at(_available_addrs.at(link_idx)).link_rate;
}
const uhd::rfnoc::chdr::chdr_packet_factory&
mpmd_link_if_ctrl_udp::get_packet_factory() const
{
return _pkt_factory;
}
|