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  *                         OVS DB API internal architecture diagram
  28  * +------------------------------------------------------------------------------+
  29  * |OVS plugin      |OVS utils                                                    |
  30  * |                |     +------------------------+                              |
  31  * |                |     |      echo handler      |                JSON request/ |
  32  * |                |  +--+ (ovs_db_table_echo_cb) +<---+---------+ update event/ |
  33  * |                |  |  |                        |    |         | result        |
  34  * |                |  |  +------------------------+    |         |               |
  35  * |                |  |                                |    +----+---+--------+  |
  36  * |  +----------+  |  |  +------------------------+    |    |        |        |  |
  37  * |  |  update  |  |  |  |     update handler     |    |    |  YAJL  |  JSON  |  |
  38  * |  | callback +<-------+(ovs_db_table_update_cp)+<---+    | parser | reader |  |
  39  * |  +----------+  |  |  |                        |    |    |        |        |  |
  40  * |                |  |  +------------------------+    |    +--------+---+----+  |
  41  * |                |  |                                |                 ^       |
  42  * |  +----------+  |  |  +------------------------+    |                 |       |
  43  * |  |  result  |  |  |  |     result handler     |    |                 |       |
  44  * |  | callback +<-------+   (ovs_db_result_cb)   +<---+        JSON raw |       |
  45  * |  +----------+  |  |  |                        |               data   |       |
  46  * |                |  |  +------------------------+                      |       |
  47  * |                |  |                                                  |       |
  48  * |                |  |    +------------------+             +------------+----+  |
  49  * |  +----------+  |  |    |thread|           |             |thread|          |  |
  50  * |  |   init   |  |  |    |                  |  reconnect  |                 |  |
  51  * |  | callback +<---------+   EVENT WORKER   +<------------+   POLL WORKER   |  |
  52  * |  +----------+  |  |    +------------------+             +--------+--------+  |
  53  * |                |  |                                              ^           |
  54  * +----------------+-------------------------------------------------------------+
  55  *                     |                                              |
  56  *                 JSON|echo reply                                 raw|data
  57  *                     v                                              v
  58  * +-------------------+----------------------------------------------+-----------+
  59  * |                                 TCP/UNIX socket                              |
  60  * +-------------------------------------------------------------------------------
  61  *
  62  **/
  63
  64 /* collectd headers */
  65 #include "common.h"
  66
  67 /* private headers */
  68 #include "utils_ovs.h"
  69
  70 /* system libraries */
  71 #include <semaphore.h>
  72 #include <arpa/inet.h>
  73 #include <poll.h>
  74 #include <sys/un.h>
  75
  76 #define OVS_ERROR(fmt, ...) do { \
  77   ERROR("ovs_utils: "fmt, ## __VA_ARGS__); } while (0)
  78 #define OVS_DEBUG(fmt, ...) do { \
  79   DEBUG("%s:%d:%s(): "fmt, __FILE__, __LINE__, __FUNCTION__, \
  80         ## __VA_ARGS__); } while (0)
  81
  82 #define OVS_DB_POLL_TIMEOUT          1  /* poll receive timeout (sec) */
  83 #define OVS_DB_POLL_READ_BLOCK_SIZE  5  /* read block size (bytes) */
  84 #define OVS_DB_DEFAULT_DB_NAME       "Open_vSwitch"
  85 #define OVS_DB_RECONNECT_TIMEOUT     1  /* reconnect timeout (sec) */
  86
  87 #define OVS_DB_EVENT_TIMEOUT         5  /* event thread timeout (sec) */
  88 #define OVS_DB_EVENT_TERMINATE       1
  89 #define OVS_DB_EVENT_CONN_ESTABLISHED     2
  90 #define OVS_DB_EVENT_CONN_TERMINATED      3
  91
  92 #define OVS_DB_POLL_STATE_RUNNING    1
  93 #define OVS_DB_POLL_STATE_EXITING    2
  94
  95 #define OVS_DB_SEND_REQ_TIMEOUT      5  /* send request timeout (sec) */
  96
  97 #define OVS_YAJL_CALL(func, ...) \
  98   do { \
  99     yajl_gen_ret = yajl_gen_status_ok; \
 100     if ((yajl_gen_ret = func(__VA_ARGS__)) != yajl_gen_status_ok) \
 101       goto yajl_gen_failure; \
 102   } while (0)
 103 #define OVS_YAJL_ERROR_BUFFER_SIZE       1024
 104 #define OVS_ERROR_BUFF_SIZE              512
 105 #define OVS_UID_STR_SIZE                 17     /* 64-bit HEX string len + '\0' */
 106
 107 /* JSON reader internal data */
 108 struct ovs_json_reader_s {
 109   char *buff_ptr;
 110   size_t buff_size;
 111   size_t buff_offset;
 112   size_t json_offset;
 113 };
 114 typedef struct ovs_json_reader_s ovs_json_reader_t;
 115
 116 /* Result callback declaration */
 117 struct ovs_result_cb_s {
 118   sem_t sync;
 119   ovs_db_result_cb_t call;
 120 };
 121 typedef struct ovs_result_cb_s ovs_result_cb_t;
 122
 123 /* Table callback declaration */
 124 struct ovs_table_cb_s {
 125   ovs_db_table_cb_t call;
 126 };
 127 typedef struct ovs_table_cb_s ovs_table_cb_t;
 128
 129 /* Callback declaration */
 130 struct ovs_callback_s {
 131   uint64_t uid;
 132   union {
 133     ovs_result_cb_t result;
 134     ovs_table_cb_t table;
 135   };
 136   struct ovs_callback_s *next;
 137   struct ovs_callback_s *prev;
 138 };
 139 typedef struct ovs_callback_s ovs_callback_t;
 140
 141 /* Connection declaration */
 142 struct ovs_conn_s {
 143   int sock;
 144   int domain;
 145   int type;
 146   int addr_size;
 147   union {
 148     struct sockaddr_in s_inet;
 149     struct sockaddr_un s_unix;
 150   } addr;
 151 };
 152 typedef struct ovs_conn_s ovs_conn_t;
 153
 154 /* Event thread data declaration */
 155 struct ovs_event_thread_s {
 156   pthread_t tid;
 157   pthread_mutex_t mutex;
 158   pthread_cond_t cond;
 159   int value;
 160 };
 161 typedef struct ovs_event_thread_s ovs_event_thread_t;
 162
 163 /* Poll thread data declaration */
 164 struct ovs_poll_thread_s {
 165   pthread_t tid;
 166   pthread_mutex_t mutex;
 167   int state;
 168 };
 169 typedef struct ovs_poll_thread_s ovs_poll_thread_t;
 170
 171 /* OVS DB internal data declaration */
 172 struct ovs_db_s {
 173   ovs_poll_thread_t poll_thread;
 174   ovs_event_thread_t event_thread;
 175   pthread_mutex_t mutex;
 176   ovs_callback_t *remote_cb;
 177   ovs_db_callback_t cb;
 178   ovs_conn_t conn;
 179 };
 180 typedef struct ovs_db_s ovs_db_t;
 181
 182 /* Post an event to event thread.
 183  * Possible events are:
 184  *  OVS_DB_EVENT_TERMINATE
 185  *  OVS_DB_EVENT_CONN_ESTABLISHED
 186  *  OVS_DB_EVENT_CONN_TERMINATED
 187  */
 188 static void
 189 ovs_db_event_post(ovs_db_t *pdb, int event)
 190 {
 191   pthread_mutex_lock(&pdb->event_thread.mutex);
 192   pdb->event_thread.value = event;
 193   pthread_mutex_unlock(&pdb->event_thread.mutex);
 194   pthread_cond_signal(&pdb->event_thread.cond);
 195 }
 196
 197 /* Check if POLL thread is still running. Returns
 198  * 1 if running otherwise 0 is returned */
 199 static inline int
 200 ovs_db_poll_is_running(ovs_db_t *pdb)
 201 {
 202   int state = 0;
 203   pthread_mutex_lock(&pdb->poll_thread.mutex);
 204   state = pdb->poll_thread.state;
 205   pthread_mutex_unlock(&pdb->poll_thread.mutex);
 206   return (state == OVS_DB_POLL_STATE_RUNNING);
 207 }
 208
 209 /* Terminate POLL thread */
 210 static inline void
 211 ovs_db_poll_terminate(ovs_db_t *pdb)
 212 {
 213   pthread_mutex_lock(&pdb->poll_thread.mutex);
 214   pdb->poll_thread.state = OVS_DB_POLL_STATE_EXITING;
 215   pthread_mutex_unlock(&pdb->poll_thread.mutex);
 216 }
 217
 218 /* Generate unique identifier (UID). It is used by OVS DB API
 219  * to set "id" field for any OVS DB JSON request. */
 220 static uint64_t
 221 ovs_uid_generate()
 222 {
 223   struct timespec ts;
 224   clock_gettime(CLOCK_MONOTONIC, &ts);
 225   return ((ts.tv_sec << 32) | (ts.tv_nsec & UINT32_MAX));
 226 }
 227
 228 /*
 229  * Callback API. These function are used to store
 230  * registered callbacks in OVS DB API.
 231  */
 232
 233 /* Add new callback into OVS DB object */
 234 static void
 235 ovs_db_callback_add(ovs_db_t *pdb, ovs_callback_t *new_cb)
 236 {
 237   pthread_mutex_lock(&pdb->mutex);
 238   if (pdb->remote_cb)
 239     pdb->remote_cb->prev = new_cb;
 240   new_cb->next = pdb->remote_cb;
 241   new_cb->prev = NULL;
 242   pdb->remote_cb = new_cb;
 243   pthread_mutex_unlock(&pdb->mutex);
 244 }
 245
 246 /* Remove callback from OVS DB object */
 247 static void
 248 ovs_db_callback_remove(ovs_db_t *pdb, ovs_callback_t *del_cb)
 249 {
 250   ovs_callback_t *pre_cb = del_cb->prev;
 251   ovs_callback_t *next_cb = del_cb->next;
 252
 253   pthread_mutex_lock(&pdb->mutex);
 254   if (next_cb)
 255     next_cb->prev = del_cb->prev;
 256
 257   if (pre_cb)
 258     pre_cb->next = del_cb->next;
 259   else
 260     pdb->remote_cb = del_cb->next;
 261
 262   free(del_cb);
 263   pthread_mutex_unlock(&pdb->mutex);
 264 }
 265
 266 /* Remove all callbacks form OVS DB object */
 267 static void
 268 ovs_db_callback_remove_all(ovs_db_t *pdb)
 269 {
 270   pthread_mutex_lock(&pdb->mutex);
 271   for (ovs_callback_t *del_cb = pdb->remote_cb; pdb->remote_cb;
 272        del_cb = pdb->remote_cb) {
 273     pdb->remote_cb = pdb->remote_cb->next;
 274     free(del_cb);
 275   }
 276   pdb->remote_cb = NULL;
 277   pthread_mutex_unlock(&pdb->mutex);
 278 }
 279
 280 /* Get/find callback in OVS DB object by UID. Returns pointer
 281  * to requested callback otherwise NULL is returned */
 282 static ovs_callback_t *
 283 ovs_db_callback_get(ovs_db_t *pdb, uint64_t uid)
 284 {
 285   pthread_mutex_lock(&pdb->mutex);
 286   for (ovs_callback_t *cb = pdb->remote_cb; cb != NULL; cb = cb->next)
 287     if (cb->uid == uid) {
 288       pthread_mutex_unlock(&pdb->mutex);
 289       return cb;
 290     }
 291   pthread_mutex_unlock(&pdb->mutex);
 292   return NULL;
 293 }
 294
 295 /* Send all requested data to the socket. Returns 0 if
 296  * ALL request data has been sent otherwise negative value
 297  * is returned */
 298 static int
 299 ovs_db_data_send(const ovs_db_t *pdb, const char *data, size_t len)
 300 {
 301   ssize_t nbytes = 0;
 302   size_t rem = len;
 303   size_t off = 0;
 304
 305   while (rem > 0) {
 306     if ((nbytes = send(pdb->conn.sock, data + off, rem, 0)) <= 0)
 307       return (-1);
 308     rem -= (size_t)nbytes;
 309     off += (size_t)nbytes;
 310   }
 311   return (0);
 312 }
 313
 314 /* Parse OVS server URL.
 315  * Format of the URL:
 316  *   "tcp:a.b.c.d:port" - define TCP connection (INET domain)
 317  *   "unix:file" - define UNIX socket file (UNIX domain)
 318  */
 319 static int
 320 ovs_db_url_parse(const char *surl, ovs_conn_t *conn)
 321 {
 322   ovs_conn_t tmp_conn;
 323   char *nexttok = NULL;
 324   char *in_str = NULL;
 325   char *saveptr;
 326   int ret = 0;
 327
 328   /* sanity check */
 329   if ((surl == NULL) || (strlen(surl) < 1))
 330     return (-1);
 331
 332   /* parse domain */
 333   tmp_conn = *conn;
 334   in_str = sstrdup(surl);
 335   if ((nexttok = strtok_r(in_str, ":", &saveptr)) != NULL) {
 336     if (strcmp("tcp", nexttok) == 0) {
 337       tmp_conn.domain = AF_INET;
 338       tmp_conn.type = SOCK_STREAM;
 339       tmp_conn.addr_size = sizeof(tmp_conn.addr.s_inet);
 340     } else if (strcmp("unix", nexttok) == 0) {
 341       tmp_conn.domain = AF_UNIX;
 342       tmp_conn.type = SOCK_STREAM;
 343       tmp_conn.addr_size = sizeof(tmp_conn.addr.s_unix);
 344     } else
 345       goto failure;
 346   } else
 347     goto failure;
 348
 349   /* parse url depending on domain */
 350   if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL) {
 351     if (tmp_conn.domain == AF_UNIX) {
 352       /* <UNIX-NAME> */
 353       tmp_conn.addr.s_inet.sin_family = AF_UNIX;
 354       sstrncpy(tmp_conn.addr.s_unix.sun_path, nexttok, strlen(nexttok) + 1);
 355     } else {
 356       /* <IP:PORT> */
 357       tmp_conn.addr.s_inet.sin_family = AF_INET;
 358       ret =
 359         inet_pton(AF_INET, nexttok, (void *)&tmp_conn.addr.s_inet.sin_addr);
 360       if (ret == 1) {
 361         if ((nexttok = strtok_r(NULL, ":", &saveptr)) != NULL)
 362           tmp_conn.addr.s_inet.sin_port = htons(atoi(nexttok));
 363         else
 364           goto failure;
 365       } else
 366         goto failure;
 367     }
 368   }
 369
 370   /* save result and return success */
 371   *conn = tmp_conn;
 372   sfree(in_str);
 373   return (0);
 374
 375 failure:
 376   OVS_ERROR("%s() : invalid OVS DB URL provided");
 377   sfree(in_str);
 378   return (-1);
 379 }
 380
 381 /*
 382  * YAJL (Yet Another JSON Library) helper functions
 383  * Documentation (https://lloyd.github.io/yajl/)
 384  */
 385
 386 /* Add null-terminated string into YAJL generator handle (JSON object).
 387  * Similar function to yajl_gen_string() but takes null-terminated string
 388  * instead of string and its length.
 389  *
 390  * jgen   - YAJL generator handle allocated by yajl_gen_alloc()
 391  * string - Null-terminated string
 392  */
 393 static inline yajl_gen_status
 394 ovs_yajl_gen_tstring(yajl_gen hander, const char *string)
 395 {
 396   return yajl_gen_string(hander, string, strlen(string));
 397 }
 398
 399 /* Add YAJL value into YAJL generator handle (JSON object)
 400  *
 401  * jgen - YAJL generator handle allocated by yajl_gen_alloc()
 402  * jval - YAJL value usually returned by yajl_tree_get()
 403  */
 404 static yajl_gen_status
 405 ovs_yajl_gen_val(yajl_gen jgen, yajl_val jval)
 406 {
 407   size_t array_len = 0;
 408   yajl_val *jvalues = NULL;
 409   yajl_val jobj_value = NULL;
 410   const char *obj_key = NULL;
 411   size_t obj_len = 0;
 412   yajl_gen_status yajl_gen_ret;
 413
 414   if (YAJL_IS_STRING(jval))
 415     OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, YAJL_GET_STRING(jval));
 416   else if (YAJL_IS_DOUBLE(jval))
 417     OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_DOUBLE(jval));
 418   else if (YAJL_IS_INTEGER(jval))
 419     OVS_YAJL_CALL(yajl_gen_double, jgen, YAJL_GET_INTEGER(jval));
 420   else if (YAJL_IS_TRUE(jval))
 421     OVS_YAJL_CALL(yajl_gen_bool, jgen, 1);
 422   else if (YAJL_IS_FALSE(jval))
 423     OVS_YAJL_CALL(yajl_gen_bool, jgen, 0);
 424   else if (YAJL_IS_NULL(jval))
 425     OVS_YAJL_CALL(yajl_gen_null, jgen);
 426   else if (YAJL_IS_ARRAY(jval)) {
 427     /* create new array and add all elements into the array */
 428     array_len = YAJL_GET_ARRAY(jval)->len;
 429     jvalues = YAJL_GET_ARRAY(jval)->values;
 430     OVS_YAJL_CALL(yajl_gen_array_open, jgen);
 431     for (int i = 0; i < array_len; i++)
 432       OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jvalues[i]);
 433     OVS_YAJL_CALL(yajl_gen_array_close, jgen);
 434   } else if (YAJL_IS_OBJECT(jval)) {
 435     /* create new object and add all elements into the object */
 436     OVS_YAJL_CALL(yajl_gen_map_open, jgen);
 437     obj_len = YAJL_GET_OBJECT(jval)->len;
 438     for (int i = 0; i < obj_len; i++) {
 439       obj_key = YAJL_GET_OBJECT(jval)->keys[i];
 440       jobj_value = YAJL_GET_OBJECT(jval)->values[i];
 441       OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, obj_key);
 442       OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jobj_value);
 443     }
 444     OVS_YAJL_CALL(yajl_gen_map_close, jgen);
 445   } else {
 446     OVS_ERROR("%s() unsupported value type %d (skip)", __FUNCTION__,
 447               (int)(jval)->type);
 448     goto yajl_gen_failure;
 449   }
 450   return yajl_gen_status_ok;
 451
 452 yajl_gen_failure:
 453   OVS_ERROR("%s() error to generate value", __FUNCTION__);
 454   return yajl_gen_ret;
 455 }
 456
 457 /* OVS DB echo request handler. When OVS DB sends
 458  * "echo" request to the client, client should generate
 459  * "echo" replay with the same content received in the
 460  * request */
 461 static int
 462 ovs_db_table_echo_cb(const ovs_db_t *pdb, yajl_val jnode)
 463 {
 464   yajl_val jparams;
 465   yajl_val jid;
 466   yajl_gen jgen;
 467   size_t resp_len = 0;
 468   const char *resp = NULL;
 469   const char *params_path[] = {"params", NULL};
 470   const char *id_path[] = {"id", NULL};
 471   yajl_gen_status yajl_gen_ret;
 472
 473   if ((jgen = yajl_gen_alloc(NULL)) == NULL)
 474     return (-1);
 475
 476   /* check & get request attributes */
 477   if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
 478       ((jid = yajl_tree_get(jnode, id_path, yajl_t_any)) == NULL)) {
 479     OVS_ERROR("parse echo request failed");
 480     goto yajl_gen_failure;
 481   }
 482
 483   /* generate JSON echo response */
 484   OVS_YAJL_CALL(yajl_gen_map_open, jgen);
 485
 486   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "result");
 487   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
 488
 489   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "error");
 490   OVS_YAJL_CALL(yajl_gen_null, jgen);
 491
 492   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
 493   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jid);
 494
 495   OVS_YAJL_CALL(yajl_gen_map_close, jgen);
 496   OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&resp,
 497                 &resp_len);
 498
 499   /* send the response */
 500   OVS_DEBUG("response: %s", resp);
 501   if (ovs_db_data_send(pdb, resp, resp_len) < 0) {
 502     OVS_ERROR("send echo reply failed");
 503     goto yajl_gen_failure;
 504   }
 505   /* clean up and return success */
 506   yajl_gen_clear(jgen);
 507   return (0);
 508
 509 yajl_gen_failure:
 510   /* release memory */
 511   yajl_gen_clear(jgen);
 512   return (-1);
 513 }
 514
 515 /* Get OVS DB registered callback by YAJL val. The YAJL
 516  * value should be YAJL string (UID). Returns NULL if
 517  * callback hasn't been found.
 518  */
 519 static ovs_callback_t *
 520 ovs_db_table_callback_get(ovs_db_t *pdb, yajl_val jid)
 521 {
 522   char *endptr = NULL;
 523   const char *suid = NULL;
 524   uint64_t uid;
 525
 526   if (jid && YAJL_IS_STRING(jid)) {
 527     suid = YAJL_GET_STRING(jid);
 528     uid = (uint64_t) strtoul(suid, &endptr, 16);
 529     if (*endptr == '\0' && uid)
 530       return ovs_db_callback_get(pdb, uid);
 531   }
 532
 533   return NULL;
 534 }
 535
 536 /* OVS DB table update event handler.
 537  * This callback is called by POLL thread if OVS DB
 538  * table update callback is received from the DB
 539  * server. Once registered callback found, it's called
 540  * by this handler. */
 541 static int
 542 ovs_db_table_update_cb(ovs_db_t *pdb, yajl_val jnode)
 543 {
 544   ovs_callback_t *cb = NULL;
 545   yajl_val jvalue;
 546   yajl_val jparams;
 547   yajl_val jtable_updates;
 548   yajl_val jtable_update;
 549   size_t obj_len = 0;
 550   const char *table_name = NULL;
 551   const char *params_path[] = {"params", NULL};
 552   const char *id_path[] = {"id", NULL};
 553
 554   /* check & get request attributes */
 555   if ((jparams = yajl_tree_get(jnode, params_path, yajl_t_array)) == NULL ||
 556       (yajl_tree_get(jnode, id_path, yajl_t_null) == NULL))
 557     goto ovs_failure;
 558
 559   /* check array length: [<json-value>, <table-updates>] */
 560   if (YAJL_GET_ARRAY(jparams)->len != 2)
 561     goto ovs_failure;
 562
 563   jvalue = YAJL_GET_ARRAY(jparams)->values[0];
 564   jtable_updates = YAJL_GET_ARRAY(jparams)->values[1];
 565   if ((!YAJL_IS_OBJECT(jtable_updates)) || (!YAJL_IS_STRING(jvalue)))
 566     goto ovs_failure;
 567
 568   /* find registered callback based on <json-value> */
 569   cb = ovs_db_table_callback_get(pdb, jvalue);
 570   if (cb == NULL || cb->table.call == NULL)
 571     goto ovs_failure;
 572
 573   /* call registered callback */
 574   cb->table.call(jtable_updates);
 575   return 0;
 576
 577 ovs_failure:
 578   OVS_ERROR("invalid OVS DB table update event");
 579   return (-1);
 580 }
 581
 582 /* OVS DB result request handler.
 583  * This callback is called by POLL thread if OVS DB
 584  * result reply is received from the DB server.
 585  * Once registered callback found, it's called
 586  * by this handler. */
 587 static int
 588 ovs_db_result_cb(ovs_db_t *pdb, yajl_val jnode)
 589 {
 590   ovs_callback_t *cb = NULL;
 591   yajl_val jresult;
 592   yajl_val jerror;
 593   yajl_val jid;
 594   const char *result_path[] = {"result", NULL};
 595   const char *error_path[] = {"error", NULL};
 596   const char *id_path[] = {"id", NULL};
 597
 598   jresult = yajl_tree_get(jnode, result_path, yajl_t_any);
 599   jerror = yajl_tree_get(jnode, error_path, yajl_t_any);
 600   jid = yajl_tree_get(jnode, id_path, yajl_t_string);
 601
 602   /* check & get result attributes */
 603   if (!jresult || !jerror || !jid)
 604     return (-1);
 605
 606   /* try to find registered callback */
 607   cb = ovs_db_table_callback_get(pdb, jid);
 608   if (cb != NULL && cb->result.call != NULL) {
 609     /* call registered callback */
 610     cb->result.call(jresult, jerror);
 611     /* unlock owner of the reply */
 612     sem_post(&cb->result.sync);
 613   }
 614
 615   return (0);
 616 }
 617
 618 /* Handle JSON data (one request) and call
 619  * appropriate event OVS DB handler. Currently,
 620  * update callback 'ovs_db_table_update_cb' and
 621  * result callback 'ovs_db_result_cb' is supported.
 622  */
 623 static int
 624 ovs_db_json_data_process(ovs_db_t *pdb, const char *data, size_t len)
 625 {
 626   const char *method = NULL;
 627   char yajl_errbuf[OVS_YAJL_ERROR_BUFFER_SIZE];
 628   const char *method_path[] = {"method", NULL};
 629   const char *result_path[] = {"result", NULL};
 630   char *sjson = NULL;
 631   yajl_val jnode, jval;
 632
 633   /* duplicate the data to make null-terminated string
 634    * required for yajl_tree_parse() */
 635   if ((sjson = strndup(data, len)) == NULL)
 636     return (-1);
 637
 638   OVS_DEBUG("%s", sjson);
 639
 640   /* parse json data */
 641   jnode = yajl_tree_parse(sjson, yajl_errbuf, sizeof(yajl_errbuf));
 642   if (jnode == NULL) {
 643     OVS_ERROR("yajl_tree_parse() %s", yajl_errbuf);
 644     return (-1);
 645   }
 646
 647   /* get method name */
 648   if (jval = yajl_tree_get(jnode, method_path, yajl_t_string)) {
 649     method = YAJL_GET_STRING(jval);
 650     if (strcmp("echo", method) == 0) {
 651       /* echo request from the server */
 652       if (ovs_db_table_echo_cb(pdb, jnode) < 0)
 653         OVS_ERROR("handle echo request failed");
 654     } else if (strcmp("update", method) == 0) {
 655       /* update notification */
 656       if (ovs_db_table_update_cb(pdb, jnode) < 0)
 657         OVS_ERROR("handle update notification failed");
 658     }
 659   } else if (jval = yajl_tree_get(jnode, result_path, yajl_t_object)) {
 660     /* result notification */
 661     if (ovs_db_result_cb(pdb, jnode) < 0)
 662       OVS_ERROR("handle result reply failed");
 663   }
 664
 665   /* release memory */
 666   yajl_tree_free(jnode);
 667   sfree(sjson);
 668   return (0);
 669 }
 670
 671 /*
 672  * JSON reader implementation.
 673  *
 674  * This module process raw JSON data (byte stream) and
 675  * returns fully-fledged JSON data which can be processed
 676  * (parsed) by YAJL later.
 677  */
 678
 679 /* Allocate JSON reader instance */
 680 static inline ovs_json_reader_t *
 681 ovs_json_reader_alloc()
 682 {
 683   ovs_json_reader_t *jreader = NULL;
 684
 685   if ((jreader = calloc(sizeof(ovs_json_reader_t), 1)) == NULL)
 686     return NULL;
 687
 688   return jreader;
 689 }
 690
 691 /* Push raw data into into the JSON reader for processing */
 692 static inline int
 693 ovs_json_reader_push_data(ovs_json_reader_t *jreader,
 694                           const char *data, size_t data_len)
 695 {
 696   char *new_buff = NULL;
 697   size_t available = jreader->buff_size - jreader->buff_offset;
 698
 699   /* check/update required memory space */
 700   if (available < data_len) {
 701     OVS_DEBUG("Reallocate buffer [size=%d, available=%d required=%d]",
 702               (int)jreader->buff_size, (int)available, (int)data_len);
 703
 704     /* allocate new chunk of memory */
 705     new_buff = realloc(jreader->buff_ptr, (jreader->buff_size + data_len));
 706     if (new_buff == NULL)
 707       return (-1);
 708
 709     /* point to new allocated memory */
 710     jreader->buff_ptr = new_buff;
 711     jreader->buff_size += data_len;
 712   }
 713
 714   /* store input data */
 715   memcpy(jreader->buff_ptr + jreader->buff_offset, data, data_len);
 716   jreader->buff_offset += data_len;
 717   return (0);
 718 }
 719
 720 /* Pop one fully-fledged JSON if already exists. Returns 0 if
 721  * completed JSON already exists otherwise negative value is
 722  * returned */
 723 static inline int
 724 ovs_json_reader_pop(ovs_json_reader_t *jreader,
 725                     const char **json_ptr, size_t *json_len_ptr)
 726 {
 727   size_t nbraces = 0;
 728   size_t json_len = 0;
 729   char *json = NULL;
 730
 731   /* search open/close brace */
 732   for (int i = jreader->json_offset; i < jreader->buff_offset; i++) {
 733     if (jreader->buff_ptr[i] == '{') {
 734       nbraces++;
 735     } else if (jreader->buff_ptr[i] == '}')
 736       if (nbraces)
 737         if (!(--nbraces)) {
 738           /* JSON data */
 739           *json_ptr = jreader->buff_ptr + jreader->json_offset;
 740           *json_len_ptr = json_len + 1;
 741           jreader->json_offset = i + 1;
 742           return (0);
 743         }
 744
 745     /* increase JSON data length */
 746     if (nbraces)
 747       json_len++;
 748   }
 749
 750   if (jreader->json_offset) {
 751     if (jreader->json_offset < jreader->buff_offset) {
 752       /* shift data to the beginning of the buffer
 753        * and zero rest of the buffer data */
 754       json = &jreader->buff_ptr[jreader->json_offset];
 755       json_len = jreader->buff_offset - jreader->json_offset;
 756       for (int i = 0; i < jreader->buff_size; i++)
 757         jreader->buff_ptr[i] = ((i < json_len) ? (json[i]) : (0));
 758       jreader->buff_offset = json_len;
 759     } else
 760       /* reset the buffer */
 761       jreader->buff_offset = 0;
 762
 763     /* data is at the beginning of the buffer */
 764     jreader->json_offset = 0;
 765   }
 766
 767   return (-1);
 768 }
 769
 770 /* Reset JSON reader. It is useful when start processing
 771  * new raw data. E.g.: in case of lost stream connection.
 772  */
 773 static inline void
 774 ovs_json_reader_reset(ovs_json_reader_t *jreader)
 775 {
 776   if (jreader) {
 777     jreader->buff_offset = 0;
 778     jreader->json_offset = 0;
 779   }
 780 }
 781
 782 /* Release internal data allocated for JSON reader */
 783 static inline void
 784 ovs_json_reader_free(ovs_json_reader_t *jreader)
 785 {
 786   if (jreader) {
 787     free(jreader->buff_ptr);
 788     free(jreader);
 789   }
 790 }
 791
 792 /* Reconnect to OVD DB and call init OVS DB callback
 793  * 'init_cb' if connection has been established.
 794  */
 795 static int
 796 ovs_db_reconnect(ovs_db_t *pdb)
 797 {
 798   char errbuff[OVS_ERROR_BUFF_SIZE];
 799
 800   /* remove all registered OVS DB table/result callbacks */
 801   ovs_db_callback_remove_all(pdb);
 802
 803   /* open new socket */
 804   if ((pdb->conn.sock = socket(pdb->conn.domain, pdb->conn.type, 0)) < 0) {
 805     sstrerror(errno, errbuff, sizeof(errbuff));
 806     OVS_ERROR("socket(): %s", errbuff);
 807     return (-1);
 808   }
 809
 810   /* try to connect to server */
 811   if (connect(pdb->conn.sock, (struct sockaddr *)&pdb->conn.addr,
 812               pdb->conn.addr_size) < 0) {
 813     sstrerror(errno, errbuff, sizeof(errbuff));
 814     OVS_ERROR("connect(): %s", errbuff);
 815     close(pdb->conn.sock);
 816     return (-1);
 817   }
 818
 819   /* send notification to event thread */
 820   ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_ESTABLISHED);
 821   return (0);
 822 }
 823
 824 /* POLL worker thread.
 825  * It listens on OVS DB connection for incoming
 826  * requests/reply/events etc. Also, it reconnects to OVS DB
 827  * if connection has been lost.
 828  */
 829 static void *
 830 ovs_poll_worker(void *arg)
 831 {
 832   ovs_db_t *pdb = (ovs_db_t *)arg;      /* pointer to OVS DB */
 833   ovs_json_reader_t *jreader = NULL;
 834   const char *json;
 835   size_t json_len;
 836   ssize_t nbytes = 0;
 837   char buff[OVS_DB_POLL_READ_BLOCK_SIZE];
 838   struct pollfd poll_fd;
 839   int poll_ret = 0;
 840
 841   if ((jreader = ovs_json_reader_alloc()) == NULL) {
 842     OVS_ERROR("initialize json reader failed");
 843     goto thread_exit;
 844   }
 845
 846   /* start polling data */
 847   poll_fd.fd = pdb->conn.sock;
 848   poll_fd.events = POLLIN | POLLPRI;
 849   poll_fd.revents = 0;
 850
 851   /* poll data */
 852   while (ovs_db_poll_is_running(pdb)) {
 853     poll_ret = poll(&poll_fd, 1, /* ms */ OVS_DB_POLL_TIMEOUT * 1000);
 854     if (poll_ret > 0) {
 855       if (poll_fd.revents & POLLNVAL) {
 856         /* invalid file descriptor, reconnect */
 857         if (ovs_db_reconnect(pdb) != 0) {
 858           /* sleep awhile until next reconnect */
 859           usleep(OVS_DB_RECONNECT_TIMEOUT * 1000000);
 860         }
 861         ovs_json_reader_reset(jreader);
 862         poll_fd.fd = pdb->conn.sock;
 863       } else if ((poll_fd.revents & POLLERR) || (poll_fd.revents & POLLHUP)) {
 864         /* connection is broken */
 865         OVS_ERROR("poll() peer closed its end of the channel");
 866         ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
 867         close(poll_fd.fd);
 868       } else if ((poll_fd.revents & POLLIN) || (poll_fd.revents & POLLPRI)) {
 869         /* read incoming data */
 870         nbytes = recv(poll_fd.fd, buff, OVS_DB_POLL_READ_BLOCK_SIZE, 0);
 871         if (nbytes > 0) {
 872           OVS_DEBUG("recv(): received %d bytes of data", (int)nbytes);
 873           ovs_json_reader_push_data(jreader, buff, nbytes);
 874           while (!ovs_json_reader_pop(jreader, &json, &json_len))
 875             /* process JSON data */
 876             ovs_db_json_data_process(pdb, json, json_len);
 877         } else if (nbytes == 0) {
 878           OVS_ERROR("recv() peer has performed an orderly shutdown");
 879           ovs_db_event_post(pdb, OVS_DB_EVENT_CONN_TERMINATED);
 880           close(poll_fd.fd);
 881         } else {
 882           OVS_ERROR("recv() receive data error");
 883           break;
 884         }
 885       }                         /* poll() POLLIN & POLLPRI */
 886     } else if (poll_ret == 0)
 887       OVS_DEBUG("poll() timeout");
 888     else {
 889       OVS_ERROR("poll() error");
 890       break;
 891     }
 892   }
 893
 894 thread_exit:
 895   OVS_DEBUG("poll thread has been completed");
 896   ovs_json_reader_free(jreader);
 897   pthread_exit((void *)0);
 898   return ((void *)0);
 899 }
 900
 901 /* EVENT worker thread.
 902  * Perform task based on incoming events. This
 903  * task can be done asynchronously which allows to
 904  * handle OVD DB callback like 'init_cb'.
 905  */
 906 static void *
 907 ovs_event_worker(void *arg)
 908 {
 909   int ret = 0;
 910   ovs_db_t *pdb = (ovs_db_t *)arg;
 911   struct timespec ts;
 912
 913   while (pdb->event_thread.value != OVS_DB_EVENT_TERMINATE) {
 914     /* wait for an event */
 915     clock_gettime(CLOCK_REALTIME, &ts);
 916     ts.tv_sec += (OVS_DB_EVENT_TIMEOUT);
 917     ret = pthread_cond_timedwait(&pdb->event_thread.cond,
 918                                  &pdb->event_thread.mutex, &ts);
 919     if (!ret) {
 920       /* handle the event */
 921       OVS_DEBUG("handle event %d", pdb->event_thread.value);
 922       switch (pdb->event_thread.value) {
 923       case OVS_DB_EVENT_CONN_ESTABLISHED:
 924         if (pdb->cb.post_conn_init)
 925           pdb->cb.post_conn_init(pdb);
 926         break;
 927       case OVS_DB_EVENT_CONN_TERMINATED:
 928         if (pdb->cb.post_conn_terminate)
 929           pdb->cb.post_conn_terminate();
 930         break;
 931       default:
 932         OVS_DEBUG("unknown event received");
 933         break;
 934       }
 935     } else if (ret == ETIMEDOUT) {
 936       /* wait timeout */
 937       OVS_DEBUG("no event received (timeout)");
 938       continue;
 939     } else {
 940       /* unexpected error */
 941       OVS_ERROR("pthread_cond_timedwait() failed");
 942       break;
 943     }
 944   }
 945
 946 thread_exit:
 947   OVS_DEBUG("event thread has been completed");
 948   pthread_exit((void *)0);
 949   return ((void *)0);
 950 }
 951
 952 /* Stop EVENT thread */
 953 static int
 954 ovs_db_event_thread_stop(ovs_db_t *pdb)
 955 {
 956   ovs_db_event_post(pdb, OVS_DB_EVENT_TERMINATE);
 957   if (pthread_join(pdb->event_thread.tid, NULL) != 0)
 958     return (-1);
 959   pthread_mutex_unlock(&pdb->event_thread.mutex);
 960   pthread_mutex_destroy(&pdb->event_thread.mutex);
 961   return (0);
 962 }
 963
 964 /* Stop POLL thread */
 965 static int
 966 ovs_db_poll_thread_stop(ovs_db_t *pdb)
 967 {
 968   ovs_db_poll_terminate(pdb);
 969   if (pthread_join(pdb->poll_thread.tid, NULL) != 0)
 970     return (-1);
 971   pthread_mutex_destroy(&pdb->poll_thread.mutex);
 972   return (0);
 973 }
 974
 975 /*
 976  * Public OVS DB API implementation
 977  */
 978
 979 ovs_db_t *
 980 ovs_db_init(const char *surl, ovs_db_callback_t *cb)
 981 {
 982   pthread_mutexattr_t mutex_attr;
 983   ovs_db_t *pdb = NULL;
 984
 985   /* allocate db data & fill it */
 986   if ((pdb = calloc(1, sizeof(*pdb))) == NULL)
 987     return (NULL);
 988
 989   /* convert string url to socket addr */
 990   if (ovs_db_url_parse(surl, &pdb->conn) < 0)
 991     goto failure;
 992
 993   /* setup OVS DB callbacks */
 994   if (cb)
 995     pdb->cb = *cb;
 996
 997   /* prepare event thread */
 998   pthread_cond_init(&pdb->event_thread.cond, NULL);
 999   pthread_mutex_init(&pdb->event_thread.mutex, NULL);
1000   pthread_mutex_lock(&pdb->event_thread.mutex);
1001   if (plugin_thread_create(&pdb->event_thread.tid, NULL,
1002                            ovs_event_worker, pdb) != 0) {
1003     OVS_ERROR("event worker start failed");
1004     goto failure;
1005   }
1006
1007   /* prepare polling thread */
1008   ovs_db_reconnect(pdb);
1009   pdb->poll_thread.state = OVS_DB_POLL_STATE_RUNNING;
1010   pthread_mutex_init(&pdb->poll_thread.mutex, NULL);
1011   if (plugin_thread_create(&pdb->poll_thread.tid, NULL,
1012                            ovs_poll_worker, pdb) != 0) {
1013     OVS_ERROR("pull worker start failed");
1014     goto failure;
1015   }
1016
1017   /* init OVS DB mutex */
1018   if (pthread_mutexattr_init(&mutex_attr) ||
1019       pthread_mutexattr_settype(&mutex_attr, PTHREAD_MUTEX_RECURSIVE) ||
1020       pthread_mutex_init(&pdb->mutex, &mutex_attr)) {
1021     OVS_ERROR("OVS DB mutex init failed");
1022     goto failure;
1023   }
1024
1025   /* return db to the caller */
1026   return pdb;
1027
1028 failure:
1029   if (pdb->conn.sock)
1030     /* close connection */
1031     close(pdb->conn.sock);
1032   if (pdb->event_thread.tid != 0)
1033     /* stop event thread */
1034     if (ovs_db_event_thread_stop(pdb) < 0)
1035       OVS_ERROR("stop event thread failed");
1036   if (pdb->poll_thread.tid != 0)
1037     /* stop poll thread */
1038     if (ovs_db_poll_thread_stop(pdb) < 0)
1039       OVS_ERROR("stop poll thread failed");
1040   sfree(pdb);
1041   return NULL;
1042 }
1043
1044 int
1045 ovs_db_send_request(ovs_db_t *pdb, const char *method,
1046                     const char *params, ovs_db_result_cb_t cb)
1047 {
1048   int ret = 0;
1049   yajl_gen_status yajl_gen_ret;
1050   yajl_val jparams;
1051   yajl_gen jgen;
1052   ovs_callback_t *new_cb = NULL;
1053   uint64_t uid;
1054   char uid_buff[OVS_UID_STR_SIZE];
1055   const char *req = NULL;
1056   size_t req_len = 0;
1057   struct timespec ts;
1058
1059   /* sanity check */
1060   if (!pdb || !method || !params)
1061     return (-1);
1062
1063   if ((jgen = yajl_gen_alloc(NULL)) == NULL)
1064     return (-1);
1065
1066   /* try to parse params */
1067   if ((jparams = yajl_tree_parse(params, NULL, 0)) == NULL) {
1068     OVS_ERROR("params is not a JSON string");
1069     yajl_gen_clear(jgen);
1070     return (-1);
1071   }
1072
1073   /* generate method field */
1074   OVS_YAJL_CALL(yajl_gen_map_open, jgen);
1075
1076   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "method");
1077   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, method);
1078
1079   /* generate params field */
1080   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "params");
1081   OVS_YAJL_CALL(ovs_yajl_gen_val, jgen, jparams);
1082   yajl_tree_free(jparams);
1083
1084   /* generate id field */
1085   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, "id");
1086   uid = ovs_uid_generate();
1087   ssnprintf(uid_buff, sizeof(uid_buff), "%" PRIX64, uid);
1088   OVS_YAJL_CALL(ovs_yajl_gen_tstring, jgen, uid_buff);
1089
1090   OVS_YAJL_CALL(yajl_gen_map_close, jgen);
1091
1092   if (cb) {
1093     /* register result callback */
1094     if ((new_cb = malloc(sizeof(ovs_callback_t))) == NULL)
1095       goto yajl_gen_failure;
1096
1097     /* add new callback to front */
1098     sem_init(&new_cb->result.sync, 0, 0);
1099     new_cb->result.call = cb;
1100     new_cb->uid = uid;
1101     ovs_db_callback_add(pdb, new_cb);
1102   }
1103
1104   /* send the request */
1105   OVS_YAJL_CALL(yajl_gen_get_buf, jgen, (const unsigned char **)&req,
1106                 &req_len);
1107   OVS_DEBUG("%s", req);
1108   if (!ovs_db_data_send(pdb, req, req_len)) {
1109     if (cb) {
1110       /* wait for result */
1111       clock_gettime(CLOCK_REALTIME, &ts);
1112       ts.tv_sec += OVS_DB_SEND_REQ_TIMEOUT;
1113       if (sem_timedwait(&new_cb->result.sync, &ts) < 0) {
1114         OVS_ERROR("%s() no replay received within %d sec", __FUNCTION__,
1115                   OVS_DB_SEND_REQ_TIMEOUT);
1116         ret = (-1);
1117       }
1118     }
1119   } else {
1120     OVS_ERROR("ovs_db_data_send() failed");
1121     ret = (-1);
1122   }
1123
1124 yajl_gen_failure:
1125   if (new_cb) {
1126     /* destroy callback */
1127     sem_destroy(&new_cb->result.sync);
1128     ovs_db_callback_remove(pdb, new_cb);
1129   }
1130
1131   /* release memory */
1132   yajl_gen_clear(jgen);
1133   return (yajl_gen_ret != yajl_gen_status_ok) ? (-1) : ret;
1134 }
1135
1136 int
1137 ovs_db_table_cb_register(ovs_db_t *pdb, const char *tb_name,
1138                          const char **tb_column, ovs_db_table_cb_t update_cb,
1139                          ovs_db_result_cb_t result_cb, unsigned int flags)
1140 {
1141   yajl_gen jgen;
1142   yajl_gen_status yajl_gen_ret;
1143   ovs_callback_t *new_cb = NULL;
1144   char uid_str[OVS_UID_STR_SIZE];
1145   char *params;
1146   size_t params_len;
1147   int ovs_db_ret = 0;
1148
1149   /* sanity check */
1150   if (pdb == NULL || tb_name == NULL || update_cb == NULL)
1151     return (-1);
1152
1153   if ((jgen = yajl_gen_alloc(NULL)) == NULL)
1154     return (-1);
1155
1156   /* register table update callback */
1157   if ((new_cb = malloc(sizeof(ovs_callback_t))) == NULL)
1158     return (-1);
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
1236 ovs_db_destroy(ovs_db_t *pdb)
1237 {
1238   int ovs_db_ret = 0;
1239   int ret = 0;
1240
1241   /* sanity check */
1242   if (pdb == NULL)
1243     return (-1);
1244
1245   /* try to lock the structure before releasing */
1246   if (ret = pthread_mutex_lock(&pdb->mutex)) {
1247     OVS_ERROR("pthread_mutex_lock() DB mutext lock failed (%d)", ret);
1248     return (-1);
1249   }
1250
1251   /* stop poll thread */
1252   if (ovs_db_event_thread_stop(pdb) < 0) {
1253     OVS_ERROR("stop poll thread failed");
1254     ovs_db_ret = (-1);
1255   }
1256
1257   /* stop event thread */
1258   if (ovs_db_poll_thread_stop(pdb) < 0) {
1259     OVS_ERROR("stop event thread failed");
1260     ovs_db_ret = (-1);
1261   }
1262
1263   /* unsubscribe callbacks */
1264   ovs_db_callback_remove_all(pdb);
1265
1266   /* close connection */
1267   if (pdb->conn.sock)
1268     close(pdb->conn.sock);
1269
1270   /* release DB handler */
1271   pthread_mutex_unlock(&pdb->mutex);
1272   pthread_mutex_destroy(&pdb->mutex);
1273   sfree(pdb);
1274   return ovs_db_ret;
1275 }
1276
1277 /*
1278  * Public OVS utils API implementation
1279  */
1280
1281 /* Get YAJL value by key from YAJL dictionary */
1282 yajl_val
1283 ovs_utils_get_value_by_key(yajl_val jval, const char *key)
1284 {
1285   const char *obj_key = NULL;
1286
1287   /* check params */
1288   if (!YAJL_IS_OBJECT(jval) || !key)
1289     return NULL;
1290
1291   /* find a value by key */
1292   for (int i = 0; i < YAJL_GET_OBJECT(jval)->len; i++) {
1293     obj_key = YAJL_GET_OBJECT(jval)->keys[i];
1294     if (strcmp(obj_key, key) == 0)
1295       return YAJL_GET_OBJECT(jval)->values[i];
1296   }
1297
1298   return NULL;
1299 }