vhost: refactor code structure
[dpdk.git] / lib / librte_vhost / socket.c
1 /*-
2  *   BSD LICENSE
3  *
4  *   Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
5  *   All rights reserved.
6  *
7  *   Redistribution and use in source and binary forms, with or without
8  *   modification, are permitted provided that the following conditions
9  *   are met:
10  *
11  *     * Redistributions of source code must retain the above copyright
12  *       notice, this list of conditions and the following disclaimer.
13  *     * Redistributions in binary form must reproduce the above copyright
14  *       notice, this list of conditions and the following disclaimer in
15  *       the documentation and/or other materials provided with the
16  *       distribution.
17  *     * Neither the name of Intel Corporation nor the names of its
18  *       contributors may be used to endorse or promote products derived
19  *       from this software without specific prior written permission.
20  *
21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32  */
33
34 #include <stdint.h>
35 #include <stdio.h>
36 #include <stdbool.h>
37 #include <limits.h>
38 #include <stdlib.h>
39 #include <unistd.h>
40 #include <string.h>
41 #include <sys/types.h>
42 #include <sys/socket.h>
43 #include <sys/un.h>
44 #include <sys/queue.h>
45 #include <errno.h>
46 #include <fcntl.h>
47 #include <pthread.h>
48
49 #include <rte_log.h>
50
51 #include "fd_man.h"
52 #include "vhost.h"
53 #include "vhost_user.h"
54
55 /*
56  * Every time rte_vhost_driver_register() is invoked, an associated
57  * vhost_user_socket struct will be created.
58  */
59 struct vhost_user_socket {
60         char *path;
61         int listenfd;
62         int connfd;
63         bool is_server;
64         bool reconnect;
65 };
66
67 struct vhost_user_connection {
68         struct vhost_user_socket *vsocket;
69         int vid;
70 };
71
72 #define MAX_VHOST_SOCKET 1024
73 struct vhost_user {
74         struct vhost_user_socket *vsockets[MAX_VHOST_SOCKET];
75         struct fdset fdset;
76         int vsocket_cnt;
77         pthread_mutex_t mutex;
78 };
79
80 #define MAX_VIRTIO_BACKLOG 128
81
82 static void vhost_user_server_new_connection(int fd, void *data, int *remove);
83 static void vhost_user_read_cb(int fd, void *dat, int *remove);
84 static int vhost_user_create_client(struct vhost_user_socket *vsocket);
85
86 static struct vhost_user vhost_user = {
87         .fdset = {
88                 .fd = { [0 ... MAX_FDS - 1] = {-1, NULL, NULL, NULL, 0} },
89                 .fd_mutex = PTHREAD_MUTEX_INITIALIZER,
90                 .num = 0
91         },
92         .vsocket_cnt = 0,
93         .mutex = PTHREAD_MUTEX_INITIALIZER,
94 };
95
96 /* return bytes# of read on success or negative val on failure. */
97 int
98 read_fd_message(int sockfd, char *buf, int buflen, int *fds, int fd_num)
99 {
100         struct iovec iov;
101         struct msghdr msgh;
102         size_t fdsize = fd_num * sizeof(int);
103         char control[CMSG_SPACE(fdsize)];
104         struct cmsghdr *cmsg;
105         int ret;
106
107         memset(&msgh, 0, sizeof(msgh));
108         iov.iov_base = buf;
109         iov.iov_len  = buflen;
110
111         msgh.msg_iov = &iov;
112         msgh.msg_iovlen = 1;
113         msgh.msg_control = control;
114         msgh.msg_controllen = sizeof(control);
115
116         ret = recvmsg(sockfd, &msgh, 0);
117         if (ret <= 0) {
118                 RTE_LOG(ERR, VHOST_CONFIG, "recvmsg failed\n");
119                 return ret;
120         }
121
122         if (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC)) {
123                 RTE_LOG(ERR, VHOST_CONFIG, "truncted msg\n");
124                 return -1;
125         }
126
127         for (cmsg = CMSG_FIRSTHDR(&msgh); cmsg != NULL;
128                 cmsg = CMSG_NXTHDR(&msgh, cmsg)) {
129                 if ((cmsg->cmsg_level == SOL_SOCKET) &&
130                         (cmsg->cmsg_type == SCM_RIGHTS)) {
131                         memcpy(fds, CMSG_DATA(cmsg), fdsize);
132                         break;
133                 }
134         }
135
136         return ret;
137 }
138
139 int
140 send_fd_message(int sockfd, char *buf, int buflen, int *fds, int fd_num)
141 {
142
143         struct iovec iov;
144         struct msghdr msgh;
145         size_t fdsize = fd_num * sizeof(int);
146         char control[CMSG_SPACE(fdsize)];
147         struct cmsghdr *cmsg;
148         int ret;
149
150         memset(&msgh, 0, sizeof(msgh));
151         iov.iov_base = buf;
152         iov.iov_len = buflen;
153
154         msgh.msg_iov = &iov;
155         msgh.msg_iovlen = 1;
156
157         if (fds && fd_num > 0) {
158                 msgh.msg_control = control;
159                 msgh.msg_controllen = sizeof(control);
160                 cmsg = CMSG_FIRSTHDR(&msgh);
161                 cmsg->cmsg_len = CMSG_LEN(fdsize);
162                 cmsg->cmsg_level = SOL_SOCKET;
163                 cmsg->cmsg_type = SCM_RIGHTS;
164                 memcpy(CMSG_DATA(cmsg), fds, fdsize);
165         } else {
166                 msgh.msg_control = NULL;
167                 msgh.msg_controllen = 0;
168         }
169
170         do {
171                 ret = sendmsg(sockfd, &msgh, 0);
172         } while (ret < 0 && errno == EINTR);
173
174         if (ret < 0) {
175                 RTE_LOG(ERR, VHOST_CONFIG,  "sendmsg error\n");
176                 return ret;
177         }
178
179         return ret;
180 }
181
182 static void
183 vhost_user_add_connection(int fd, struct vhost_user_socket *vsocket)
184 {
185         int vid;
186         size_t size;
187         struct vhost_user_connection *conn;
188         int ret;
189
190         conn = malloc(sizeof(*conn));
191         if (conn == NULL) {
192                 close(fd);
193                 return;
194         }
195
196         vid = vhost_new_device();
197         if (vid == -1) {
198                 close(fd);
199                 free(conn);
200                 return;
201         }
202
203         size = strnlen(vsocket->path, PATH_MAX);
204         vhost_set_ifname(vid, vsocket->path, size);
205
206         RTE_LOG(INFO, VHOST_CONFIG, "new device, handle is %d\n", vid);
207
208         vsocket->connfd = fd;
209         conn->vsocket = vsocket;
210         conn->vid = vid;
211         ret = fdset_add(&vhost_user.fdset, fd, vhost_user_read_cb,
212                         NULL, conn);
213         if (ret < 0) {
214                 vsocket->connfd = -1;
215                 free(conn);
216                 close(fd);
217                 RTE_LOG(ERR, VHOST_CONFIG,
218                         "failed to add fd %d into vhost server fdset\n",
219                         fd);
220         }
221 }
222
223 /* call back when there is new vhost-user connection from client  */
224 static void
225 vhost_user_server_new_connection(int fd, void *dat, int *remove __rte_unused)
226 {
227         struct vhost_user_socket *vsocket = dat;
228
229         fd = accept(fd, NULL, NULL);
230         if (fd < 0)
231                 return;
232
233         RTE_LOG(INFO, VHOST_CONFIG, "new vhost user connection is %d\n", fd);
234         vhost_user_add_connection(fd, vsocket);
235 }
236
237 static void
238 vhost_user_read_cb(int connfd, void *dat, int *remove)
239 {
240         struct vhost_user_connection *conn = dat;
241         struct vhost_user_socket *vsocket = conn->vsocket;
242         int ret;
243
244         ret = vhost_user_msg_handler(conn->vid, connfd);
245         if (ret < 0) {
246                 vsocket->connfd = -1;
247                 close(connfd);
248                 *remove = 1;
249                 free(conn);
250                 vhost_destroy_device(conn->vid);
251
252                 if (vsocket->reconnect)
253                         vhost_user_create_client(vsocket);
254         }
255 }
256
257 static int
258 create_unix_socket(const char *path, struct sockaddr_un *un, bool is_server)
259 {
260         int fd;
261
262         fd = socket(AF_UNIX, SOCK_STREAM, 0);
263         if (fd < 0)
264                 return -1;
265         RTE_LOG(INFO, VHOST_CONFIG, "vhost-user %s: socket created, fd: %d\n",
266                 is_server ? "server" : "client", fd);
267
268         if (!is_server && fcntl(fd, F_SETFL, O_NONBLOCK)) {
269                 RTE_LOG(ERR, VHOST_CONFIG,
270                         "vhost-user: can't set nonblocking mode for socket, fd: "
271                         "%d (%s)\n", fd, strerror(errno));
272                 close(fd);
273                 return -1;
274         }
275
276         memset(un, 0, sizeof(*un));
277         un->sun_family = AF_UNIX;
278         strncpy(un->sun_path, path, sizeof(un->sun_path));
279         un->sun_path[sizeof(un->sun_path) - 1] = '\0';
280
281         return fd;
282 }
283
284 static int
285 vhost_user_create_server(struct vhost_user_socket *vsocket)
286 {
287         int fd;
288         int ret;
289         struct sockaddr_un un;
290         const char *path = vsocket->path;
291
292         fd = create_unix_socket(path, &un, vsocket->is_server);
293         if (fd < 0)
294                 return -1;
295
296         ret = bind(fd, (struct sockaddr *)&un, sizeof(un));
297         if (ret < 0) {
298                 RTE_LOG(ERR, VHOST_CONFIG,
299                         "failed to bind to %s: %s; remove it and try again\n",
300                         path, strerror(errno));
301                 goto err;
302         }
303         RTE_LOG(INFO, VHOST_CONFIG, "bind to %s\n", path);
304
305         ret = listen(fd, MAX_VIRTIO_BACKLOG);
306         if (ret < 0)
307                 goto err;
308
309         vsocket->listenfd = fd;
310         ret = fdset_add(&vhost_user.fdset, fd, vhost_user_server_new_connection,
311                   NULL, vsocket);
312         if (ret < 0) {
313                 RTE_LOG(ERR, VHOST_CONFIG,
314                         "failed to add listen fd %d to vhost server fdset\n",
315                         fd);
316                 goto err;
317         }
318
319         return 0;
320
321 err:
322         close(fd);
323         return -1;
324 }
325
326 struct vhost_user_reconnect {
327         struct sockaddr_un un;
328         int fd;
329         struct vhost_user_socket *vsocket;
330
331         TAILQ_ENTRY(vhost_user_reconnect) next;
332 };
333
334 TAILQ_HEAD(vhost_user_reconnect_tailq_list, vhost_user_reconnect);
335 struct vhost_user_reconnect_list {
336         struct vhost_user_reconnect_tailq_list head;
337         pthread_mutex_t mutex;
338 };
339
340 static struct vhost_user_reconnect_list reconn_list;
341 static pthread_t reconn_tid;
342
343 static int
344 vhost_user_connect_nonblock(int fd, struct sockaddr *un, size_t sz)
345 {
346         int ret, flags;
347
348         ret = connect(fd, un, sz);
349         if (ret < 0 && errno != EISCONN)
350                 return -1;
351
352         flags = fcntl(fd, F_GETFL, 0);
353         if (flags < 0) {
354                 RTE_LOG(ERR, VHOST_CONFIG,
355                         "can't get flags for connfd %d\n", fd);
356                 return -2;
357         }
358         if ((flags & O_NONBLOCK) && fcntl(fd, F_SETFL, flags & ~O_NONBLOCK)) {
359                 RTE_LOG(ERR, VHOST_CONFIG,
360                                 "can't disable nonblocking on fd %d\n", fd);
361                 return -2;
362         }
363         return 0;
364 }
365
366 static void *
367 vhost_user_client_reconnect(void *arg __rte_unused)
368 {
369         int ret;
370         struct vhost_user_reconnect *reconn, *next;
371
372         while (1) {
373                 pthread_mutex_lock(&reconn_list.mutex);
374
375                 /*
376                  * An equal implementation of TAILQ_FOREACH_SAFE,
377                  * which does not exist on all platforms.
378                  */
379                 for (reconn = TAILQ_FIRST(&reconn_list.head);
380                      reconn != NULL; reconn = next) {
381                         next = TAILQ_NEXT(reconn, next);
382
383                         ret = vhost_user_connect_nonblock(reconn->fd,
384                                                 (struct sockaddr *)&reconn->un,
385                                                 sizeof(reconn->un));
386                         if (ret == -2) {
387                                 close(reconn->fd);
388                                 RTE_LOG(ERR, VHOST_CONFIG,
389                                         "reconnection for fd %d failed\n",
390                                         reconn->fd);
391                                 goto remove_fd;
392                         }
393                         if (ret == -1)
394                                 continue;
395
396                         RTE_LOG(INFO, VHOST_CONFIG,
397                                 "%s: connected\n", reconn->vsocket->path);
398                         vhost_user_add_connection(reconn->fd, reconn->vsocket);
399 remove_fd:
400                         TAILQ_REMOVE(&reconn_list.head, reconn, next);
401                         free(reconn);
402                 }
403
404                 pthread_mutex_unlock(&reconn_list.mutex);
405                 sleep(1);
406         }
407
408         return NULL;
409 }
410
411 static int
412 vhost_user_reconnect_init(void)
413 {
414         int ret;
415
416         pthread_mutex_init(&reconn_list.mutex, NULL);
417         TAILQ_INIT(&reconn_list.head);
418
419         ret = pthread_create(&reconn_tid, NULL,
420                              vhost_user_client_reconnect, NULL);
421         if (ret < 0)
422                 RTE_LOG(ERR, VHOST_CONFIG, "failed to create reconnect thread");
423
424         return ret;
425 }
426
427 static int
428 vhost_user_create_client(struct vhost_user_socket *vsocket)
429 {
430         int fd;
431         int ret;
432         struct sockaddr_un un;
433         const char *path = vsocket->path;
434         struct vhost_user_reconnect *reconn;
435
436         fd = create_unix_socket(path, &un, vsocket->is_server);
437         if (fd < 0)
438                 return -1;
439
440         ret = vhost_user_connect_nonblock(fd, (struct sockaddr *)&un,
441                                           sizeof(un));
442         if (ret == 0) {
443                 vhost_user_add_connection(fd, vsocket);
444                 return 0;
445         }
446
447         RTE_LOG(ERR, VHOST_CONFIG,
448                 "failed to connect to %s: %s\n",
449                 path, strerror(errno));
450
451         if (ret == -2 || !vsocket->reconnect) {
452                 close(fd);
453                 return -1;
454         }
455
456         RTE_LOG(ERR, VHOST_CONFIG, "%s: reconnecting...\n", path);
457         reconn = malloc(sizeof(*reconn));
458         if (reconn == NULL) {
459                 RTE_LOG(ERR, VHOST_CONFIG,
460                         "failed to allocate memory for reconnect\n");
461                 close(fd);
462                 return -1;
463         }
464         reconn->un = un;
465         reconn->fd = fd;
466         reconn->vsocket = vsocket;
467         pthread_mutex_lock(&reconn_list.mutex);
468         TAILQ_INSERT_TAIL(&reconn_list.head, reconn, next);
469         pthread_mutex_unlock(&reconn_list.mutex);
470
471         return 0;
472 }
473
474 /*
475  * Register a new vhost-user socket; here we could act as server
476  * (the default case), or client (when RTE_VHOST_USER_CLIENT) flag
477  * is set.
478  */
479 int
480 rte_vhost_driver_register(const char *path, uint64_t flags)
481 {
482         int ret = -1;
483         struct vhost_user_socket *vsocket;
484
485         if (!path)
486                 return -1;
487
488         pthread_mutex_lock(&vhost_user.mutex);
489
490         if (vhost_user.vsocket_cnt == MAX_VHOST_SOCKET) {
491                 RTE_LOG(ERR, VHOST_CONFIG,
492                         "error: the number of vhost sockets reaches maximum\n");
493                 goto out;
494         }
495
496         vsocket = malloc(sizeof(struct vhost_user_socket));
497         if (!vsocket)
498                 goto out;
499         memset(vsocket, 0, sizeof(struct vhost_user_socket));
500         vsocket->path = strdup(path);
501         vsocket->connfd = -1;
502
503         if ((flags & RTE_VHOST_USER_CLIENT) != 0) {
504                 vsocket->reconnect = !(flags & RTE_VHOST_USER_NO_RECONNECT);
505                 if (vsocket->reconnect && reconn_tid == 0) {
506                         if (vhost_user_reconnect_init() < 0) {
507                                 free(vsocket->path);
508                                 free(vsocket);
509                                 goto out;
510                         }
511                 }
512                 ret = vhost_user_create_client(vsocket);
513         } else {
514                 vsocket->is_server = true;
515                 ret = vhost_user_create_server(vsocket);
516         }
517         if (ret < 0) {
518                 free(vsocket->path);
519                 free(vsocket);
520                 goto out;
521         }
522
523         vhost_user.vsockets[vhost_user.vsocket_cnt++] = vsocket;
524
525 out:
526         pthread_mutex_unlock(&vhost_user.mutex);
527
528         return ret;
529 }
530
531 static bool
532 vhost_user_remove_reconnect(struct vhost_user_socket *vsocket)
533 {
534         int found = false;
535         struct vhost_user_reconnect *reconn, *next;
536
537         pthread_mutex_lock(&reconn_list.mutex);
538
539         for (reconn = TAILQ_FIRST(&reconn_list.head);
540              reconn != NULL; reconn = next) {
541                 next = TAILQ_NEXT(reconn, next);
542
543                 if (reconn->vsocket == vsocket) {
544                         TAILQ_REMOVE(&reconn_list.head, reconn, next);
545                         close(reconn->fd);
546                         free(reconn);
547                         found = true;
548                         break;
549                 }
550         }
551         pthread_mutex_unlock(&reconn_list.mutex);
552         return found;
553 }
554
555 /**
556  * Unregister the specified vhost socket
557  */
558 int
559 rte_vhost_driver_unregister(const char *path)
560 {
561         int i;
562         int count;
563         struct vhost_user_connection *conn;
564
565         pthread_mutex_lock(&vhost_user.mutex);
566
567         for (i = 0; i < vhost_user.vsocket_cnt; i++) {
568                 struct vhost_user_socket *vsocket = vhost_user.vsockets[i];
569
570                 if (!strcmp(vsocket->path, path)) {
571                         if (vsocket->is_server) {
572                                 fdset_del(&vhost_user.fdset, vsocket->listenfd);
573                                 close(vsocket->listenfd);
574                                 unlink(path);
575                         } else if (vsocket->reconnect) {
576                                 vhost_user_remove_reconnect(vsocket);
577                         }
578
579                         conn = fdset_del(&vhost_user.fdset, vsocket->connfd);
580                         if (conn) {
581                                 RTE_LOG(INFO, VHOST_CONFIG,
582                                         "free connfd = %d for device '%s'\n",
583                                         vsocket->connfd, path);
584                                 close(vsocket->connfd);
585                                 vhost_destroy_device(conn->vid);
586                                 free(conn);
587                         }
588
589                         free(vsocket->path);
590                         free(vsocket);
591
592                         count = --vhost_user.vsocket_cnt;
593                         vhost_user.vsockets[i] = vhost_user.vsockets[count];
594                         vhost_user.vsockets[count] = NULL;
595                         pthread_mutex_unlock(&vhost_user.mutex);
596
597                         return 0;
598                 }
599         }
600         pthread_mutex_unlock(&vhost_user.mutex);
601
602         return -1;
603 }
604
605 int
606 rte_vhost_driver_session_start(void)
607 {
608         fdset_event_dispatch(&vhost_user.fdset);
609         return 0;
610 }