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