Skip to content

Commit c4e66dc

Browse files
authored
Merge pull request #10135 from amastbaum/check_reachability_using_routing_table
UCT: check reachability using the routing table
2 parents 4ff3ae7 + 46f8e6f commit c4e66dc

17 files changed

Lines changed: 470 additions & 108 deletions

File tree

AUTHORS

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ Alex Margolin <alex.margolin@huawei.com>
44
Alex Mikheev <alexm@mellanox.com>
55
Alexey Rivkin <arivkin@nvidia.com>
66
Alina Sklarevich <alinas@mellanox.com>
7+
Alma Mastbaum <amastbaum@nvidia.com>
78
Anatoly Vildemanov <anatolyv@nvidia.com>
89
Andrey Maslennikov <andreyma@mellanox.com>
910
Artem Polyakov <artemp@nvidia.com>

buildlib/tools/build_static.sh

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,6 @@ az_init_modules
6464
prepare_build
6565

6666
# Don't cross-connect RoCE devices
67-
export UCX_IB_ROCE_LOCAL_SUBNET=y
6867
export UCX_IB_ROCE_SUBNET_PREFIX_LEN=inf
6968
build_static
7069

buildlib/tools/test_wire_compat.sh

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ common_opt=$3
1111
port=$((10000 + 1000 * ${AZP_AGENT_ID} + 100 * ${WIRE_COMPAT_STAGE_ID}))
1212

1313
export UCX_CM_REUSEADDR=y UCX_LOG_LEVEL=info UCX_WARN_UNUSED_ENV_VARS=n
14-
export UCX_IB_ROCE_LOCAL_SUBNET=y
1514

1615
exe_cmd="stdbuf -oL ${exe_name} -p ${port}"
1716

