/* Copyright (c) 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017 Nicira, Inc.

 * Copyright (C) 2016 Hewlett Packard Enterprise Development LP

 *

 * Licensed under the Apache License, Version 2.0 (the "License");

 * you may not use this file except in compliance with the License.

 * You may obtain a copy of the License at:

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

#include <config.h>

#include "ovsdb-idl.h"

#include <errno.h>

#include <inttypes.h>

#include <limits.h>

#include <stdlib.h>

#include "bitmap.h"

#include "coverage.h"

#include "hash.h"

#include "openvswitch/dynamic-string.h"

#include "fatal-signal.h"

#include "openvswitch/json.h"

#include "jsonrpc.h"

#include "ovsdb/ovsdb.h"

#include "ovsdb/table.h"

#include "ovsdb-cs.h"

#include "ovsdb-data.h"

#include "ovsdb-error.h"

#include "ovsdb-idl-provider.h"

#include "ovsdb-parser.h"

#include "ovsdb-server-idl.h"

#include "ovsdb-session.h"

#include "openvswitch/poll-loop.h"

#include "openvswitch/shash.h"

#include "skiplist.h"

#include "simap.h"

#include "sset.h"

#include "svec.h"

#include "util.h"

#include "uuid.h"

#include "openvswitch/vlog.h"

VLOG_DEFINE_THIS_MODULE(ovsdb_idl);

COVERAGE_DEFINE(txn_uncommitted);

COVERAGE_DEFINE(txn_unchanged);

COVERAGE_DEFINE(txn_incomplete);

COVERAGE_DEFINE(txn_aborted);

COVERAGE_DEFINE(txn_success);

COVERAGE_DEFINE(txn_try_again);

COVERAGE_DEFINE(txn_not_locked);

COVERAGE_DEFINE(txn_error);

/* An arc from one idl_row to another.  When row A contains a UUID that

 * references row B, this is represented by an arc from A (the source) to B

 * (the destination).

 *

 * Arcs from a row to itself are omitted, that is, src and dst are always

 * different.

 *

 * Arcs are never duplicated, that is, even if there are multiple references

 * from A to B, there is only a single arc from A to B.

 *

 * Arcs are directed: an arc from A to B is the converse of an an arc from B to

 * A.  Both an arc and its converse may both be present, if each row refers

 * to the other circularly.

 *

 * The source and destination row may be in the same table or in different

 * tables.

 */

struct ovsdb_idl_arc {

    struct ovs_list src_node;   /* In src->src_arcs list. */

    struct ovs_list dst_node;   /* In dst->dst_arcs list. */

    struct ovsdb_idl_row *src;  /* Source row. */

    struct ovsdb_idl_row *dst;  /* Destination row. */

};

struct ovsdb_idl {

    struct ovsdb_cs *cs;

    const struct ovsdb_idl_class *class_;

    struct shash table_by_name; /* Contains "struct ovsdb_idl_table *"s.*/

    struct ovsdb_idl_table *tables; /* Array of ->class_->n_tables elements. */

    unsigned int change_seqno;

    struct ovsdb_idl_txn *txn;

    struct hmap outstanding_txns;

    bool verify_write_only;

    struct ovs_list deleted_untracked_rows; /* Stores rows deleted in the

                                             * current run, that are not yet

                                             * added to the track_list. */

    struct ovs_list rows_to_reparse; /* Stores rows that might need to be

                                      * re-parsed due to insertion of a

                                      * referenced row. */

    bool server_schema_received; /* Set to true when the IDL has received

                                  * the DB schema from the server. */

};

static struct ovsdb_cs_ops ovsdb_idl_cs_ops;

struct ovsdb_idl_txn {

    struct hmap_node hmap_node;

    struct json *request_id;

    struct ovsdb_idl *idl;

    struct hmap txn_rows;

    enum ovsdb_idl_txn_status status;

    char *error;

    bool dry_run;

    bool assert_read_only;

    struct ds comment;

    /* Increments. */

    const char *inc_table;

    const char *inc_column;

    struct uuid inc_row;

    bool inc_force;

    unsigned int inc_index;

    int64_t inc_new_value;

    /* Inserted rows. */

    struct hmap inserted_rows;  /* Contains "struct ovsdb_idl_txn_insert"s. */

};

struct ovsdb_idl_txn_insert {

    struct hmap_node hmap_node; /* In struct ovsdb_idl_txn's inserted_rows. */

    struct uuid dummy;          /* Dummy UUID used locally. */

    int op_index;               /* Index into transaction's operation array. */

    struct uuid real;           /* Real UUID used by database server. */

};

