src/utils_ovs.c

   1 /**
   2  * collectd - src/utils_ovs.c
   3  *
   4  * Copyright(c) 2016 Intel Corporation. All rights reserved.
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy of
   7  * this software and associated documentation files (the "Software"), to deal in
   8  * the Software without restriction, including without limitation the rights to
   9  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  10  * of the Software, and to permit persons to whom the Software is furnished to do
  11  * so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in all
  14  * copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  22  * SOFTWARE.
  23  *
  24  * Authors:
  25  *   Volodymyr Mytnyk <volodymyrx.mytnyk@intel.com>
  26  **/
  27
  28 /* clang-format off */
  29 /*
  30  *                         OVS DB API internal architecture diagram
  31  * +------------------------------------------------------------------------------+
  32  * |OVS plugin      |OVS utils                                                    |
  33  * |                |     +------------------------+                              |
  34  * |                |     |      echo handler      |                JSON request/ |
  35  * |                |  +--+ (ovs_db_table_echo_cb) +<---+---------+ update event/ |
  36  * |                |  |  |                        |    |         | result        |
  37  * |                |  |  +------------------------+    |         |               |
  38  * |                |  |                                |    +----+---+--------+  |
  39  * |  +----------+  |  |  +------------------------+    |    |        |        |  |
  40  * |  |  update  |  |  |  |     update handler     |    |    |  YAJL  |  JSON  |  |
  41  * |  | callback +<-------+(ovs_db_table_update_cp)+<---+    | parser | reader |  |
  42  * |  +----------+  |  |  |                        |    |    |        |        |  |
  43  * |                |  |  +------------------------+    |    +--------+---+----+  |
  44  * |                |  |                                |                 ^       |
  45  * |  +----------+  |  |  +------------------------+    |                 |       |
  46  * |  |  result  |  |  |  |     result handler     |    |                 |       |
  47  * |  | callback +<-------+   (ovs_db_result_cb)   +<---+        JSON raw |       |
  48  * |  +----------+  |  |  |                        |               data   |       |
  49  * |                |  |  +------------------------+                      |       |
  50  * |                |  |                                                  |       |
  51  * |                |  |    +------------------+             +------------+----+  |
  52  * |  +----------+  |  |    |thread|           |             |thread|          |  |
  53  * |  |   init   |  |  |    |                  |  reconnect  |                 |  |
  54  * |  | callback +<---------+   EVENT WORKER   +<------------+   POLL WORKER   |  |
  55  * |  +----------+  |  |    +------------------+             +--------+--------+  |
  56  * |                |  |                                              ^           |
  57  * +----------------+-------------------------------------------------------------+
  58  *                     |                                              |
  59  *                 JSON|echo reply                                 raw|data
  60  *                     v                                              v
  61  * +-------------------+----------------------------------------------+-----------+
  62  * |                                 TCP/UNIX socket                              |
  63  * +-------------------------------------------------------------------------------
  64  */
  65 /* clang-format on */
  66
  67 /* collectd headers */
  68 #include "collectd.h"
  69
  70 #include "common.h"
  71
  72 /* private headers */
  73 #include "utils_ovs.h"
  74
  75 /* system libraries */
  76 #if HAVE_NETDB_H
  77 #include <netdb.h>
  78 #endif
  79 #if HAVE_ARPA_INET_H
  80 #include <arpa/inet.h>
  81 #endif
  82 #if HAVE_POLL_H
  83 #include <poll.h>
  84 #endif
  85 #if HAVE_SYS_UN_H
  86 #include <sys/un.h>
  87 #endif
  88
  89 #include <semaphore.h>
  90
  91 #define OVS_ERROR(fmt, ...)                                                    \
  92   do {                                                                         \
  93     ERROR("ovs_utils: " fmt, ##__VA_ARGS__);                                   \
  94   } while (0)
  95 #define OVS_DEBUG(fmt, ...)                                                    \
  96   do {                                                                         \
  97     DEBUG("%s:%d:%s(): " fmt, __FILE__, __LINE__, __FUNCTION__,                \
  98           ##__VA_ARGS__);                                                      \
  99   } while (0)
 100
 101 #define OVS_DB_POLL_TIMEOUT 1           /* poll receive timeout (sec) */
 102 #define OVS_DB_POLL_READ_BLOCK_SIZE 512 /* read block size (bytes) */
 103 #define OVS_DB_DEFAULT_DB_NAME "Open_vSwitch"
 104
 105 #define OVS_DB_EVENT_TIMEOUT 5 /* event thread timeout (sec) */
 106 #define OVS_DB_EVENT_TERMINATE 1
 107 #define OVS_DB_EVENT_CONN_ESTABLISHED 2
 108 #define OVS_DB_EVENT_CONN_TERMINATED 3
 109
 110 #define OVS_DB_POLL_STATE_RUNNING 1
 111 #define OVS_DB_POLL_STATE_EXITING 2
 112
 113 #define OVS_DB_SEND_REQ_TIMEOUT 5 /* send request timeout (sec) */
 114
 115 #define OVS_YAJL_CALL(func, ...)                                               \
 116   do {                                                                         \
 117     yajl_gen_ret = yajl_gen_status_ok;                                         \
 118     if ((yajl_gen_ret = func(__VA_ARGS__)) != yajl_gen_status_ok)              \
 119       goto yajl_gen_failure;                                                   \
 120   } while (0)
 121 #define OVS_YAJL_ERROR_BUFFER_SIZE 1024
 122 #define OVS_ERROR_BUFF_SIZE 512
 123 #define OVS_UID_STR_SIZE 17 /* 64-bit HEX string len + '\0' */
 124
 125 /* JSON reader internal data */
 126 struct ovs_json_reader_s {
 127   char *buff_ptr;
 128   size_t buff_size;
 129   size_t buff_offset;
 130   size_t json_offset;
 131 };
 132 typedef struct ovs_json_reader_s ovs_json_reader_t;
 133
 134 /* Result callback declaration */
 135 struct ovs_result_cb_s {
 136   sem_t sync;
 137   ovs_db_result_cb_t call;
 138 };
 139 typedef struct ovs_result_cb_s ovs_result_cb_t;
 140
 141 /* Table callback declaration */
 142 struct ovs_table_cb_s {
 143   ovs_db_table_cb_t call;
 144 };
 145 typedef struct ovs_table_cb_s ovs_table_cb_t;
 146
 147 /* Callback declaration */
 148 struct ovs_callback_s {
 149   uint64_t uid;
 150   union {
 151     ovs_result_cb_t result;
 152     ovs_table_cb_t table;
 153   };
 154   struct ovs_callback_s *next;
 155   struct ovs_callback_s *prev;
 156 };
 157 typedef struct ovs_callback_s ovs_callback_t;
 158
 159 /* Event thread data declaration */
 160 struct ovs_event_thread_s {
 161   pthread_t tid;
 162   pthread_mutex_t mutex;
 163   pthread_cond_t cond;
 164   int value;
 165 };
 166 typedef struct ovs_event_thread_s ovs_event_thread_t;
 167
 168 /* Poll thread data declaration */
 169 struct ovs_poll_thread_s {
 170   pthread_t tid;
 171   pthread_mutex_t mutex;
 172   int state;
 173 };
 174 typedef struct ovs_poll_thread_s ovs_poll_thread_t;
 175
 176 /* OVS DB internal data declaration */
 177 struct ovs_db_s {
 178   ovs_poll_thread_t poll_thread;
 179   ovs_event_thread_t event_thread;
 180   pthread_mutex_t mutex;
 181   ovs_callback_t *remote_cb;
 182   ovs_db_callback_t cb;
 183   char service[OVS_DB_ADDR_SERVICE_SIZE];
 184   char node[OVS_DB_ADDR_NODE_SIZE];
 185   char unix_path[OVS_DB_ADDR_NODE_SIZE];
 186   int sock;
 187 };
 188
 189 /* Global variables */
 190 static uint64_t ovs_uid = 0;
 191 static pthread_mutex_t ovs_uid_mutex = PTHREAD_MUTEX_INITIALIZER;
 192
 193 /* Post an event to event thread.
 194  * Possible events are:
 195  *  OVS_DB_EVENT_TERMINATE
 196  *  OVS_DB_EVENT_CONN_ESTABLISHED
 197  *  OVS_DB_EVENT_CONN_TERMINATED
 198  */
 199 static void ovs_db_event_post(ovs_db_t *pdb, int event) {
 200   pthread_mutex_lock(&pdb->event_thread.mutex);
 201   pdb->event_thread.value = event;
 202   pthread_mutex_unlock(&pdb->event_thread.mutex);
 203   pthread_cond_signal(&pdb->event_thread.cond);
 204 }
 205
 206 /* Check if POLL thread is still running. Returns
 207  * 1 if running otherwise 0 is returned */
 208 static _Bool ovs_db_poll_is_running(ovs_db_t *pdb) {
 209   int state = 0;
 210   pthread_mutex_lock(&pdb->poll_thread.mutex);
 211   state = pdb->poll_thread.state;
 212   pthread_mutex_unlock(&pdb->poll_thread.mutex);
 213   return (state == OVS_DB_POLL_STATE_RUNNING);
 214 }
 215
 216 /* Generate unique identifier (UID). It is used by OVS DB API
 217  * to set "id" field for any OVS DB JSON request. */
 218 static uint64_t ovs_uid_generate() {
 219   uint64_t new_uid;
 220   pthread_mutex_lock(&ovs_uid_mutex);
 221   new_uid = ++ovs_uid;
 222   pthread_mutex_unlock(&ovs_uid_mutex);
 223   return new_uid;
 224 }
 225
 226 /*
 227  * Callback API. These function are used to store
 228  * registered callbacks in OVS DB API.
 229  */
 230
 231 /* Add new callback into OVS DB object */
 232 static void ovs_db_callback_add(ovs_db_t *pdb, ovs_callback_t *new_cb) {
 233   pthread_mutex_lock(&pdb->mutex);
 234   if (pdb->remote_cb)
 235     pdb->remote_cb->prev = new_cb;
 236   new_cb->next = pdb->remote_cb;
 237   new_cb->prev = NULL;
 238   pdb->remote_cb = new_cb;
 239   pthread_mutex_unlock(&pdb->mutex);
 240 }
 241
 242 /* Remove callback from OVS DB object */
 243 static void ovs_db_callback_remove(ovs_db_t *pdb, ovs_callback_t *del_cb) {
 244   pthread_mutex_lock(&pdb->mutex);
 245   ovs_callback_t *pre_cb = del_cb->prev;
 246   ovs_callback_t *next_cb = del_cb->next;
 247
 248   if (next_cb)
 249     next_cb->prev = del_cb->prev;
 250
 251   if (pre_cb)
 252     pre_cb->next = del_cb->next;
 253   else
 254     pdb->remote_cb = del_cb->next;
 255
 256   free(del_cb);
 257   pthread_mutex_unlock(&pdb->mutex);
 258 }
 259
 260 /* Remove all callbacks form OVS DB object */
 261 static void ovs_db_callback_remove_all(ovs_db_t *pdb) {
 262   pthread_mutex_lock(&pdb->mutex);
 263   for (ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb;
 264        del_cb = pdb->remote_cb) {
 265     pdb->remote_cb = del_cb->next;
 266     free(del_cb);
 267   }
 268   pdb->remote_cb = NULL;
 269   pthread_mutex_unlock(&pdb->mutex);
 270 }
 271
 272 /* Get/find callback in OVS DB object by UID. Returns pointer
 273  * to requested callback otherwise NULL is returned.
 274  *
 275  * IMPORTANT NOTE:
 276  *   The OVS DB mutex MUST be locked by the caller
 277  *   to make sure that returned callback is still valid.
 278  */
 279 static ovs_callback_t *ovs_db_callback_get(ovs_db_t *pdb, uint64_t uid) {
 280   for (ovs_callback_t *cb = pdb->remote_cb; cb != NULL; cb = cb->next)
 281     if (cb->uid == uid)
 282       return cb;
 283   return NULL;
 284 }
 285
 286 /* Send all requested data to the socket. Returns 0 if
 287  * ALL request data has been sent otherwise negative value
 288  * is returned */
 289 static int ovs_db_data_send(const ovs_db_t *pdb, const char *data, size_t len) {
 290   ssize_t nbytes = 0;
 291   size_t rem = len;
 292   size_t off = 0;
 293
 294   while (rem > 0) {
 295     if ((nbytes = send(pdb->sock, data + off, rem, 0)) <= 0)
 296       return (-1);
 297     rem -= (size_t)nbytes;
 298     off += (size_t)nbytes;
 299   }
 300   return (0);
 301 }
 302
 303 /*
 304  * YAJL (Yet Another JSON Library) helper functions
 305  * Documentation (https://lloyd.github.io/yajl/)
 306  */
 307
 308 /* Add null-terminated string into YAJL generator handle (JSON object).
 309  * Similar function to yajl_gen_string() but takes null-terminated string
 310  * instead of string and its length.
 311  *
 312  * jgen   - YAJL generator handle allocated by yajl_gen_alloc()
 313  * string - Null-terminated string
 314  */
 315 static yajl_gen_status ovs_yajl_gen_tstring(yajl_gen hander,
 316                                             const char *string) {
 317   return yajl_gen_string(hander, (const unsigned char *)string, strlen(string));
 318 }
 319
 320 /* Add YAJL value into YAJL generator handle (JSON object)
 321  *
 322  * jgen - YAJL generator handle allocated by yajl_gen_alloc()
 323  * jval - YAJL value usually returned by yajl_tree_get()
 324  */
 325 static yajl_gen_status ovs_yajl_gen_val(yajl_gen jgen, yajl_val jval) {
 326   size_t array_len = 0;
 327   yajl_val *jvalues = NULL;
 328   yajl_val jobj_value = NULL;
 329   const char *obj_key = NULL;
 330   size_t obj_len = 0;
 331   yajl_gen_status yajl_gen_ret = yajl_gen_status_ok;
 332
 333   if (YAJL_IS_STRING(jval))
 334     OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, YAJL_GET_STRING(jval));
 335   else if (YAJL_IS_DOUBLE(jval))
 336     OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_DOUBLE(jval));
 337   else if (YAJL_IS_INTEGER(jval))
 338     OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_INTEGER(jval));
 339   else if (YAJL_IS_TRUE(jval))
 340     OVS_YAJL_CALL(yajl_gen_bool, jgen, 1);
 341   else if (YAJL_IS_FALSE(jval))
 342     OVS_YAJL_CALL(yajl_gen_bool, jgen, 0);
 343   else if (YAJL_IS_NULL(jval))
 344     OVS_YAJL_CALL(yajl_gen_null, jgen);
 345   else if (YAJL_IS_ARRAY(jval)) {
 346     /* create new array and add all elements into the array */
 347     array_len = YAJL_GET_ARRAY(jval)->len;
 348     jvalues = YAJL_GET_ARRAY(jval)->values;
 349     OVS_YAJL_CALL(yajl_gen_array_open, jgen);
 350     for (int i = 0; i < array_len; i++)
 351       OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jvalues[i]);
 352     OVS_YAJL_CALL(yajl_gen_array_close, jgen);
 353   } else if (YAJL_IS_OBJECT(jval)) {
 354     /* create new object and add all elements into the object */
 355     OVS_YAJL_CALL(yajl_gen_map_open, jgen);
 356     obj_len = YAJL_GET_OBJECT(jval)->len;
 357     for (int i = 0; i < obj_len; i++) {
 358       obj_key = YAJL_GET_OBJECT(jval)->keys[i];
 359       jobj_value = YAJL_GET_OBJECT(jval)->values[i];
 360       OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, obj_key);
 361       OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jobj_value);
 362     }
 363     OVS_YAJL_CALL(yajl_gen_map_close, jgen);
 364   } else {
 365     OVS_ERROR("%s() unsupported value type %d (skip)", __FUNCTION__,
 366               (int)(jval)->type);
 367     goto yajl_gen_failure;
 368   }
 369   return yajl_gen_status_ok;
 370
 371 yajl_gen_failure:
 372   OVS_ERROR("%s() error to generate value", __FUNCTION__);
 373   return yajl_gen_ret;
 374 }
 375
 376 /* OVS DB echo request handler. When OVS DB sends
 377  * "echo" request to the client, client should generate
 378  * "echo" replay with the same content received in the
 379  * request */
 380 static int ovs_db_table_echo_cb(const ovs_db_t *pdb, yajl_val jnode) {
 381   yajl_val jparams;
 382   yajl_val jid;
 383   yajl_gen jgen;
 384   size_t resp_len = 0;
 385   const char *resp = NULL;
 386   const char *params_path[] = {"params", NULL};
 387   const char *id_path[] = {"id", NULL};
 388   yajl_gen_status yajl_gen_ret;
 389
 390   if ((jgen = yajl_gen_alloc(NULL)) == NULL)
 391     return (-1);
 392
 393   /* check & get request attributes */
 394   if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
 395       ((jid = yajl_tree_get(jnode, id_path, yajl_t_any)) == NULL)) {
 396     OVS_ERROR("parse echo request failed");
 397     goto yajl_gen_failure;
 398   }
 399
 400   /* generate JSON echo response */
 401   OVS_YAJL_CALL(yajl_gen_map_open, jgen);
 402
 403   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "result");
 404   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
 405
 406   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "error");
 407   OVS_YAJL_CALL(yajl_gen_null, jgen);
 408
 409   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
 410   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jid);
 411
 412   OVS_YAJL_CALL(yajl_gen_map_close, jgen);
 413   OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&resp,
 414                 &resp_len);
 415
 416   /* send the response */
 417   OVS_DEBUG("response: %s", resp);
 418   if (ovs_db_data_send(pdb, resp, resp_len) < 0) {
 419     OVS_ERROR("send echo reply failed");
 420     goto yajl_gen_failure;
 421   }
 422   /* clean up and return success */
 423   yajl_gen_clear(jgen);
 424   return (0);
 425
 426 yajl_gen_failure:
 427   /* release memory */
 428   yajl_gen_clear(jgen);
 429   return (-1);
 430 }
 431
 432 /* Get OVS DB registered callback by YAJL val. The YAJL
 433  * value should be YAJL string (UID). Returns NULL if
 434  * callback hasn't been found. See also ovs_db_callback_get()
 435  * description for addition info.
 436  */
 437 static ovs_callback_t *ovs_db_table_callback_get(ovs_db_t *pdb, yajl_val jid) {
 438   char *endptr = NULL;
 439   const char *suid = NULL;
 440   uint64_t uid;
 441
 442   if (jid && YAJL_IS_STRING(jid)) {
 443     suid = YAJL_GET_STRING(jid);
 444     uid = (uint64_t)strtoul(suid, &endptr, 16);
 445     if (*endptr == '\0' && uid)
 446       return ovs_db_callback_get(pdb, uid);
 447   }
 448
 449   return NULL;
 450 }
 451
 452 /* OVS DB table update event handler.
 453  * This callback is called by POLL thread if OVS DB
 454  * table update callback is received from the DB
 455  * server. Once registered callback found, it's called
 456  * by this handler. */
 457 static int ovs_db_table_update_cb(ovs_db_t *pdb, yajl_val jnode) {
 458   ovs_callback_t *cb = NULL;
 459   yajl_val jvalue;
 460   yajl_val jparams;
 461   yajl_val jtable_updates;
 462   const char *params_path[] = {"params", NULL};
 463   const char *id_path[] = {"id", NULL};
 464
 465   /* check & get request attributes */
 466   if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
 467       (yajl_tree_get(jnode, id_path, yajl_t_null) == NULL)) {
 468     OVS_ERROR("invalid OVS DB request received");
 469     return (-1);
 470   }
 471
 472   /* check array length: [<json-value>, <table-updates>] */
 473   if ((YAJL_GET_ARRAY(jparams) == NULL) ||
 474       (YAJL_GET_ARRAY(jparams)->len != 2)) {
 475     OVS_ERROR("invalid OVS DB request received");
 476     return (-1);
 477   }
 478
 479   jvalue = YAJL_GET_ARRAY(jparams)->values[0];
 480   jtable_updates = YAJL_GET_ARRAY(jparams)->values[1];
 481   if ((!YAJL_IS_OBJECT(jtable_updates)) || (!YAJL_IS_STRING(jvalue))) {
 482     OVS_ERROR("invalid OVS DB request id or table update received");
 483     return (-1);
 484   }
 485
 486   /* find registered callback based on <json-value> */
 487   pthread_mutex_lock(&pdb->mutex);
 488   cb = ovs_db_table_callback_get(pdb, jvalue);
 489   if (cb == NULL || cb->table.call == NULL) {
 490     OVS_ERROR("No OVS DB table update callback found");
 491     pthread_mutex_unlock(&pdb->mutex);
 492     return (-1);
 493   }
 494
 495   /* call registered callback */
 496   cb->table.call(jtable_updates);
 497   pthread_mutex_unlock(&pdb->mutex);
 498   return 0;
 499 }
 500
 501 /* OVS DB result request handler.
 502  * This callback is called by POLL thread if OVS DB
 503  * result reply is received from the DB server.
 504  * Once registered callback found, it's called
 505  * by this handler. */
 506 static int ovs_db_result_cb(ovs_db_t *pdb, yajl_val jnode) {
 507   ovs_callback_t *cb = NULL;
 508   yajl_val jresult;
 509   yajl_val jerror;
 510   yajl_val jid;
 511   const char *result_path[] = {"result", NULL};
 512   const char *error_path[] = {"error", NULL};
 513   const char *id_path[] = {"id", NULL};
 514
 515   jresult = yajl_tree_get(jnode, result_path, yajl_t_any);
 516   jerror = yajl_tree_get(jnode, error_path, yajl_t_any);
 517   jid = yajl_tree_get(jnode, id_path, yajl_t_string);
 518
 519   /* check & get result attributes */
 520   if (!jresult || !jerror || !jid)
 521     return (-1);
 522
 523   /* try to find registered callback */
 524   pthread_mutex_lock(&pdb->mutex);
 525   cb = ovs_db_table_callback_get(pdb, jid);
 526   if (cb != NULL && cb->result.call != NULL) {
 527     /* call registered callback */
 528     cb->result.call(jresult, jerror);
 529     /* unlock owner of the reply */
 530     sem_post(&cb->result.sync);
 531   }
 532
 533   pthread_mutex_unlock(&pdb->mutex);
 534   return (0);
 535 }
 536
 537 /* Handle JSON data (one request) and call
 538  * appropriate event OVS DB handler. Currently,
 539  * update callback 'ovs_db_table_update_cb' and
 540  * result callback 'ovs_db_result_cb' is supported.
 541  */
 542 static int ovs_db_json_data_process(ovs_db_t *pdb, const char *data,
 543                                     size_t len) {
 544   const char *method = NULL;
 545   char yajl_errbuf[OVS_YAJL_ERROR_BUFFER_SIZE];
 546   const char *method_path[] = {"method", NULL};
 547   const char *result_path[] = {"result", NULL};
 548   char *sjson = NULL;
 549   yajl_val jnode, jval;
 550
 551   /* duplicate the data to make null-terminated string
 552    * required for yajl_tree_parse() */
 553   if ((sjson = calloc(1, len + 1)) == NULL)
 554     return (-1);
 555
 556   sstrncpy(sjson, data, len + 1);
 557   OVS_DEBUG("[len=%zu] %s", len, sjson);
 558
 559   /* parse json data */
 560   jnode = yajl_tree_parse(sjson, yajl_errbuf, sizeof(yajl_errbuf));
 561   if (jnode == NULL) {
 562     OVS_ERROR("yajl_tree_parse() %s", yajl_errbuf);
 563     sfree(sjson);
 564     return (-1);
 565   }
 566
 567   /* get method name */
 568   if ((jval = yajl_tree_get(jnode, method_path, yajl_t_string)) != NULL) {
 569     method = YAJL_GET_STRING(jval);
 570     if (strcmp("echo", method) == 0) {
 571       /* echo request from the server */
 572       if (ovs_db_table_echo_cb(pdb, jnode) < 0)
 573         OVS_ERROR("handle echo request failed");
 574     } else if (strcmp("update", method) == 0) {
 575       /* update notification */
 576       if (ovs_db_table_update_cb(pdb, jnode) < 0)
 577         OVS_ERROR("handle update notification failed");
 578     }
 579   } else if ((jval = yajl_tree_get(jnode, result_path, yajl_t_any)) != NULL) {
 580     /* result notification */
 581     if (ovs_db_result_cb(pdb, jnode) < 0)
 582       OVS_ERROR("handle result reply failed");
 583   } else
 584     OVS_ERROR("connot find method or result failed");
 585
 586   /* release memory */
 587   yajl_tree_free(jnode);
 588   sfree(sjson);
 589   return (0);
 590 }
 591
 592 /*
 593  * JSON reader implementation.
 594  *
 595  * This module process raw JSON data (byte stream) and
 596  * returns fully-fledged JSON data which can be processed
 597  * (parsed) by YAJL later.
 598  */
 599
 600 /* Allocate JSON reader instance */
 601 static ovs_json_reader_t *ovs_json_reader_alloc() {
 602   ovs_json_reader_t *jreader = NULL;
 603
 604   if ((jreader = calloc(sizeof(ovs_json_reader_t), 1)) == NULL)
 605     return NULL;
 606
 607   return jreader;
 608 }
 609
 610 /* Push raw data into into the JSON reader for processing */
 611 static int ovs_json_reader_push_data(ovs_json_reader_t *jreader,
 612                                      const char *data, size_t data_len) {
 613   char *new_buff = NULL;
 614   size_t available = jreader->buff_size - jreader->buff_offset;
 615
 616   /* check/update required memory space */
 617   if (available < data_len) {
 618     OVS_DEBUG("Reallocate buffer [size=%d, available=%d required=%d]",
 619               (int)jreader->buff_size, (int)available, (int)data_len);
 620
 621     /* allocate new chunk of memory */
 622     new_buff = realloc(jreader->buff_ptr, (jreader->buff_size + data_len));
 623     if (new_buff == NULL)
 624       return (-1);
 625
 626     /* point to new allocated memory */
 627     jreader->buff_ptr = new_buff;
 628     jreader->buff_size += data_len;
 629   }
 630
 631   /* store input data */
 632   memcpy(jreader->buff_ptr + jreader->buff_offset, data, data_len);
 633   jreader->buff_offset += data_len;
 634   return (0);
 635 }
 636
 637 /* Pop one fully-fledged JSON if already exists. Returns 0 if
 638  * completed JSON already exists otherwise negative value is
 639  * returned */
 640 static int ovs_json_reader_pop(ovs_json_reader_t *jreader,
 641                                const char **json_ptr, size_t *json_len_ptr) {
 642   size_t nbraces = 0;
 643   size_t json_len = 0;
 644   char *json = NULL;
 645
 646   /* search open/close brace */
 647   for (int i = jreader->json_offset; i < jreader->buff_offset; i++) {
 648     if (jreader->buff_ptr[i] == '{') {
 649       nbraces++;
 650     } else if (jreader->buff_ptr[i] == '}')
 651       if (nbraces)
 652         if (!(--nbraces)) {
 653           /* JSON data */
 654           *json_ptr = jreader->buff_ptr + jreader->json_offset;
 655           *json_len_ptr = json_len + 1;
 656           jreader->json_offset = i + 1;
 657           return (0);
 658         }
 659
 660     /* increase JSON data length */
 661     if (nbraces)
 662       json_len++;
 663   }
 664
 665   if (jreader->json_offset) {
 666     if (jreader->json_offset < jreader->buff_offset) {
 667       /* shift data to the beginning of the buffer
 668        * and zero rest of the buffer data */
 669       json = &jreader->buff_ptr[jreader->json_offset];
 670       json_len = jreader->buff_offset - jreader->json_offset;
 671       for (int i = 0; i < jreader->buff_size; i++)
 672         jreader->buff_ptr[i] = ((i < json_len) ? (json[i]) : (0));
 673       jreader->buff_offset = json_len;
 674     } else
 675       /* reset the buffer */
 676       jreader->buff_offset = 0;
 677
 678     /* data is at the beginning of the buffer */
 679     jreader->json_offset = 0;
 680   }
 681
 682   return (-1);
 683 }
 684
 685 /* Reset JSON reader. It is useful when start processing
 686  * new raw data. E.g.: in case of lost stream connection.
 687  */
 688 static void ovs_json_reader_reset(ovs_json_reader_t *jreader) {
 689   if (jreader) {
 690     jreader->buff_offset = 0;
 691     jreader->json_offset = 0;
 692   }
 693 }
 694
 695 /* Release internal data allocated for JSON reader */
 696 static void ovs_json_reader_free(ovs_json_reader_t *jreader) {
 697   if (jreader) {
 698     free(jreader->buff_ptr);
 699     free(jreader);
 700   }
 701 }
 702
 703 /* Reconnect to OVS DB and call the OVS DB post connection init callback
 704  * if connection has been established.
 705  */
 706 static void ovs_db_reconnect(ovs_db_t *pdb) {
 707   const char *node_info = pdb->node;
 708   struct addrinfo *result;
 709
 710   if (pdb->unix_path[0] != '\0') {
 711     /* use UNIX socket instead of INET address */
 712     node_info = pdb->unix_path;
 713     result = calloc(1, sizeof(struct addrinfo));
 714     struct sockaddr_un *sa_unix = calloc(1, sizeof(struct sockaddr_un));
 715     if (result == NULL || sa_unix == NULL) {
 716       sfree(result);
 717       sfree(sa_unix);
 718       return;
 719     }
 720     result->ai_family = AF_UNIX;
 721     result->ai_socktype = SOCK_STREAM;
 722     result->ai_addrlen = sizeof(*sa_unix);
 723     result->ai_addr = (struct sockaddr *)sa_unix;
 724     sa_unix->sun_family = result->ai_family;
 725     sstrncpy(sa_unix->sun_path, pdb->unix_path, sizeof(sa_unix->sun_path));
 726   } else {
 727     /* inet socket address */
 728     struct addrinfo hints;
 729
 730     /* setup criteria for selecting the socket address */
 731     memset(&hints, 0, sizeof(hints));
 732     hints.ai_family = AF_UNSPEC;
 733     hints.ai_socktype = SOCK_STREAM;
 734
 735     /* get socket addresses */
 736     int ret = getaddrinfo(pdb->node, pdb->service, &hints, &result);
 737     if (ret != 0) {
 738       OVS_ERROR("getaddrinfo(): %s", gai_strerror(ret));
 739       return;
 740     }
 741   }
 742   /* try to connect to the server */
 743   for (struct addrinfo *rp = result; rp != NULL; rp = rp->ai_next) {
 744     char errbuff[OVS_ERROR_BUFF_SIZE];
 745     int sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
 746     if (sock < 0) {
 747       sstrerror(errno, errbuff, sizeof(errbuff));
 748       OVS_DEBUG("socket(): %s", errbuff);
 749       continue;
 750     }
 751     if (connect(sock, rp->ai_addr, rp->ai_addrlen) < 0) {
 752       close(sock);
 753       sstrerror(errno, errbuff, sizeof(errbuff));
 754       OVS_DEBUG("connect(): %s [family=%d]", errbuff, rp->ai_family);
 755     } else {
 756       /* send notification to event thread */
 757       ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED);
 758       pdb->sock = sock;
 759       break;
 760     }
 761   }
 762
 763   if (pdb->sock < 0)
 764     OVS_ERROR("connect to \"%s\" failed", node_info);
 765
 766   freeaddrinfo(result);
 767 }
 768
 769 /* POLL worker thread.
 770  * It listens on OVS DB connection for incoming
 771  * requests/reply/events etc. Also, it reconnects to OVS DB
 772  * if connection has been lost.
 773  */
 774 static void *ovs_poll_worker(void *arg) {
 775   ovs_db_t *pdb = (ovs_db_t *)arg; /* pointer to OVS DB */
 776   ovs_json_reader_t *jreader = NULL;
 777   struct pollfd poll_fd = {
 778       .fd = pdb->sock, .events = POLLIN | POLLPRI, .revents = 0,
 779   };
 780
 781   /* create JSON reader instance */
 782   if ((jreader = ovs_json_reader_alloc()) == NULL) {
 783     OVS_ERROR("initialize json reader failed");
 784     return (NULL);
 785   }
 786
 787   /* poll data */
 788   while (ovs_db_poll_is_running(pdb)) {
 789     char errbuff[OVS_ERROR_BUFF_SIZE];
 790     poll_fd.fd = pdb->sock;
 791     int poll_ret = poll(&poll_fd, 1, /* ms */ OVS_DB_POLL_TIMEOUT * 1000);
 792     if (poll_ret < 0) {
 793       sstrerror(errno, errbuff, sizeof(errbuff));
 794       OVS_ERROR("poll(): %s", errbuff);
 795       break;
 796     } else if (poll_ret == 0) {
 797       OVS_DEBUG("poll(): timeout");
 798       if (pdb->sock < 0)
 799         /* invalid fd, so try to reconnect */
 800         ovs_db_reconnect(pdb);
 801       continue;
 802     }
 803     if (poll_fd.revents & POLLNVAL) {
 804       /* invalid file descriptor, clean-up */
 805       ovs_db_callback_remove_all(pdb);
 806       ovs_json_reader_reset(jreader);
 807       /* setting poll FD to -1 tells poll() call to ignore this FD.
 808        * In that case poll() call will return timeout all the time */
 809       pdb->sock = (-1);
 810     } else if ((poll_fd.revents & POLLERR) || (poll_fd.revents & POLLHUP)) {
 811       /* connection is broken */
 812       close(poll_fd.fd);
 813       ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
 814       OVS_ERROR("poll() peer closed its end of the channel");
 815     } else if ((poll_fd.revents & POLLIN) || (poll_fd.revents & POLLPRI)) {
 816       /* read incoming data */
 817       char buff[OVS_DB_POLL_READ_BLOCK_SIZE];
 818       ssize_t nbytes = recv(poll_fd.fd, buff, sizeof(buff), 0);
 819       if (nbytes < 0) {
 820         sstrerror(errno, errbuff, sizeof(errbuff));
 821         OVS_ERROR("recv(): %s", errbuff);
 822         /* read error? Try to reconnect */
 823         close(poll_fd.fd);
 824         continue;
 825       } else if (nbytes == 0) {
 826         close(poll_fd.fd);
 827         ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
 828         OVS_ERROR("recv() peer has performed an orderly shutdown");
 829         continue;
 830       }
 831       /* read incoming data */
 832       size_t json_len = 0;
 833       const char *json = NULL;
 834       OVS_DEBUG("recv(): received %zd bytes of data", nbytes);
 835       ovs_json_reader_push_data(jreader, buff, nbytes);
 836       while (!ovs_json_reader_pop(jreader, &json, &json_len))
 837         /* process JSON data */
 838         ovs_db_json_data_process(pdb, json, json_len);
 839     }
 840   }
 841
 842   OVS_DEBUG("poll thread has been completed");
 843   ovs_json_reader_free(jreader);
 844   return (NULL);
 845 }
 846
 847 /* EVENT worker thread.
 848  * Perform task based on incoming events. This
 849  * task can be done asynchronously which allows to
 850  * handle OVS DB callback like 'init_cb'.
 851  */
 852 static void *ovs_event_worker(void *arg) {
 853   ovs_db_t *pdb = (ovs_db_t *)arg;
 854
 855   while (pdb->event_thread.value != OVS_DB_EVENT_TERMINATE) {
 856     /* wait for an event */
 857     struct timespec ts;
 858     clock_gettime(CLOCK_REALTIME, &ts);
 859     ts.tv_sec += (OVS_DB_EVENT_TIMEOUT);
 860     int ret = pthread_cond_timedwait(&pdb->event_thread.cond,
 861                                      &pdb->event_thread.mutex, &ts);
 862     if (!ret) {
 863       /* handle the event */
 864       OVS_DEBUG("handle event %d", pdb->event_thread.value);
 865       switch (pdb->event_thread.value) {
 866       case OVS_DB_EVENT_CONN_ESTABLISHED:
 867         if (pdb->cb.post_conn_init)
 868           pdb->cb.post_conn_init(pdb);
 869         break;
 870       case OVS_DB_EVENT_CONN_TERMINATED:
 871         if (pdb->cb.post_conn_terminate)
 872           pdb->cb.post_conn_terminate();
 873         break;
 874       default:
 875         OVS_DEBUG("unknown event received");
 876         break;
 877       }
 878     } else if (ret == ETIMEDOUT) {
 879       /* wait timeout */
 880       OVS_DEBUG("no event received (timeout)");
 881       continue;
 882     } else {
 883       /* unexpected error */
 884       OVS_ERROR("pthread_cond_timedwait() failed");
 885       break;
 886     }
 887   }
 888
 889   OVS_DEBUG("event thread has been completed");
 890   return (NULL);
 891 }
 892
 893 /* Initialize EVENT thread */
 894 static int ovs_db_event_thread_init(ovs_db_t *pdb) {
 895   pdb->event_thread.tid = -1;
 896   /* init event thread condition variable */
 897   if (pthread_cond_init(&pdb->event_thread.cond, NULL)) {
 898     return (-1);
 899   }
 900   /* init event thread mutex */
 901   if (pthread_mutex_init(&pdb->event_thread.mutex, NULL)) {
 902     pthread_cond_destroy(&pdb->event_thread.cond);
 903     return (-1);
 904   }
 905   /* Hold the event thread mutex. It ensures that no events
 906    * will be lost while thread is still starting. Once event
 907    * thread is started and ready to accept events, it will release
 908    * the mutex */
 909   if (pthread_mutex_lock(&pdb->event_thread.mutex)) {
 910     pthread_mutex_destroy(&pdb->event_thread.mutex);
 911     pthread_cond_destroy(&pdb->event_thread.cond);
 912     return (-1);
 913   }
 914   /* start event thread */
 915   pthread_t tid;
 916   if (plugin_thread_create(&tid, NULL, ovs_event_worker, pdb,
 917                            "utils_ovs:event") != 0) {
 918     pthread_mutex_unlock(&pdb->event_thread.mutex);
 919     pthread_mutex_destroy(&pdb->event_thread.mutex);
 920     pthread_cond_destroy(&pdb->event_thread.cond);
 921     return (-1);
 922   }
 923   pdb->event_thread.tid = tid;
 924   return (0);
 925 }
 926
 927 /* Destroy EVENT thread */
 928 static int ovs_db_event_thread_destroy(ovs_db_t *pdb) {
 929   if (pdb->event_thread.tid < 0)
 930     /* already destroyed */
 931     return (0);
 932   ovs_db_event_post(pdb, OVS_DB_EVENT_TERMINATE);
 933   if (pthread_join(pdb->event_thread.tid, NULL) != 0)
 934     return (-1);
 935   /* Event thread always holds the thread mutex when
 936    * performs some task (handles event) and releases it when
 937    * while sleeping. Thus, if event thread exits, the mutex
 938    * remains locked */
 939   pthread_mutex_unlock(&pdb->event_thread.mutex);
 940   pthread_mutex_destroy(&pdb->event_thread.mutex);
 941   pthread_cond_destroy(&pdb->event_thread.cond);
 942   pdb->event_thread.tid = -1;
 943   return (0);
 944 }
 945
 946 /* Initialize POLL thread */
 947 static int ovs_db_poll_thread_init(ovs_db_t *pdb) {
 948   pdb->poll_thread.tid = -1;
 949   /* init event thread mutex */
 950   if (pthread_mutex_init(&pdb->poll_thread.mutex, NULL)) {
 951     return (-1);
 952   }
 953   /* start poll thread */
 954   pthread_t tid;
 955   pdb->poll_thread.state = OVS_DB_POLL_STATE_RUNNING;
 956   if (plugin_thread_create(&tid, NULL, ovs_poll_worker, pdb,
 957                            "utils_ovs:poll") != 0) {
 958     pthread_mutex_destroy(&pdb->poll_thread.mutex);
 959     return (-1);
 960   }
 961   pdb->poll_thread.tid = tid;
 962   return (0);
 963 }
 964
 965 /* Destroy POLL thread */
 966 static int ovs_db_poll_thread_destroy(ovs_db_t *pdb) {
 967   if (pdb->poll_thread.tid < 0)
 968     /* already destroyed */
 969     return (0);
 970   /* change thread state */
 971   pthread_mutex_lock(&pdb->poll_thread.mutex);
 972   pdb->poll_thread.state = OVS_DB_POLL_STATE_EXITING;
 973   pthread_mutex_unlock(&pdb->poll_thread.mutex);
 974   /* join the thread */
 975   if (pthread_join(pdb->poll_thread.tid, NULL) != 0)
 976     return (-1);
 977   pthread_mutex_destroy(&pdb->poll_thread.mutex);
 978   pdb->poll_thread.tid = -1;
 979   return (0);
 980 }
 981
 982 /*
 983  * Public OVS DB API implementation
 984  */
 985
 986 ovs_db_t *ovs_db_init(const char *node, const char *service,
 987                       const char *unix_path, ovs_db_callback_t *cb) {
 988   /* sanity check */
 989   if (node == NULL || service == NULL || unix_path == NULL)
 990     return (NULL);
 991
 992   /* allocate db data & fill it */
 993   ovs_db_t *pdb = pdb = calloc(1, sizeof(*pdb));
 994   if (pdb == NULL)
 995     return (NULL);
 996
 997   /* store the OVS DB address */
 998   sstrncpy(pdb->node, node, sizeof(pdb->node));
 999   sstrncpy(pdb->service, service, sizeof(pdb->service));
1000   sstrncpy(pdb->unix_path, unix_path, sizeof(pdb->unix_path));
1001
1002   /* setup OVS DB callbacks */
1003   if (cb)
1004     pdb->cb = *cb;
1005
1006   /* init OVS DB mutex attributes */
1007   pthread_mutexattr_t mutex_attr;
1008   if (pthread_mutexattr_init(&mutex_attr)) {
1009     OVS_ERROR("OVS DB mutex attribute init failed");
1010     sfree(pdb);
1011     return (NULL);
1012   }
1013   /* set OVS DB mutex as recursive */
1014   if (pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_RECURSIVE)) {
1015     OVS_ERROR("Failed to set OVS DB mutex as recursive");
1016     pthread_mutexattr_destroy(&mutex_attr);
1017     sfree(pdb);
1018     return (NULL);
1019   }
1020   /* init OVS DB mutex */
1021   if (pthread_mutex_init(&pdb->mutex, &mutex_attr)) {
1022     OVS_ERROR("OVS DB mutex init failed");
1023     pthread_mutexattr_destroy(&mutex_attr);
1024     sfree(pdb);
1025     return (NULL);
1026   }
1027   /* destroy mutex attributes */
1028   pthread_mutexattr_destroy(&mutex_attr);
1029
1030   /* init event thread */
1031   if (ovs_db_event_thread_init(pdb) < 0) {
1032     ovs_db_destroy(pdb);
1033     return (NULL);
1034   }
1035
1036   /* init polling thread */
1037   pdb->sock = -1;
1038   if (ovs_db_poll_thread_init(pdb) < 0) {
1039     ovs_db_destroy(pdb);
1040     return (NULL);
1041   }
1042   return pdb;
1043 }
1044
1045 int ovs_db_send_request(ovs_db_t *pdb, const char *method, const char *params,
1046                         ovs_db_result_cb_t cb) {
1047   int ret = 0;
1048   yajl_gen_status yajl_gen_ret;
1049   yajl_val jparams;
1050   yajl_gen jgen;
1051   ovs_callback_t *new_cb = NULL;
1052   uint64_t uid;
1053   char uid_buff[OVS_UID_STR_SIZE];
1054   const char *req = NULL;
1055   size_t req_len = 0;
1056   struct timespec ts;
1057
1058   /* sanity check */
1059   if (!pdb || !method || !params)
1060     return (-1);
1061
1062   if ((jgen = yajl_gen_alloc(NULL)) == NULL)
1063     return (-1);
1064
1065   /* try to parse params */
1066   if ((jparams = yajl_tree_parse(params, NULL, 0)) == NULL) {
1067     OVS_ERROR("params is not a JSON string");
1068     yajl_gen_clear(jgen);
1069     return (-1);
1070   }
1071
1072   /* generate method field */
1073   OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1074
1075   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "method");
1076   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, method);
1077
1078   /* generate params field */
1079   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "params");
1080   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
1081   yajl_tree_free(jparams);
1082
1083   /* generate id field */
1084   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
1085   uid = ovs_uid_generate();
1086   ssnprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid);
1087   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_buff);
1088
1089   OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1090
1091   if (cb) {
1092     /* register result callback */
1093     if ((new_cb = calloc(1, sizeof(ovs_callback_t))) == NULL)
1094       goto yajl_gen_failure;
1095
1096     /* add new callback to front */
1097     sem_init(&new_cb->result.sync, 0, 0);
1098     new_cb->result.call = cb;
1099     new_cb->uid = uid;
1100     ovs_db_callback_add(pdb, new_cb);
1101   }
1102
1103   /* send the request */
1104   OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&req, &req_len);
1105   OVS_DEBUG("%s", req);
1106   if (!ovs_db_data_send(pdb, req, req_len)) {
1107     if (cb) {
1108       /* wait for result */
1109       clock_gettime(CLOCK_REALTIME, &ts);
1110       ts.tv_sec += OVS_DB_SEND_REQ_TIMEOUT;
1111       if (sem_timedwait(&new_cb->result.sync, &ts) < 0) {
1112         OVS_ERROR("%s() no replay received within %d sec", __FUNCTION__,
1113                   OVS_DB_SEND_REQ_TIMEOUT);
1114         ret = (-1);
1115       }
1116     }
1117   } else {
1118     OVS_ERROR("ovs_db_data_send() failed");
1119     ret = (-1);
1120   }
1121
1122 yajl_gen_failure:
1123   if (new_cb) {
1124     /* destroy callback */
1125     sem_destroy(&new_cb->result.sync);
1126     ovs_db_callback_remove(pdb, new_cb);
1127   }
1128
1129   /* release memory */
1130   yajl_gen_clear(jgen);
1131   return (yajl_gen_ret != yajl_gen_status_ok) ? (-1) : ret;
1132 }
1133
1134 int ovs_db_table_cb_register(ovs_db_t *pdb, const char *tb_name,
1135                              const char **tb_column,
1136                              ovs_db_table_cb_t update_cb,
1137                              ovs_db_result_cb_t result_cb, unsigned int flags) {
1138   yajl_gen jgen;
1139   yajl_gen_status yajl_gen_ret;
1140   ovs_callback_t *new_cb = NULL;
1141   char uid_str[OVS_UID_STR_SIZE];
1142   char *params;
1143   size_t params_len;
1144   int ovs_db_ret = 0;
1145
1146   /* sanity check */
1147   if (pdb == NULL || tb_name == NULL || update_cb == NULL)
1148     return (-1);
1149
1150   /* allocate new update callback */
1151   if ((new_cb = calloc(1, sizeof(ovs_callback_t))) == NULL)
1152     return (-1);
1153
1154   /* init YAJL generator */
1155   if ((jgen = yajl_gen_alloc(NULL)) == NULL) {
1156     sfree(new_cb);
1157     return (-1);
1158   }
1159
1160   /* add new callback to front */
1161   new_cb->table.call = update_cb;
1162   new_cb->uid = ovs_uid_generate();
1163   ovs_db_callback_add(pdb, new_cb);
1164
1165   /* make update notification request
1166    * [<db-name>, <json-value>, <monitor-requests>] */
1167   OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1168   {
1169     OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, OVS_DB_DEFAULT_DB_NAME);
1170
1171     /* uid string <json-value> */
1172     ssnprintf(uid_str, sizeof(uid_str), "%" PRIX64, new_cb->uid);
1173     OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_str);
1174
1175     /* <monitor-requests> */
1176     OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1177     {
1178       OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, tb_name);
1179       OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1180       {
1181         /* <monitor-request> */
1182         OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1183         {
1184           if (tb_column) {
1185             /* columns within the table to be monitored */
1186             OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "columns");
1187             OVS_YAJL_CALL(yajl_gen_array_open, jgen);
1188             for (; *tb_column; tb_column++)
1189               OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, *tb_column);
1190             OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1191           }
1192           /* specify select option */
1193           OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "select");
1194           {
1195             OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1196             {
1197               OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "initial");
1198               OVS_YAJL_CALL(yajl_gen_bool, jgen,
1199                             flags & OVS_DB_TABLE_CB_FLAG_INITIAL);
1200               OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "insert");
1201               OVS_YAJL_CALL(yajl_gen_bool, jgen,
1202                             flags & OVS_DB_TABLE_CB_FLAG_INSERT);
1203               OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "delete");
1204               OVS_YAJL_CALL(yajl_gen_bool, jgen,
1205                             flags & OVS_DB_TABLE_CB_FLAG_DELETE);
1206               OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "modify");
1207               OVS_YAJL_CALL(yajl_gen_bool, jgen,
1208                             flags & OVS_DB_TABLE_CB_FLAG_MODIFY);
1209             }
1210             OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1211           }
1212         }
1213         OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1214       }
1215       OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1216     }
1217     OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1218   }
1219   OVS_YAJL_CALL(yajl_gen_array_close, jgen);
1220
1221   /* make a request to subscribe to given table */
1222   OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&params,
1223                 &params_len);
1224   if (ovs_db_send_request(pdb, "monitor", params, result_cb) < 0) {
1225     OVS_ERROR("Failed to subscribe to \"%s\" table", tb_name);
1226     ovs_db_ret = (-1);
1227   }
1228
1229 yajl_gen_failure:
1230   /* release memory */
1231   yajl_gen_clear(jgen);
1232   return ovs_db_ret;
1233 }
1234
1235 int ovs_db_destroy(ovs_db_t *pdb) {
1236   int ovs_db_ret = 0;
1237   int ret = 0;
1238
1239   /* sanity check */
1240   if (pdb == NULL)
1241     return (-1);
1242
1243   /* try to lock the structure before releasing */
1244   if ((ret = pthread_mutex_lock(&pdb->mutex))) {
1245     OVS_ERROR("pthread_mutex_lock() DB mutex lock failed (%d)", ret);
1246     return (-1);
1247   }
1248
1249   /* stop poll thread */
1250   if (ovs_db_event_thread_destroy(pdb) < 0) {
1251     OVS_ERROR("destroy poll thread failed");
1252     ovs_db_ret = (-1);
1253   }
1254
1255   /* stop event thread */
1256   if (ovs_db_poll_thread_destroy(pdb) < 0) {
1257     OVS_ERROR("stop event thread failed");
1258     ovs_db_ret = (-1);
1259   }
1260
1261   /* unsubscribe callbacks */
1262   ovs_db_callback_remove_all(pdb);
1263
1264   /* close connection */
1265   if (pdb->sock >= 0)
1266     close(pdb->sock);
1267
1268   /* release DB handler */
1269   pthread_mutex_unlock(&pdb->mutex);
1270   pthread_mutex_destroy(&pdb->mutex);
1271   sfree(pdb);
1272   return ovs_db_ret;
1273 }
1274
1275 /*
1276  * Public OVS utils API implementation
1277  */
1278
1279 /* Get YAJL value by key from YAJL dictionary
1280  *
1281  * EXAMPLE:
1282  *  {
1283  *    "key_a" : <YAJL return value>
1284  *    "key_b" : <YAJL return value>
1285  *  }
1286  */
1287 yajl_val ovs_utils_get_value_by_key(yajl_val jval, const char *key) {
1288   const char *obj_key = NULL;
1289
1290   /* check params */
1291   if (!YAJL_IS_OBJECT(jval) || (key == NULL))
1292     return NULL;
1293
1294   /* find a value by key */
1295   for (int i = 0; i < YAJL_GET_OBJECT(jval)->len; i++) {
1296     obj_key = YAJL_GET_OBJECT(jval)->keys[i];
1297     if (strcmp(obj_key, key) == 0)
1298       return YAJL_GET_OBJECT(jval)->values[i];
1299   }
1300
1301   return NULL;
1302 }
1303
1304 /* Get OVS DB map value by given map key
1305  *
1306  * FROM RFC7047:
1307  *
1308  *   <pair>
1309  *     A 2-element JSON array that represents a pair within a database
1310  *     map.  The first element is an <atom> that represents the key, and
1311  *     the second element is an <atom> that represents the value.
1312  *
1313  *   <map>
1314  *     A 2-element JSON array that represents a database map value.  The
1315  *     first element of the array must be the string "map", and the
1316  *     second element must be an array of zero or more <pair>s giving the
1317  *     values in the map.  All of the <pair>s must have the same key and
1318  *     value types.
1319  *
1320  * EXAMPLE:
1321  *  [
1322  *    "map", [
1323  *             [ "key_a", <YAJL value>], [ "key_b", <YAJL value>], ...
1324  *           ]
1325  *  ]
1326  */
1327 yajl_val ovs_utils_get_map_value(yajl_val jval, const char *key) {
1328   size_t map_len = 0;
1329   size_t array_len = 0;
1330   yajl_val *map_values = NULL;
1331   yajl_val *array_values = NULL;
1332   const char *str_val = NULL;
1333
1334   /* check YAJL array */
1335   if (!YAJL_IS_ARRAY(jval) || (key == NULL))
1336     return NULL;
1337
1338   /* check a database map value (2-element, first one should be a string */
1339   array_len = YAJL_GET_ARRAY(jval)->len;
1340   array_values = YAJL_GET_ARRAY(jval)->values;
1341   if ((array_len != 2) || (!YAJL_IS_STRING(array_values[0])) ||
1342       (!YAJL_IS_ARRAY(array_values[1])))
1343     return NULL;
1344
1345   /* check first element of the array */
1346   str_val = YAJL_GET_STRING(array_values[0]);
1347   if (strcmp("map", str_val) != 0)
1348     return NULL;
1349
1350   /* try to find map value by map key */
1351   map_len = YAJL_GET_ARRAY(array_values[1])->len;
1352   map_values = YAJL_GET_ARRAY(array_values[1])->values;
1353   for (int i = 0; i < map_len; i++) {
1354     /* check YAJL array */
1355     if (!YAJL_IS_ARRAY(map_values[i]))
1356       break;
1357
1358     /* check a database pair value (2-element, first one represents a key
1359      * and it should be a string in our case */
1360     array_len = YAJL_GET_ARRAY(map_values[i])->len;
1361     array_values = YAJL_GET_ARRAY(map_values[i])->values;
1362     if ((array_len != 2) || (!YAJL_IS_STRING(array_values[0])))
1363       break;
1364
1365     /* return map value if given key equals map key */
1366     str_val = YAJL_GET_STRING(array_values[0]);
1367     if (strcmp(key, str_val) == 0)
1368       return array_values[1];
1369   }
1370   return NULL;
1371 }