X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Futils_ovs.c;h=0ee05e0044344fe818b135d65631a49240f1a674;hp=57da628e33c722b416435c75d9979d16e219cf1d;hb=a9e50e9e30ecde17e167e271060c8183bfcbf407;hpb=2412d3ca76f7ac24e67543720c1d385188da0ce3 diff --git a/src/utils_ovs.c b/src/utils_ovs.c index 57da628e..0ee05e00 100644 --- a/src/utils_ovs.c +++ b/src/utils_ovs.c @@ -3,14 +3,17 @@ * * Copyright(c) 2016 Intel Corporation. All rights reserved. * - * Permission is hereby granted, free of charge, to any person obtaining a copy of + * Permission is hereby granted, free of charge, to any person obtaining a copy + *of * this software and associated documentation files (the "Software"), to deal in * the Software without restriction, including without limitation the rights to * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies - * of the Software, and to permit persons to whom the Software is furnished to do + * of the Software, and to permit persons to whom the Software is furnished to + *do * so, subject to the following conditions: * - * The above copyright notice and this permission notice shall be included in all + * The above copyright notice and this permission notice shall be included in + *all * copies or substantial portions of the Software. * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR @@ -102,6 +105,7 @@ #define OVS_DB_POLL_READ_BLOCK_SIZE 512 /* read block size (bytes) */ #define OVS_DB_DEFAULT_DB_NAME "Open_vSwitch" +#define OVS_DB_EVENT_NONE 0 #define OVS_DB_EVENT_TIMEOUT 5 /* event thread timeout (sec) */ #define OVS_DB_EVENT_TERMINATE 1 #define OVS_DB_EVENT_CONN_ESTABLISHED 2 @@ -210,7 +214,7 @@ static _Bool ovs_db_poll_is_running(ovs_db_t *pdb) { pthread_mutex_lock(&pdb->poll_thread.mutex); state = pdb->poll_thread.state; pthread_mutex_unlock(&pdb->poll_thread.mutex); - return (state == OVS_DB_POLL_STATE_RUNNING); + return state == OVS_DB_POLL_STATE_RUNNING; } /* Generate unique identifier (UID). It is used by OVS DB API @@ -260,12 +264,11 @@ static void ovs_db_callback_remove(ovs_db_t *pdb, ovs_callback_t *del_cb) { /* Remove all callbacks form OVS DB object */ static void ovs_db_callback_remove_all(ovs_db_t *pdb) { pthread_mutex_lock(&pdb->mutex); - for (ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb; - del_cb = pdb->remote_cb) { + while (pdb->remote_cb != NULL) { + ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb = del_cb->next; - free(del_cb); + sfree(del_cb); } - pdb->remote_cb = NULL; pthread_mutex_unlock(&pdb->mutex); } @@ -293,11 +296,11 @@ static int ovs_db_data_send(const ovs_db_t *pdb, const char *data, size_t len) { while (rem > 0) { if ((nbytes = send(pdb->sock, data + off, rem, 0)) <= 0) - return (-1); + return -1; rem -= (size_t)nbytes; off += (size_t)nbytes; } - return (0); + return 0; } /* @@ -330,6 +333,9 @@ static yajl_gen_status ovs_yajl_gen_val(yajl_gen jgen, yajl_val jval) { size_t obj_len = 0; yajl_gen_status yajl_gen_ret = yajl_gen_status_ok; + if (jval == NULL) + return yajl_gen_generation_complete; + if (YAJL_IS_STRING(jval)) OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, YAJL_GET_STRING(jval)); else if (YAJL_IS_DOUBLE(jval)) @@ -388,7 +394,7 @@ static int ovs_db_table_echo_cb(const ovs_db_t *pdb, yajl_val jnode) { yajl_gen_status yajl_gen_ret; if ((jgen = yajl_gen_alloc(NULL)) == NULL) - return (-1); + return -1; /* check & get request attributes */ if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL || @@ -421,12 +427,12 @@ static int ovs_db_table_echo_cb(const ovs_db_t *pdb, yajl_val jnode) { } /* clean up and return success */ yajl_gen_clear(jgen); - return (0); + return 0; yajl_gen_failure: /* release memory */ yajl_gen_clear(jgen); - return (-1); + return -1; } /* Get OVS DB registered callback by YAJL val. The YAJL @@ -466,21 +472,21 @@ static int ovs_db_table_update_cb(ovs_db_t *pdb, yajl_val jnode) { if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL || (yajl_tree_get(jnode, id_path, yajl_t_null) == NULL)) { OVS_ERROR("invalid OVS DB request received"); - return (-1); + return -1; } /* check array length: [, ] */ if ((YAJL_GET_ARRAY(jparams) == NULL) || (YAJL_GET_ARRAY(jparams)->len != 2)) { OVS_ERROR("invalid OVS DB request received"); - return (-1); + return -1; } jvalue = YAJL_GET_ARRAY(jparams)->values[0]; jtable_updates = YAJL_GET_ARRAY(jparams)->values[1]; if ((!YAJL_IS_OBJECT(jtable_updates)) || (!YAJL_IS_STRING(jvalue))) { OVS_ERROR("invalid OVS DB request id or table update received"); - return (-1); + return -1; } /* find registered callback based on */ @@ -489,7 +495,7 @@ static int ovs_db_table_update_cb(ovs_db_t *pdb, yajl_val jnode) { if (cb == NULL || cb->table.call == NULL) { OVS_ERROR("No OVS DB table update callback found"); pthread_mutex_unlock(&pdb->mutex); - return (-1); + return -1; } /* call registered callback */ @@ -518,7 +524,7 @@ static int ovs_db_result_cb(ovs_db_t *pdb, yajl_val jnode) { /* check & get result attributes */ if (!jresult || !jerror || !jid) - return (-1); + return -1; /* try to find registered callback */ pthread_mutex_lock(&pdb->mutex); @@ -531,7 +537,7 @@ static int ovs_db_result_cb(ovs_db_t *pdb, yajl_val jnode) { } pthread_mutex_unlock(&pdb->mutex); - return (0); + return 0; } /* Handle JSON data (one request) and call @@ -551,22 +557,26 @@ static int ovs_db_json_data_process(ovs_db_t *pdb, const char *data, /* duplicate the data to make null-terminated string * required for yajl_tree_parse() */ if ((sjson = calloc(1, len + 1)) == NULL) - return (-1); + return -1; sstrncpy(sjson, data, len + 1); - OVS_DEBUG("[len=%zu] %s", len, sjson); + OVS_DEBUG("[len=%" PRIsz "] %s", len, sjson); /* parse json data */ jnode = yajl_tree_parse(sjson, yajl_errbuf, sizeof(yajl_errbuf)); if (jnode == NULL) { OVS_ERROR("yajl_tree_parse() %s", yajl_errbuf); sfree(sjson); - return (-1); + return -1; } /* get method name */ if ((jval = yajl_tree_get(jnode, method_path, yajl_t_string)) != NULL) { - method = YAJL_GET_STRING(jval); + if ((method = YAJL_GET_STRING(jval)) == NULL) { + yajl_tree_free(jnode); + sfree(sjson); + return -1; + } if (strcmp("echo", method) == 0) { /* echo request from the server */ if (ovs_db_table_echo_cb(pdb, jnode) < 0) @@ -586,7 +596,7 @@ static int ovs_db_json_data_process(ovs_db_t *pdb, const char *data, /* release memory */ yajl_tree_free(jnode); sfree(sjson); - return (0); + return 0; } /* @@ -621,7 +631,7 @@ static int ovs_json_reader_push_data(ovs_json_reader_t *jreader, /* allocate new chunk of memory */ new_buff = realloc(jreader->buff_ptr, (jreader->buff_size + data_len)); if (new_buff == NULL) - return (-1); + return -1; /* point to new allocated memory */ jreader->buff_ptr = new_buff; @@ -631,7 +641,7 @@ static int ovs_json_reader_push_data(ovs_json_reader_t *jreader, /* store input data */ memcpy(jreader->buff_ptr + jreader->buff_offset, data, data_len); jreader->buff_offset += data_len; - return (0); + return 0; } /* Pop one fully-fledged JSON if already exists. Returns 0 if @@ -654,7 +664,7 @@ static int ovs_json_reader_pop(ovs_json_reader_t *jreader, *json_ptr = jreader->buff_ptr + jreader->json_offset; *json_len_ptr = json_len + 1; jreader->json_offset = i + 1; - return (0); + return 0; } /* increase JSON data length */ @@ -679,7 +689,7 @@ static int ovs_json_reader_pop(ovs_json_reader_t *jreader, jreader->json_offset = 0; } - return (-1); + return -1; } /* Reset JSON reader. It is useful when start processing @@ -741,21 +751,18 @@ static void ovs_db_reconnect(ovs_db_t *pdb) { } /* try to connect to the server */ for (struct addrinfo *rp = result; rp != NULL; rp = rp->ai_next) { - char errbuff[OVS_ERROR_BUFF_SIZE]; int sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); if (sock < 0) { - sstrerror(errno, errbuff, sizeof(errbuff)); - OVS_DEBUG("socket(): %s", errbuff); + OVS_DEBUG("socket(): %s", STRERRNO); continue; } if (connect(sock, rp->ai_addr, rp->ai_addrlen) < 0) { close(sock); - sstrerror(errno, errbuff, sizeof(errbuff)); - OVS_DEBUG("connect(): %s [family=%d]", errbuff, rp->ai_family); + OVS_DEBUG("connect(): %s [family=%d]", STRERRNO, rp->ai_family); } else { /* send notification to event thread */ - ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED); pdb->sock = sock; + ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED); break; } } @@ -781,17 +788,15 @@ static void *ovs_poll_worker(void *arg) { /* create JSON reader instance */ if ((jreader = ovs_json_reader_alloc()) == NULL) { OVS_ERROR("initialize json reader failed"); - return (NULL); + return NULL; } /* poll data */ while (ovs_db_poll_is_running(pdb)) { - char errbuff[OVS_ERROR_BUFF_SIZE]; poll_fd.fd = pdb->sock; int poll_ret = poll(&poll_fd, 1, /* ms */ OVS_DB_POLL_TIMEOUT * 1000); if (poll_ret < 0) { - sstrerror(errno, errbuff, sizeof(errbuff)); - OVS_ERROR("poll(): %s", errbuff); + OVS_ERROR("poll(): %s", STRERRNO); break; } else if (poll_ret == 0) { OVS_DEBUG("poll(): timeout"); @@ -817,8 +822,7 @@ static void *ovs_poll_worker(void *arg) { char buff[OVS_DB_POLL_READ_BLOCK_SIZE]; ssize_t nbytes = recv(poll_fd.fd, buff, sizeof(buff), 0); if (nbytes < 0) { - sstrerror(errno, errbuff, sizeof(errbuff)); - OVS_ERROR("recv(): %s", errbuff); + OVS_ERROR("recv(): %s", STRERRNO); /* read error? Try to reconnect */ close(poll_fd.fd); continue; @@ -841,7 +845,7 @@ static void *ovs_poll_worker(void *arg) { OVS_DEBUG("poll thread has been completed"); ovs_json_reader_free(jreader); - return (NULL); + return NULL; } /* EVENT worker thread. @@ -859,26 +863,30 @@ static void *ovs_event_worker(void *arg) { ts.tv_sec += (OVS_DB_EVENT_TIMEOUT); int ret = pthread_cond_timedwait(&pdb->event_thread.cond, &pdb->event_thread.mutex, &ts); - if (!ret) { + if (!ret || ret == ETIMEDOUT) { /* handle the event */ OVS_DEBUG("handle event %d", pdb->event_thread.value); switch (pdb->event_thread.value) { case OVS_DB_EVENT_CONN_ESTABLISHED: if (pdb->cb.post_conn_init) pdb->cb.post_conn_init(pdb); + /* reset event */ + pdb->event_thread.value = OVS_DB_EVENT_NONE; break; case OVS_DB_EVENT_CONN_TERMINATED: if (pdb->cb.post_conn_terminate) pdb->cb.post_conn_terminate(); + /* reset event */ + pdb->event_thread.value = OVS_DB_EVENT_NONE; + break; + case OVS_DB_EVENT_NONE: + /* wait timeout */ + OVS_DEBUG("no event received (timeout)"); break; default: OVS_DEBUG("unknown event received"); break; } - } else if (ret == ETIMEDOUT) { - /* wait timeout */ - OVS_DEBUG("no event received (timeout)"); - continue; } else { /* unexpected error */ OVS_ERROR("pthread_cond_timedwait() failed"); @@ -887,20 +895,20 @@ static void *ovs_event_worker(void *arg) { } OVS_DEBUG("event thread has been completed"); - return (NULL); + return NULL; } /* Initialize EVENT thread */ static int ovs_db_event_thread_init(ovs_db_t *pdb) { - pdb->event_thread.tid = (pthread_t)-1; + pdb->event_thread.tid = (pthread_t){0}; /* init event thread condition variable */ if (pthread_cond_init(&pdb->event_thread.cond, NULL)) { - return (-1); + return -1; } /* init event thread mutex */ if (pthread_mutex_init(&pdb->event_thread.mutex, NULL)) { pthread_cond_destroy(&pdb->event_thread.cond); - return (-1); + return -1; } /* Hold the event thread mutex. It ensures that no events * will be lost while thread is still starting. Once event @@ -909,7 +917,7 @@ static int ovs_db_event_thread_init(ovs_db_t *pdb) { if (pthread_mutex_lock(&pdb->event_thread.mutex)) { pthread_mutex_destroy(&pdb->event_thread.mutex); pthread_cond_destroy(&pdb->event_thread.cond); - return (-1); + return -1; } /* start event thread */ pthread_t tid; @@ -918,37 +926,43 @@ static int ovs_db_event_thread_init(ovs_db_t *pdb) { pthread_mutex_unlock(&pdb->event_thread.mutex); pthread_mutex_destroy(&pdb->event_thread.mutex); pthread_cond_destroy(&pdb->event_thread.cond); - return (-1); + return -1; } pdb->event_thread.tid = tid; - return (0); + return 0; } -/* Destroy EVENT thread */ -static int ovs_db_event_thread_destroy(ovs_db_t *pdb) { - if (pdb->event_thread.tid == (pthread_t)-1) - /* already destroyed */ - return (0); +/* Terminate EVENT thread */ +static int ovs_db_event_thread_terminate(ovs_db_t *pdb) { + if (pthread_equal(pdb->event_thread.tid, (pthread_t){0})) { + /* already terminated */ + return 0; + } ovs_db_event_post(pdb, OVS_DB_EVENT_TERMINATE); if (pthread_join(pdb->event_thread.tid, NULL) != 0) - return (-1); + return -1; /* Event thread always holds the thread mutex when * performs some task (handles event) and releases it when * while sleeping. Thus, if event thread exits, the mutex * remains locked */ + pdb->event_thread.tid = (pthread_t){0}; pthread_mutex_unlock(&pdb->event_thread.mutex); + return 0; +} + +/* Destroy EVENT thread private data */ +static void ovs_db_event_thread_data_destroy(ovs_db_t *pdb) { + /* destroy mutex */ pthread_mutex_destroy(&pdb->event_thread.mutex); pthread_cond_destroy(&pdb->event_thread.cond); - pdb->event_thread.tid = (pthread_t)-1; - return (0); } /* Initialize POLL thread */ static int ovs_db_poll_thread_init(ovs_db_t *pdb) { - pdb->poll_thread.tid = (pthread_t)-1; + pdb->poll_thread.tid = (pthread_t){0}; /* init event thread mutex */ if (pthread_mutex_init(&pdb->poll_thread.mutex, NULL)) { - return (-1); + return -1; } /* start poll thread */ pthread_t tid; @@ -956,27 +970,29 @@ static int ovs_db_poll_thread_init(ovs_db_t *pdb) { if (plugin_thread_create(&tid, NULL, ovs_poll_worker, pdb, "utils_ovs:poll") != 0) { pthread_mutex_destroy(&pdb->poll_thread.mutex); - return (-1); + return -1; } pdb->poll_thread.tid = tid; - return (0); + return 0; } /* Destroy POLL thread */ +/* XXX: Must hold pdb->mutex when calling! */ static int ovs_db_poll_thread_destroy(ovs_db_t *pdb) { - if (pdb->poll_thread.tid == (pthread_t)-1) + if (pthread_equal(pdb->poll_thread.tid, (pthread_t){0})) { /* already destroyed */ - return (0); + return 0; + } /* change thread state */ pthread_mutex_lock(&pdb->poll_thread.mutex); pdb->poll_thread.state = OVS_DB_POLL_STATE_EXITING; pthread_mutex_unlock(&pdb->poll_thread.mutex); /* join the thread */ if (pthread_join(pdb->poll_thread.tid, NULL) != 0) - return (-1); + return -1; pthread_mutex_destroy(&pdb->poll_thread.mutex); - pdb->poll_thread.tid = (pthread_t)-1; - return (0); + pdb->poll_thread.tid = (pthread_t){0}; + return 0; } /* @@ -987,12 +1003,13 @@ ovs_db_t *ovs_db_init(const char *node, const char *service, const char *unix_path, ovs_db_callback_t *cb) { /* sanity check */ if (node == NULL || service == NULL || unix_path == NULL) - return (NULL); + return NULL; /* allocate db data & fill it */ - ovs_db_t *pdb = pdb = calloc(1, sizeof(*pdb)); + ovs_db_t *pdb = calloc(1, sizeof(*pdb)); if (pdb == NULL) - return (NULL); + return NULL; + pdb->sock = -1; /* store the OVS DB address */ sstrncpy(pdb->node, node, sizeof(pdb->node)); @@ -1008,21 +1025,21 @@ ovs_db_t *ovs_db_init(const char *node, const char *service, if (pthread_mutexattr_init(&mutex_attr)) { OVS_ERROR("OVS DB mutex attribute init failed"); sfree(pdb); - return (NULL); + return NULL; } /* set OVS DB mutex as recursive */ if (pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_RECURSIVE)) { OVS_ERROR("Failed to set OVS DB mutex as recursive"); pthread_mutexattr_destroy(&mutex_attr); sfree(pdb); - return (NULL); + return NULL; } /* init OVS DB mutex */ if (pthread_mutex_init(&pdb->mutex, &mutex_attr)) { OVS_ERROR("OVS DB mutex init failed"); pthread_mutexattr_destroy(&mutex_attr); sfree(pdb); - return (NULL); + return NULL; } /* destroy mutex attributes */ pthread_mutexattr_destroy(&mutex_attr); @@ -1030,14 +1047,13 @@ ovs_db_t *ovs_db_init(const char *node, const char *service, /* init event thread */ if (ovs_db_event_thread_init(pdb) < 0) { ovs_db_destroy(pdb); - return (NULL); + return NULL; } /* init polling thread */ - pdb->sock = -1; if (ovs_db_poll_thread_init(pdb) < 0) { ovs_db_destroy(pdb); - return (NULL); + return NULL; } return pdb; } @@ -1057,16 +1073,16 @@ int ovs_db_send_request(ovs_db_t *pdb, const char *method, const char *params, /* sanity check */ if (!pdb || !method || !params) - return (-1); + return -1; if ((jgen = yajl_gen_alloc(NULL)) == NULL) - return (-1); + return -1; /* try to parse params */ if ((jparams = yajl_tree_parse(params, NULL, 0)) == NULL) { OVS_ERROR("params is not a JSON string"); yajl_gen_clear(jgen); - return (-1); + return -1; } /* generate method field */ @@ -1083,7 +1099,7 @@ int ovs_db_send_request(ovs_db_t *pdb, const char *method, const char *params, /* generate id field */ OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id"); uid = ovs_uid_generate(); - ssnprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid); + snprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid); OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_buff); OVS_YAJL_CALL(yajl_gen_map_close, jgen); @@ -1145,16 +1161,16 @@ int ovs_db_table_cb_register(ovs_db_t *pdb, const char *tb_name, /* sanity check */ if (pdb == NULL || tb_name == NULL || update_cb == NULL) - return (-1); + return -1; /* allocate new update callback */ if ((new_cb = calloc(1, sizeof(ovs_callback_t))) == NULL) - return (-1); + return -1; /* init YAJL generator */ if ((jgen = yajl_gen_alloc(NULL)) == NULL) { sfree(new_cb); - return (-1); + return -1; } /* add new callback to front */ @@ -1169,7 +1185,7 @@ int ovs_db_table_cb_register(ovs_db_t *pdb, const char *tb_name, OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, OVS_DB_DEFAULT_DB_NAME); /* uid string */ - ssnprintf(uid_str, sizeof(uid_str), "%" PRIX64, new_cb->uid); + snprintf(uid_str, sizeof(uid_str), "%" PRIX64, new_cb->uid); OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_str); /* */ @@ -1238,25 +1254,30 @@ int ovs_db_destroy(ovs_db_t *pdb) { /* sanity check */ if (pdb == NULL) - return (-1); + return -1; + + /* stop event thread */ + if (ovs_db_event_thread_terminate(pdb) < 0) { + OVS_ERROR("stop event thread failed"); + ovs_db_ret = -1; + } /* try to lock the structure before releasing */ if ((ret = pthread_mutex_lock(&pdb->mutex))) { OVS_ERROR("pthread_mutex_lock() DB mutex lock failed (%d)", ret); - return (-1); + return -1; } - /* stop poll thread */ - if (ovs_db_event_thread_destroy(pdb) < 0) { + /* stop poll thread and destroy thread's private data */ + if (ovs_db_poll_thread_destroy(pdb) < 0) { OVS_ERROR("destroy poll thread failed"); - ovs_db_ret = (-1); + ovs_db_ret = -1; } - /* stop event thread */ - if (ovs_db_poll_thread_destroy(pdb) < 0) { - OVS_ERROR("stop event thread failed"); - ovs_db_ret = (-1); - } + /* destroy event thread private data */ + ovs_db_event_thread_data_destroy(pdb); + + pthread_mutex_unlock(&pdb->mutex); /* unsubscribe callbacks */ ovs_db_callback_remove_all(pdb); @@ -1266,7 +1287,6 @@ int ovs_db_destroy(ovs_db_t *pdb) { close(pdb->sock); /* release DB handler */ - pthread_mutex_unlock(&pdb->mutex); pthread_mutex_destroy(&pdb->mutex); sfree(pdb); return ovs_db_ret;