static struct vlog_rate_limit syntax_rl = VLOG_RATE_LIMIT_INIT(1, 5);

static struct vlog_rate_limit semantic_rl = VLOG_RATE_LIMIT_INIT(1, 5);

static struct vlog_rate_limit other_rl = VLOG_RATE_LIMIT_INIT(1, 5);

enum update_result {

    OVSDB_IDL_UPDATE_DB_CHANGED,

    OVSDB_IDL_UPDATE_NO_CHANGES,

    OVSDB_IDL_UPDATE_INCONSISTENT,

};

static void ovsdb_idl_clear(struct ovsdb_idl *);

static enum update_result ovsdb_idl_process_update(

    struct ovsdb_idl_table *, const struct ovsdb_cs_row_update *);

static void ovsdb_idl_insert_row(struct ovsdb_idl_row *,

                                 const struct shash *values);

static void ovsdb_idl_delete_row(struct ovsdb_idl_row *);

static bool ovsdb_idl_modify_row(struct ovsdb_idl_row *,

                                 const struct shash *values, bool xor);

static void ovsdb_idl_parse_update(struct ovsdb_idl *,

                                   const struct ovsdb_cs_update_event *);

static void ovsdb_idl_reparse_deleted(struct ovsdb_idl *);

static void ovsdb_idl_reparse_refs_to_inserted(struct ovsdb_idl *);

static void ovsdb_idl_txn_process_reply(struct ovsdb_idl *,

                                        const struct jsonrpc_msg *);

static bool ovsdb_idl_row_is_orphan(const struct ovsdb_idl_row *);

static struct ovsdb_idl_row *ovsdb_idl_row_create__(

    const struct ovsdb_idl_table_class *);

static struct ovsdb_idl_row *ovsdb_idl_row_create(struct ovsdb_idl_table *,

                                                  const struct uuid *);

static void ovsdb_idl_row_destroy(struct ovsdb_idl_row *);

static void ovsdb_idl_row_destroy_postprocess(struct ovsdb_idl *);

static void ovsdb_idl_destroy_all_map_op_lists(struct ovsdb_idl_row *);

static void ovsdb_idl_destroy_all_set_op_lists(struct ovsdb_idl_row *);

static void ovsdb_idl_row_parse(struct ovsdb_idl_row *);

static void ovsdb_idl_row_unparse(struct ovsdb_idl_row *);

static void ovsdb_idl_row_clear_old(struct ovsdb_idl_row *);

static void ovsdb_idl_row_clear_new(struct ovsdb_idl_row *);

static void ovsdb_idl_row_clear_arcs(struct ovsdb_idl_row *, bool destroy_dsts);

static void ovsdb_idl_row_reparse_backrefs(struct ovsdb_idl_row *);

static void ovsdb_idl_row_mark_backrefs_for_reparsing(struct ovsdb_idl_row *);

static void ovsdb_idl_row_track_change(struct ovsdb_idl_row *,

                                       enum ovsdb_idl_change);

static void ovsdb_idl_row_untrack_change(struct ovsdb_idl_row *);

static void ovsdb_idl_row_clear_changeseqno(struct ovsdb_idl_row *);

static void ovsdb_idl_txn_abort_all(struct ovsdb_idl *);

static bool ovsdb_idl_txn_extract_mutations(struct ovsdb_idl_row *,

                                            struct json *);

static void ovsdb_idl_txn_add_map_op(struct ovsdb_idl_row *,

                                     const struct ovsdb_idl_column *,

                                     struct ovsdb_datum *,

                                     enum map_op_type);

static void ovsdb_idl_txn_add_set_op(struct ovsdb_idl_row *,

                                     const struct ovsdb_idl_column *,

                                     struct ovsdb_datum *,

                                     enum set_op_type);

static struct ovsdb_idl_table *

ovsdb_idl_table_from_class(const struct ovsdb_idl *,

                              const struct ovsdb_idl_table_class *);

static struct ovsdb_idl_table *

ovsdb_idl_table_from_class(const struct ovsdb_idl *,

                           const struct ovsdb_idl_table_class *);

static void ovsdb_idl_track_clear__(struct ovsdb_idl *, bool flush_all);

static void ovsdb_idl_destroy_indexes(struct ovsdb_idl_table *);

static void ovsdb_idl_add_to_indexes(const struct ovsdb_idl_row *);

static void ovsdb_idl_remove_from_indexes(const struct ovsdb_idl_row *);

static int ovsdb_idl_try_commit_loop_txn(struct ovsdb_idl_loop *loop,

                                         bool *may_need_wakeup);

