ovs-ofctl: Bug fix in flow_format()
[sliver-openvswitch.git] / ovsdb / file.c
1 /* Copyright (c) 2009, 2010, 2011, 2012 Nicira, Inc.
2  *
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15
16 #include <config.h>
17
18 #include "file.h"
19
20 #include <errno.h>
21 #include <fcntl.h>
22 #include <unistd.h>
23
24 #include "bitmap.h"
25 #include "column.h"
26 #include "log.h"
27 #include "json.h"
28 #include "lockfile.h"
29 #include "ovsdb.h"
30 #include "ovsdb-error.h"
31 #include "row.h"
32 #include "socket-util.h"
33 #include "table.h"
34 #include "timeval.h"
35 #include "transaction.h"
36 #include "uuid.h"
37 #include "util.h"
38 #include "vlog.h"
39
40 VLOG_DEFINE_THIS_MODULE(ovsdb_file);
41
42 /* Minimum number of milliseconds between database compactions. */
43 #define COMPACT_MIN_MSEC        (10 * 60 * 1000) /* 10 minutes. */
44
45 /* Minimum number of milliseconds between trying to compact the database if
46  * compacting fails. */
47 #define COMPACT_RETRY_MSEC      (60 * 1000)      /* 1 minute. */
48
49 /* A transaction being converted to JSON for writing to a file. */
50 struct ovsdb_file_txn {
51     struct json *json;          /* JSON for the whole transaction. */
52     struct json *table_json;    /* JSON for 'table''s transaction. */
53     struct ovsdb_table *table;  /* Table described in 'table_json'.  */
54 };
55
56 static void ovsdb_file_txn_init(struct ovsdb_file_txn *);
57 static void ovsdb_file_txn_add_row(struct ovsdb_file_txn *,
58                                    const struct ovsdb_row *old,
59                                    const struct ovsdb_row *new,
60                                    const unsigned long int *changed);
61 static struct ovsdb_error *ovsdb_file_txn_commit(struct json *,
62                                                  const char *comment,
63                                                  bool durable,
64                                                  struct ovsdb_log *);
65
66 static struct ovsdb_error *ovsdb_file_open__(const char *file_name,
67                                              const struct ovsdb_schema *,
68                                              bool read_only, struct ovsdb **,
69                                              struct ovsdb_file **);
70 static struct ovsdb_error *ovsdb_file_txn_from_json(
71     struct ovsdb *, const struct json *, bool converting,
72     long long int *date, struct ovsdb_txn **);
73 static struct ovsdb_error *ovsdb_file_create(struct ovsdb *,
74                                              struct ovsdb_log *,
75                                              const char *file_name,
76                                              long long int oldest_commit,
77                                              unsigned int n_transactions,
78                                              struct ovsdb_file **filep);
79
80 /* Opens database 'file_name' and stores a pointer to the new database in
81  * '*dbp'.  If 'read_only' is false, then the database will be locked and
82  * changes to the database will be written to disk.  If 'read_only' is true,
83  * the database will not be locked and changes to the database will persist
84  * only as long as the "struct ovsdb".
85  *
86  * If 'filep' is nonnull and 'read_only' is false, then on success sets
87  * '*filep' to an ovsdb_file that represents the open file.  This ovsdb_file
88  * persists until '*dbp' is destroyed.
89  *
90  * On success, returns NULL.  On failure, returns an ovsdb_error (which the
91  * caller must destroy) and sets '*dbp' and '*filep' to NULL. */
92 struct ovsdb_error *
93 ovsdb_file_open(const char *file_name, bool read_only,
94                 struct ovsdb **dbp, struct ovsdb_file **filep)
95 {
96     return ovsdb_file_open__(file_name, NULL, read_only, dbp, filep);
97 }
98
99 /* Opens database 'file_name' with an alternate schema.  The specified 'schema'
100  * is used to interpret the data in 'file_name', ignoring the schema actually
101  * stored in the file.  Data in the file for tables or columns that do not
102  * exist in 'schema' are ignored, but the ovsdb file format must otherwise be
103  * observed, including column constraints.
104  *
105  * This function can be useful for upgrading or downgrading databases to
106  * "almost-compatible" formats.
107  *
108  * The database will not be locked.  Changes to the database will persist only
109  * as long as the "struct ovsdb".
110  *
111  * On success, stores a pointer to the new database in '*dbp' and returns a
112  * null pointer.  On failure, returns an ovsdb_error (which the caller must
113  * destroy) and sets '*dbp' to NULL. */
114 struct ovsdb_error *
115 ovsdb_file_open_as_schema(const char *file_name,
116                           const struct ovsdb_schema *schema,
117                           struct ovsdb **dbp)
118 {
119     return ovsdb_file_open__(file_name, schema, true, dbp, NULL);
120 }
121
122 static struct ovsdb_error *
123 ovsdb_file_open_log(const char *file_name, enum ovsdb_log_open_mode open_mode,
124                     struct ovsdb_log **logp, struct ovsdb_schema **schemap)
125 {
126     struct ovsdb_schema *schema = NULL;
127     struct ovsdb_log *log = NULL;
128     struct ovsdb_error *error;
129     struct json *json = NULL;
130
131     ovs_assert(logp || schemap);
132
133     error = ovsdb_log_open(file_name, open_mode, -1, &log);
134     if (error) {
135         goto error;
136     }
137
138     error = ovsdb_log_read(log, &json);
139     if (error) {
140         goto error;
141     } else if (!json) {
142         error = ovsdb_io_error(EOF, "%s: database file contains no schema",
143                                file_name);
144         goto error;
145     }
146
147     if (schemap) {
148         error = ovsdb_schema_from_json(json, &schema);
149         if (error) {
150             error = ovsdb_wrap_error(error,
151                                      "failed to parse \"%s\" as ovsdb schema",
152                                      file_name);
153             goto error;
154         }
155     }
156     json_destroy(json);
157
158     if (logp) {
159         *logp = log;
160     } else {
161         ovsdb_log_close(log);
162     }
163     if (schemap) {
164         *schemap = schema;
165     }
166     return NULL;
167
168 error:
169     ovsdb_log_close(log);
170     json_destroy(json);
171     if (logp) {
172         *logp = NULL;
173     }
174     if (schemap) {
175         *schemap = NULL;
176     }
177     return error;
178 }
179
180 static struct ovsdb_error *
181 ovsdb_file_open__(const char *file_name,
182                   const struct ovsdb_schema *alternate_schema,
183                   bool read_only, struct ovsdb **dbp,
184                   struct ovsdb_file **filep)
185 {
186     enum ovsdb_log_open_mode open_mode;
187     long long int oldest_commit;
188     unsigned int n_transactions;
189     struct ovsdb_schema *schema = NULL;
190     struct ovsdb_error *error;
191     struct ovsdb_log *log;
192     struct json *json;
193     struct ovsdb *db = NULL;
194
195     /* In read-only mode there is no ovsdb_file so 'filep' must be null. */
196     ovs_assert(!(read_only && filep));
197
198     open_mode = read_only ? OVSDB_LOG_READ_ONLY : OVSDB_LOG_READ_WRITE;
199     error = ovsdb_file_open_log(file_name, open_mode, &log,
200                                 alternate_schema ? NULL : &schema);
201     if (error) {
202         goto error;
203     }
204
205     db = ovsdb_create(schema ? schema : ovsdb_schema_clone(alternate_schema));
206
207     oldest_commit = LLONG_MAX;
208     n_transactions = 0;
209     while ((error = ovsdb_log_read(log, &json)) == NULL && json) {
210         struct ovsdb_txn *txn;
211         long long int date;
212
213         error = ovsdb_file_txn_from_json(db, json, alternate_schema != NULL,
214                                          &date, &txn);
215         json_destroy(json);
216         if (error) {
217             ovsdb_log_unread(log);
218             break;
219         }
220
221         n_transactions++;
222         if (date < oldest_commit) {
223             oldest_commit = date;
224         }
225
226         error = ovsdb_txn_commit(txn, false);
227         if (error) {
228             ovsdb_log_unread(log);
229             break;
230         }
231     }
232     if (error) {
233         /* Log error but otherwise ignore it.  Probably the database just got
234          * truncated due to power failure etc. and we should use its current
235          * contents. */
236         char *msg = ovsdb_error_to_string(error);
237         VLOG_ERR("%s", msg);
238         free(msg);
239
240         ovsdb_error_destroy(error);
241     }
242
243     if (!read_only) {
244         struct ovsdb_file *file;
245
246         error = ovsdb_file_create(db, log, file_name, oldest_commit,
247                                   n_transactions, &file);
248         if (error) {
249             goto error;
250         }
251         if (filep) {
252             *filep = file;
253         }
254     } else {
255         ovsdb_log_close(log);
256     }
257
258     *dbp = db;
259     return NULL;
260
261 error:
262     *dbp = NULL;
263     if (filep) {
264         *filep = NULL;
265     }
266     ovsdb_destroy(db);
267     ovsdb_log_close(log);
268     return error;
269 }
270
271 static struct ovsdb_error *
272 ovsdb_file_update_row_from_json(struct ovsdb_row *row, bool converting,
273                                 const struct json *json)
274 {
275     struct ovsdb_table_schema *schema = row->table->schema;
276     struct ovsdb_error *error;
277     struct shash_node *node;
278
279     if (json->type != JSON_OBJECT) {
280         return ovsdb_syntax_error(json, NULL, "row must be JSON object");
281     }
282
283     SHASH_FOR_EACH (node, json_object(json)) {
284         const char *column_name = node->name;
285         const struct ovsdb_column *column;
286         struct ovsdb_datum datum;
287
288         column = ovsdb_table_schema_get_column(schema, column_name);
289         if (!column) {
290             if (converting) {
291                 continue;
292             }
293             return ovsdb_syntax_error(json, "unknown column",
294                                       "No column %s in table %s.",
295                                       column_name, schema->name);
296         }
297
298         error = ovsdb_datum_from_json(&datum, &column->type, node->data, NULL);
299         if (error) {
300             return error;
301         }
302         ovsdb_datum_swap(&row->fields[column->index], &datum);
303         ovsdb_datum_destroy(&datum, &column->type);
304     }
305
306     return NULL;
307 }
308
309 static struct ovsdb_error *
310 ovsdb_file_txn_row_from_json(struct ovsdb_txn *txn, struct ovsdb_table *table,
311                              bool converting,
312                              const struct uuid *row_uuid, struct json *json)
313 {
314     const struct ovsdb_row *row = ovsdb_table_get_row(table, row_uuid);
315     if (json->type == JSON_NULL) {
316         if (!row) {
317             return ovsdb_syntax_error(NULL, NULL, "transaction deletes "
318                                       "row "UUID_FMT" that does not exist",
319                                       UUID_ARGS(row_uuid));
320         }
321         ovsdb_txn_row_delete(txn, row);
322         return NULL;
323     } else if (row) {
324         return ovsdb_file_update_row_from_json(ovsdb_txn_row_modify(txn, row),
325                                                converting, json);
326     } else {
327         struct ovsdb_error *error;
328         struct ovsdb_row *new;
329
330         new = ovsdb_row_create(table);
331         *ovsdb_row_get_uuid_rw(new) = *row_uuid;
332         error = ovsdb_file_update_row_from_json(new, converting, json);
333         if (error) {
334             ovsdb_row_destroy(new);
335         } else {
336             ovsdb_txn_row_insert(txn, new);
337         }
338         return error;
339     }
340 }
341
342 static struct ovsdb_error *
343 ovsdb_file_txn_table_from_json(struct ovsdb_txn *txn,
344                                struct ovsdb_table *table,
345                                bool converting, struct json *json)
346 {
347     struct shash_node *node;
348
349     if (json->type != JSON_OBJECT) {
350         return ovsdb_syntax_error(json, NULL, "object expected");
351     }
352
353     SHASH_FOR_EACH (node, json->u.object) {
354         const char *uuid_string = node->name;
355         struct json *txn_row_json = node->data;
356         struct ovsdb_error *error;
357         struct uuid row_uuid;
358
359         if (!uuid_from_string(&row_uuid, uuid_string)) {
360             return ovsdb_syntax_error(json, NULL, "\"%s\" is not a valid UUID",
361                                       uuid_string);
362         }
363
364         error = ovsdb_file_txn_row_from_json(txn, table, converting,
365                                              &row_uuid, txn_row_json);
366         if (error) {
367             return error;
368         }
369     }
370
371     return NULL;
372 }
373
374 /* Converts 'json' to an ovsdb_txn for 'db', storing the new transaction in
375  * '*txnp'.  Returns NULL if successful, otherwise an error.
376  *
377  * If 'converting' is true, then unknown table and column names are ignored
378  * (which can ease upgrading and downgrading schemas); otherwise, they are
379  * treated as errors.
380  *
381  * If successful, the date associated with the transaction, as the number of
382  * milliseconds since the epoch, is stored in '*date'.  If the transaction does
383  * not include a date, LLONG_MAX is stored. */
384 static struct ovsdb_error *
385 ovsdb_file_txn_from_json(struct ovsdb *db, const struct json *json,
386                          bool converting, long long int *date,
387                          struct ovsdb_txn **txnp)
388 {
389     struct ovsdb_error *error;
390     struct shash_node *node;
391     struct ovsdb_txn *txn;
392
393     *txnp = NULL;
394     *date = LLONG_MAX;
395
396     if (json->type != JSON_OBJECT) {
397         return ovsdb_syntax_error(json, NULL, "object expected");
398     }
399
400     txn = ovsdb_txn_create(db);
401     SHASH_FOR_EACH (node, json->u.object) {
402         const char *table_name = node->name;
403         struct json *node_json = node->data;
404         struct ovsdb_table *table;
405
406         table = shash_find_data(&db->tables, table_name);
407         if (!table) {
408             if (!strcmp(table_name, "_date")
409                 && node_json->type == JSON_INTEGER) {
410                 *date = json_integer(node_json);
411                 continue;
412             } else if (!strcmp(table_name, "_comment") || converting) {
413                 continue;
414             }
415
416             error = ovsdb_syntax_error(json, "unknown table",
417                                        "No table named %s.", table_name);
418             goto error;
419         }
420
421         error = ovsdb_file_txn_table_from_json(txn, table, converting,
422                                                node_json);
423         if (error) {
424             goto error;
425         }
426     }
427     *txnp = txn;
428     return NULL;
429
430 error:
431     ovsdb_txn_abort(txn);
432     return error;
433 }
434
435 static struct ovsdb_error *
436 ovsdb_file_save_copy__(const char *file_name, int locking,
437                        const char *comment, const struct ovsdb *db,
438                        struct ovsdb_log **logp)
439 {
440     const struct shash_node *node;
441     struct ovsdb_file_txn ftxn;
442     struct ovsdb_error *error;
443     struct ovsdb_log *log;
444     struct json *json;
445
446     error = ovsdb_log_open(file_name, OVSDB_LOG_CREATE, locking, &log);
447     if (error) {
448         return error;
449     }
450
451     /* Write schema. */
452     json = ovsdb_schema_to_json(db->schema);
453     error = ovsdb_log_write(log, json);
454     json_destroy(json);
455     if (error) {
456         goto exit;
457     }
458
459     /* Write data. */
460     ovsdb_file_txn_init(&ftxn);
461     SHASH_FOR_EACH (node, &db->tables) {
462         const struct ovsdb_table *table = node->data;
463         const struct ovsdb_row *row;
464
465         HMAP_FOR_EACH (row, hmap_node, &table->rows) {
466             ovsdb_file_txn_add_row(&ftxn, NULL, row, NULL);
467         }
468     }
469     error = ovsdb_file_txn_commit(ftxn.json, comment, true, log);
470
471 exit:
472     if (logp) {
473         if (!error) {
474             *logp = log;
475             log = NULL;
476         } else {
477             *logp = NULL;
478         }
479     }
480     ovsdb_log_close(log);
481     if (error) {
482         remove(file_name);
483     }
484     return error;
485 }
486
487 /* Saves a snapshot of 'db''s current contents as 'file_name'.  If 'comment' is
488  * nonnull, then it is added along with the data contents and can be viewed
489  * with "ovsdb-tool show-log".
490  *
491  * 'locking' is passed along to ovsdb_log_open() untouched. */
492 struct ovsdb_error *
493 ovsdb_file_save_copy(const char *file_name, int locking,
494                      const char *comment, const struct ovsdb *db)
495 {
496     return ovsdb_file_save_copy__(file_name, locking, comment, db, NULL);
497 }
498
499 /* Opens database 'file_name', reads its schema, and closes it.  On success,
500  * stores the schema into '*schemap' and returns NULL; the caller then owns the
501  * schema.  On failure, returns an ovsdb_error (which the caller must destroy)
502  * and sets '*dbp' to NULL. */
503 struct ovsdb_error *
504 ovsdb_file_read_schema(const char *file_name, struct ovsdb_schema **schemap)
505 {
506     ovs_assert(schemap != NULL);
507     return ovsdb_file_open_log(file_name, OVSDB_LOG_READ_ONLY, NULL, schemap);
508 }
509 \f
510 /* Replica implementation. */
511
512 struct ovsdb_file {
513     struct ovsdb_replica replica;
514     struct ovsdb *db;
515     struct ovsdb_log *log;
516     char *file_name;
517     long long int oldest_commit;
518     long long int next_compact;
519     unsigned int n_transactions;
520 };
521
522 static const struct ovsdb_replica_class ovsdb_file_class;
523
524 static struct ovsdb_error *
525 ovsdb_file_create(struct ovsdb *db, struct ovsdb_log *log,
526                   const char *file_name,
527                   long long int oldest_commit,
528                   unsigned int n_transactions,
529                   struct ovsdb_file **filep)
530 {
531     long long int now = time_msec();
532     struct ovsdb_file *file;
533     char *deref_name;
534     char *abs_name;
535
536     /* Use the absolute name of the file because ovsdb-server opens its
537      * database before daemonize() chdirs to "/". */
538     deref_name = follow_symlinks(file_name);
539     abs_name = abs_file_name(NULL, deref_name);
540     free(deref_name);
541     if (!abs_name) {
542         *filep = NULL;
543         return ovsdb_io_error(0, "could not determine current "
544                               "working directory");
545     }
546
547     file = xmalloc(sizeof *file);
548     ovsdb_replica_init(&file->replica, &ovsdb_file_class);
549     file->db = db;
550     file->log = log;
551     file->file_name = abs_name;
552     file->oldest_commit = MIN(oldest_commit, now);
553     file->next_compact = file->oldest_commit + COMPACT_MIN_MSEC;
554     file->n_transactions = n_transactions;
555     ovsdb_add_replica(db, &file->replica);
556
557     *filep = file;
558     return NULL;
559 }
560
561 static struct ovsdb_file *
562 ovsdb_file_cast(struct ovsdb_replica *replica)
563 {
564     ovs_assert(replica->class == &ovsdb_file_class);
565     return CONTAINER_OF(replica, struct ovsdb_file, replica);
566 }
567
568 static bool
569 ovsdb_file_change_cb(const struct ovsdb_row *old,
570                      const struct ovsdb_row *new,
571                      const unsigned long int *changed,
572                      void *ftxn_)
573 {
574     struct ovsdb_file_txn *ftxn = ftxn_;
575     ovsdb_file_txn_add_row(ftxn, old, new, changed);
576     return true;
577 }
578
579 static struct ovsdb_error *
580 ovsdb_file_commit(struct ovsdb_replica *replica,
581                   const struct ovsdb_txn *txn, bool durable)
582 {
583     struct ovsdb_file *file = ovsdb_file_cast(replica);
584     struct ovsdb_file_txn ftxn;
585     struct ovsdb_error *error;
586
587     ovsdb_file_txn_init(&ftxn);
588     ovsdb_txn_for_each_change(txn, ovsdb_file_change_cb, &ftxn);
589     if (!ftxn.json) {
590         /* Nothing to commit. */
591         return NULL;
592     }
593
594     error = ovsdb_file_txn_commit(ftxn.json, ovsdb_txn_get_comment(txn),
595                                   durable, file->log);
596     if (error) {
597         return error;
598     }
599     file->n_transactions++;
600
601     /* If it has been at least COMPACT_MIN_MSEC ms since the last time we
602      * compacted (or at least COMPACT_RETRY_MSEC ms since the last time we
603      * tried), and if there are at least 100 transactions in the database, and
604      * if the database is at least 10 MB, then compact the database. */
605     if (time_msec() >= file->next_compact
606         && file->n_transactions >= 100
607         && ovsdb_log_get_offset(file->log) >= 10 * 1024 * 1024)
608     {
609         error = ovsdb_file_compact(file);
610         if (error) {
611             char *s = ovsdb_error_to_string(error);
612             ovsdb_error_destroy(error);
613             VLOG_WARN("%s: compacting database failed (%s), retrying in "
614                       "%d seconds",
615                       file->file_name, s, COMPACT_RETRY_MSEC / 1000);
616             free(s);
617
618             file->next_compact = time_msec() + COMPACT_RETRY_MSEC;
619         }
620     }
621
622     return NULL;
623 }
624
625 struct ovsdb_error *
626 ovsdb_file_compact(struct ovsdb_file *file)
627 {
628     struct ovsdb_log *new_log = NULL;
629     struct lockfile *tmp_lock = NULL;
630     struct ovsdb_error *error;
631     char *tmp_name = NULL;
632     char *comment = NULL;
633     int retval;
634
635     comment = xasprintf("compacting database online "
636                         "(%.3f seconds old, %u transactions, %llu bytes)",
637                         (time_msec() - file->oldest_commit) / 1000.0,
638                         file->n_transactions,
639                         (unsigned long long) ovsdb_log_get_offset(file->log));
640     VLOG_INFO("%s: %s", file->file_name, comment);
641
642     /* Commit the old version, so that we can be assured that we'll eventually
643      * have either the old or the new version. */
644     error = ovsdb_log_commit(file->log);
645     if (error) {
646         goto exit;
647     }
648
649     /* Lock temporary file. */
650     tmp_name = xasprintf("%s.tmp", file->file_name);
651     retval = lockfile_lock(tmp_name, &tmp_lock);
652     if (retval) {
653         error = ovsdb_io_error(retval, "could not get lock on %s", tmp_name);
654         goto exit;
655     }
656
657     /* Remove temporary file.  (It might not exist.) */
658     if (unlink(tmp_name) < 0 && errno != ENOENT) {
659         error = ovsdb_io_error(errno, "failed to remove %s", tmp_name);
660         goto exit;
661     }
662
663     /* Save a copy. */
664     error = ovsdb_file_save_copy__(tmp_name, false, comment, file->db,
665                                    &new_log);
666     if (error) {
667         goto exit;
668     }
669
670     /* Replace original by temporary. */
671     if (rename(tmp_name, file->file_name)) {
672         error = ovsdb_io_error(errno, "failed to rename \"%s\" to \"%s\"",
673                                tmp_name, file->file_name);
674         goto exit;
675     }
676     fsync_parent_dir(file->file_name);
677
678 exit:
679     if (!error) {
680         ovsdb_log_close(file->log);
681         file->log = new_log;
682         file->oldest_commit = time_msec();
683         file->next_compact = file->oldest_commit + COMPACT_MIN_MSEC;
684         file->n_transactions = 1;
685     } else {
686         ovsdb_log_close(new_log);
687         if (tmp_lock) {
688             unlink(tmp_name);
689         }
690     }
691
692     lockfile_unlock(tmp_lock);
693     free(tmp_name);
694     free(comment);
695
696     return error;
697 }
698
699 static void
700 ovsdb_file_destroy(struct ovsdb_replica *replica)
701 {
702     struct ovsdb_file *file = ovsdb_file_cast(replica);
703
704     ovsdb_log_close(file->log);
705     free(file->file_name);
706     free(file);
707 }
708
709 static const struct ovsdb_replica_class ovsdb_file_class = {
710     ovsdb_file_commit,
711     ovsdb_file_destroy
712 };
713 \f
714 static void
715 ovsdb_file_txn_init(struct ovsdb_file_txn *ftxn)
716 {
717     ftxn->json = NULL;
718     ftxn->table_json = NULL;
719     ftxn->table = NULL;
720 }
721
722 static void
723 ovsdb_file_txn_add_row(struct ovsdb_file_txn *ftxn,
724                        const struct ovsdb_row *old,
725                        const struct ovsdb_row *new,
726                        const unsigned long int *changed)
727 {
728     struct json *row;
729
730     if (!new) {
731         row = json_null_create();
732     } else {
733         struct shash_node *node;
734
735         row = old ? NULL : json_object_create();
736         SHASH_FOR_EACH (node, &new->table->schema->columns) {
737             const struct ovsdb_column *column = node->data;
738             const struct ovsdb_type *type = &column->type;
739             unsigned int idx = column->index;
740
741             if (idx != OVSDB_COL_UUID && column->persistent
742                 && (old
743                     ? bitmap_is_set(changed, idx)
744                     : !ovsdb_datum_is_default(&new->fields[idx], type)))
745             {
746                 if (!row) {
747                     row = json_object_create();
748                 }
749                 json_object_put(row, column->name,
750                                 ovsdb_datum_to_json(&new->fields[idx], type));
751             }
752         }
753     }
754
755     if (row) {
756         struct ovsdb_table *table = new ? new->table : old->table;
757         char uuid[UUID_LEN + 1];
758
759         if (table != ftxn->table) {
760             /* Create JSON object for transaction overall. */
761             if (!ftxn->json) {
762                 ftxn->json = json_object_create();
763             }
764
765             /* Create JSON object for transaction on this table. */
766             ftxn->table_json = json_object_create();
767             ftxn->table = table;
768             json_object_put(ftxn->json, table->schema->name, ftxn->table_json);
769         }
770
771         /* Add row to transaction for this table. */
772         snprintf(uuid, sizeof uuid,
773                  UUID_FMT, UUID_ARGS(ovsdb_row_get_uuid(new ? new : old)));
774         json_object_put(ftxn->table_json, uuid, row);
775     }
776 }
777
778 static struct ovsdb_error *
779 ovsdb_file_txn_commit(struct json *json, const char *comment,
780                       bool durable, struct ovsdb_log *log)
781 {
782     struct ovsdb_error *error;
783
784     if (!json) {
785         json = json_object_create();
786     }
787     if (comment) {
788         json_object_put_string(json, "_comment", comment);
789     }
790     json_object_put(json, "_date", json_integer_create(time_wall()));
791
792     error = ovsdb_log_write(log, json);
793     json_destroy(json);
794     if (error) {
795         return ovsdb_wrap_error(error, "writing transaction failed");
796     }
797
798     if (durable) {
799         error = ovsdb_log_commit(log);
800         if (error) {
801             return ovsdb_wrap_error(error, "committing transaction failed");
802         }
803     }
804
805     return NULL;
806 }