X-Git-Url: https://gerrit.fd.io/r/gitweb?p=vpp.git;a=blobdiff_plain;f=src%2Fvpp-api%2Fclient%2Fclient.c;h=7a30792402c70370dbb966aad64db0217aa909b3;hp=8bdcda01c4d15d34a2b31be9a0fbb7567f897cf5;hb=36217e3ca;hpb=5fec1e8b2282f4d3d1d02556020254a84c3b6e3d diff --git a/src/vpp-api/client/client.c b/src/vpp-api/client/client.c index 8bdcda01c4d..7a30792402c 100644 --- a/src/vpp-api/client/client.c +++ b/src/vpp-api/client/client.c @@ -12,6 +12,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +#include #include #include #include @@ -29,10 +30,15 @@ #include #include -#include +#include +#include #include "vppapiclient.h" +bool timeout_cancelled; +bool timeout_in_progress; +bool rx_thread_done; + /* * Asynchronous mode: * Client registers a callback. All messages are sent to the callback. @@ -43,17 +49,6 @@ * vac_read() -> resumes RX thread */ -#define vl_typedefs /* define message structures */ -#include -#undef vl_typedefs - -#define vl_endianfun /* define message structures */ -#include -#undef vl_endianfun - -vlib_main_t vlib_global_main; -vlib_main_t **vlib_mains; - typedef struct { u8 connected_to_vlib; pthread_t rx_thread_handle; @@ -62,6 +57,7 @@ typedef struct { pthread_cond_t suspend_cv; pthread_cond_t resume_cv; pthread_mutex_t timeout_lock; + u8 timeout_loop; pthread_cond_t timeout_cv; pthread_cond_t timeout_cancel_cv; pthread_cond_t terminate_cv; @@ -71,16 +67,21 @@ vac_main_t vac_main; vac_callback_t vac_callback; u16 read_timeout = 0; bool rx_is_running = false; +bool timeout_thread_cancelled = false; + +/* Only ever allocate one heap */ +bool mem_initialized = false; static void init (void) { vac_main_t *pm = &vac_main; - memset(pm, 0, sizeof(*pm)); + clib_memset(pm, 0, sizeof(*pm)); pthread_mutex_init(&pm->queue_lock, NULL); pthread_cond_init(&pm->suspend_cv, NULL); pthread_cond_init(&pm->resume_cv, NULL); pthread_mutex_init(&pm->timeout_lock, NULL); + pm->timeout_loop = 1; pthread_cond_init(&pm->timeout_cv, NULL); pthread_cond_init(&pm->timeout_cancel_cv, NULL); pthread_cond_init(&pm->terminate_cv, NULL); @@ -90,22 +91,14 @@ static void cleanup (void) { vac_main_t *pm = &vac_main; + pthread_mutex_destroy(&pm->queue_lock); pthread_cond_destroy(&pm->suspend_cv); pthread_cond_destroy(&pm->resume_cv); + pthread_mutex_destroy(&pm->timeout_lock); pthread_cond_destroy(&pm->timeout_cv); pthread_cond_destroy(&pm->timeout_cancel_cv); pthread_cond_destroy(&pm->terminate_cv); - pthread_mutex_destroy(&pm->queue_lock); - pthread_mutex_destroy(&pm->timeout_lock); - memset (pm, 0, sizeof (*pm)); -} - -/* - * Satisfy external references when -lvlib is not available. - */ -void vlib_cli_output (struct vlib_main_t * vm, char * fmt, ...) -{ - clib_warning ("vlib_cli_output called..."); + clib_memset(pm, 0, sizeof(*pm)); } void @@ -132,22 +125,27 @@ vac_api_handler (void *msg) static void * vac_rx_thread_fn (void *arg) { - unix_shared_memory_queue_t *q; + svm_queue_t *q; + vl_api_memclnt_keepalive_t *mp; + vl_api_memclnt_keepalive_reply_t *rmp; vac_main_t *pm = &vac_main; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); + vl_shmem_hdr_t *shmem_hdr; uword msg; q = am->vl_input_queue; while (1) - while (!unix_shared_memory_queue_sub(q, (u8 *)&msg, 0)) + while (!svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0)) { + VL_MSG_API_UNPOISON((void *)msg); u16 id = ntohs(*((u16 *)msg)); switch (id) { case VL_API_RX_THREAD_EXIT: vl_msg_api_free((void *) msg); /* signal waiting threads that this thread is about to terminate */ pthread_mutex_lock(&pm->queue_lock); + rx_thread_done = true; pthread_cond_signal(&pm->terminate_cv); pthread_mutex_unlock(&pm->queue_lock); pthread_exit(0); @@ -169,6 +167,17 @@ vac_rx_thread_fn (void *arg) vl_msg_api_free((void *) msg); break; + case VL_API_MEMCLNT_KEEPALIVE: + mp = (void *)msg; + rmp = vl_msg_api_alloc (sizeof (*rmp)); + clib_memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs(VL_API_MEMCLNT_KEEPALIVE_REPLY); + rmp->context = mp->context; + shmem_hdr = am->shmem_hdr; + vl_msg_api_send_shmem(shmem_hdr->vl_input_queue, (u8 *)&rmp); + vl_msg_api_free((void *) msg); + break; + default: vac_api_handler((void *)msg); } @@ -180,30 +189,34 @@ vac_timeout_thread_fn (void *arg) { vl_api_memclnt_read_timeout_t *ep; vac_main_t *pm = &vac_main; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); struct timespec ts; struct timeval tv; - u16 timeout; int rv; - while (1) + while (pm->timeout_loop) { /* Wait for poke */ pthread_mutex_lock(&pm->timeout_lock); - pthread_cond_wait (&pm->timeout_cv, &pm->timeout_lock); - timeout = read_timeout; + while (!timeout_in_progress) + pthread_cond_wait (&pm->timeout_cv, &pm->timeout_lock); + + /* Starting timer */ gettimeofday(&tv, NULL); - ts.tv_sec = tv.tv_sec + timeout; + ts.tv_sec = tv.tv_sec + read_timeout; ts.tv_nsec = 0; - rv = pthread_cond_timedwait (&pm->timeout_cancel_cv, - &pm->timeout_lock, &ts); - pthread_mutex_unlock(&pm->timeout_lock); - if (rv == ETIMEDOUT) - { + + if (!timeout_cancelled) { + rv = pthread_cond_timedwait (&pm->timeout_cancel_cv, + &pm->timeout_lock, &ts); + if (rv == ETIMEDOUT && !timeout_thread_cancelled) { ep = vl_msg_api_alloc (sizeof (*ep)); ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_READ_TIMEOUT); vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep); } + } + + pthread_mutex_unlock(&pm->timeout_lock); } pthread_exit(0); } @@ -211,7 +224,7 @@ vac_timeout_thread_fn (void *arg) void vac_rx_suspend (void) { - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); vac_main_t *pm = &vac_main; vl_api_memclnt_rx_thread_suspend_t *ep; @@ -222,7 +235,7 @@ vac_rx_suspend (void) ep = vl_msg_api_alloc (sizeof (*ep)); ep->_vl_msg_id = ntohs(VL_API_MEMCLNT_RX_THREAD_SUSPEND); vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep); - /* Wait for RX thread to tell us it has suspendend */ + /* Wait for RX thread to tell us it has suspended */ pthread_cond_wait(&pm->suspend_cv, &pm->queue_lock); rx_is_running = false; } @@ -245,30 +258,32 @@ vac_rx_resume (void) static uword * vac_msg_table_get_hash (void) { - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); return (am->msg_index_by_name_and_crc); } int vac_msg_table_size(void) { - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); return hash_elts(am->msg_index_by_name_and_crc); } int vac_connect (char * name, char * chroot_prefix, vac_callback_t cb, - int rx_qlen) + int rx_qlen) { + rx_thread_done = false; int rv = 0; vac_main_t *pm = &vac_main; + assert (clib_mem_get_heap ()); init(); if (chroot_prefix != NULL) vl_set_memory_root_path (chroot_prefix); if ((rv = vl_client_api_map("/vpe-api"))) { - clib_warning ("vl_client_api map rv %d", rv); + clib_warning ("vl_client_api_map returned %d", rv); return rv; } @@ -302,18 +317,41 @@ vac_connect (char * name, char * chroot_prefix, vac_callback_t cb, return (0); } +static void +set_timeout (unsigned short timeout) +{ + vac_main_t *pm = &vac_main; + pthread_mutex_lock(&pm->timeout_lock); + read_timeout = timeout; + timeout_in_progress = true; + timeout_cancelled = false; + pthread_cond_signal(&pm->timeout_cv); + pthread_mutex_unlock(&pm->timeout_lock); +} + +static void +unset_timeout (void) +{ + vac_main_t *pm = &vac_main; + pthread_mutex_lock(&pm->timeout_lock); + timeout_in_progress = false; + timeout_cancelled = true; + pthread_cond_signal(&pm->timeout_cancel_cv); + pthread_mutex_unlock(&pm->timeout_lock); +} int vac_disconnect (void) { - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); vac_main_t *pm = &vac_main; + uword junk; + int rv = 0; if (!pm->connected_to_vlib) return 0; if (pm->rx_thread_handle) { vl_api_rx_thread_exit_t *ep; - uword junk; ep = vl_msg_api_alloc (sizeof (*ep)); ep->_vl_msg_id = ntohs(VL_API_RX_THREAD_EXIT); vl_msg_api_send_shmem(am->vl_input_queue, (u8 *)&ep); @@ -324,137 +362,157 @@ vac_disconnect (void) gettimeofday(&tv, NULL); ts.tv_sec = tv.tv_sec + 5; ts.tv_nsec = 0; + pthread_mutex_lock(&pm->queue_lock); - int rv = pthread_cond_timedwait(&pm->terminate_cv, &pm->queue_lock, &ts); + if (rx_thread_done == false) + rv = pthread_cond_timedwait(&pm->terminate_cv, &pm->queue_lock, &ts); pthread_mutex_unlock(&pm->queue_lock); + /* now join so we wait until thread has -really- finished */ if (rv == ETIMEDOUT) pthread_cancel(pm->rx_thread_handle); else pthread_join(pm->rx_thread_handle, (void **) &junk); } - if (pm->timeout_thread_handle) - pthread_cancel(pm->timeout_thread_handle); + if (pm->timeout_thread_handle) { + /* cancel, wake then join the timeout thread */ + pm->timeout_loop = 0; + timeout_thread_cancelled = true; + set_timeout(0); + pthread_join(pm->timeout_thread_handle, (void **) &junk); + } vl_client_disconnect(); vl_client_api_unmap(); - vac_callback = 0; + //vac_callback = 0; cleanup(); return (0); } -static void -set_timeout (unsigned short timeout) -{ - vac_main_t *pm = &vac_main; - pthread_mutex_lock(&pm->timeout_lock); - read_timeout = timeout; - pthread_cond_signal(&pm->timeout_cv); - pthread_mutex_unlock(&pm->timeout_lock); -} - -static void -unset_timeout (void) -{ - vac_main_t *pm = &vac_main; - pthread_mutex_lock(&pm->timeout_lock); - pthread_cond_signal(&pm->timeout_cancel_cv); - pthread_mutex_unlock(&pm->timeout_lock); -} - int vac_read (char **p, int *l, u16 timeout) { - unix_shared_memory_queue_t *q; - api_main_t *am = &api_main; + svm_queue_t *q; + api_main_t *am = vlibapi_get_main(); vac_main_t *pm = &vac_main; + vl_api_memclnt_keepalive_t *mp; + vl_api_memclnt_keepalive_reply_t *rmp; uword msg; msgbuf_t *msgbuf; + int rv; + vl_shmem_hdr_t *shmem_hdr; - if (!pm->connected_to_vlib) return -1; + /* svm_queue_sub(below) returns {-1, -2} */ + if (!pm->connected_to_vlib) + return VAC_NOT_CONNECTED; *l = 0; - if (am->our_pid == 0) return (-1); + /* svm_queue_sub(below) returns {-1, -2} */ + if (am->our_pid == 0) + return (VAC_SHM_NOT_READY); /* Poke timeout thread */ if (timeout) set_timeout(timeout); q = am->vl_input_queue; - int rv = unix_shared_memory_queue_sub(q, (u8 *)&msg, 0); + + again: + rv = svm_queue_sub(q, (u8 *)&msg, SVM_Q_WAIT, 0); + if (rv == 0) { + VL_MSG_API_UNPOISON((void *)msg); u16 msg_id = ntohs(*((u16 *)msg)); switch (msg_id) { case VL_API_RX_THREAD_EXIT: - printf("Received thread exit\n"); - return -1; + vl_msg_api_free((void *) msg); + goto error; case VL_API_MEMCLNT_RX_THREAD_SUSPEND: - printf("Received thread suspend\n"); goto error; case VL_API_MEMCLNT_READ_TIMEOUT: - printf("Received read timeout %ds\n", timeout); goto error; + case VL_API_MEMCLNT_KEEPALIVE: + /* Handle an alive-check ping from vpp. */ + mp = (void *)msg; + rmp = vl_msg_api_alloc (sizeof (*rmp)); + clib_memset (rmp, 0, sizeof (*rmp)); + rmp->_vl_msg_id = ntohs(VL_API_MEMCLNT_KEEPALIVE_REPLY); + rmp->context = mp->context; + shmem_hdr = am->shmem_hdr; + vl_msg_api_send_shmem(shmem_hdr->vl_input_queue, (u8 *)&rmp); + vl_msg_api_free((void *) msg); + /* + * Python code is blissfully unaware of these pings, so + * act as if it never happened... + */ + goto again; default: msgbuf = (msgbuf_t *)(((u8 *)msg) - offsetof(msgbuf_t, data)); *l = ntohl(msgbuf->data_len); if (*l == 0) { - printf("Unregistered API message: %d\n", msg_id); + fprintf(stderr, "Unregistered API message: %d\n", msg_id); goto error; } } *p = (char *)msg; - /* Let timeout notification thread know we're done */ - unset_timeout(); } else { - printf("Read failed with %d\n", rv); + fprintf(stderr, "Read failed with %d\n", rv); } + /* Let timeout notification thread know we're done */ + if (timeout) + unset_timeout(); + return (rv); error: + if (timeout) + unset_timeout(); vl_msg_api_free((void *) msg); /* Client might forget to resume RX thread on failure */ vac_rx_resume (); - return -1; + return VAC_TIMEOUT; } /* * XXX: Makes the assumption that client_index is the first member */ -typedef VL_API_PACKED(struct _vl_api_header { +typedef struct _vl_api_header +{ u16 _vl_msg_id; u32 client_index; -}) vl_api_header_t; +} __attribute__ ((packed)) vl_api_header_t; -static unsigned int +static u32 vac_client_index (void) { - return (api_main.my_client_index); + return (vlibapi_get_main()->my_client_index); } int vac_write (char *p, int l) { int rv = -1; - api_main_t *am = &api_main; + api_main_t *am = vlibapi_get_main(); vl_api_header_t *mp = vl_msg_api_alloc(l); - unix_shared_memory_queue_t *q; + svm_queue_t *q; vac_main_t *pm = &vac_main; - if (!pm->connected_to_vlib) return -1; + if (!pm->connected_to_vlib) + return VAC_NOT_CONNECTED; if (!mp) return (-1); memcpy(mp, p, l); mp->client_index = vac_client_index(); q = am->shmem_hdr->vl_input_queue; - rv = unix_shared_memory_queue_add(q, (u8 *)&mp, 0); + rv = svm_queue_add(q, (u8 *)&mp, 0); if (rv != 0) { - clib_warning("vpe_api_write fails: %d\n", rv); + fprintf(stderr, "vpe_api_write fails: %d\n", rv); /* Clear message */ vac_free(mp); } @@ -462,9 +520,9 @@ vac_write (char *p, int l) } int -vac_get_msg_index (unsigned char * name) +vac_get_msg_index (char * name) { - return vl_api_get_msg_index (name); + return vl_msg_api_get_msg_index ((u8 *)name); } int @@ -485,5 +543,21 @@ vac_msg_table_max_index(void) void vac_set_error_handler (vac_error_callback_t cb) { + assert (clib_mem_get_heap ()); if (cb) clib_error_register_handler (cb, 0); } + +/* + * Required if application doesn't use a VPP heap. + */ +void +vac_mem_init (size_t size) +{ + if (mem_initialized) + return; + if (size == 0) + clib_mem_init (0, 1 << 30); // default + else + clib_mem_init (0, size); + mem_initialized = true; +}