static void ovsdb_idl_condition_clone(struct ovsdb_idl_condition *dest,

                                      const struct ovsdb_idl_condition *);

static void ovsdb_idl_create_req_condition(

        struct ovsdb_idl *,

        const struct ovsdb_idl_table_class *,

        const struct ovsdb_idl_condition *);

static void ovsdb_idl_destroy_req_condition(struct ovsdb_idl_table *);

static unsigned int ovsdb_idl_set_condition__(

        struct ovsdb_idl *,

        const struct ovsdb_idl_table_class *,

        const struct ovsdb_idl_condition *);

static void add_tracked_change_for_references(struct ovsdb_idl_row *);

/* Creates and returns a connection to database 'remote', which should be in a

 * form acceptable to jsonrpc_session_open().  The connection will maintain an

 * in-memory replica of the remote database whose schema is described by

 * 'class'.  (Ordinarily 'class' is compiled from an OVSDB schema automatically

 * by ovsdb-idlc.)

 *

 * Passes 'retry' to jsonrpc_session_open().  See that function for

 * documentation.

 *

 * If 'monitor_everything_by_default' is true, then everything in the remote

 * database will be replicated by default.  ovsdb_idl_omit() and

 * ovsdb_idl_omit_alert() may be used to selectively drop some columns from

 * monitoring.

 *

 * If 'monitor_everything_by_default' is false, then no columns or tables will

 * be replicated by default.  ovsdb_idl_add_column() and ovsdb_idl_add_table()

 * must be used to choose some columns or tables to replicate.

 */

struct ovsdb_idl *

ovsdb_idl_create(const char *remote, const struct ovsdb_idl_class *class,

                 bool monitor_everything_by_default, bool retry)

{

    struct ovsdb_idl *idl = ovsdb_idl_create_unconnected(

        class, monitor_everything_by_default);

    ovsdb_idl_set_remote(idl, remote, retry);

    return idl;

}

/* Creates and returns a connection to an in-memory replica of the remote

 * database whose schema is described by 'class'.  (Ordinarily 'class' is

 * compiled from an OVSDB schema automatically by ovsdb-idlc.)

 *

 * Use ovsdb_idl_set_remote() to configure the database to which to connect.

 * Until a remote is configured, no data can be retrieved.

 *

 * If 'monitor_everything_by_default' is true, then everything in the remote

 * database will be replicated by default.  ovsdb_idl_omit() and

 * ovsdb_idl_omit_alert() may be used to selectively drop some columns from

 * monitoring.

 *

 * If 'monitor_everything_by_default' is false, then no columns or tables will

 * be replicated by default.  ovsdb_idl_add_column() and ovsdb_idl_add_table()

 * must be used to choose some columns or tables to replicate.

 */

struct ovsdb_idl *

ovsdb_idl_create_unconnected(const struct ovsdb_idl_class *class,

                             bool monitor_everything_by_default)

{

    struct ovsdb_idl *idl = xmalloc(sizeof *idl);

    *idl = (struct ovsdb_idl) {

        .cs = ovsdb_cs_create(class->database, 3, &ovsdb_idl_cs_ops, idl),

        .class_ = class,

        .table_by_name = SHASH_INITIALIZER(&idl->table_by_name),

        .tables = xmalloc(class->n_tables * sizeof *idl->tables),

        .change_seqno = 0,

        .txn = NULL,

        .outstanding_txns = HMAP_INITIALIZER(&idl->outstanding_txns),

        .verify_write_only = false,

        .server_schema_received = false,

        .deleted_untracked_rows

            = OVS_LIST_INITIALIZER(&idl->deleted_untracked_rows),

        .rows_to_reparse

            = OVS_LIST_INITIALIZER(&idl->rows_to_reparse),

    };

    uint8_t default_mode = (monitor_everything_by_default

                            ? OVSDB_IDL_MONITOR | OVSDB_IDL_ALERT

                            : 0);

    for (size_t i = 0; i < class->n_tables; i++) {

        const struct ovsdb_idl_table_class *tc = &class->tables[i];

        struct ovsdb_idl_table *table = &idl->tables[i];

        shash_add_assert(&idl->table_by_name, tc->name, table);

        table->class_ = tc;

        table->modes = xmalloc(tc->n_columns);

        memset(table->modes, default_mode, tc->n_columns);

        table->need_table = false;

        shash_init(&table->columns);

        ovs_list_init(&table->indexes);

        for (size_t j = 0; j < tc->n_columns; j++) {

            const struct ovsdb_idl_column *column = &tc->columns[j];

            shash_add_assert(&table->columns, column->name, column);

        }

        hmap_init(&table->rows);

        ovs_list_init(&table->track_list);

        table->change_seqno[OVSDB_IDL_CHANGE_INSERT]

            = table->change_seqno[OVSDB_IDL_CHANGE_MODIFY]

            = table->change_seqno[OVSDB_IDL_CHANGE_DELETE] = 0;

        table->idl = idl;

        table->in_server_schema = false;

        table->req_cond = NULL;

        shash_init(&table->schema_columns);

    }

    return idl;

}

