ovsdb-data: Deduplicate string atoms.

ovsdb-server spends a lot of time cloning atoms for various reasons,
e.g. to create a diff of two rows or to clone a row to the transaction.
All atoms, except for strings, contains a simple value that could be
copied in efficient way, but duplicating strings every time has a
significant performance impact.

Introducing a new reference-counted structure 'ovsdb_atom_string'
that allows to not copy strings every time, but just increase a
reference counter.

This change allows to increase transaction throughput in benchmarks
up to 2x for standalone databases and 3x for clustered databases, i.e.
number of transactions that ovsdb-server can handle per second.
It also noticeably reduces memory consumption of ovsdb-server.

Next step will be to consolidate this structure with json strings,
so we will not need to duplicate strings while converting database
objects to json and back.

Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
Acked-by: Dumitru Ceara <dceara@redhat.com>
Acked-by: Mark D. Gray <mark.d.gray@redhat.com>

5 files changed, 59 insertions, 31 deletions

diff --git a/lib/db-ctl-base.c b/lib/db-ctl-base.c
index 3923683fd..707456158 100644
--- a/lib/db-ctl-base.c
+++ b/lib/db-ctl-base.c
@@ -247,15 +247,15 @@ record_id_equals(const union ovsdb_atom *name, enum ovsdb_atomic_type type,
                  const char *record_id)
 {
     if (type == OVSDB_TYPE_STRING) {
-        if (!strcmp(name->string, record_id)) {
+        if (!strcmp(name->s->string, record_id)) {
             return true;
         }
 
         struct uuid uuid;
         size_t len = strlen(record_id);
         if (len >= 4
-            && uuid_from_string(&uuid, name->string)
-            && !strncmp(name->string, record_id, len)) {
+            && uuid_from_string(&uuid, name->s->string)
+            && !strncmp(name->s->string, record_id, len)) {
             return true;
         }
 
@@ -318,14 +318,15 @@ get_row_by_id(struct ctl_context *ctx,
         if (!id->key) {
             name = datum->n == 1 ? &datum->keys[0] : NULL;
         } else {
-            const union ovsdb_atom key_atom
-                = { .string = CONST_CAST(char *, id->key) };
+            union ovsdb_atom key_atom = {
+                .s = ovsdb_atom_string_create(CONST_CAST(char *, id->key)) };
             unsigned int i;
 
             if (ovsdb_datum_find_key(datum, &key_atom,
                                      OVSDB_TYPE_STRING, &i)) {
                 name = &datum->values[i];
             }
+            ovsdb_atom_destroy(&key_atom, OVSDB_TYPE_STRING);
         }
         if (!name) {
             continue;
diff --git a/lib/ovsdb-cs.c b/lib/ovsdb-cs.c
index 659d49dbf..fcb6fe1b3 100644
--- a/lib/ovsdb-cs.c
+++ b/lib/ovsdb-cs.c
@@ -1833,7 +1833,7 @@ server_column_get_string(const struct server_row *row,
 {
     ovs_assert(server_columns[index].type.key.type == OVSDB_TYPE_STRING);
     const struct ovsdb_datum *d = &row->data[index];
-    return d->n == 1 ? d->keys[0].string : default_value;
+    return d->n == 1 ? d->keys[0].s->string : default_value;
 }
 
 static bool
diff --git a/lib/ovsdb-data.c b/lib/ovsdb-data.c
index 04e0306e2..6654ed6de 100644
--- a/lib/ovsdb-data.c
+++ b/lib/ovsdb-data.c
@@ -74,7 +74,7 @@ ovsdb_atom_init_default(union ovsdb_atom *atom, enum ovsdb_atomic_type type)
         break;
 
     case OVSDB_TYPE_STRING:
-        atom->string = xmemdup("", 1);
+        atom->s = ovsdb_atom_string_create_nocopy(xmemdup("", 1));
         break;
 
     case OVSDB_TYPE_UUID:
@@ -136,7 +136,7 @@ ovsdb_atom_is_default(const union ovsdb_atom *atom,
         return atom->boolean == false;
 
     case OVSDB_TYPE_STRING:
-        return atom->string[0] == '\0';
+        return atom->s->string[0] == '\0';
 
     case OVSDB_TYPE_UUID:
         return uuid_is_zero(&atom->uuid);
@@ -172,7 +172,8 @@ ovsdb_atom_clone(union ovsdb_atom *new, const union ovsdb_atom *old,
         break;
 
     case OVSDB_TYPE_STRING:
-        new->string = xstrdup(old->string);
+        new->s = old->s;
+        new->s->n_refs++;
         break;
 
     case OVSDB_TYPE_UUID:
@@ -214,7 +215,7 @@ ovsdb_atom_hash(const union ovsdb_atom *atom, enum ovsdb_atomic_type type,
         return hash_boolean(atom->boolean, basis);
 
     case OVSDB_TYPE_STRING:
-        return hash_string(atom->string, basis);
+        return hash_string(atom->s->string, basis);
 
     case OVSDB_TYPE_UUID:
         return hash_int(uuid_hash(&atom->uuid), basis);
@@ -246,7 +247,7 @@ ovsdb_atom_compare_3way(const union ovsdb_atom *a,
         return a->boolean - b->boolean;
 
     case OVSDB_TYPE_STRING:
-        return strcmp(a->string, b->string);
+        return a->s == b->s ? 0 : strcmp(a->s->string, b->s->string);
 
     case OVSDB_TYPE_UUID:
         return uuid_compare_3way(&a->uuid, &b->uuid);
@@ -404,7 +405,7 @@ ovsdb_atom_from_json__(union ovsdb_atom *atom,
 
     case OVSDB_TYPE_STRING:
         if (json->type == JSON_STRING) {
-            atom->string = xstrdup(json->string);
+            atom->s = ovsdb_atom_string_create(json->string);
             return NULL;
         }
         break;
@@ -473,7 +474,7 @@ ovsdb_atom_to_json(const union ovsdb_atom *atom, enum ovsdb_atomic_type type)
         return json_boolean_create(atom->boolean);
 
     case OVSDB_TYPE_STRING:
-        return json_string_create(atom->string);
+        return json_string_create(atom->s->string);
 
     case OVSDB_TYPE_UUID:
         return wrap_json("uuid", json_string_create_nocopy(
@@ -551,14 +552,18 @@ ovsdb_atom_from_string__(union ovsdb_atom *atom,
             if (s_len < 2 || s[s_len - 1] != '"') {
                 return xasprintf("%s: missing quote at end of "
                                  "quoted string", s);
-            } else if (!json_string_unescape(s + 1, s_len - 2,
-                                             &atom->string)) {
-                char *error = xasprintf("%s: %s", s, atom->string);
-                free(atom->string);
-                return error;
+            } else {
+                char *res;
+                if (json_string_unescape(s + 1, s_len - 2, &res)) {
+                    atom->s = ovsdb_atom_string_create_nocopy(res);
+                } else {
+                    char *error = xasprintf("%s: %s", s, res);
+                    free(res);
+                    return error;
+                }
             }
         } else {
-            atom->string = xstrdup(s);
+            atom->s = ovsdb_atom_string_create(s);
         }
         break;
 
@@ -721,14 +726,14 @@ ovsdb_atom_to_string(const union ovsdb_atom *atom, enum ovsdb_atomic_type type,
         break;
 
     case OVSDB_TYPE_STRING:
-        if (string_needs_quotes(atom->string)) {
+        if (string_needs_quotes(atom->s->string)) {
             struct json json;
 
             json.type = JSON_STRING;
-            json.string = atom->string;
+            json.string = atom->s->string;
             json_to_ds(&json, 0, out);
         } else {
-            ds_put_cstr(out, atom->string);
+            ds_put_cstr(out, atom->s->string);
         }
         break;
 
@@ -750,7 +755,7 @@ ovsdb_atom_to_bare(const union ovsdb_atom *atom, enum ovsdb_atomic_type type,
                    struct ds *out)
 {
     if (type == OVSDB_TYPE_STRING) {
-        ds_put_cstr(out, atom->string);
+        ds_put_cstr(out, atom->s->string);
     } else {
         ovsdb_atom_to_string(atom, type, out);
     }
@@ -877,7 +882,7 @@ ovsdb_atom_check_constraints(const union ovsdb_atom *atom,
         return NULL;
 
     case OVSDB_TYPE_STRING:
-        return check_string_constraints(atom->string, &base->string);
+        return check_string_constraints(atom->s->string, &base->string);
 
     case OVSDB_TYPE_UUID:
         return NULL;
@@ -1691,8 +1696,8 @@ ovsdb_datum_from_smap(struct ovsdb_datum *datum, const struct smap *smap)
     struct smap_node *node;
     size_t i = 0;
     SMAP_FOR_EACH (node, smap) {
-        datum->keys[i].string = xstrdup(node->key);
-        datum->values[i].string = xstrdup(node->value);
+        datum->keys[i].s = ovsdb_atom_string_create(node->key);
+        datum->values[i].s = ovsdb_atom_string_create(node->value);
         i++;
     }
     ovs_assert(i == datum->n);
diff --git a/lib/ovsdb-data.h b/lib/ovsdb-data.h
index b634e97ff..f66ed3472 100644
--- a/lib/ovsdb-data.h
+++ b/lib/ovsdb-data.h
@@ -20,6 +20,7 @@
 #include "compiler.h"
 #include "ovsdb-types.h"
 #include "openvswitch/shash.h"
+#include "util.h"
 
 #ifdef  __cplusplus
 extern "C" {
@@ -31,12 +32,33 @@ struct ds;
 struct ovsdb_symbol_table;
 struct smap;
 
+struct ovsdb_atom_string {
+    char *string;
+    size_t n_refs;
+};
+
+static inline struct ovsdb_atom_string *
+ovsdb_atom_string_create_nocopy(char *str)
+{
+    struct ovsdb_atom_string *s = xzalloc(sizeof *s);
+
+    s->string = str;
+    s->n_refs = 1;
+    return s;
+}
+
+static inline struct ovsdb_atom_string *
+ovsdb_atom_string_create(const char *str)
+{
+    return ovsdb_atom_string_create_nocopy(xstrdup(str));
+}
+
 /* One value of an atomic type (given by enum ovs_atomic_type). */
 union ovsdb_atom {
     int64_t integer;
     double real;
     bool boolean;
-    char *string;
+    struct ovsdb_atom_string *s;
     struct uuid uuid;
 };
 
@@ -66,8 +88,9 @@ ovsdb_atom_needs_destruction(enum ovsdb_atomic_type type)
 static inline void
 ovsdb_atom_destroy(union ovsdb_atom *atom, enum ovsdb_atomic_type type)
 {
-    if (type == OVSDB_TYPE_STRING) {
-        free(atom->string);
+    if (type == OVSDB_TYPE_STRING && !--atom->s->n_refs) {
+        free(atom->s->string);
+        free(atom->s);
     }
 }
 
diff --git a/lib/ovsdb-idl.c b/lib/ovsdb-idl.c
index c9c567e2c..383a601f6 100644
--- a/lib/ovsdb-idl.c
+++ b/lib/ovsdb-idl.c
@@ -1951,8 +1951,7 @@ ovsdb_idl_index_destroy_row(const struct ovsdb_idl_row *row_)
     BITMAP_FOR_EACH_1 (i, class->n_columns, row->written) {
         c = &class->columns[i];
         (c->unparse) (row);
-        free(row->new_datum[i].values);
-        free(row->new_datum[i].keys);
+        ovsdb_datum_destroy(&row->new_datum[i], &c->type);
     }
     free(row->new_datum);
     free(row->written);