contrib/test_jenkins.sh

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1143,7 +1143,6 @@ set_ucx_common_test_env() {
11431143
export UCX_TCP_CM_REUSEADDR=y
11441144

11451145
# Don't cross-connect RoCE devices
1146-
export UCX_IB_ROCE_LOCAL_SUBNET=y
11471146
export UCX_IB_ROCE_SUBNET_PREFIX_LEN=inf
11481147

11491148
export LSAN_OPTIONS=suppressions=${WORKSPACE}/contrib/lsan.supp

src/ucs/Makefile.am

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,7 @@ noinst_HEADERS = \
131131
sys/iovec.h \
132132
sys/iovec.inl \
133133
sys/ptr_arith.h \
134+
sys/netlink.h \
134135
time/time.h \
135136
time/timerq.h \
136137
time/timer_wheel.h \
@@ -205,6 +206,7 @@ libucs_la_SOURCES = \
205206
sys/sock.c \
206207
sys/topo/base/topo.c \
207208
sys/stubs.c \
209+
sys/netlink.c \
208210
sys/uid.c \
209211
time/time.c \
210212
time/timer_wheel.c \

src/ucs/sys/netlink.c

Lines changed: 237 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,237 @@
1+
/**
2+
* Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2025. ALL RIGHTS RESERVED.
3+
*
4+
* See file LICENSE for terms.
5+
*/
6+
7+
#ifdef HAVE_CONFIG_H
8+
#include "config.h"
9+
#endif
10+
11+
#include "netlink.h"
12+
13+
#include <ucs/debug/log.h>
14+
#include <ucs/sys/compiler.h>
15+
#include <ucs/sys/sock.h>
16+
#include <ucs/type/status.h>
17+
#include <ucs/debug/memtrack_int.h>
18+
19+
#include <errno.h>
20+
#include <linux/netlink.h>
21+
#include <linux/rtnetlink.h>
22+
#include <sys/socket.h>
23+
#include <unistd.h>
24+
25+
26+
typedef struct {
27+
const struct sockaddr *sa_remote;
28+
int if_index;
29+
int found;
30+
} ucs_netlink_route_info_t;
31+
32+
33+
/**
34+
* Callback function for parsing individual netlink messages.
35+
*
36+
* @param [in] nlh Pointer to the netlink message header.
37+
* @param [in] nl_msg Pointer to the netlink message payload.
38+
* @param [in] arg User-provided argument passed through from the caller.
39+
*
40+
* @return UCS_OK if parsing is complete, UCS_INPROGRESS if there are more
41+
* messages to be parsed, or error code otherwise.
42+
*/
43+
typedef ucs_status_t (*ucs_netlink_parse_cb_t)(const struct nlmsghdr *nlh,
44+
const void *nl_msg, void *arg);
45+
46+
static ucs_status_t ucs_netlink_socket_init(int *fd_p, int protocol)
47+
{
48+
struct sockaddr_nl sa = {.nl_family = AF_NETLINK};
49+
ucs_status_t status;
50+
51+
status = ucs_socket_create(AF_NETLINK, SOCK_RAW, protocol, fd_p);
52+
if (status != UCS_OK) {
53+
ucs_error("failed to create netlink socket: %s",
54+
ucs_status_string(status));
55+
goto err;
56+
}
57+
58+
if (bind(*fd_p, (struct sockaddr *)&sa, sizeof(sa)) < 0) {
59+
ucs_error("failed to bind netlink socket %d: %m", *fd_p);
60+
status = UCS_ERR_IO_ERROR;
61+
goto err_close_socket;
62+
}
63+
64+
return UCS_OK;
65+
66+
err_close_socket:
67+
ucs_close_fd(fd_p);
68+
err:
69+
return status;
70+
}
71+
72+
static ucs_status_t
73+
ucs_netlink_parse_msg(const void *msg, size_t msg_len,
74+
ucs_netlink_parse_cb_t parse_cb, void *arg)
75+
{
76+
ucs_status_t status = UCS_INPROGRESS;
77+
const struct nlmsghdr *nlh = (const struct nlmsghdr *)msg;
78+
79+
while ((status == UCS_INPROGRESS) && NLMSG_OK(nlh, msg_len) &&
80+
(nlh->nlmsg_type != NLMSG_DONE)) {
81+
if (nlh->nlmsg_type == NLMSG_ERROR) {
82+
struct nlmsgerr *err = (struct nlmsgerr *)NLMSG_DATA(nlh);
83+
ucs_error("received error response from netlink err=%d: %s\n",
84+
err->error, strerror(-err->error));
85+
return UCS_ERR_IO_ERROR;
86+
}
87+
88+
status = parse_cb(nlh, NLMSG_DATA(nlh), arg);
89+
nlh = NLMSG_NEXT(nlh, msg_len);
90+
}
91+
92+
return UCS_OK;
93+
}
94+
95+
static ucs_status_t
96+
ucs_netlink_send_request(int protocol, unsigned short nlmsg_type,
97+
const void *protocol_header, size_t header_length,
98+
ucs_netlink_parse_cb_t parse_cb, void *arg)
99+
{
100+
struct nlmsghdr nlh = {0};
101+
char *recv_msg = NULL;
102+
size_t recv_msg_len = 0;
103+
int netlink_fd = -1;
104+
ucs_status_t status;
105+
struct iovec iov[2];
106+
size_t bytes_sent;
107+
108+
status = ucs_netlink_socket_init(&netlink_fd, protocol);
109+
if (status != UCS_OK) {
110+
goto out;
111+
}
112+
113+
nlh.nlmsg_len = NLMSG_LENGTH(header_length);
114+
nlh.nlmsg_type = nlmsg_type;
115+
nlh.nlmsg_flags = NLM_F_REQUEST | NLM_F_DUMP;
116+
iov[0].iov_base = &nlh;
117+
iov[0].iov_len = sizeof(nlh);
118+
iov[1].iov_base = (void *)protocol_header;
119+
iov[1].iov_len = header_length;
120+
121+
do {
122+
status = ucs_socket_sendv_nb(netlink_fd, iov, 2, &bytes_sent);
123+
} while (status == UCS_ERR_NO_PROGRESS);
124+
125+
if (status != UCS_OK) {
126+
ucs_error("failed to send netlink message on fd=%d: %s",
127+
netlink_fd, ucs_status_string(status));
128+
goto out;
129+
}
130+
131+
/* get message size */
132+
status = ucs_socket_recv_nb(netlink_fd, NULL, MSG_PEEK | MSG_TRUNC,
133+
&recv_msg_len);
134+
if (status != UCS_OK) {
135+
ucs_error("failed to get netlink message size %d (%s)",
136+
status, ucs_status_string(status));
137+
goto out;
138+
}
139+
140+
recv_msg = ucs_malloc(recv_msg_len, "netlink recv message");
141+
if (recv_msg == NULL) {
142+
ucs_error("failed to allocate a buffer for netlink receive message of"
143+
" size %zu", recv_msg_len);
144+
goto out;
145+
}
146+
147+
status = ucs_socket_recv(netlink_fd, recv_msg, recv_msg_len);
148+
if (status != UCS_OK) {
149+
ucs_error("failed to receive netlink message on fd=%d: %s",
150+
netlink_fd, ucs_status_string(status));
151+
goto out;
152+
}
153+
154+
status = ucs_netlink_parse_msg(recv_msg, recv_msg_len, parse_cb, arg);
155+
156+
out:
157+
ucs_close_fd(&netlink_fd);
158+
ucs_free(recv_msg);
159+
return status;
160+
}
161+
162+
static ucs_status_t
163+
ucs_netlink_get_route_info(const struct rtattr *rta, int len, int *if_index_p,
164+
const void **dst_in_addr)
165+
{
166+
*if_index_p = -1;
167+
*dst_in_addr = NULL;
168+
169+
for (; RTA_OK(rta, len); rta = RTA_NEXT(rta, len)) {
170+
if (rta->rta_type == RTA_OIF) {
171+
*if_index_p = *((const int *)RTA_DATA(rta));
172+
} else if (rta->rta_type == RTA_DST) {
173+
*dst_in_addr = RTA_DATA(rta);
174+
}
175+
}
176+
177+
if ((*if_index_p == -1) || (*dst_in_addr == NULL)) {
178+
ucs_diag("invalid routing table entry");
179+
return UCS_ERR_INVALID_PARAM;
180+
}
181+
182+
return UCS_OK;
183+
}
184+
185+
static ucs_status_t
186+
ucs_netlink_parse_rt_entry_cb(const struct nlmsghdr *nlh, const void *nl_msg,
187+
void *arg)
188+
{
189+
ucs_netlink_route_info_t *info = (ucs_netlink_route_info_t *)arg;
190+
int rule_iface;
191+
const void *dst_in_addr;
192+
193+
if (ucs_netlink_get_route_info(RTM_RTA((const struct rtmsg *)nl_msg),
194+
RTM_PAYLOAD(nlh), &rule_iface,
195+
&dst_in_addr) != UCS_OK) {
196+
return UCS_INPROGRESS;
197+
}
198+
199+
if (rule_iface != info->if_index) {
200+
return UCS_INPROGRESS;
201+
}
202+
203+
if (ucs_bitwise_is_equal(ucs_sockaddr_get_inet_addr(info->sa_remote),
204+
dst_in_addr,
205+
((const struct rtmsg *)nl_msg)->rtm_dst_len)) {
206+
info->found = 1;
207+
return UCS_OK;
208+
}
209+
210+
return UCS_INPROGRESS;
211+
}
212+
213+
int ucs_netlink_route_exists(const char *if_name,
214+
const struct sockaddr *sa_remote)
215+
{
216+
ucs_netlink_route_info_t info = {0};
217+
struct rtmsg rtm = {0};
218+
int iface_index;
219+
220+
iface_index = if_nametoindex(if_name);
221+
if (iface_index == 0) {
222+
ucs_error("failed to get interface index (errno %d)", errno);
223+
goto out;
224+
}
225+
226+
rtm.rtm_family = sa_remote->sa_family;
227+
rtm.rtm_table = RT_TABLE_MAIN;
228+
229+
info.if_index = iface_index;
230+
info.sa_remote = sa_remote;
231+
232+
ucs_netlink_send_request(NETLINK_ROUTE, RTM_GETROUTE, &rtm, sizeof(rtm),
233+
ucs_netlink_parse_rt_entry_cb, &info);
234+
235+
out:
236+
return info.found;
237+
}

src/ucs/sys/netlink.h

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
/**
2+
* Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2025. ALL RIGHTS RESERVED.
3+
*
4+
* See file LICENSE for terms.
5+
*/
6+
7+
#ifndef UCS_NETLINK_H
8+
#define UCS_NETLINK_H
9+
10+
#include <ucs/type/status.h>
11+
12+
#include <netinet/in.h>
13+
14+
BEGIN_C_DECLS
15+
16+
17+
/**
18+
* Check whether a routing table rule exists for a given network
19+
* interface name and a destination address.
20+
*
21+
* @param [in] if_name Pointer to the name of the interface.
22+
* @param [in] sa_remote Pointer to the destination address.
23+
*
24+
* @return 1 if rule exists, or 0 otherwise.
25+
*/
26+
int ucs_netlink_route_exists(const char *if_name,
27+
const struct sockaddr *sa_remote);
28+
29+
END_C_DECLS
30+
31+
#endif /* UCS_NETLINK_H */

src/ucs/sys/sock.c

Lines changed: 11 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ ucs_status_t ucs_netif_ioctl(const char *if_name, unsigned long request,
8080

8181
ucs_strncpy_zero(if_req->ifr_name, if_name, sizeof(if_req->ifr_name));
8282

83-
status = ucs_socket_create(AF_INET, SOCK_STREAM, &fd);
83+
status = ucs_socket_create(AF_INET, SOCK_STREAM, 0, &fd);
8484
if (status != UCS_OK) {
8585
goto out;
8686
}
@@ -209,9 +209,9 @@ unsigned ucs_netif_bond_ad_num_ports(const char *bond_name)
209209
return ret;
210210
}
211211

212-
ucs_status_t ucs_socket_create(int domain, int type, int *fd_p)
212+
ucs_status_t ucs_socket_create(int domain, int type, int protocol, int *fd_p)
213213
{
214-
int fd = socket(domain, type, 0);
214+
int fd = socket(domain, type, protocol);
215215
if (fd < 0) {
216216
ucs_socket_print_error_info("socket create failed", errno);
217217
return UCS_ERR_IO_ERROR;
@@ -465,7 +465,7 @@ ucs_status_t ucs_socket_server_init(const struct sockaddr *saddr, socklen_t sock
465465

466466
/* Create the server socket for accepting incoming connections */
467467
fd = -1; /* Suppress compiler warning */
468-
status = ucs_socket_create(saddr->sa_family, SOCK_STREAM, &fd);
468+
status = ucs_socket_create(saddr->sa_family, SOCK_STREAM, 0, &fd);
469469
if (status != UCS_OK) {
470470
goto err;
471471
}
@@ -596,9 +596,9 @@ ucs_socket_handle_io(int fd, const void *data, size_t count,
596596

597597
static inline ucs_status_t
598598
ucs_socket_do_io_nb(int fd, void *data, size_t *length_p,
599-
ucs_socket_io_func_t io_func, const char *name)
599+
ucs_socket_io_func_t io_func, const char *name, int flags)
600600
{
601-
ssize_t ret = io_func(fd, data, *length_p, MSG_NOSIGNAL);
601+
ssize_t ret = io_func(fd, data, *length_p, MSG_NOSIGNAL | flags);
602602
return ucs_socket_handle_io(fd, data, *length_p, length_p, 0,
603603
ret, errno, name);
604604
}
@@ -611,7 +611,7 @@ ucs_socket_do_io_b(int fd, void *data, size_t length,
611611
ucs_status_t status;
612612

613613
do {
614-
status = ucs_socket_do_io_nb(fd, data, &cur_cnt, io_func, name);
614+
status = ucs_socket_do_io_nb(fd, data, &cur_cnt, io_func, name, 0);
615615
done_cnt += cur_cnt;
616616
ucs_assert(done_cnt <= length);
617617
cur_cnt = length - done_cnt;
@@ -638,7 +638,7 @@ ucs_socket_do_iov_nb(int fd, struct iovec *iov, size_t iov_cnt, size_t *length_p
638638
ucs_status_t ucs_socket_send_nb(int fd, const void *data, size_t *length_p)
639639
{
640640
return ucs_socket_do_io_nb(fd, (void*)data, length_p,
641-
(ucs_socket_io_func_t)send, "send");
641+
(ucs_socket_io_func_t)send, "send", 0);
642642
}
643643

644644
/* recv is declared as 'always_inline' on some platforms, it leads to
@@ -648,9 +648,10 @@ static ssize_t ucs_socket_recv_io(int fd, void *data, size_t size, int flags)
648648
return recv(fd, data, size, flags);
649649
}
650650

651-
ucs_status_t ucs_socket_recv_nb(int fd, void *data, size_t *length_p)
651+
ucs_status_t ucs_socket_recv_nb(int fd, void *data, int flags, size_t *length_p)
652652
{
653-
return ucs_socket_do_io_nb(fd, data, length_p, ucs_socket_recv_io, "recv");
653+
return ucs_socket_do_io_nb(fd, data, length_p, ucs_socket_recv_io,
654+
"recv", flags);
654655
}
655656

656657
ucs_status_t ucs_socket_send(int fd, const void *data, size_t length)

0 commit comments

Comments
 (0)