/* Changes the remote and creates a new session.

 *

 * If 'retry' is true, the connection to the remote will automatically retry

 * when it fails.  If 'retry' is false, the connection is one-time. */

void

ovsdb_idl_set_remote(struct ovsdb_idl *idl, const char *remote, bool retry)

{

    ovsdb_cs_set_remote(idl->cs, remote, retry);

}

/* Set whether the order of remotes should be shuffled, when there

 * are more than one remotes.  The setting doesn't take effect

 * until the next time when ovsdb_idl_set_remote() is called. */

void

ovsdb_idl_set_shuffle_remotes(struct ovsdb_idl *idl, bool shuffle)

{

    ovsdb_cs_set_shuffle_remotes(idl->cs, shuffle);

}

/* Passes 'set_db_change_aware' to ovsdb_cs_set_db_change_aware().  See that

 * function for documentation. */

void

ovsdb_idl_set_db_change_aware(struct ovsdb_idl *idl, bool set_db_change_aware)

{

    ovsdb_cs_set_db_change_aware(idl->cs, set_db_change_aware);

}

/* Reset min_index to 0. This prevents a situation where the client

 * thinks all databases have stale data, when they actually have all

 * been destroyed and rebuilt from scratch.

 */

void

ovsdb_idl_reset_min_index(struct ovsdb_idl *idl)

{

    ovsdb_cs_reset_min_index(idl->cs);

}

static void

ovsdb_idl_schema_columns_clear(struct shash *schema_columns)

{

    struct shash_node *node;

    SHASH_FOR_EACH (node, schema_columns) {

        ovsdb_type_destroy(node->data);

    }

    shash_clear_free_data(schema_columns);

}

/* Destroys 'idl' and all of the data structures that it manages. */

void

ovsdb_idl_destroy(struct ovsdb_idl *idl)

{

    if (idl) {

        ovs_assert(!idl->txn);

        ovsdb_idl_txn_abort_all(idl);

        hmap_destroy(&idl->outstanding_txns);

        ovsdb_idl_clear(idl);

        ovsdb_cs_destroy(idl->cs);

        for (size_t i = 0; i < idl->class_->n_tables; i++) {

            struct ovsdb_idl_table *table = &idl->tables[i];

            ovsdb_idl_destroy_indexes(table);

            shash_destroy(&table->columns);

            ovsdb_idl_schema_columns_clear(&table->schema_columns);

            shash_destroy(&table->schema_columns);

            ovsdb_idl_destroy_req_condition(table);

            hmap_destroy(&table->rows);

            free(table->modes);

        }

        shash_destroy(&idl->table_by_name);

        free(idl->tables);

        free(idl);

    }

}

/* By default, or if 'leader_only' is true, when 'idl' connects to a clustered

 * database, the IDL will avoid servers other than the cluster leader. This

 * ensures that any data that it reads and reports is up-to-date.  If

 * 'leader_only' is false, the IDL will accept any server in the cluster, which

 * means that for read-only transactions it can report and act on stale data

 * (transactions that modify the database are always serialized even with false

 * 'leader_only').  Refer to Understanding Cluster Consistency in ovsdb(7) for

 * more information. */

void

ovsdb_idl_set_leader_only(struct ovsdb_idl *idl, bool leader_only)

{

    ovsdb_cs_set_leader_only(idl->cs, leader_only);

}

static void

ovsdb_idl_clear(struct ovsdb_idl *db)

