X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvlibmemory%2Fsocket_api.c;h=6238746525af005ee72dfbd222837868f5edade0;hb=a6ef36b2c25de47824a1b45e147ab2fbf67c3a33;hp=0f8c0f046a3f76c208562f439d9f08e2541ef529;hpb=178cf493d009995b28fdf220f04c98860ff79a9b;p=vpp.git diff --git a/src/vlibmemory/socket_api.c b/src/vlibmemory/socket_api.c index 0f8c0f046a3..6238746525a 100644 --- a/src/vlibmemory/socket_api.c +++ b/src/vlibmemory/socket_api.c @@ -118,7 +118,7 @@ vl_socket_api_send (vl_api_registration_t * rp, u8 * elem) #endif socket_main_t *sm = &socket_main; u16 msg_id = ntohs (*(u16 *) elem); - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main (); msgbuf_t *mb = (msgbuf_t *) (elem - offsetof (msgbuf_t, data)); vl_api_registration_t *sock_rp; clib_file_main_t *fm = &file_main; @@ -190,56 +190,69 @@ vl_socket_free_registration_index (u32 pool_index) } void -vl_socket_process_api_msg (clib_file_t * uf, vl_api_registration_t * rp, - i8 * input_v) +vl_socket_process_api_msg (vl_api_registration_t * rp, i8 * input_v) { msgbuf_t *mbp = (msgbuf_t *) input_v; u8 *the_msg = (u8 *) (mbp->data); - socket_main.current_uf = uf; socket_main.current_rp = rp; vl_msg_api_socket_handler (the_msg); - socket_main.current_uf = 0; socket_main.current_rp = 0; } +/* + * Read function for API socket. + * + * Read data from socket, invoke SOCKET_READ_EVENT + * for each fully read API message, return 0. + * Store incomplete data for next invocation to continue. + * + * On severe read error, the file is closed. + * + * As reading is single threaded, + * socket_main.input_buffer is used temporarily. + * Even its length is modified, but always restored before return. + * + * Incomplete data is copied into a vector, + * pointer saved in registration's unprocessed_input. + */ clib_error_t * vl_socket_read_ready (clib_file_t * uf) { clib_file_main_t *fm = &file_main; vlib_main_t *vm = vlib_get_main (); vl_api_registration_t *rp; + /* n is the size of data read to input_buffer */ int n; + /* msg_buffer vector can point to input_buffer or unprocessed_input */ i8 *msg_buffer = 0; + /* data_for_process is a vector containing one full message, incl msgbuf_t */ u8 *data_for_process; - u32 msg_len; + /* msgbuf_len is the size of one message, including sizeof (msgbuf_t) */ + u32 msgbuf_len; u32 save_input_buffer_length = vec_len (socket_main.input_buffer); vl_socket_args_for_process_t *a; - msgbuf_t *mbp; - int mbp_set = 0; + u32 reg_index = uf->private_data; - rp = pool_elt_at_index (socket_main.registration_pool, uf->private_data); + rp = vl_socket_get_registration (reg_index); + /* Ignore unprocessed_input for now, n describes input_buffer for now. */ n = read (uf->file_descriptor, socket_main.input_buffer, vec_len (socket_main.input_buffer)); - if (n <= 0 && errno != EAGAIN) + if (n <= 0) { - clib_file_del (fm, uf); - - if (!pool_is_free (socket_main.registration_pool, rp)) - { - u32 index = rp - socket_main.registration_pool; - vl_socket_free_registration_index (index); - } - else + if (errno != EAGAIN) { - clib_warning ("client index %d already free?", - rp->vl_api_registration_pool_index); + /* Severe error, close the file. */ + clib_file_del (fm, uf); + vl_socket_free_registration_index (reg_index); } + /* EAGAIN means we do not close the file, but no data to process anyway. */ return 0; } + /* Fake smaller length teporarily, so input_buffer can be used as msg_buffer. */ _vec_len (socket_main.input_buffer) = n; /* @@ -248,37 +261,38 @@ vl_socket_read_ready (clib_file_t * uf) * boundaries. In the case of a long message (>4K bytes) * we have to do (at least) 2 reads, etc. */ + /* Determine msg_buffer. */ + if (vec_len (rp->unprocessed_input)) + { + vec_append (rp->unprocessed_input, socket_main.input_buffer); + msg_buffer = rp->unprocessed_input; + } + else + { + msg_buffer = socket_main.input_buffer; + } + /* Loop to process any full messages. */ + ASSERT (vec_len (msg_buffer) > 0); do { - if (vec_len (rp->unprocessed_input)) - { - vec_append (rp->unprocessed_input, socket_main.input_buffer); - msg_buffer = rp->unprocessed_input; - } - else - { - msg_buffer = socket_main.input_buffer; - mbp_set = 0; - } - - if (mbp_set == 0) - { - /* Any chance that we have a complete message? */ - if (vec_len (msg_buffer) <= sizeof (msgbuf_t)) - goto save_and_split; - - mbp = (msgbuf_t *) msg_buffer; - msg_len = ntohl (mbp->data_len); - mbp_set = 1; - } - - /* We don't have the entire message yet. */ - if (mbp_set == 0 - || (msg_len + sizeof (msgbuf_t)) > vec_len (msg_buffer)) + /* Here, we are not sure how big a chunk of message we have left. */ + /* Do we at least know how big the full message will be? */ + if (vec_len (msg_buffer) <= sizeof (msgbuf_t)) + /* No, so fragment is not a full message. */ + goto save_and_split; + + /* Now we know how big the full message will be. */ + msgbuf_len = + ntohl (((msgbuf_t *) msg_buffer)->data_len) + sizeof (msgbuf_t); + + /* But do we have a full message? */ + if (msgbuf_len > vec_len (msg_buffer)) { save_and_split: - /* if we were using the input buffer save the fragment */ + /* We don't have the entire message yet. */ + /* If msg_buffer is unprocessed_input, nothing needs to be done. */ if (msg_buffer == socket_main.input_buffer) + /* But if we were using the input buffer, save the fragment. */ { ASSERT (vec_len (rp->unprocessed_input) == 0); vec_validate (rp->unprocessed_input, vec_len (msg_buffer) - 1); @@ -286,32 +300,37 @@ vl_socket_read_ready (clib_file_t * uf) vec_len (msg_buffer)); _vec_len (rp->unprocessed_input) = vec_len (msg_buffer); } + /* No more full messages, restore original input_buffer length. */ _vec_len (socket_main.input_buffer) = save_input_buffer_length; return 0; } + /* + * We have at least one full message. + * But msg_buffer can contain more data, so copy one message data + * so we can overwrite its length to what single message has. + */ data_for_process = (u8 *) vec_dup (msg_buffer); - _vec_len (data_for_process) = (msg_len + sizeof (msgbuf_t)); + _vec_len (data_for_process) = msgbuf_len; + /* Everything is ready to signal the SOCKET_READ_EVENT. */ pool_get (socket_main.process_args, a); - a->clib_file = uf; - a->regp = rp; + a->reg_index = reg_index; a->data = data_for_process; vlib_process_signal_event (vm, vl_api_clnt_node.index, SOCKET_READ_EVENT, a - socket_main.process_args); - if (n > (msg_len + sizeof (*mbp))) - vec_delete (msg_buffer, msg_len + sizeof (*mbp), 0); + if (vec_len (msg_buffer) > msgbuf_len) + /* There are some fragments left. Shrink the msg_buffer to simplify logic. */ + vec_delete (msg_buffer, msgbuf_len, 0); else + /* We are done with msg_buffer. */ _vec_len (msg_buffer) = 0; - n -= msg_len + sizeof (msgbuf_t); - msg_len = 0; - mbp_set = 0; } - while (n > 0); + while (vec_len (msg_buffer) > 0); + /* Restore input_buffer, it could have been msg_buffer. */ _vec_len (socket_main.input_buffer) = save_input_buffer_length; - return 0; } @@ -426,7 +445,7 @@ vl_api_sockclnt_create_t_handler (vl_api_sockclnt_create_t * mp) { vl_api_registration_t *regp; vl_api_sockclnt_create_reply_t *rp; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main (); hash_pair_t *hp; int rv = 0; u32 nmsg = hash_elts (am->msg_index_by_name_and_crc); @@ -439,7 +458,7 @@ vl_api_sockclnt_create_t_handler (vl_api_sockclnt_create_t * mp) regp->name = format (0, "%s%c", mp->name, 0); u32 size = sizeof (*rp) + (nmsg * sizeof (vl_api_message_table_entry_t)); - rp = vl_msg_api_alloc (size); + rp = vl_msg_api_alloc_zero (size); rp->_vl_msg_id = htons (VL_API_SOCKCLNT_CREATE_REPLY); rp->index = htonl (sock_api_registration_handle (regp)); rp->context = mp->context; @@ -450,7 +469,10 @@ vl_api_sockclnt_create_t_handler (vl_api_sockclnt_create_t * mp) hash_foreach_pair (hp, am->msg_index_by_name_and_crc, ({ rp->message_table[i].index = htons(hp->value[0]); - strncpy((char *)rp->message_table[i].name, (char *)hp->key, 64-1); + (void) strncpy_s((char *)rp->message_table[i].name, + 64 /* bytes of space at dst */, + (char *)hp->key, + 64-1 /* chars to copy, without zero byte. */); i++; })); /* *INDENT-ON* */ @@ -470,7 +492,7 @@ vl_api_sockclnt_delete_t_handler (vl_api_sockclnt_delete_t * mp) if (!regp) return; - u32 reg_index = ntohl (mp->index); + u32 reg_index = socket_api_registration_handle_to_index (ntohl (mp->index)); rp = vl_msg_api_alloc (sizeof (*rp)); rp->_vl_msg_id = htons (VL_API_SOCKCLNT_DELETE_REPLY); rp->context = mp->context; @@ -586,12 +608,12 @@ vl_api_sock_init_shm_t_handler (vl_api_sock_init_shm_t * mp) ssvm_private_t _memfd_private, *memfd = &_memfd_private; svm_map_region_args_t _args, *a = &_args; vl_api_registration_t *regp; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main (); svm_region_t *vlib_rp; clib_file_t *cf; vl_api_shm_elem_config_t *config = 0; vl_shmem_hdr_t *shmem_hdr; - int rv; + int rv, tries = 1000; regp = vl_api_client_index_to_registration (mp->client_index); if (regp == 0) @@ -671,13 +693,29 @@ reply: /* Send the magic "here's your sign (aka fd)" socket message */ cf = vl_api_registration_file (regp); + + /* Wait for reply to be consumed before sending the fd */ + while (tries-- > 0) + { + int bytes; + rv = ioctl (cf->file_descriptor, TIOCOUTQ, &bytes); + if (rv < 0) + { + clib_unix_warning ("ioctl returned"); + break; + } + if (bytes == 0) + break; + usleep (1e3); + } + vl_sock_api_send_fd_msg (cf->file_descriptor, &memfd->fd, 1); } #define foreach_vlib_api_msg \ -_(SOCKCLNT_CREATE, sockclnt_create) \ -_(SOCKCLNT_DELETE, sockclnt_delete) \ -_(SOCK_INIT_SHM, sock_init_shm) + _(SOCKCLNT_CREATE, sockclnt_create, 1) \ + _(SOCKCLNT_DELETE, sockclnt_delete, 1) \ + _(SOCK_INIT_SHM, sock_init_shm, 1) clib_error_t * vl_sock_api_init (vlib_main_t * vm) @@ -693,37 +731,20 @@ vl_sock_api_init (vlib_main_t * vm) if (sm->socket_name == 0) return 0; -#define _(N,n) \ +#define _(N,n,t) \ vl_msg_api_set_handlers(VL_API_##N, #n, \ vl_api_##n##_t_handler, \ vl_noop_handler, \ vl_api_##n##_t_endian, \ vl_api_##n##_t_print, \ - sizeof(vl_api_##n##_t), 1); + sizeof(vl_api_##n##_t), t); foreach_vlib_api_msg; #undef _ vec_resize (sm->input_buffer, 4096); sock->config = (char *) sm->socket_name; - - /* mkdir of file socket, only under /run */ - if (strncmp (sock->config, "/run", 4) == 0) - { - u8 *tmp = format (0, "%s", sock->config); - int i = vec_len (tmp); - while (i && tmp[--i] != '/') - ; - - tmp[i] = 0; - - if (i) - vlib_unix_recursive_mkdir ((char *) tmp); - vec_free (tmp); - } - - sock->flags = CLIB_SOCKET_F_IS_SERVER | CLIB_SOCKET_F_SEQPACKET | - CLIB_SOCKET_F_ALLOW_GROUP_WRITE; + sock->flags = CLIB_SOCKET_F_IS_SERVER | CLIB_SOCKET_F_ALLOW_GROUP_WRITE; error = clib_socket_init (sock); if (error) return error; @@ -775,29 +796,31 @@ socksvr_config (vlib_main_t * vm, unformat_input_t * input) { if (unformat (input, "socket-name %s", &sm->socket_name)) ; + /* DEPRECATE: default keyword is ignored */ else if (unformat (input, "default")) - { - sm->socket_name = format (0, "%s%c", API_SOCKET_FILE, 0); - } + ; else { return clib_error_return (0, "unknown input '%U'", format_unformat_error, input); } } + + if (!vec_len (sm->socket_name)) + sm->socket_name = format (0, "%s/%s", vlib_unix_get_runtime_dir (), + API_SOCKET_FILENAME); + vec_terminate_c_string (sm->socket_name); + return 0; } VLIB_CONFIG_FUNCTION (socksvr_config, "socksvr"); -clib_error_t * -vlibsocket_init (vlib_main_t * vm) +void +vlibsocket_reference () { - return 0; } -VLIB_INIT_FUNCTION (vlibsocket_init); - /* * fd.io coding-style-patch-verification: ON *