Line data Source code
1 : /*
2 : * vr_dpdk_usocket.c -- library to deal with packet0 and netlink tcp/unix
3 : * sockets
4 : *
5 : * Copyright(c) 2014, Juniper Networks Inc.
6 : * All rights reserved
7 : */
8 :
9 : #include <fcntl.h>
10 : #include <poll.h>
11 : #include <linux/netlink.h>
12 : #include <sys/eventfd.h>
13 : #include <sys/stat.h>
14 : #include <sys/un.h>
15 : #include <stdint.h>
16 : #include <netinet/tcp.h>
17 :
18 : #include "nl_util.h"
19 : #include "vr_dpdk.h"
20 : #include "vr_dpdk_usocket.h"
21 : #include "vr_message.h"
22 :
23 : #include <rte_byteorder.h>
24 : #include <rte_errno.h>
25 : #include <rte_hexdump.h>
26 : #include <rte_timer.h>
27 :
28 : #define INFINITE_TIMEOUT -1
29 :
30 : extern void dpdk_burst_rx(unsigned int, struct rte_mbuf *[],
31 : struct vr_interface *, const char *, unsigned int);
32 : extern struct nlmsghdr *dpdk_nl_message_hdr(struct vr_message *);
33 : extern unsigned int dpdk_nl_message_len(struct vr_message *);
34 :
35 : static int vr_usocket_accept(struct vr_usocket *);
36 : static int vr_usocket_connect(struct vr_usocket *);
37 : static int vr_usocket_bind(struct vr_usocket *);
38 : static int usock_write(struct vr_usocket *);
39 : static int usock_read_init(struct vr_usocket *);
40 :
41 : static char vr_packet_unix_file[VR_UNIX_PATH_MAX];
42 : char *vr_socket_dir = VR_DEF_SOCKET_DIR;
43 : uint16_t vr_netlink_port = VR_DEF_NETLINK_PORT;
44 :
45 : /*
46 : * mark the error in socket for somebody to process/see
47 : */
48 : static void
49 0 : usock_set_error(struct vr_usocket *usockp, int error)
50 : {
51 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d error %d\n", __func__, pthread_self(),
52 : usockp->usock_fd, error);
53 0 : usockp->usock_error = error;
54 0 : usockp->usock_errno = errno;
55 :
56 0 : return;
57 : }
58 :
59 : /*
60 : * free the poll descriptor array, if it was allocated
61 : */
62 : static void
63 3057 : usock_deinit_poll(struct vr_usocket *usockp)
64 : {
65 3057 : if (!usockp)
66 0 : return;
67 :
68 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(), usockp->usock_fd);
69 3057 : if (usockp->usock_pfds) {
70 53 : vr_free(usockp->usock_pfds, VR_USOCK_POLL_OBJECT);
71 53 : usockp->usock_pfds = NULL;
72 : }
73 :
74 3057 : return;
75 : }
76 :
77 : /*
78 : * for both netlink and packet protocol sockets, we will need to poll. in case
79 : * of netlink, the poll in on tcp sockets to accept a connection (from agent,
80 : * utilities etc.:). for packet socket, the poll is on unix socket to receive
81 : * packets from agent and to be passed to vrouter. packet sockets also will
82 : * have an event usocket to dequeue packets from the packet_mbuf_ring, where
83 : * packets to be trapped will be enqueued.
84 : *
85 : * alloc the poll array
86 : */
87 : static int
88 3057 : usock_init_poll(struct vr_usocket *usockp)
89 : {
90 : unsigned int i;
91 : unsigned int proto;
92 :
93 3057 : if (!usockp)
94 0 : return -EINVAL;
95 :
96 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(), usockp->usock_fd);
97 3057 : proto = usockp->usock_proto;
98 3057 : if ((proto != NETLINK) && (proto != PACKET)) {
99 0 : usock_set_error(usockp, -EINVAL);
100 0 : goto error_return;
101 : }
102 :
103 3057 : if (!usockp->usock_max_cfds) {
104 0 : usock_set_error(usockp, -EINVAL);
105 0 : goto error_return;
106 : }
107 :
108 3057 : if (!usockp->usock_pfds) {
109 106 : usockp->usock_pfds = vr_zalloc(sizeof(struct pollfd) *
110 53 : usockp->usock_max_cfds + 1, VR_USOCK_POLL_OBJECT);
111 53 : if (!usockp->usock_pfds) {
112 0 : usock_set_error(usockp, -ENOMEM);
113 0 : goto error_return;
114 : }
115 :
116 3445 : for (i = 1; i <= usockp->usock_max_cfds; i++) {
117 3392 : usockp->usock_pfds[i].fd = -1;
118 : }
119 : }
120 :
121 3057 : return 0;
122 :
123 0 : error_return:
124 0 : return usockp->usock_error;
125 : }
126 :
127 : /*
128 : * bind a child socket to the parent. binding in this context means adding
129 : * a child usocket to parent poll list. an example where this will be required
130 : * is when one has created an event usocket. An event usocket by itself cannot
131 : * do anything useful in the context of dpdk vrouter application. Hence it needs
132 : * to be bound to the parent socket that does something useful, in this case
133 : * the packet socket. Another example is that of netlink socket. when thexi
134 : * netlink socket accepts new connection and the new connected socket has to be
135 : * polled, in which case we will need to bind it to the parent socket poll list
136 : */
137 : static int
138 3004 : usock_bind_usockets(struct vr_usocket *parent, struct vr_usocket *child)
139 : {
140 : unsigned int i;
141 : int ret;
142 : struct vr_usocket *child_pair;
143 :
144 3004 : if (parent->usock_state == LIMITED)
145 0 : return -ENOSPC;
146 :
147 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: parent FD %d child FD %d\n", __func__,
148 : pthread_self(), parent->usock_fd, child->usock_fd);
149 :
150 3004 : if (child->usock_proto == EVENT) {
151 53 : child_pair = vr_usocket(EVENT, RAW);
152 53 : if (!child_pair)
153 0 : return -ENOMEM;
154 :
155 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: parent FD %d closing child FD %d\n",
156 : __func__, pthread_self(), parent->usock_fd, child->usock_fd);
157 53 : close(child->usock_fd);
158 53 : child->usock_fd = child_pair->usock_fd;
159 53 : child = child_pair;
160 : }
161 :
162 3004 : ret = usock_init_poll(parent);
163 3004 : if (ret)
164 0 : return ret;
165 :
166 3004 : if (!parent->usock_children) {
167 53 : parent->usock_children = vr_zalloc(sizeof(struct vr_usocket *) *
168 : USOCK_MAX_CHILD_FDS + 1, VR_USOCK_OBJECT);
169 53 : if (!parent->usock_children) {
170 0 : usock_set_error(parent, -ENOMEM);
171 0 : return -ENOMEM;
172 : }
173 : }
174 :
175 3004 : child->usock_parent = parent;
176 3004 : parent->usock_cfds++;
177 3004 : if (parent->usock_cfds >= USOCK_MAX_CHILD_FDS){
178 0 : parent->usock_state = LIMITED;
179 0 : RTE_LOG(ERR, USOCK, "%s:[%lx] Netlink socket reached max connection %d\n",
180 : __func__, pthread_self(), USOCK_MAX_CHILD_FDS);
181 0 : RTE_LOG(ERR, USOCK, "Netlink state: %x\n", parent->usock_state);
182 : }
183 :
184 5955 : for (i = 1; i <= parent->usock_max_cfds; i++) {
185 5955 : if (!parent->usock_children[i]) {
186 3004 : parent->usock_children[i] = child;
187 3004 : parent->usock_pfds[i].fd = child->usock_fd;
188 3004 : parent->usock_pfds[i].events = POLLIN;
189 3004 : child->usock_child_index = i;
190 3004 : break;
191 : }
192 : }
193 :
194 3004 : if (child->usock_proto == EVENT)
195 53 : child->usock_state = READING_DATA;
196 :
197 3004 : usock_read_init(child);
198 :
199 3004 : return 0;
200 : }
201 :
202 : int
203 53 : vr_usocket_bind_usockets(void *usock1, void *usock2)
204 : {
205 53 : struct vr_usocket *parent = (struct vr_usocket *)usock1;
206 53 : struct vr_usocket *child = (struct vr_usocket *)usock2;
207 :
208 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: usock1 FD %d usock2 FD %d\n", __func__,
209 : pthread_self(), parent->usock_fd, child->usock_fd);
210 53 : return usock_bind_usockets(parent, child);
211 : }
212 :
213 : static int
214 2951 : usock_clone(struct vr_usocket *parent, int cfd)
215 : {
216 : struct vr_usocket *child;
217 :
218 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: parent FD %d cfd %d\n", __func__,
219 : pthread_self(), parent->usock_fd, cfd);
220 2951 : child = vr_zalloc(sizeof(struct vr_usocket), VR_USOCK_OBJECT);
221 2951 : if (!child) {
222 0 : usock_set_error(parent, -ENOMEM);
223 0 : goto error_return;
224 : }
225 :
226 2951 : child->usock_rx_buf = vr_malloc(USOCK_RX_BUF_LEN, VR_USOCK_BUF_OBJECT);
227 2951 : if (!child->usock_rx_buf) {
228 0 : usock_set_error(parent, -ENOMEM);
229 0 : goto error_return;
230 : }
231 2951 : child->usock_buf_len = USOCK_RX_BUF_LEN;
232 :
233 2951 : child->usock_type = parent->usock_type;
234 2951 : child->usock_proto = parent->usock_proto;
235 2951 : child->usock_fd = cfd;
236 :
237 2951 : if (usock_bind_usockets(parent, child))
238 0 : goto error_return;
239 :
240 2951 : return 0;
241 :
242 0 : error_return:
243 0 : if (child) {
244 0 : if (child->usock_rx_buf)
245 0 : vr_free(child->usock_rx_buf, VR_USOCK_BUF_OBJECT);
246 0 : vr_free(child, VR_USOCK_OBJECT);
247 : }
248 :
249 0 : return parent->usock_error;
250 : }
251 :
252 :
253 : static void
254 3004 : usock_unbind(struct vr_usocket *child)
255 : {
256 : struct vr_usocket *parent;
257 :
258 3004 : if (!child)
259 0 : return;
260 :
261 3004 : parent = child->usock_parent;
262 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: child FD %d parent %p\n", __func__,
263 : pthread_self(), child->usock_fd, parent);
264 3004 : if (!parent)
265 53 : return;
266 :
267 2951 : parent->usock_children[child->usock_child_index] = NULL;
268 2951 : if (parent->usock_pfds)
269 2951 : parent->usock_pfds[child->usock_child_index].fd = -1;
270 :
271 2951 : parent->usock_disconnects++;
272 2951 : parent->usock_cfds--;
273 2951 : if((parent->usock_state == LIMITED) &&
274 0 : (parent->usock_cfds < USOCK_MAX_CHILD_FDS))
275 0 : parent->usock_state = LISTENING;
276 :
277 2951 : child->usock_parent = NULL;
278 :
279 2951 : return;
280 : }
281 :
282 : static void
283 6008 : usock_close(struct vr_usocket *usockp)
284 : {
285 : int i;
286 : struct vr_usocket *parent;
287 :
288 : RTE_SET_USED(parent);
289 :
290 6008 : if (!usockp)
291 3004 : return;
292 :
293 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(), usockp->usock_fd);
294 3004 : usock_unbind(usockp);
295 3004 : usock_deinit_poll(usockp);
296 :
297 3057 : for (i = 0; i < usockp->usock_cfds; i++) {
298 53 : usock_close(usockp->usock_children[i]);
299 : }
300 :
301 : RTE_LOG_DP(DEBUG, USOCK, "%s: closing FD %d\n", __func__, usockp->usock_fd);
302 3004 : close(usockp->usock_fd);
303 :
304 3004 : if (!usockp->usock_mbuf_pool && usockp->usock_rx_buf) {
305 2951 : vr_free(usockp->usock_rx_buf, VR_USOCK_BUF_OBJECT);
306 2951 : usockp->usock_rx_buf = NULL;
307 : }
308 :
309 3004 : if (usockp->usock_iovec) {
310 0 : vr_free(usockp->usock_iovec, VR_USOCK_IOVEC_OBJECT);
311 0 : usockp->usock_iovec = NULL;
312 : }
313 :
314 3004 : if (usockp->usock_mbuf_pool) {
315 : /* no api to destroy a pool */
316 : }
317 :
318 3004 : if (usockp->usock_proto == PACKET) {
319 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: unlinking %s\n", __func__,
320 : pthread_self(), vr_packet_unix_file);
321 0 : unlink(vr_packet_unix_file);
322 : }
323 :
324 3004 : usockp->usock_io_in_progress = 0;
325 :
326 3004 : vr_free(usockp, VR_USOCK_OBJECT);
327 :
328 3004 : return;
329 : }
330 :
331 : static int
332 2863 : __usock_write(struct vr_usocket *usockp)
333 : {
334 : int ret;
335 : unsigned int len;
336 : unsigned char *buf;
337 2863 : struct vr_usocket *parent = NULL;
338 :
339 2863 : if (usockp->usock_proto != EVENT) {
340 2863 : parent = usockp->usock_parent;
341 2863 : if (!parent)
342 0 : return -1;
343 : }
344 :
345 2863 : buf = usockp->usock_tx_buf;
346 2863 : if (!buf || !usockp->usock_write_len)
347 0 : return 0;
348 :
349 2863 : len = usockp->usock_write_len;
350 :
351 2863 : buf += usockp->usock_write_offset;
352 2863 : len -= usockp->usock_write_offset;
353 :
354 2863 : retry_write:
355 : #ifdef VR_DPDK_USOCK_DUMP
356 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d writing %d bytes\n",
357 : __func__, pthread_self(), usockp->usock_fd, len);
358 : rte_hexdump(stdout, "usock buffer dump:", buf, len);
359 : #endif
360 2863 : if (usockp->usock_owner != pthread_self()) {
361 0 : if (usockp->usock_owner)
362 0 : RTE_LOG(WARNING, USOCK, "WARNING: thread %lx (lcore %u) is trying to write %u bytes"
363 : " to usocket FD %d owned by thread %lx\n",
364 : pthread_self(), rte_lcore_id(), len, usockp->usock_fd, usockp->usock_owner);
365 0 : usockp->usock_owner = pthread_self();
366 : }
367 2863 : ret = write(usockp->usock_fd, buf, len);
368 2863 : if (ret > 0) {
369 2863 : usockp->usock_write_offset += ret;
370 2863 : if (usockp->usock_write_offset == usockp->usock_write_len) {
371 : /* remove from output poll */
372 2863 : if (parent)
373 2863 : parent->usock_pfds[usockp->usock_child_index].events = POLLIN;
374 2863 : usockp->usock_tx_buf = NULL;
375 : } else {
376 0 : if (parent)
377 0 : parent->usock_pfds[usockp->usock_child_index].events = POLLOUT;
378 : }
379 0 : } else if (ret < 0) {
380 : RTE_LOG_DP(DEBUG, VROUTER, "%s[%lx]: write error FD %d\n", __func__, pthread_self(),
381 : usockp->usock_fd);
382 0 : usock_set_error(usockp, ret);
383 :
384 0 : if (errno == EINTR)
385 0 : goto retry_write;
386 :
387 0 : if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
388 0 : if (parent) {
389 0 : parent->usock_pfds[usockp->usock_child_index].events = POLLOUT;
390 0 : return 0;
391 : }
392 : }
393 0 : usockp->usock_tx_buf = NULL;
394 : }
395 :
396 2863 : return ret;
397 : }
398 :
399 : static void
400 0 : usock_netlink_write_responses(struct vr_usocket *usockp)
401 : {
402 : int ret;
403 : struct vr_message *resp;
404 :
405 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(), usockp->usock_fd);
406 0 : while ((resp =
407 0 : (struct vr_message *)vr_queue_dequeue(&usockp->usock_nl_responses))) {
408 0 : usockp->usock_tx_buf = (unsigned char *)dpdk_nl_message_hdr(resp);
409 0 : usockp->usock_write_len = dpdk_nl_message_len(resp);
410 0 : usockp->usock_write_offset = 0;
411 0 : ret = __usock_write(usockp);
412 0 : if ((ret < 0) || (ret == usockp->usock_write_len)) {
413 0 : vr_message_free(resp);
414 : } else {
415 : break;
416 : }
417 : }
418 :
419 0 : return;
420 : }
421 :
422 :
423 : static int
424 0 : usock_mbuf_write(struct vr_usocket *usockp, struct rte_mbuf *mbuf)
425 : {
426 : unsigned int i, pkt_len;
427 : struct msghdr mhdr;
428 : struct rte_mbuf *m;
429 : struct iovec *iov;
430 :
431 0 : if (!mbuf)
432 0 : return 0;
433 :
434 0 : pkt_len = rte_pktmbuf_pkt_len(mbuf);
435 0 : if (!pkt_len)
436 0 : return 0;
437 :
438 0 : iov = usockp->usock_iovec;
439 :
440 0 : m = mbuf;
441 0 : for (i = 0; (m && (i < PKT0_MAX_IOV_LEN)); i++) {
442 0 : iov->iov_base = rte_pktmbuf_mtod(m, unsigned char *);
443 0 : iov->iov_len = rte_pktmbuf_data_len(m);
444 0 : m = m->next;
445 0 : iov++;
446 : }
447 :
448 0 : if ((i == PKT0_MAX_IOV_LEN) && m)
449 0 : usockp->usock_pkt_truncated++;
450 :
451 0 : mhdr.msg_name = NULL;
452 0 : mhdr.msg_namelen = 0;
453 0 : mhdr.msg_iov = usockp->usock_iovec;
454 0 : mhdr.msg_iovlen = i;
455 0 : mhdr.msg_control = NULL;
456 0 : mhdr.msg_controllen = 0;
457 0 : mhdr.msg_flags = 0;
458 :
459 : #ifdef VR_DPDK_USOCK_DUMP
460 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d sending message\n", __func__,
461 : pthread_self(), usockp->usock_fd);
462 : rte_hexdump(stdout, "usock message dump:", &mhdr, sizeof(mhdr));
463 : #endif
464 0 : return sendmsg(usockp->usock_fd, &mhdr, MSG_DONTWAIT);
465 : }
466 :
467 : static void
468 0 : vr_dpdk_packet_receive(struct vr_usocket *usockp)
469 : {
470 0 : const unsigned lcore_id = rte_lcore_id();
471 0 : struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
472 : struct vr_interface_stats *stats;
473 :
474 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
475 : usockp->usock_fd);
476 : /**
477 : * Packets is read from the agent's socket here. On success, a counter for
478 : * packets dequeued from the interface is incremented.
479 : */
480 0 : stats = vif_get_stats(usockp->usock_vif, lcore_id);
481 0 : if (usockp->usock_vif) {
482 0 : stats->vis_port_ipackets++;
483 : /* buf_addr and data_off do not change */
484 0 : usockp->usock_mbuf->data_len = usockp->usock_read_len;
485 0 : usockp->usock_mbuf->pkt_len = usockp->usock_read_len;
486 : /* convert mbuf to vr_packet */
487 0 : vr_dpdk_packet_get(usockp->usock_mbuf, usockp->usock_vif);
488 : /* send the mbuf to vRouter */
489 0 : vr_dpdk_lcore_vroute(lcore, usockp->usock_vif, &usockp->usock_mbuf, 1);
490 : /* flush packet TX queues immediately */
491 0 : vr_dpdk_lcore_flush(lcore);
492 : } else {
493 : /**
494 : * If reading from socket failed, increment counter for interface
495 : * dequeue drops.
496 : */
497 0 : RTE_LOG(ERR, VROUTER, "Error receiving from packet socket: no vif attached\n");
498 0 : vr_dpdk_pfree(usockp->usock_mbuf, NULL, VP_DROP_INTERFACE_DROP);
499 0 : stats->vis_port_ierrors++;
500 : }
501 :
502 0 : usockp->usock_mbuf = NULL;
503 0 : usockp->usock_rx_buf = NULL;
504 0 : usockp->usock_buf_len = 0;
505 :
506 0 : return;
507 : }
508 :
509 : static void
510 53 : vr_dpdk_packet_ring_drain(struct vr_usocket *usockp)
511 : {
512 : int i;
513 : unsigned nb_pkts;
514 : struct rte_mbuf *mbuf_arr[VR_DPDK_RX_BURST_SZ];
515 53 : const unsigned lcore_id = rte_lcore_id();
516 : struct vr_interface_stats *stats;
517 :
518 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: draining packet ring...\n", __func__,
519 : pthread_self());
520 :
521 53 : if (unlikely(usockp->usock_parent->usock_vif == NULL))
522 53 : return;
523 :
524 0 : rcu_thread_offline();
525 :
526 0 : stats = vif_get_stats(usockp->usock_parent->usock_vif, lcore_id);
527 : do {
528 0 : nb_pkts = rte_ring_sc_dequeue_burst(vr_dpdk.packet_ring,
529 : (void **)&mbuf_arr, VR_DPDK_RX_BURST_SZ, NULL);
530 0 : for (i = 0; i < nb_pkts; i++) {
531 0 : if (usock_mbuf_write(usockp->usock_parent, mbuf_arr[i]) >= 0)
532 0 : stats->vis_port_opackets++;
533 : else {
534 0 : stats->vis_port_oerrors++;
535 : RTE_LOG_DP(DEBUG, USOCK,
536 : "%s: Error writing mbuf to packet socket: %s (%d)\n",
537 : __func__, rte_strerror(errno), errno);
538 : }
539 :
540 0 : rte_pktmbuf_free(mbuf_arr[i]);
541 : }
542 0 : } while (nb_pkts > 0);
543 :
544 0 : rcu_thread_online();
545 : }
546 :
547 : static int
548 2915 : usock_read_done(struct vr_usocket *usockp)
549 : {
550 2915 : if (usockp->usock_state == READING_FAULTY_DATA)
551 0 : return 0;
552 :
553 2915 : switch (usockp->usock_proto) {
554 0 : case PACKET:
555 0 : vr_dpdk_packet_receive(usockp);
556 0 : break;
557 :
558 53 : case EVENT:
559 53 : vr_dpdk_packet_ring_drain(usockp);
560 53 : break;
561 :
562 2862 : case NETLINK:
563 2862 : dpdk_netlink_receive(usockp, usockp->usock_rx_buf,
564 : usockp->usock_read_len);
565 2862 : break;
566 :
567 0 : default:
568 0 : break;
569 : }
570 :
571 2915 : return 0;
572 : }
573 :
574 : static int
575 6025 : usock_read_init(struct vr_usocket *usockp)
576 : {
577 6025 : usockp->usock_read_offset = 0;
578 :
579 6025 : switch (usockp->usock_proto) {
580 5813 : case NETLINK:
581 5813 : if (usockp->usock_parent) {
582 5813 : usockp->usock_read_len = NLMSG_HDRLEN;
583 5813 : usockp->usock_state = READING_HEADER;
584 : }
585 5813 : break;
586 :
587 212 : case EVENT:
588 212 : usockp->usock_read_len = USOCK_EVENT_BUF_LEN;
589 212 : usockp->usock_state = READING_DATA;
590 212 : break;
591 :
592 0 : case PACKET:
593 0 : if (usockp->usock_mbuf) {
594 0 : RTE_LOG(ERR, VROUTER, "Error initing usock read: mbuf is already exist\n");
595 0 : return -EINVAL;
596 : }
597 :
598 0 : usockp->usock_mbuf = rte_pktmbuf_alloc(usockp->usock_mbuf_pool);
599 0 : if (!usockp->usock_mbuf) {
600 0 : RTE_LOG(ERR, VROUTER, "Error initing usock read: cannot allocate mbuf\n");
601 0 : return -ENOMEM;
602 : }
603 :
604 0 : usockp->usock_rx_buf = rte_pktmbuf_mtod(usockp->usock_mbuf, char *);
605 0 : usockp->usock_buf_len = usockp->usock_mbuf->buf_len
606 0 : - rte_pktmbuf_headroom(usockp->usock_mbuf);
607 0 : usockp->usock_read_len = usockp->usock_buf_len;
608 0 : usockp->usock_state = READING_DATA;
609 0 : break;
610 :
611 0 : default:
612 0 : break;
613 : }
614 :
615 6025 : return 0;
616 : }
617 :
618 : static int
619 8728 : __usock_read(struct vr_usocket *usockp)
620 : {
621 : int ret;
622 8728 : unsigned int offset = usockp->usock_read_offset;
623 8728 : unsigned int len = usockp->usock_read_len;
624 8728 : unsigned int toread = len - offset;
625 :
626 : struct nlmsghdr *nlh;
627 8728 : unsigned int proto = usockp->usock_proto;
628 8728 : char *buf = usockp->usock_rx_buf;
629 :
630 8728 : if (toread > usockp->usock_buf_len) {
631 0 : toread = usockp->usock_buf_len - offset;
632 : }
633 :
634 8728 : retry_read:
635 8728 : if (usockp->usock_owner != pthread_self()) {
636 3004 : if (usockp->usock_owner)
637 0 : RTE_LOG(WARNING, USOCK, "WARNING: thread %lx is trying to read"
638 : " usocket FD %d owned by thread %lx\n",
639 : pthread_self(), usockp->usock_fd, usockp->usock_owner);
640 3004 : usockp->usock_owner = pthread_self();
641 : }
642 8728 : ret = read(usockp->usock_fd, buf + offset, toread);
643 : #ifdef VR_DPDK_USOCK_DUMP
644 : if (ret > 0) {
645 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d read %d bytes\n", __func__,
646 : pthread_self(), usockp->usock_fd, ret);
647 : rte_hexdump(stdout, "usock buffer dump:", buf + offset, ret);
648 : } else if (ret < 0) {
649 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d read returned error %d: %s (%d)\n", __func__,
650 : pthread_self(), usockp->usock_fd, ret, rte_strerror(errno), errno);
651 : }
652 : #endif
653 8728 : if (ret <= 0) {
654 2951 : if (!ret)
655 2951 : return -1;
656 :
657 0 : if (errno == EINTR)
658 0 : goto retry_read;
659 :
660 0 : if ((errno == EAGAIN) ||
661 0 : (errno == EWOULDBLOCK))
662 0 : return 0;
663 :
664 0 : RTE_LOG(ERR, USOCK, "Error reading FD %d: %s (%d)\n",
665 : usockp->usock_fd, rte_strerror(errno), errno);
666 0 : return ret;
667 : }
668 :
669 5777 : offset += ret;
670 5777 : usockp->usock_read_offset = offset;
671 :
672 5777 : if (proto == NETLINK) {
673 5724 : if (usockp->usock_state == READING_HEADER) {
674 2862 : if (usockp->usock_read_offset == usockp->usock_read_len) {
675 2862 : usockp->usock_state = READING_DATA;
676 2862 : nlh = (struct nlmsghdr *)(usockp->usock_rx_buf);
677 2862 : usockp->usock_read_len = nlh->nlmsg_len;
678 : }
679 : }
680 :
681 5724 : if (usockp->usock_buf_len < usockp->usock_read_len) {
682 0 : usockp->usock_rx_buf = vr_malloc(usockp->usock_read_len,
683 : VR_USOCK_BUF_OBJECT);
684 0 : if (!usockp->usock_rx_buf) {
685 : /* bad, but let's recover */
686 0 : usockp->usock_rx_buf = buf;
687 0 : usockp->usock_read_len -= usockp->usock_read_offset;
688 0 : usockp->usock_read_offset = 0;
689 0 : usockp->usock_state = READING_FAULTY_DATA;
690 : } else {
691 0 : memcpy(usockp->usock_rx_buf, buf, usockp->usock_read_offset);
692 0 : vr_free(buf, VR_USOCK_BUF_OBJECT);
693 0 : usockp->usock_buf_len = usockp->usock_read_len;
694 0 : buf = usockp->usock_rx_buf;
695 : }
696 : }
697 53 : } else if (proto == PACKET) {
698 0 : usockp->usock_read_len = ret;
699 : }
700 :
701 5777 : return ret;
702 : }
703 :
704 :
705 : static struct vr_usocket *
706 159 : usock_alloc(unsigned short proto, unsigned short type)
707 : {
708 159 : unsigned int pkt0_mempool_sz = PKT0_MBUF_POOL_SIZE;
709 159 : int sock_fd = -1, domain, ret;
710 : /* socket TX buffer size = (hold flow table entries * size of jumbo frame) */
711 159 : int setsocksndbuff = vr_flow_hold_limit * vr_packet_sz;
712 : int getsocksndbuff;
713 159 : socklen_t getsocksndbufflen = sizeof(getsocksndbuff);
714 159 : int error = 0, flags;
715 : unsigned int buf_len;
716 159 : struct vr_usocket *usockp = NULL, *child;
717 159 : bool is_socket = true;
718 : unsigned short sock_type;
719 159 : int flag = 1;
720 :
721 : RTE_SET_USED(child);
722 :
723 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: proto %u type %u\n", __func__,
724 : pthread_self(), proto, type);
725 159 : switch (type) {
726 0 : case TCP:
727 0 : domain = AF_INET;
728 0 : sock_type = SOCK_STREAM;
729 0 : break;
730 :
731 53 : case UNIX:
732 53 : domain = AF_UNIX;
733 53 : sock_type = SOCK_STREAM;
734 53 : break;
735 :
736 106 : case RAW:
737 106 : domain = AF_UNIX;
738 106 : sock_type = SOCK_DGRAM;
739 106 : break;
740 :
741 0 : default:
742 0 : return NULL;
743 : }
744 :
745 159 : if (proto == EVENT) {
746 106 : is_socket = false;
747 106 : sock_fd = eventfd(0, 0);
748 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: new event FD %d\n", __func__,
749 : pthread_self(), sock_fd);
750 106 : if (sock_fd < 0)
751 0 : return NULL;
752 : }
753 :
754 159 : if (is_socket) {
755 53 : sock_fd = socket(domain, sock_type, 0);
756 53 : RTE_LOG(INFO, USOCK, "%s[%lx]: new socket FD %d\n", __func__,
757 : pthread_self(), sock_fd);
758 53 : if (sock_fd < 0)
759 0 : return NULL;
760 :
761 53 : if (type == TCP) {
762 0 : RTE_LOG(INFO, USOCK, "%s[%lx]: setting socket FD %d nodelay.\n"
763 : , __func__, pthread_self(), sock_fd);
764 0 : ret = setsockopt(sock_fd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag,
765 : sizeof(int));
766 0 : if (ret != 0) {
767 0 : RTE_LOG(ERR, USOCK, "%s[%lx]: setting socket FD %d nodelay failed (%d).\n"
768 : , __func__, pthread_self(), sock_fd, errno);
769 : }
770 : }
771 :
772 : /* set socket send buffer size */
773 53 : ret = setsockopt(sock_fd, SOL_SOCKET, SO_SNDBUF, &setsocksndbuff,
774 : sizeof(setsocksndbuff));
775 53 : if (ret == 0) {
776 : /* check if setting buffer succeeded */
777 53 : ret = getsockopt(sock_fd, SOL_SOCKET, SO_SNDBUF, &getsocksndbuff,
778 : &getsocksndbufflen);
779 53 : if (ret == 0) {
780 53 : if (getsocksndbuff >= setsocksndbuff) {
781 0 : RTE_LOG(INFO, USOCK, "%s[%lx]: setting socket FD %d send buff size.\n"
782 : "Buffer size set to %d (requested %d)\n", __func__,
783 : pthread_self(), sock_fd, getsocksndbuff, setsocksndbuff);
784 : } else { /* set other than requested */
785 53 : RTE_LOG(ERR, USOCK, "%s[%lx]: setting socket FD %d send buff size failed.\n"
786 : "Buffer size set to %d (requested %d)\n", __func__,
787 : pthread_self(), sock_fd, getsocksndbuff, setsocksndbuff);
788 : }
789 : } else { /* requesting buffer size failed */
790 0 : RTE_LOG(ERR, USOCK, "%s[%lx]: getting socket FD %d send buff size failed (%d)\n",
791 : __func__, pthread_self(), sock_fd, errno);
792 : }
793 : } else { /* setting buffer size failed */
794 0 : RTE_LOG(ERR, USOCK, "%s[%lx]: setting socket FD %d send buff size %d failed (%d)\n",
795 : __func__, pthread_self(), sock_fd, setsocksndbuff, errno);
796 : }
797 : }
798 :
799 159 : usockp = vr_zalloc(sizeof(*usockp), VR_USOCK_OBJECT);
800 159 : if (!usockp)
801 0 : goto error_exit;
802 :
803 159 : usockp->usock_type = type;
804 159 : usockp->usock_proto = proto;
805 159 : usockp->usock_fd = sock_fd;
806 159 : usockp->usock_state = INITED;
807 :
808 159 : if (is_socket) {
809 53 : error = vr_usocket_bind(usockp);
810 53 : if (error < 0)
811 0 : goto error_exit;
812 :
813 53 : if (usockp->usock_proto == PACKET) {
814 0 : error = vr_usocket_connect(usockp);
815 0 : if (error < 0)
816 0 : goto error_exit;
817 : }
818 : }
819 :
820 159 : switch (proto) {
821 53 : case NETLINK:
822 53 : usockp->usock_max_cfds = USOCK_MAX_CHILD_FDS;
823 53 : buf_len = 0;
824 53 : break;
825 :
826 0 : case PACKET:
827 0 : usockp->usock_max_cfds = USOCK_MAX_CHILD_FDS;
828 0 : buf_len = 0;
829 0 : break;
830 :
831 106 : case EVENT:
832 : /* TODO: we don't need the buf since we use stack to send an event */
833 106 : buf_len = USOCK_EVENT_BUF_LEN;
834 106 : break;
835 :
836 0 : default:
837 0 : buf_len = 0;
838 0 : break;
839 : }
840 :
841 159 : if (buf_len) {
842 106 : usockp->usock_rx_buf = vr_zalloc(buf_len, VR_USOCK_BUF_OBJECT);
843 106 : if (!usockp->usock_rx_buf)
844 0 : goto error_exit;
845 :
846 106 : usockp->usock_buf_len = buf_len;
847 106 : usock_read_init(usockp);
848 : }
849 :
850 159 : if (proto == PACKET) {
851 0 : usockp->usock_mbuf_pool = rte_mempool_lookup("packet_mbuf_pool");
852 0 : if (!usockp->usock_mbuf_pool) {
853 0 : usockp->usock_mbuf_pool = rte_mempool_create("packet_mbuf_pool",
854 : pkt0_mempool_sz, PKT0_MBUF_PACKET_SIZE,
855 : PKT0_MBUF_POOL_CACHE_SZ, sizeof(struct rte_pktmbuf_pool_private),
856 : vr_dpdk_pktmbuf_pool_init, NULL, vr_dpdk_pktmbuf_init, NULL,
857 0 : rte_socket_id(), 0);
858 0 : if (!usockp->usock_mbuf_pool)
859 0 : goto error_exit;
860 : }
861 :
862 0 : usockp->usock_iovec = vr_zalloc(sizeof(struct iovec) *
863 : PKT0_MAX_IOV_LEN, VR_USOCK_IOVEC_OBJECT);
864 0 : if (!usockp->usock_iovec)
865 0 : goto error_exit;
866 :
867 0 : usock_read_init(usockp);
868 : }
869 :
870 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d F_GETFL\n", __func__, pthread_self(),
871 : usockp->usock_fd);
872 159 : flags = fcntl(usockp->usock_fd, F_GETFL);
873 159 : if (flags == -1)
874 0 : goto error_exit;
875 :
876 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d F_SETFL\n", __func__, pthread_self(),
877 : usockp->usock_fd);
878 159 : error = fcntl(usockp->usock_fd, F_SETFL, flags | O_NONBLOCK);
879 159 : if (error == -1)
880 0 : goto error_exit;
881 :
882 159 : usockp->usock_poll_block = 1;
883 :
884 159 : return usockp;
885 :
886 0 : error_exit:
887 :
888 0 : error = errno;
889 0 : if (sock_fd >= 0) {
890 0 : close(sock_fd);
891 0 : sock_fd = -1;
892 : }
893 :
894 0 : usock_close(usockp);
895 0 : usockp = NULL;
896 0 : errno = error;
897 :
898 0 : return usockp;
899 : }
900 :
901 : /*
902 : * currently defined protocols are netlink and packet.
903 : * for packet, only raw socket type is accepted
904 : * for netlink, both tcp and unix socket types are accepted
905 : */
906 : static bool
907 159 : valid_usock(int proto, int type)
908 : {
909 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: proto %u type %u\n", __func__,
910 : pthread_self(), proto, type);
911 159 : if ((proto != PACKET) &&
912 106 : (proto != NETLINK) &&
913 : (proto != EVENT))
914 0 : return -EINVAL;
915 :
916 159 : if (((proto == PACKET) || (proto == EVENT)) && (type != RAW)) {
917 0 : return -EINVAL;
918 : } else {
919 159 : if (type != TCP && type != UNIX)
920 106 : return -EINVAL;
921 : }
922 :
923 53 : return true;
924 : }
925 :
926 : void
927 0 : vr_usocket_detach_vif(void *usockp)
928 : {
929 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
930 : ((struct vr_usocket *)usockp)->usock_fd);
931 0 : ((struct vr_usocket *)usockp)->usock_vif = NULL;
932 0 : return;
933 : }
934 :
935 : void
936 0 : vr_usocket_attach_vif(void *usockp, struct vr_interface *vif)
937 : {
938 0 : if (!vif)
939 0 : return;
940 :
941 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
942 : ((struct vr_usocket *)usockp)->usock_fd);
943 0 : ((struct vr_usocket *)usockp)->usock_vif = vif;
944 0 : return;
945 : }
946 :
947 : void
948 0 : vr_usocket_non_blocking(struct vr_usocket *usockp)
949 : {
950 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
951 : usockp->usock_fd);
952 0 : usockp->usock_poll_block = 0;
953 0 : return;
954 : }
955 :
956 : int
957 2863 : vr_usocket_write(struct vr_usocket *usockp, unsigned char *buf,
958 : unsigned int len)
959 : {
960 2863 : if (usockp->usock_tx_buf)
961 0 : return -1;
962 :
963 2863 : usockp->usock_tx_buf = buf;
964 2863 : usockp->usock_write_offset = 0;
965 2863 : usockp->usock_write_len = len;
966 :
967 2863 : return __usock_write(usockp);
968 : }
969 :
970 : int
971 53 : vr_usocket_eventfd_write(struct vr_usocket *usockp)
972 : {
973 53 : if (usockp->usock_proto != EVENT)
974 0 : return -1;
975 :
976 53 : return eventfd_write(usockp->usock_fd, 1);
977 : }
978 :
979 : int
980 2863 : vr_usocket_message_write(struct vr_usocket *usockp,
981 : struct vr_message *message)
982 : {
983 : int ret;
984 : unsigned int len;
985 : unsigned char *buf;
986 :
987 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
988 : usockp->usock_fd);
989 2863 : if ((usockp->usock_proto != NETLINK) && (usockp->usock_type != TCP))
990 0 : return -EINVAL;
991 :
992 2863 : if (usockp->usock_tx_buf || !vr_queue_empty(&usockp->usock_nl_responses)) {
993 0 : vr_queue_enqueue(&usockp->usock_nl_responses,
994 : &message->vr_message_queue);
995 0 : return 0;
996 : }
997 :
998 2863 : buf = (unsigned char *)dpdk_nl_message_hdr(message);
999 2863 : len = dpdk_nl_message_len(message);
1000 2863 : ret = vr_usocket_write(usockp, buf, len);
1001 2863 : if (ret == len) {
1002 2863 : vr_message_free(message);
1003 : }
1004 :
1005 2863 : return ret;
1006 : }
1007 :
1008 : static int
1009 11679 : vr_usocket_read(struct vr_usocket *usockp)
1010 : {
1011 : int ret;
1012 :
1013 11679 : if (!usockp || usockp->usock_fd < 0)
1014 0 : return -1;
1015 :
1016 11679 : switch (usockp->usock_state) {
1017 2951 : case LISTENING:
1018 : case LIMITED:
1019 2951 : ret = vr_usocket_accept(usockp);
1020 2951 : if (ret < 0)
1021 0 : return ret;
1022 :
1023 2951 : break;
1024 :
1025 8728 : case READING_HEADER:
1026 : case READING_DATA:
1027 : case READING_FAULTY_DATA:
1028 8728 : ret = __usock_read(usockp);
1029 8728 : if (ret < 0) {
1030 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: read error FD %d\n", __func__, pthread_self(),
1031 : usockp->usock_fd);
1032 2951 : usock_close(usockp);
1033 2951 : return ret;
1034 : }
1035 :
1036 5777 : if (usockp->usock_read_offset == usockp->usock_read_len) {
1037 2915 : usock_read_done(usockp);
1038 : /* we have the complete message */
1039 2915 : usock_read_init(usockp);
1040 : }
1041 :
1042 5777 : break;
1043 :
1044 0 : default:
1045 0 : return -1;
1046 : }
1047 :
1048 8728 : return ret;
1049 : }
1050 :
1051 : static int
1052 0 : vr_usocket_connect(struct vr_usocket *usockp)
1053 : {
1054 : struct sockaddr_un sun;
1055 :
1056 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
1057 : usockp->usock_fd);
1058 0 : if (usockp->usock_proto != PACKET)
1059 0 : return -EINVAL;
1060 :
1061 0 : sun.sun_family = AF_UNIX;
1062 0 : memset(sun.sun_path, 0, sizeof(sun.sun_path));
1063 0 : strncpy(sun.sun_path, vr_socket_dir, sizeof(sun.sun_path) - 1);
1064 0 : strncat(sun.sun_path, "/"VR_PACKET_AGENT_UNIX_NAME, sizeof(sun.sun_path)
1065 0 : - strlen(sun.sun_path) - 1);
1066 :
1067 : #ifdef VR_DPDK_USOCK_DUMP
1068 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d retry connecting\n", __func__,
1069 : pthread_self(), usockp->usock_fd);
1070 : rte_hexdump(stdout, "usock address dump:", &sun, sizeof(sun));
1071 : #endif
1072 0 : return vr_dpdk_retry_connect(usockp->usock_fd, (struct sockaddr *)&sun,
1073 : sizeof(sun));
1074 : }
1075 :
1076 : static int
1077 2951 : vr_usocket_accept(struct vr_usocket *usockp)
1078 : {
1079 : int ret;
1080 :
1081 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
1082 : usockp->usock_fd);
1083 2951 : ret = accept(usockp->usock_fd, NULL, NULL);
1084 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d accepted %d\n", __func__, pthread_self(),
1085 : usockp->usock_fd, ret);
1086 2951 : if (ret < 0) {
1087 0 : usock_set_error(usockp, ret);
1088 0 : return ret;
1089 : }
1090 :
1091 5902 : if ((usockp->usock_state == LIMITED) ||
1092 2951 : (usock_clone(usockp, ret)))
1093 0 : close(ret);
1094 :
1095 2951 : return 0;
1096 : }
1097 :
1098 : static int
1099 53 : vr_usocket_bind(struct vr_usocket *usockp)
1100 : {
1101 53 : int error = 0;
1102 : struct sockaddr_in sin;
1103 : struct sockaddr_un sun;
1104 53 : struct sockaddr *addr = NULL;
1105 53 : socklen_t addrlen = 0;
1106 : int optval;
1107 : bool server;
1108 :
1109 53 : optval = 1;
1110 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d setting option\n", __func__,
1111 : pthread_self(), usockp->usock_fd);
1112 53 : if (setsockopt(usockp->usock_fd, SOL_SOCKET, SO_REUSEADDR, &optval,
1113 : sizeof(optval)))
1114 0 : return -errno;
1115 :
1116 53 : switch (usockp->usock_type) {
1117 0 : case TCP:
1118 0 : sin.sin_family = AF_INET;
1119 0 : sin.sin_port = rte_cpu_to_be_16(vr_netlink_port);
1120 0 : sin.sin_addr.s_addr = INADDR_ANY;
1121 0 : addr = (struct sockaddr *)&sin;
1122 0 : addrlen = sizeof(sin);
1123 0 : server = true;
1124 :
1125 0 : break;
1126 :
1127 53 : case UNIX:
1128 53 : sun.sun_family = AF_UNIX;
1129 53 : memset(sun.sun_path, 0, sizeof(sun.sun_path));
1130 53 : strncpy(sun.sun_path, vr_socket_dir, sizeof(sun.sun_path) - 1);
1131 53 : strncat(sun.sun_path, "/"VR_NETLINK_UNIX_NAME, sizeof(sun.sun_path)
1132 53 : - strlen(sun.sun_path) - 1);
1133 :
1134 53 : addr = (struct sockaddr *)&sun;
1135 53 : addrlen = sizeof(sun);
1136 53 : server = true;
1137 53 : mkdir(vr_socket_dir, VR_DEF_SOCKET_DIR_MODE);
1138 53 : unlink(sun.sun_path);
1139 :
1140 53 : break;
1141 :
1142 0 : case RAW:
1143 0 : sun.sun_family = AF_UNIX;
1144 0 : strncpy(vr_packet_unix_file, vr_socket_dir, sizeof(vr_packet_unix_file)
1145 : - 1);
1146 0 : strncat(vr_packet_unix_file, "/"VR_PACKET_UNIX_NAME,
1147 0 : sizeof(vr_packet_unix_file) - strlen(vr_packet_unix_file) - 1);
1148 0 : memset(sun.sun_path, 0, sizeof(sun.sun_path));
1149 0 : memcpy(sun.sun_path, vr_packet_unix_file, sizeof(sun.sun_path) - 1);
1150 :
1151 0 : addr = (struct sockaddr *)&sun;
1152 0 : addrlen = sizeof(sun);
1153 0 : server = false;
1154 0 : mkdir(vr_socket_dir, VR_DEF_SOCKET_DIR_MODE);
1155 0 : unlink(sun.sun_path);
1156 :
1157 0 : break;
1158 :
1159 0 : default:
1160 0 : return -EINVAL;
1161 : }
1162 :
1163 : #ifdef VR_DPDK_USOCK_DUMP
1164 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d binding\n", __func__, pthread_self(),
1165 : usockp->usock_fd);
1166 : rte_hexdump(stdout, "usock address dump:", addr, addrlen);
1167 : #endif
1168 53 : error = bind(usockp->usock_fd, addr, addrlen);
1169 53 : if (error < 0)
1170 0 : return error;
1171 :
1172 53 : if (server) {
1173 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d listening\n", __func__,
1174 : pthread_self(), usockp->usock_fd);
1175 53 : error = listen(usockp->usock_fd, 1);
1176 53 : if (error < 0)
1177 0 : return error;
1178 53 : usockp->usock_state = LISTENING;
1179 : }
1180 :
1181 53 : return 0;
1182 : }
1183 :
1184 : void
1185 53 : vr_usocket_close(void *sock)
1186 : {
1187 53 : struct vr_usocket *usockp = (struct vr_usocket *)sock;
1188 :
1189 53 : if (usockp == NULL)
1190 0 : return;
1191 :
1192 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
1193 : usockp->usock_fd);
1194 53 : if (usockp->usock_io_in_progress) {
1195 0 : usockp->usock_should_close = 1;
1196 0 : return;
1197 : }
1198 :
1199 53 : usock_close(usockp);
1200 53 : return;
1201 : }
1202 :
1203 : /*
1204 : * create a usocket which is of type (TCP/UNIX/RAW). the messages
1205 : * read in those sockets will follow protocol 'proto'. protocol is
1206 : * netlink, packet or event
1207 : */
1208 : void *
1209 159 : vr_usocket(int proto, int type)
1210 : {
1211 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: proto %u type %u\n", __func__,
1212 : pthread_self(), proto, type);
1213 :
1214 159 : if (!valid_usock(proto, type))
1215 0 : return NULL;
1216 :
1217 159 : return (void *)usock_alloc(proto, type);
1218 : }
1219 :
1220 : static int
1221 0 : usock_write(struct vr_usocket *usockp)
1222 : {
1223 : int ret;
1224 :
1225 0 : if (!usockp || usockp->usock_fd < 0)
1226 0 : return 0;
1227 :
1228 0 : ret = __usock_write(usockp);
1229 0 : if (ret < 0) {
1230 0 : usock_close(usockp);
1231 0 : return ret;
1232 : }
1233 :
1234 0 : if (usockp->usock_proto == NETLINK) {
1235 0 : if (usockp->usock_write_offset == usockp->usock_write_len) {
1236 0 : usock_netlink_write_responses(usockp);
1237 : }
1238 : }
1239 :
1240 0 : return 0;
1241 : }
1242 :
1243 :
1244 : /*
1245 : * start io on socket
1246 : */
1247 : int
1248 53 : vr_usocket_io(void *transport)
1249 : {
1250 : int ret, i, processed;
1251 : int timeout;
1252 : struct pollfd *pfd;
1253 53 : struct vr_usocket *usockp = (struct vr_usocket *)transport;
1254 53 : unsigned lcore_id = rte_lcore_id();
1255 53 : struct vr_dpdk_lcore *lcore = vr_dpdk.lcores[lcore_id];
1256 :
1257 53 : if (!usockp)
1258 0 : return -1;
1259 :
1260 : RTE_LOG_DP(DEBUG, USOCK, "%s[%lx]: FD %d\n", __func__, pthread_self(),
1261 : usockp->usock_fd);
1262 53 : if ((ret = usock_init_poll(usockp)))
1263 0 : goto return_from_io;
1264 :
1265 53 : pfd = &usockp->usock_pfds[0];
1266 53 : pfd->fd = usockp->usock_fd;
1267 53 : pfd->events = POLLIN;
1268 :
1269 53 : usockp->usock_io_in_progress = 1;
1270 :
1271 53 : timeout = usockp->usock_poll_block ? INFINITE_TIMEOUT : 0;
1272 : while (1) {
1273 11732 : if (usockp->usock_should_close) {
1274 0 : usock_close(usockp);
1275 0 : return -1;
1276 : }
1277 :
1278 : /*
1279 : * Handle an IPC commands for IO_LCORE_ID up
1280 : * and just check the stop flag for the rest.
1281 : */
1282 11732 : if (lcore_id >= VR_DPDK_IO_LCORE_ID) {
1283 11732 : if (unlikely(vr_dpdk_lcore_cmd_handle(lcore)))
1284 53 : break;
1285 : } else {
1286 0 : if (unlikely(vr_dpdk_is_stop_flag_set()))
1287 0 : break;
1288 : }
1289 :
1290 11679 : rcu_thread_offline();
1291 11679 : ret = poll(usockp->usock_pfds, usockp->usock_max_cfds,
1292 : timeout);
1293 11679 : if (ret < 0) {
1294 0 : usock_set_error(usockp, ret);
1295 : /* all other errors are fatal */
1296 0 : if (errno != EINTR)
1297 0 : goto return_from_io;
1298 : }
1299 :
1300 11679 : rcu_thread_online();
1301 :
1302 11679 : processed = 0;
1303 11679 : pfd = usockp->usock_pfds;
1304 40761 : for (i = 0; (i < usockp->usock_max_cfds) && (processed < ret);
1305 29082 : i++, pfd++) {
1306 29082 : if ((pfd->fd >= 0)) {
1307 29082 : if (pfd->revents & POLLIN) {
1308 11679 : if (i == 0) {
1309 2951 : ret = vr_usocket_read(usockp);
1310 2951 : if (ret < 0)
1311 0 : return ret;
1312 : } else {
1313 8728 : vr_usocket_read(usockp->usock_children[i]);
1314 : }
1315 : }
1316 :
1317 29082 : if (pfd->revents & POLLOUT) {
1318 0 : usock_write(usockp->usock_children[i]);
1319 : }
1320 :
1321 29082 : if (pfd->revents & POLLHUP) {
1322 2951 : if (i) {
1323 2951 : usock_close(usockp->usock_children[i]);
1324 : } else {
1325 0 : break;
1326 : }
1327 : }
1328 :
1329 29082 : if (pfd->revents)
1330 11679 : processed++;
1331 : }
1332 : }
1333 :
1334 11679 : if (!timeout)
1335 0 : return 0;
1336 : }
1337 :
1338 53 : return_from_io:
1339 53 : usockp->usock_io_in_progress = 0;
1340 53 : usock_deinit_poll(usockp);
1341 :
1342 53 : return ret;
1343 : }
1344 :
|