{

    /* Process deleted rows, removing them from the 'deleted_untracked_rows'

     * list and reparsing their backrefs.

     */

    ovsdb_idl_reparse_deleted(db);

    /* Process backrefs of inserted rows, removing them from the

     * 'rows_to_reparse' list.

     */

    ovsdb_idl_reparse_refs_to_inserted(db);

    /* Cleanup all rows; each row gets added to its own table's

     * 'track_list'.

     */

    for (size_t i = 0; i < db->class_->n_tables; i++) {

        struct ovsdb_idl_table *table = &db->tables[i];

        struct ovsdb_idl_row *row;

        if (hmap_is_empty(&table->rows)) {

            continue;

        }

        HMAP_FOR_EACH_SAFE (row, hmap_node, &table->rows) {

            struct ovsdb_idl_arc *arc;

            if (!ovsdb_idl_row_is_orphan(row)) {

                ovsdb_idl_remove_from_indexes(row);

                ovsdb_idl_row_unparse(row);

            }

            LIST_FOR_EACH_SAFE (arc, src_node, &row->src_arcs) {

                ovs_list_remove(&arc->src_node);

                ovs_list_remove(&arc->dst_node);

                free(arc);

            }

            LIST_FOR_EACH_SAFE (arc, dst_node, &row->dst_arcs) {

                ovs_list_remove(&arc->src_node);

                ovs_list_remove(&arc->dst_node);

                free(arc);

            }

            ovsdb_idl_row_destroy(row);

        }

    }

    /* Free rows deleted from tables with change tracking disabled. */

    ovsdb_idl_row_destroy_postprocess(db);

    /* Free rows deleted from tables with change tracking enabled. */

    ovsdb_idl_track_clear__(db, true);

    ovs_assert(ovs_list_is_empty(&db->deleted_untracked_rows));

    ovs_assert(ovs_list_is_empty(&db->rows_to_reparse));

    db->change_seqno++;

}

/* Processes a batch of messages from the database server on 'idl'.  This may

 * cause the IDL's contents to change.  The client may check for that with

 * ovsdb_idl_get_seqno(). */

void

ovsdb_idl_run(struct ovsdb_idl *idl)

{

    ovs_assert(!idl->txn);

    struct ovs_list events;

    ovsdb_cs_run(idl->cs, &events);

    struct ovsdb_cs_event *event;

    LIST_FOR_EACH_POP (event, list_node, &events) {

        switch (event->type) {

        case OVSDB_CS_EVENT_TYPE_RECONNECT:

            ovsdb_idl_txn_abort_all(idl);

            break;

        case OVSDB_CS_EVENT_TYPE_LOCKED:

            if (ovsdb_cs_may_send_transaction(idl->cs)) {

                /* If the client couldn't run a transaction because it didn't

                 * have the lock, this will encourage it to try again. */

                idl->change_seqno++;

            } else {

                /* We're setting up a session, so don't signal that the

                 * database changed.  Finalizing the session will increment

                 * change_seqno anyhow. */

            }

            break;

        case OVSDB_CS_EVENT_TYPE_UPDATE:

            ovsdb_idl_parse_update(idl, &event->update);

            break;

        case OVSDB_CS_EVENT_TYPE_TXN_REPLY:

            ovsdb_idl_txn_process_reply(idl, event->txn_reply);

            break;

        }

        ovsdb_cs_event_destroy(event);

    }

    ovsdb_idl_reparse_refs_to_inserted(idl);

    ovsdb_idl_reparse_deleted(idl);

    ovsdb_idl_row_destroy_postprocess(idl);

}

/* Arranges for poll_block() to wake up when ovsdb_idl_run() has something to

 * do or when activity occurs on a transaction on 'idl'. */

void

ovsdb_idl_wait(struct ovsdb_idl *idl)

{

    ovsdb_cs_wait(idl->cs);

}

/* Returns memory usage statistics. */

void

ovsdb_idl_get_memory_usage(struct ovsdb_idl *idl, struct simap *usage)

{

    unsigned int cells = 0;

    if (!idl) {

        return;

    }

    for (size_t i = 0; i < idl->class_->n_tables; i++) {

        struct ovsdb_idl_table *table = &idl->tables[i];

        unsigned int n_columns = table->class_->n_columns;

        unsigned int n_rows = hmap_count(&table->rows);

        cells += n_rows * n_columns;

    }

    struct {

        const char *name;

        unsigned int val;

    } idl_mem_stats[] = {

        {"idl-outstanding-txns", hmap_count(&idl->outstanding_txns)},

        {"idl-cells", cells},

    };

    for (size_t i = 0; i < ARRAY_SIZE(idl_mem_stats); i++) {

        char *stat_name = xasprintf("%s-%s", idl_mem_stats[i].name,

                                             idl->class_->database);

        simap_increase(usage, stat_name, idl_mem_stats[i].val);

        free(stat_name);

    }

}

/* Returns a "sequence number" that represents the state of 'idl'.  When

 * ovsdb_idl_run() changes the database, the sequence number changes.  The

 * initial fetch of the entire contents of the remote database is considered to

 * be one kind of change.  Successfully acquiring a lock, if one has been

 * configured with ovsdb_idl_set_lock(), is also considered to be a change.

 *

 * As long as the sequence number does not change, the client may continue to

 * use any data structures it obtains from 'idl'.  But when it changes, the

 * client must not access any of these data structures again, because they

 * could have freed or reused for other purposes.

 *

 * The sequence number can occasionally change even if the database does not.

 * This happens if the connection to the database drops and reconnects, which

 * causes the database contents to be reloaded even if they didn't change.  (It

 * could also happen if the database server sends out a "change" that reflects

 * what the IDL already thought was in the database.  The database server is

 * not supposed to do that, but bugs could in theory cause it to do so.) */

