X-Git-Url: https://gerrit.fd.io/r/gitweb?p=vpp.git;a=blobdiff_plain;f=src%2Fvlibmemory%2Fsocket_client.c;h=2fb6b8a0c4e9ac43c92f7e259f30ecfc99e40ccd;hp=44d21228bfe6af823961b877a00e9e1c55c9bf2a;hb=36217e3ca;hpb=4d9b9d8e74f12a26404ccdd8baf46c61b44584db diff --git a/src/vlibmemory/socket_client.c b/src/vlibmemory/socket_client.c index 44d21228bfe..2fb6b8a0c4e 100644 --- a/src/vlibmemory/socket_client.c +++ b/src/vlibmemory/socket_client.c @@ -19,6 +19,7 @@ #include #define __USE_GNU +#define _GNU_SOURCE #include #include @@ -42,6 +43,7 @@ #undef vl_printfun socket_client_main_t socket_client_main; +__thread socket_client_main_t *socket_client_ctx = &socket_client_main; /* Debug aid */ u32 vl (void *p) __attribute__ ((weak)); @@ -52,10 +54,24 @@ vl (void *p) return vec_len (p); } -int -vl_socket_client_read (int wait) +static socket_client_main_t * +vl_socket_client_ctx_push (socket_client_main_t * ctx) +{ + socket_client_main_t *old = socket_client_ctx; + socket_client_ctx = ctx; + return old; +} + +static void +vl_socket_client_ctx_pop (socket_client_main_t * old_ctx) { - socket_client_main_t *scm = &socket_client_main; + socket_client_ctx = old_ctx; +} + +static int +vl_socket_client_read_internal (socket_client_main_t * scm, int wait) +{ + u32 data_len = 0, msg_size; int n, current_rx_index; msgbuf_t *mbp = 0; f64 timeout; @@ -68,10 +84,9 @@ vl_socket_client_read (int wait) while (1) { - current_rx_index = vec_len (scm->socket_rx_buffer); - while (vec_len (scm->socket_rx_buffer) < - sizeof (*mbp) + 2 /* msg id */ ) + while (vec_len (scm->socket_rx_buffer) < sizeof (*mbp)) { + current_rx_index = vec_len (scm->socket_rx_buffer); vec_validate (scm->socket_rx_buffer, current_rx_index + scm->socket_buffer_size - 1); _vec_len (scm->socket_rx_buffer) = current_rx_index; @@ -79,6 +94,9 @@ vl_socket_client_read (int wait) scm->socket_buffer_size); if (n < 0) { + if (errno == EAGAIN) + continue; + clib_unix_warning ("socket_read"); return -1; } @@ -90,20 +108,38 @@ vl_socket_client_read (int wait) clib_warning ("read %d bytes", n); #endif - if (mbp == 0) - mbp = (msgbuf_t *) (scm->socket_rx_buffer); + mbp = (msgbuf_t *) (scm->socket_rx_buffer); + data_len = ntohl (mbp->data_len); + current_rx_index = vec_len (scm->socket_rx_buffer); + vec_validate (scm->socket_rx_buffer, current_rx_index + data_len); + _vec_len (scm->socket_rx_buffer) = current_rx_index; + mbp = (msgbuf_t *) (scm->socket_rx_buffer); + msg_size = data_len + sizeof (*mbp); - if (vec_len (scm->socket_rx_buffer) >= ntohl (mbp->data_len) - + sizeof (*mbp)) + while (vec_len (scm->socket_rx_buffer) < msg_size) + { + n = read (scm->socket_fd, + scm->socket_rx_buffer + vec_len (scm->socket_rx_buffer), + msg_size - vec_len (scm->socket_rx_buffer)); + if (n < 0) + { + if (errno == EAGAIN) + continue; + + clib_unix_warning ("socket_read"); + return -1; + } + _vec_len (scm->socket_rx_buffer) += n; + } + + if (vec_len (scm->socket_rx_buffer) >= data_len + sizeof (*mbp)) { vl_msg_api_socket_handler ((void *) (mbp->data)); - if (vec_len (scm->socket_rx_buffer) == ntohl (mbp->data_len) - + sizeof (*mbp)) + if (vec_len (scm->socket_rx_buffer) == data_len + sizeof (*mbp)) _vec_len (scm->socket_rx_buffer) = 0; else - vec_delete (scm->socket_rx_buffer, ntohl (mbp->data_len) - + sizeof (*mbp), 0); + vec_delete (scm->socket_rx_buffer, data_len + sizeof (*mbp), 0); mbp = 0; /* Quit if we're out of data, and not expecting a ping reply */ @@ -111,7 +147,6 @@ vl_socket_client_read (int wait) && scm->control_pings_outstanding == 0) break; } - if (wait && clib_time_now (&scm->clib_time) >= timeout) return -1; } @@ -119,9 +154,26 @@ vl_socket_client_read (int wait) } int -vl_socket_client_write (void) +vl_socket_client_read (int wait) +{ + return vl_socket_client_read_internal (socket_client_ctx, wait); +} + +int +vl_socket_client_read2 (socket_client_main_t * scm, int wait) +{ + socket_client_main_t *old_ctx; + int rv; + + old_ctx = vl_socket_client_ctx_push (scm); + rv = vl_socket_client_read_internal (scm, wait); + vl_socket_client_ctx_pop (old_ctx); + return rv; +} + +static int +vl_socket_client_write_internal (socket_client_main_t * scm) { - socket_client_main_t *scm = &socket_client_main; int n; msgbuf_t msgbuf = { @@ -147,42 +199,87 @@ vl_socket_client_write (void) return n; } +int +vl_socket_client_write (void) +{ + return vl_socket_client_write_internal (socket_client_ctx); +} + +int +vl_socket_client_write2 (socket_client_main_t * scm) +{ + socket_client_main_t *old_ctx; + int rv; + + old_ctx = vl_socket_client_ctx_push (scm); + rv = vl_socket_client_write_internal (scm); + vl_socket_client_ctx_pop (old_ctx); + return rv; +} + +void * +vl_socket_client_msg_alloc2 (socket_client_main_t * scm, int nbytes) +{ + scm->socket_tx_nbytes = nbytes; + return ((void *) scm->socket_tx_buffer); +} + void * vl_socket_client_msg_alloc (int nbytes) { - socket_client_main.socket_tx_nbytes = nbytes; - return ((void *) socket_client_main.socket_tx_buffer); + return vl_socket_client_msg_alloc2 (socket_client_ctx, nbytes); } void -vl_socket_client_disconnect (void) +vl_socket_client_disconnect2 (socket_client_main_t * scm) { - socket_client_main_t *scm = &socket_client_main; + if (vl_mem_client_is_connected ()) + { + vl_client_disconnect_from_vlib_no_unmap (); + ssvm_delete_memfd (&scm->memfd_segment); + } if (scm->socket_fd && (close (scm->socket_fd) < 0)) clib_unix_warning ("close"); scm->socket_fd = 0; } void -vl_socket_client_enable_disable (int enable) +vl_socket_client_disconnect (void) +{ + vl_socket_client_disconnect2 (socket_client_ctx); +} + +void +vl_socket_client_enable_disable2 (socket_client_main_t * scm, int enable) { - socket_client_main_t *scm = &socket_client_main; scm->socket_enable = enable; } +void +vl_socket_client_enable_disable (int enable) +{ + vl_socket_client_enable_disable2 (socket_client_ctx, enable); +} + static clib_error_t * -receive_fd_msg (int socket_fd, int *my_fd) +vl_sock_api_recv_fd_msg_internal (socket_client_main_t * scm, int fds[], + int n_fds, u32 wait) { char msgbuf[16]; - char ctl[CMSG_SPACE (sizeof (int)) + CMSG_SPACE (sizeof (struct ucred))]; + char ctl[CMSG_SPACE (sizeof (int) * n_fds) + + CMSG_SPACE (sizeof (struct ucred))]; struct msghdr mh = { 0 }; struct iovec iov[1]; - ssize_t size; + ssize_t size = 0; struct ucred *cr = 0; struct cmsghdr *cmsg; pid_t pid __attribute__ ((unused)); uid_t uid __attribute__ ((unused)); gid_t gid __attribute__ ((unused)); + int socket_fd; + f64 timeout; + + socket_fd = scm->client_socket.fd; iov[0].iov_base = msgbuf; iov[0].iov_len = 5; @@ -191,10 +288,17 @@ receive_fd_msg (int socket_fd, int *my_fd) mh.msg_control = ctl; mh.msg_controllen = sizeof (ctl); - memset (ctl, 0, sizeof (ctl)); + clib_memset (ctl, 0, sizeof (ctl)); + + if (wait != ~0) + { + timeout = clib_time_now (&scm->clib_time) + wait; + while (size != 5 && clib_time_now (&scm->clib_time) < timeout) + size = recvmsg (socket_fd, &mh, MSG_DONTWAIT); + } + else + size = recvmsg (socket_fd, &mh, 0); - /* receive the incoming message */ - size = recvmsg (socket_fd, &mh, 0); if (size != 5) { return (size == 0) ? clib_error_return (0, "disconnected") : @@ -216,7 +320,7 @@ receive_fd_msg (int socket_fd, int *my_fd) } else if (cmsg->cmsg_type == SCM_RIGHTS) { - clib_memcpy (my_fd, CMSG_DATA (cmsg), sizeof (int)); + clib_memcpy_fast (fds, CMSG_DATA (cmsg), sizeof (int) * n_fds); } } cmsg = CMSG_NXTHDR (&mh, cmsg); @@ -224,15 +328,35 @@ receive_fd_msg (int socket_fd, int *my_fd) return 0; } +clib_error_t * +vl_sock_api_recv_fd_msg (int socket_fd, int fds[], int n_fds, u32 wait) +{ + return vl_sock_api_recv_fd_msg_internal (socket_client_ctx, fds, n_fds, + wait); +} + +clib_error_t * +vl_sock_api_recv_fd_msg2 (socket_client_main_t * scm, int socket_fd, + int fds[], int n_fds, u32 wait) +{ + socket_client_main_t *old_ctx; + clib_error_t *error; + + old_ctx = vl_socket_client_ctx_push (scm); + error = vl_sock_api_recv_fd_msg_internal (scm, fds, n_fds, wait); + vl_socket_client_ctx_pop (old_ctx); + return error; +} + static void vl_api_sock_init_shm_reply_t_handler (vl_api_sock_init_shm_reply_t * mp) { - socket_client_main_t *scm = &socket_client_main; - int my_fd = -1; - clib_error_t *error; + socket_client_main_t *scm = socket_client_ctx; + ssvm_private_t *memfd = &scm->memfd_segment; i32 retval = ntohl (mp->retval); - ssvm_private_t memfd; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main (); + clib_error_t *error; + int my_fd = -1; u8 *new_name; if (retval) @@ -244,31 +368,42 @@ static void vl_api_sock_init_shm_reply_t_handler /* * Check the socket for the magic fd */ - error = receive_fd_msg (scm->socket_fd, &my_fd); + error = vl_sock_api_recv_fd_msg (scm->socket_fd, &my_fd, 1, 5); if (error) { + clib_error_report (error); retval = -99; return; } - memset (&memfd, 0, sizeof (memfd)); - memfd.fd = my_fd; + clib_memset (memfd, 0, sizeof (*memfd)); + memfd->fd = my_fd; /* Note: this closes memfd.fd */ - retval = ssvm_slave_init_memfd (&memfd); + retval = ssvm_client_init_memfd (memfd); if (retval) clib_warning ("WARNING: segment map returned %d", retval); /* * Pivot to the memory client segment that vpp just created */ - am->vlib_rp = (void *) (memfd.requested_va + MMAP_PAGESIZE); + am->vlib_rp = (void *) (memfd->requested_va + MMAP_PAGESIZE); am->shmem_hdr = (void *) am->vlib_rp->user_ctx; new_name = format (0, "%v[shm]%c", scm->name, 0); vl_client_install_client_message_handlers (); - vl_client_connect_to_vlib_no_map ("pvt", (char *) new_name, - 32 /* input_queue_length */ ); + if (scm->want_shm_pthread) + { + vl_client_connect_to_vlib_no_map ("pvt", (char *) new_name, + 32 /* input_queue_length */ ); + } + else + { + vl_client_connect_to_vlib_no_rx_pthread_no_map ("pvt", + (char *) new_name, 32 + /* input_queue_length */ + ); + } vl_socket_client_enable_disable (0); vec_free (new_name); } @@ -276,9 +411,12 @@ static void vl_api_sock_init_shm_reply_t_handler static void vl_api_sockclnt_create_reply_t_handler (vl_api_sockclnt_create_reply_t * mp) { - socket_client_main_t *scm = &socket_client_main; + socket_client_main_t *scm = socket_client_ctx; if (!mp->response) - scm->socket_enable = 1; + { + scm->socket_enable = 1; + scm->client_index = clib_net_to_host_u32 (mp->index); + } } #define foreach_sock_client_api_msg \ @@ -294,22 +432,21 @@ void vl_sock_client_install_message_handlers (void) { -#define _(N,n) \ - vl_msg_api_set_handlers(VL_API_##N, #n, \ - vl_api_##n##_t_handler, \ - noop_handler, \ - vl_api_##n##_t_endian, \ - vl_api_##n##_t_print, \ - sizeof(vl_api_##n##_t), 1); +#define _(N, n) \ + vl_msg_api_set_handlers (VL_API_##N, #n, vl_api_##n##_t_handler, \ + noop_handler, vl_api_##n##_t_endian, \ + vl_api_##n##_t_print, sizeof (vl_api_##n##_t), 0, \ + vl_api_##n##_t_print_json, vl_api_##n##_t_tojson, \ + vl_api_##n##_t_fromjson); foreach_sock_client_api_msg; #undef _ } int -vl_socket_client_connect (char *socket_path, char *client_name, - u32 socket_buffer_size) +vl_socket_client_connect_internal (socket_client_main_t * scm, + char *socket_path, char *client_name, + u32 socket_buffer_size) { - socket_client_main_t *scm = &socket_client_main; vl_api_sockclnt_create_t *mp; clib_socket_t *sock; clib_error_t *error; @@ -324,7 +461,7 @@ vl_socket_client_connect (char *socket_path, char *client_name, sock = &scm->client_socket; sock->config = socket_path; - sock->flags = CLIB_SOCKET_F_IS_CLIENT | CLIB_SOCKET_F_SEQPACKET; + sock->flags = CLIB_SOCKET_F_IS_CLIENT; if ((error = clib_socket_init (sock))) { @@ -343,34 +480,61 @@ vl_socket_client_connect (char *socket_path, char *client_name, _vec_len (scm->socket_tx_buffer) = 0; scm->name = format (0, "%s", client_name); - mp = vl_socket_client_msg_alloc (sizeof (*mp)); + mp = vl_socket_client_msg_alloc2 (scm, sizeof (*mp)); mp->_vl_msg_id = htons (VL_API_SOCKCLNT_CREATE); strncpy ((char *) mp->name, client_name, sizeof (mp->name) - 1); mp->name[sizeof (mp->name) - 1] = 0; mp->context = 0xfeedface; - if (vl_socket_client_write () <= 0) + clib_time_init (&scm->clib_time); + + if (vl_socket_client_write_internal (scm) <= 0) return (-1); - if (vl_socket_client_read (1)) + if (vl_socket_client_read_internal (scm, 5)) return (-1); - clib_time_init (&scm->clib_time); return (0); } int -vl_socket_client_init_shm (vl_api_shm_elem_config_t * config) +vl_socket_client_connect (char *socket_path, char *client_name, + u32 socket_buffer_size) +{ + return vl_socket_client_connect_internal (socket_client_ctx, socket_path, + client_name, socket_buffer_size); +} + +int +vl_socket_client_connect2 (socket_client_main_t * scm, char *socket_path, + char *client_name, u32 socket_buffer_size) +{ + socket_client_main_t *old_ctx; + int rv; + + old_ctx = vl_socket_client_ctx_push (scm); + rv = vl_socket_client_connect_internal (socket_client_ctx, socket_path, + client_name, socket_buffer_size); + vl_socket_client_ctx_pop (old_ctx); + return rv; +} + +int +vl_socket_client_init_shm_internal (socket_client_main_t * scm, + vl_api_shm_elem_config_t * config, + int want_pthread) { vl_api_sock_init_shm_t *mp; int rv, i; u64 *cfg; - mp = vl_socket_client_msg_alloc (sizeof (*mp) + - vec_len (config) * sizeof (u64)); - memset (mp, 0, sizeof (*mp)); + scm->want_shm_pthread = want_pthread; + + mp = vl_socket_client_msg_alloc2 (scm, sizeof (*mp) + + vec_len (config) * sizeof (u64)); + clib_memset (mp, 0, sizeof (*mp)); mp->_vl_msg_id = clib_host_to_net_u16 (VL_API_SOCK_INIT_SHM); - mp->client_index = ~0; + mp->client_index = clib_host_to_net_u32 (scm->client_index); mp->requested_size = 64 << 20; if (config) @@ -382,16 +546,54 @@ vl_socket_client_init_shm (vl_api_shm_elem_config_t * config) } mp->nitems = vec_len (config); } - rv = vl_socket_client_write (); + rv = vl_socket_client_write_internal (scm); if (rv <= 0) return rv; - if (vl_socket_client_read (1)) + if (vl_socket_client_read_internal (scm, 1)) return -1; return 0; } +int +vl_socket_client_init_shm (vl_api_shm_elem_config_t * config, + int want_pthread) +{ + return vl_socket_client_init_shm_internal (socket_client_ctx, config, + want_pthread); +} + +int +vl_socket_client_init_shm2 (socket_client_main_t * scm, + vl_api_shm_elem_config_t * config, + int want_pthread) +{ + socket_client_main_t *old_ctx; + int rv; + + old_ctx = vl_socket_client_ctx_push (scm); + rv = vl_socket_client_init_shm_internal (socket_client_ctx, config, + want_pthread); + vl_socket_client_ctx_pop (old_ctx); + return rv; +} + +clib_error_t * +vl_socket_client_recv_fd_msg2 (socket_client_main_t * scm, int fds[], + int n_fds, u32 wait) +{ + if (!scm->socket_fd) + return clib_error_return (0, "no socket"); + return vl_sock_api_recv_fd_msg_internal (scm, fds, n_fds, wait); +} + +clib_error_t * +vl_socket_client_recv_fd_msg (int fds[], int n_fds, u32 wait) +{ + return vl_socket_client_recv_fd_msg2 (socket_client_ctx, fds, n_fds, wait); +} + /* * fd.io coding-style-patch-verification: ON *