unsigned int

ovsdb_idl_get_seqno(const struct ovsdb_idl *idl)

{

    return idl->change_seqno;

}

/* Returns a "sequence number" that represents the number of conditional

 * monitoring updates successfully received by the OVSDB server of an IDL

 * connection.

 *

 * ovsdb_idl_set_condition() sets a new condition that is different from

 * the current condtion, the next expected "sequence number" is returned.

 *

 * Whenever ovsdb_idl_get_cond_seqno() returns a value that matches

 * the return value of ovsdb_idl_set_condition(),  The client is

 * assured that:

 *   -  The ovsdb_idl_set_condition() changes has been acknowledged by

 *      the OVSDB sever.

 *

 *   -  'idl' now contains the content matches the new conditions.   */

unsigned int

ovsdb_idl_get_condition_seqno(const struct ovsdb_idl *idl)

{

    return ovsdb_cs_get_condition_seqno(idl->cs);

}

/* Returns true if 'idl' successfully connected to the remote database and

 * retrieved its contents (even if the connection subsequently dropped and is

 * in the process of reconnecting).  If so, then 'idl' contains an atomic

 * snapshot of the database's contents (but it might be arbitrarily old if the

 * connection dropped).

 *

 * Returns false if 'idl' has never connected or retrieved the database's

 * contents.  If so, 'idl' is empty. */

bool

ovsdb_idl_has_ever_connected(const struct ovsdb_idl *idl)

{

    return ovsdb_idl_get_seqno(idl) != 0;

}

/* Reconfigures 'idl' so that it would reconnect to the database, if

 * connection was dropped. */

void

ovsdb_idl_enable_reconnect(struct ovsdb_idl *idl)

{

    ovsdb_cs_enable_reconnect(idl->cs);

}

/* Forces 'idl' to drop its connection to the database and reconnect.  In the

 * meantime, the contents of 'idl' will not change. */

void

ovsdb_idl_force_reconnect(struct ovsdb_idl *idl)

{

    ovsdb_cs_force_reconnect(idl->cs);

}

/* Some IDL users should only write to write-only columns.  Furthermore,

 * writing to a column which is not write-only can cause serious performance

 * degradations for these users.  This function causes 'idl' to reject writes

 * to columns which are not marked write only using ovsdb_idl_omit_alert(). */

void

ovsdb_idl_verify_write_only(struct ovsdb_idl *idl)

{

    idl->verify_write_only = true;

}

/* Returns true if 'idl' is currently connected or trying to connect

 * and a negative response to a schema request has not been received */

bool

ovsdb_idl_is_alive(const struct ovsdb_idl *idl)

{

    return ovsdb_cs_is_alive(idl->cs);

}

bool

ovsdb_idl_is_connected(const struct ovsdb_idl *idl)

{

    return ovsdb_cs_is_connected(idl->cs);

}

/* Returns the last error reported on a connection by 'idl'.  The return value

 * is 0 only if no connection made by 'idl' has ever encountered an error and

 * a negative response to a schema request has never been received. See

 * jsonrpc_get_status() for jsonrpc_session_get_last_error() return value

 * interpretation. */

int

ovsdb_idl_get_last_error(const struct ovsdb_idl *idl)

{

    return ovsdb_cs_get_last_error(idl->cs);

}

/* Sets the "probe interval" for 'idl->session' to 'probe_interval', in

 * milliseconds.

 */

void

ovsdb_idl_set_probe_interval(const struct ovsdb_idl *idl, int probe_interval)

{

    ovsdb_cs_set_probe_interval(idl->cs, probe_interval);

}

static size_t

find_uuid_in_array(const struct uuid *target,

                   const struct uuid *array, size_t n)

{

    for (size_t i = 0; i < n; i++) {

        if (uuid_equals(&array[i], target)) {

            return i;

        }

    }

    return SIZE_MAX;

}

static size_t

array_contains_uuid(const struct uuid *target,

                    const struct uuid *array, size_t n)

{

    return find_uuid_in_array(target, array, n) != SIZE_MAX;

}

static bool

remove_uuid_from_array(const struct uuid *target,

                       struct uuid *array, size_t *n)

{

    size_t i = find_uuid_in_array(target, array, *n);

    if (i != SIZE_MAX) {

        array[i] = array[--*n];

        return true;

    } else {

        return false;

    }

}

static void

add_row_references(const struct ovsdb_base_type *type,

                   const union ovsdb_atom *atoms, size_t n_atoms,

                   const struct uuid *exclude_uuid,

                   struct uuid **dstsp, size_t *n_dstsp,

                   size_t *allocated_dstsp)

{

    if (type->type != OVSDB_TYPE_UUID || !type->uuid.refTableName) {

        return;

    }

    for (size_t i = 0; i < n_atoms; i++) {

        const struct uuid *uuid = &atoms[i].uuid;

        if (!uuid_equals(uuid, exclude_uuid)

            && !array_contains_uuid(uuid, *dstsp, *n_dstsp)) {

            if (*n_dstsp >= *allocated_dstsp) {

                *dstsp = x2nrealloc(*dstsp, allocated_dstsp,

                                    sizeof **dstsp);

            }

            (*dstsp)[*n_dstsp] = *uuid;

            ++*n_dstsp;

        }

    }

}

/* Checks for consistency in 'idl''s graph of arcs between database rows.  Each

 * reference from one row to a different row should be reflected as a "struct

 * ovsdb_idl_arc" between those rows.

 *

 * This function is slow, big-O wise, and aborts if it finds an inconsistency,

 * thus it is only for use in test programs. */

void

ovsdb_idl_check_consistency(const struct ovsdb_idl *idl)

{

    /* Consistency is broken while a transaction is in progress. */

    if (!idl->txn) {

        return;

    }

    bool ok = true;

    struct uuid *dsts = NULL;

    size_t allocated_dsts = 0;

    for (size_t i = 0; i < idl->class_->n_tables; i++) {

        const struct ovsdb_idl_table *table = &idl->tables[i];

        const struct ovsdb_idl_table_class *class = table->class_;

        const struct ovsdb_idl_row *row;

        HMAP_FOR_EACH (row, hmap_node, &table->rows) {

            size_t n_dsts = 0;

            if (row->new_datum) {

                size_t n_columns = shash_count(&row->table->columns);

                for (size_t j = 0; j < n_columns; j++) {

                    const struct ovsdb_type *type = &class->columns[j].type;

                    const struct ovsdb_datum *datum;

                    datum = ovsdb_idl_read(row, &class->columns[j]);

                    add_row_references(&type->key,

                                       datum->keys, datum->n, &row->uuid,

                                       &dsts, &n_dsts, &allocated_dsts);

                    add_row_references(&type->value,

                                       datum->values, datum->n, &row->uuid,

                                       &dsts, &n_dsts, &allocated_dsts);

                }

            }

            const struct ovsdb_idl_arc *arc;

            LIST_FOR_EACH (arc, src_node, &row->src_arcs) {

                if (!remove_uuid_from_array(&arc->dst->uuid,

                                            dsts, &n_dsts)) {

                    VLOG_ERR("unexpected arc from %s row "UUID_FMT" to %s "

                             "row "UUID_FMT,

                             table->class_->name,

                             UUID_ARGS(&row->uuid),

                             arc->dst->table->class_->name,

                             UUID_ARGS(&arc->dst->uuid));

                    ok = false;

                }

            }

            for (size_t j = 0; j < n_dsts; j++) {

                VLOG_ERR("%s row "UUID_FMT" missing arc to row "UUID_FMT,

                         table->class_->name, UUID_ARGS(&row->uuid),

                         UUID_ARGS(&dsts[j]));

                ok = false;

            }

        }

    }

    free(dsts);

    ovs_assert(ok);

}

static struct json *

ovsdb_idl_compose_monitor_request(const struct json *schema_json, void *idl_)

{

    struct ovsdb_idl *idl = idl_;

    struct shash *schema = ovsdb_cs_parse_schema(schema_json);

    struct json *monitor_requests = json_object_create();

    idl->server_schema_received = true;

    for (size_t i = 0; i < idl->class_->n_tables; i++) {

        struct ovsdb_idl_table *table = &idl->tables[i];

        const struct ovsdb_idl_table_class *tc = table->class_;

        struct json *monitor_request;

        struct shash *table_schema

            = schema ? shash_find_data(schema, table->class_->name) : NULL;

        struct json *columns

            = table->need_table ? json_array_create_empty() : NULL;

        ovsdb_idl_schema_columns_clear(&table->schema_columns);

        for (size_t j = 0; j < tc->n_columns; j++) {

            const struct ovsdb_idl_column *column = &tc->columns[j];

            bool idl_has_column = false;

            struct shash_node *node;

            if (table_schema) {

                node = shash_find(table_schema, column->name);

                if (node) {

                    idl_has_column = true;

                    shash_add(&table->schema_columns,

                              column->name, node->data);

                    shash_delete(table_schema, node);

                }

            }

            if (column->is_synthetic) {

                if (idl_has_column) {

                    VLOG_WARN("%s table in %s database has synthetic "

                              "column %s", table->class_->name,

                              idl->class_->database, column->name);

                }

            } else if (table->modes[j] & OVSDB_IDL_MONITOR) {

                if (table_schema && !idl_has_column) {

                    VLOG_WARN("%s table in %s database lacks %s column "

                              "(database needs upgrade?)",

                              table->class_->name, idl->class_->database,

                              column->name);

                    continue;

                }

                if (!columns) {

                    columns = json_array_create_empty();

                }

                json_array_add(columns, json_string_create(column->name));

            }

        }

        if (columns) {

            if (schema && !table_schema) {

                VLOG_WARN("%s database lacks %s table "

                          "(database needs upgrade?)",

                          idl->class_->database, table->class_->name);

                json_destroy(columns);

                table->in_server_schema = false;

                ovsdb_cs_clear_condition(idl->cs, table->class_->name);

                continue;

            } else if (schema && table_schema) {

                if (!table->in_server_schema) {

                    /* The server didn't have this table before and now it

                     * does.  It should've dropped the transaction history,

                     * but it's better if we do not ask for it in the first

                     * place, as our conditions could be out of sync. */

                    ovsdb_cs_reset_last_id(idl->cs);

                }

                table->in_server_schema = true;

            }

            monitor_request = json_object_create();

            json_object_put(monitor_request, "columns", columns);

            json_object_put(monitor_requests, tc->name,

                            json_array_create_1(monitor_request));

        }

        if (!table->in_server_schema) {

            ovsdb_cs_clear_condition(idl->cs, table->class_->name);

        } else if (table->req_cond) {

            /* Update the monitor condition request according to the

             * db schema. */

            ovsdb_idl_set_condition__(idl, tc, table->req_cond);

        }

    }

    ovsdb_cs_free_schema(schema);

    return monitor_requests;

}

static struct ovsdb_cs_ops ovsdb_idl_cs_ops = {

    ovsdb_idl_compose_monitor_request,

};

const struct ovsdb_idl_class *

ovsdb_idl_get_class(const struct ovsdb_idl *idl)

{

    return idl->class_;

}

/* Given 'column' in some table in 'class', returns the table's class. */

const struct ovsdb_idl_table_class *

ovsdb_idl_table_class_from_column(const struct ovsdb_idl_class *class,

                                  const struct ovsdb_idl_column *column)

{

    for (size_t i = 0; i < class->n_tables; i++) {

        const struct ovsdb_idl_table_class *tc = &class->tables[i];

        if (column >= tc->columns && column < &tc->columns[tc->n_columns]) {

            return tc;

        }

    }

    OVS_NOT_REACHED();

}

/* Given 'column' in some table in 'idl', returns the table. */

static struct ovsdb_idl_table *

ovsdb_idl_table_from_column(const struct ovsdb_idl *idl,

                            const struct ovsdb_idl_column *column)

{

    const struct ovsdb_idl_table_class *tc =

        ovsdb_idl_table_class_from_column(idl->class_, column);

    return &idl->tables[tc - idl->class_->tables];

}

static unsigned char *

ovsdb_idl_get_mode(struct ovsdb_idl *idl,

                   const struct ovsdb_idl_column *column)

{

    ovs_assert(!idl->change_seqno);

    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,

                                                                      column);

    return &table->modes[column - table->class_->columns];

}

static void

ovsdb_idl_set_mode(struct ovsdb_idl *idl,

                   const struct ovsdb_idl_column *column,

                   unsigned char mode)

{

    const struct ovsdb_idl_table *table = ovsdb_idl_table_from_column(idl,

                                                                      column);

    size_t column_idx = column - table->class_->columns;

    if (table->modes[column_idx] != mode) {

        *ovsdb_idl_get_mode(idl, column) = mode;

    }

}

static void

add_ref_table(struct ovsdb_idl *idl, const struct ovsdb_base_type *base)

{

    if (base->type == OVSDB_TYPE_UUID && base->uuid.refTableName) {

        struct ovsdb_idl_table *table;

        table = shash_find_data(&idl->table_by_name, base->uuid.refTableName);

        if (table) {

            table->need_table = true;

        } else {

            VLOG_WARN("%s IDL class missing referenced table %s",

                      idl->class_->database, base->uuid.refTableName);

        }

    }

}