python/ovs/db/idl.py

   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 import uuid
  16
  17 import ovs.jsonrpc
  18 import ovs.db.parser
  19 import ovs.db.schema
  20 from ovs.db import error
  21 import ovs.ovsuuid
  22 import ovs.poller
  23 import ovs.vlog
  24
  25 vlog = ovs.vlog.Vlog("idl")
  26
  27 __pychecker__ = 'no-classattr no-objattrs'
  28
  29
  30 class Idl:
  31     """Open vSwitch Database Interface Definition Language (OVSDB IDL).
  32
  33     The OVSDB IDL maintains an in-memory replica of a database.  It issues RPC
  34     requests to an OVSDB database server and parses the responses, converting
  35     raw JSON into data structures that are easier for clients to digest.
  36
  37     The IDL also assists with issuing database transactions.  The client
  38     creates a transaction, manipulates the IDL data structures, and commits or
  39     aborts the transaction.  The IDL then composes and issues the necessary
  40     JSON-RPC requests and reports to the client whether the transaction
  41     completed successfully.
  42
  43     The client is allowed to access the following attributes directly, in a
  44     read-only fashion:
  45
  46     - 'tables': This is the 'tables' map in the ovs.db.schema.DbSchema provided
  47       to the Idl constructor.  Each ovs.db.schema.TableSchema in the map is
  48       annotated with a new attribute 'rows', which is a dict from a uuid.UUID
  49       to a Row object.
  50
  51       The client may directly read and write the Row objects referenced by the
  52       'rows' map values.  Refer to Row for more details.
  53
  54     - 'change_seqno': A number that represents the IDL's state.  When the IDL
  55       is updated (by Idl.run()), its value changes.  The sequence number can
  56       occasionally change even if the database does not.  This happens if the
  57       connection to the database drops and reconnects, which causes the
  58       database contents to be reloaded even if they didn't change.  (It could
  59       also happen if the database server sends out a "change" that reflects
  60       what the IDL already thought was in the database.  The database server is
  61       not supposed to do that, but bugs could in theory cause it to do so.)
  62
  63     - 'lock_name': The name of the lock configured with Idl.set_lock(), or None
  64       if no lock is configured.
  65
  66     - 'has_lock': True, if the IDL is configured to obtain a lock and owns that
  67       lock, and False otherwise.
  68
  69       Locking and unlocking happens asynchronously from the database client's
  70       point of view, so the information is only useful for optimization
  71       (e.g. if the client doesn't have the lock then there's no point in trying
  72       to write to the database).
  73
  74     - 'is_lock_contended': True, if the IDL is configured to obtain a lock but
  75       the database server has indicated that some other client already owns the
  76       requested lock, and False otherwise.
  77
  78     - 'txn': The ovs.db.idl.Transaction object for the database transaction
  79       currently being constructed, if there is one, or None otherwise.
  80 """
  81
  82     def __init__(self, remote, schema):
  83         """Creates and returns a connection to the database named 'db_name' on
  84         'remote', which should be in a form acceptable to
  85         ovs.jsonrpc.session.open().  The connection will maintain an in-memory
  86         replica of the remote database.
  87
  88         'schema' should be the schema for the remote database.  The caller may
  89         have cut it down by removing tables or columns that are not of
  90         interest.  The IDL will only replicate the tables and columns that
  91         remain.  The caller may also add a attribute named 'alert' to selected
  92         remaining columns, setting its value to False; if so, then changes to
  93         those columns will not be considered changes to the database for the
  94         purpose of the return value of Idl.run() and Idl.change_seqno.  This is
  95         useful for columns that the IDL's client will write but not read.
  96
  97         As a convenience to users, 'schema' may also be an instance of the
  98         SchemaHelper class.
  99
 100         The IDL uses and modifies 'schema' directly."""
 101
 102         assert isinstance(schema, SchemaHelper)
 103         schema = schema.get_idl_schema()
 104
 105         self.tables = schema.tables
 106         self._db = schema
 107         self._session = ovs.jsonrpc.Session.open(remote)
 108         self._monitor_request_id = None
 109         self._last_seqno = None
 110         self.change_seqno = 0
 111
 112         # Database locking.
 113         self.lock_name = None          # Name of lock we need, None if none.
 114         self.has_lock = False          # Has db server said we have the lock?
 115         self.is_lock_contended = False  # Has db server said we can't get lock?
 116         self._lock_request_id = None   # JSON-RPC ID of in-flight lock request.
 117
 118         # Transaction support.
 119         self.txn = None
 120         self._outstanding_txns = {}
 121
 122         for table in schema.tables.itervalues():
 123             for column in table.columns.itervalues():
 124                 if not hasattr(column, 'alert'):
 125                     column.alert = True
 126             table.need_table = False
 127             table.rows = {}
 128             table.idl = self
 129
 130     def close(self):
 131         """Closes the connection to the database.  The IDL will no longer
 132         update."""
 133         self._session.close()
 134
 135     def run(self):
 136         """Processes a batch of messages from the database server.  Returns
 137         True if the database as seen through the IDL changed, False if it did
 138         not change.  The initial fetch of the entire contents of the remote
 139         database is considered to be one kind of change.  If the IDL has been
 140         configured to acquire a database lock (with Idl.set_lock()), then
 141         successfully acquiring the lock is also considered to be a change.
 142
 143         This function can return occasional false positives, that is, report
 144         that the database changed even though it didn't.  This happens if the
 145         connection to the database drops and reconnects, which causes the
 146         database contents to be reloaded even if they didn't change.  (It could
 147         also happen if the database server sends out a "change" that reflects
 148         what we already thought was in the database, but the database server is
 149         not supposed to do that.)
 150
 151         As an alternative to checking the return value, the client may check
 152         for changes in self.change_seqno."""
 153         assert not self.txn
 154         initial_change_seqno = self.change_seqno
 155         self._session.run()
 156         i = 0
 157         while i < 50:
 158             i += 1
 159             if not self._session.is_connected():
 160                 break
 161
 162             seqno = self._session.get_seqno()
 163             if seqno != self._last_seqno:
 164                 self._last_seqno = seqno
 165                 self.__txn_abort_all()
 166                 self.__send_monitor_request()
 167                 if self.lock_name:
 168                     self.__send_lock_request()
 169                 break
 170
 171             msg = self._session.recv()
 172             if msg is None:
 173                 break
 174             if (msg.type == ovs.jsonrpc.Message.T_NOTIFY
 175                 and msg.method == "update"
 176                 and len(msg.params) == 2
 177                 and msg.params[0] == None):
 178                 # Database contents changed.
 179                 self.__parse_update(msg.params[1])
 180             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
 181                   and self._monitor_request_id is not None
 182                   and self._monitor_request_id == msg.id):
 183                 # Reply to our "monitor" request.
 184                 try:
 185                     self.change_seqno += 1
 186                     self._monitor_request_id = None
 187                     self.__clear()
 188                     self.__parse_update(msg.result)
 189                 except error.Error, e:
 190                     vlog.err("%s: parse error in received schema: %s"
 191                               % (self._session.get_name(), e))
 192                     self.__error()
 193             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
 194                   and self._lock_request_id is not None
 195                   and self._lock_request_id == msg.id):
 196                 # Reply to our "lock" request.
 197                 self.__parse_lock_reply(msg.result)
 198             elif (msg.type == ovs.jsonrpc.Message.T_NOTIFY
 199                   and msg.method == "locked"):
 200                 # We got our lock.
 201                 self.__parse_lock_notify(msg.params, True)
 202             elif (msg.type == ovs.jsonrpc.Message.T_NOTIFY
 203                   and msg.method == "stolen"):
 204                 # Someone else stole our lock.
 205                 self.__parse_lock_notify(msg.params, False)
 206             elif msg.type == ovs.jsonrpc.Message.T_NOTIFY and msg.id == "echo":
 207                 # Reply to our echo request.  Ignore it.
 208                 pass
 209             elif (msg.type in (ovs.jsonrpc.Message.T_ERROR,
 210                                ovs.jsonrpc.Message.T_REPLY)
 211                   and self.__txn_process_reply(msg)):
 212                 # __txn_process_reply() did everything needed.
 213                 pass
 214             else:
 215                 # This can happen if a transaction is destroyed before we
 216                 # receive the reply, so keep the log level low.
 217                 vlog.dbg("%s: received unexpected %s message"
 218                          % (self._session.get_name(),
 219                              ovs.jsonrpc.Message.type_to_string(msg.type)))
 220
 221         return initial_change_seqno != self.change_seqno
 222
 223     def wait(self, poller):
 224         """Arranges for poller.block() to wake up when self.run() has something
 225         to do or when activity occurs on a transaction on 'self'."""
 226         self._session.wait(poller)
 227         self._session.recv_wait(poller)
 228
 229     def has_ever_connected(self):
 230         """Returns True, if the IDL successfully connected to the remote
 231         database and retrieved its contents (even if the connection
 232         subsequently dropped and is in the process of reconnecting).  If so,
 233         then the IDL contains an atomic snapshot of the database's contents
 234         (but it might be arbitrarily old if the connection dropped).
 235
 236         Returns False if the IDL has never connected or retrieved the
 237         database's contents.  If so, the IDL is empty."""
 238         return self.change_seqno != 0
 239
 240     def force_reconnect(self):
 241         """Forces the IDL to drop its connection to the database and reconnect.
 242         In the meantime, the contents of the IDL will not change."""
 243         self._session.force_reconnect()
 244
 245     def set_lock(self, lock_name):
 246         """If 'lock_name' is not None, configures the IDL to obtain the named
 247         lock from the database server and to avoid modifying the database when
 248         the lock cannot be acquired (that is, when another client has the same
 249         lock).
 250
 251         If 'lock_name' is None, drops the locking requirement and releases the
 252         lock."""
 253         assert not self.txn
 254         assert not self._outstanding_txns
 255
 256         if self.lock_name and (not lock_name or lock_name != self.lock_name):
 257             # Release previous lock.
 258             self.__send_unlock_request()
 259             self.lock_name = None
 260             self.is_lock_contended = False
 261
 262         if lock_name and not self.lock_name:
 263             # Acquire new lock.
 264             self.lock_name = lock_name
 265             self.__send_lock_request()
 266
 267     def __clear(self):
 268         changed = False
 269
 270         for table in self.tables.itervalues():
 271             if table.rows:
 272                 changed = True
 273                 table.rows = {}
 274
 275         if changed:
 276             self.change_seqno += 1
 277
 278     def __update_has_lock(self, new_has_lock):
 279         if new_has_lock and not self.has_lock:
 280             if self._monitor_request_id is None:
 281                 self.change_seqno += 1
 282             else:
 283                 # We're waiting for a monitor reply, so don't signal that the
 284                 # database changed.  The monitor reply will increment
 285                 # change_seqno anyhow.
 286                 pass
 287             self.is_lock_contended = False
 288         self.has_lock = new_has_lock
 289
 290     def __do_send_lock_request(self, method):
 291         self.__update_has_lock(False)
 292         self._lock_request_id = None
 293         if self._session.is_connected():
 294             msg = ovs.jsonrpc.Message.create_request(method, [self.lock_name])
 295             msg_id = msg.id
 296             self._session.send(msg)
 297         else:
 298             msg_id = None
 299         return msg_id
 300
 301     def __send_lock_request(self):
 302         self._lock_request_id = self.__do_send_lock_request("lock")
 303
 304     def __send_unlock_request(self):
 305         self.__do_send_lock_request("unlock")
 306
 307     def __parse_lock_reply(self, result):
 308         self._lock_request_id = None
 309         got_lock = type(result) == dict and result.get("locked") is True
 310         self.__update_has_lock(got_lock)
 311         if not got_lock:
 312             self.is_lock_contended = True
 313
 314     def __parse_lock_notify(self, params, new_has_lock):
 315         if (self.lock_name is not None
 316             and type(params) in (list, tuple)
 317             and params
 318             and params[0] == self.lock_name):
 319             self.__update_has_lock(self, new_has_lock)
 320             if not new_has_lock:
 321                 self.is_lock_contended = True
 322
 323     def __send_monitor_request(self):
 324         monitor_requests = {}
 325         for table in self.tables.itervalues():
 326             monitor_requests[table.name] = {"columns": table.columns.keys()}
 327         msg = ovs.jsonrpc.Message.create_request(
 328             "monitor", [self._db.name, None, monitor_requests])
 329         self._monitor_request_id = msg.id
 330         self._session.send(msg)
 331
 332     def __parse_update(self, update):
 333         try:
 334             self.__do_parse_update(update)
 335         except error.Error, e:
 336             vlog.err("%s: error parsing update: %s"
 337                      % (self._session.get_name(), e))
 338
 339     def __do_parse_update(self, table_updates):
 340         if type(table_updates) != dict:
 341             raise error.Error("<table-updates> is not an object",
 342                               table_updates)
 343
 344         for table_name, table_update in table_updates.iteritems():
 345             table = self.tables.get(table_name)
 346             if not table:
 347                 raise error.Error('<table-updates> includes unknown '
 348                                   'table "%s"' % table_name)
 349
 350             if type(table_update) != dict:
 351                 raise error.Error('<table-update> for table "%s" is not '
 352                                   'an object' % table_name, table_update)
 353
 354             for uuid_string, row_update in table_update.iteritems():
 355                 if not ovs.ovsuuid.is_valid_string(uuid_string):
 356                     raise error.Error('<table-update> for table "%s" '
 357                                       'contains bad UUID "%s" as member '
 358                                       'name' % (table_name, uuid_string),
 359                                       table_update)
 360                 uuid = ovs.ovsuuid.from_string(uuid_string)
 361
 362                 if type(row_update) != dict:
 363                     raise error.Error('<table-update> for table "%s" '
 364                                       'contains <row-update> for %s that '
 365                                       'is not an object'
 366                                       % (table_name, uuid_string))
 367
 368                 parser = ovs.db.parser.Parser(row_update, "row-update")
 369                 old = parser.get_optional("old", [dict])
 370                 new = parser.get_optional("new", [dict])
 371                 parser.finish()
 372
 373                 if not old and not new:
 374                     raise error.Error('<row-update> missing "old" and '
 375                                       '"new" members', row_update)
 376
 377                 if self.__process_update(table, uuid, old, new):
 378                     self.change_seqno += 1
 379
 380     def __process_update(self, table, uuid, old, new):
 381         """Returns True if a column changed, False otherwise."""
 382         row = table.rows.get(uuid)
 383         changed = False
 384         if not new:
 385             # Delete row.
 386             if row:
 387                 del table.rows[uuid]
 388                 changed = True
 389             else:
 390                 # XXX rate-limit
 391                 vlog.warn("cannot delete missing row %s from table %s"
 392                           % (uuid, table.name))
 393         elif not old:
 394             # Insert row.
 395             if not row:
 396                 row = self.__create_row(table, uuid)
 397                 changed = True
 398             else:
 399                 # XXX rate-limit
 400                 vlog.warn("cannot add existing row %s to table %s"
 401                           % (uuid, table.name))
 402             if self.__row_update(table, row, new):
 403                 changed = True
 404         else:
 405             if not row:
 406                 row = self.__create_row(table, uuid)
 407                 changed = True
 408                 # XXX rate-limit
 409                 vlog.warn("cannot modify missing row %s in table %s"
 410                           % (uuid, table.name))
 411             if self.__row_update(table, row, new):
 412                 changed = True
 413         return changed
 414
 415     def __row_update(self, table, row, row_json):
 416         changed = False
 417         for column_name, datum_json in row_json.iteritems():
 418             column = table.columns.get(column_name)
 419             if not column:
 420                 # XXX rate-limit
 421                 vlog.warn("unknown column %s updating table %s"
 422                           % (column_name, table.name))
 423                 continue
 424
 425             try:
 426                 datum = ovs.db.data.Datum.from_json(column.type, datum_json)
 427             except error.Error, e:
 428                 # XXX rate-limit
 429                 vlog.warn("error parsing column %s in table %s: %s"
 430                           % (column_name, table.name, e))
 431                 continue
 432
 433             if datum != row._data[column_name]:
 434                 row._data[column_name] = datum
 435                 if column.alert:
 436                     changed = True
 437             else:
 438                 # Didn't really change but the OVSDB monitor protocol always
 439                 # includes every value in a row.
 440                 pass
 441         return changed
 442
 443     def __create_row(self, table, uuid):
 444         data = {}
 445         for column in table.columns.itervalues():
 446             data[column.name] = ovs.db.data.Datum.default(column.type)
 447         row = table.rows[uuid] = Row(self, table, uuid, data)
 448         return row
 449
 450     def __error(self):
 451         self._session.force_reconnect()
 452
 453     def __txn_abort_all(self):
 454         while self._outstanding_txns:
 455             txn = self._outstanding_txns.popitem()[1]
 456             txn._status = Transaction.TRY_AGAIN
 457
 458     def __txn_process_reply(self, msg):
 459         txn = self._outstanding_txns.pop(msg.id, None)
 460         if txn:
 461             txn._process_reply(msg)
 462
 463
 464 def _uuid_to_row(atom, base):
 465     if base.ref_table:
 466         return base.ref_table.rows.get(atom)
 467     else:
 468         return atom
 469
 470
 471 def _row_to_uuid(value):
 472     if type(value) == Row:
 473         return value.uuid
 474     else:
 475         return value
 476
 477
 478 class Row(object):
 479     """A row within an IDL.
 480
 481     The client may access the following attributes directly:
 482
 483     - 'uuid': a uuid.UUID object whose value is the row's database UUID.
 484
 485     - An attribute for each column in the Row's table, named for the column,
 486       whose values are as returned by Datum.to_python() for the column's type.
 487
 488       If some error occurs (e.g. the database server's idea of the column is
 489       different from the IDL's idea), then the attribute values is the
 490       "default" value return by Datum.default() for the column's type.  (It is
 491       important to know this because the default value may violate constraints
 492       for the column's type, e.g. the default integer value is 0 even if column
 493       contraints require the column's value to be positive.)
 494
 495       When a transaction is active, column attributes may also be assigned new
 496       values.  Committing the transaction will then cause the new value to be
 497       stored into the database.
 498
 499       *NOTE*: In the current implementation, the value of a column is a *copy*
 500       of the value in the database.  This means that modifying its value
 501       directly will have no useful effect.  For example, the following:
 502         row.mycolumn["a"] = "b"              # don't do this
 503       will not change anything in the database, even after commit.  To modify
 504       the column, instead assign the modified column value back to the column:
 505         d = row.mycolumn
 506         d["a"] = "b"
 507         row.mycolumn = d
 508 """
 509     def __init__(self, idl, table, uuid, data):
 510         # All of the explicit references to self.__dict__ below are required
 511         # to set real attributes with invoking self.__getattr__().
 512         self.__dict__["uuid"] = uuid
 513
 514         self.__dict__["_idl"] = idl
 515         self.__dict__["_table"] = table
 516
 517         # _data is the committed data.  It takes the following values:
 518         #
 519         #   - A dictionary that maps every column name to a Datum, if the row
 520         #     exists in the committed form of the database.
 521         #
 522         #   - None, if this row is newly inserted within the active transaction
 523         #     and thus has no committed form.
 524         self.__dict__["_data"] = data
 525
 526         # _changes describes changes to this row within the active transaction.
 527         # It takes the following values:
 528         #
 529         #   - {}, the empty dictionary, if no transaction is active or if the
 530         #     row has yet not been changed within this transaction.
 531         #
 532         #   - A dictionary that maps a column name to its new Datum, if an
 533         #     active transaction changes those columns' values.
 534         #
 535         #   - A dictionary that maps every column name to a Datum, if the row
 536         #     is newly inserted within the active transaction.
 537         #
 538         #   - None, if this transaction deletes this row.
 539         self.__dict__["_changes"] = {}
 540
 541         # A dictionary whose keys are the names of columns that must be
 542         # verified as prerequisites when the transaction commits.  The values
 543         # in the dictionary are all None.
 544         self.__dict__["_prereqs"] = {}
 545
 546     def __getattr__(self, column_name):
 547         assert self._changes is not None
 548
 549         datum = self._changes.get(column_name)
 550         if datum is None:
 551             if self._data is None:
 552                 raise AttributeError("%s instance has no attribute '%s'" %
 553                                      (self.__class__.__name__, column_name))
 554             datum = self._data[column_name]
 555
 556         return datum.to_python(_uuid_to_row)
 557
 558     def __setattr__(self, column_name, value):
 559         assert self._changes is not None
 560         assert self._idl.txn
 561
 562         column = self._table.columns[column_name]
 563         try:
 564             datum = ovs.db.data.Datum.from_python(column.type, value,
 565                                                   _row_to_uuid)
 566         except error.Error, e:
 567             # XXX rate-limit
 568             vlog.err("attempting to write bad value to column %s (%s)"
 569                      % (column_name, e))
 570             return
 571         self._idl.txn._write(self, column, datum)
 572
 573     def verify(self, column_name):
 574         """Causes the original contents of column 'column_name' in this row to
 575         be verified as a prerequisite to completing the transaction.  That is,
 576         if 'column_name' changed in this row (or if this row was deleted)
 577         between the time that the IDL originally read its contents and the time
 578         that the transaction commits, then the transaction aborts and
 579         Transaction.commit() returns Transaction.TRY_AGAIN.
 580
 581         The intention is that, to ensure that no transaction commits based on
 582         dirty reads, an application should call Row.verify() on each data item
 583         read as part of a read-modify-write operation.
 584
 585         In some cases Row.verify() reduces to a no-op, because the current
 586         value of the column is already known:
 587
 588           - If this row is a row created by the current transaction (returned
 589             by Transaction.insert()).
 590
 591           - If the column has already been modified within the current
 592             transaction.
 593
 594         Because of the latter property, always call Row.verify() *before*
 595         modifying the column, for a given read-modify-write.
 596
 597         A transaction must be in progress."""
 598         assert self._idl.txn
 599         assert self._changes is not None
 600         if not self._data or column_name in self._changes:
 601             return
 602
 603         self._prereqs[column_name] = None
 604
 605     def delete(self):
 606         """Deletes this row from its table.
 607
 608         A transaction must be in progress."""
 609         assert self._idl.txn
 610         assert self._changes is not None
 611         if self._data is None:
 612             del self._idl.txn._txn_rows[self.uuid]
 613         self.__dict__["_changes"] = None
 614         del self._table.rows[self.uuid]
 615
 616     def increment(self, column_name):
 617         """Causes the transaction, when committed, to increment the value of
 618         'column_name' within this row by 1.  'column_name' must have an integer
 619         type.  After the transaction commits successfully, the client may
 620         retrieve the final (incremented) value of 'column_name' with
 621         Transaction.get_increment_new_value().
 622
 623         The client could accomplish something similar by reading and writing
 624         and verify()ing columns.  However, increment() will never (by itself)
 625         cause a transaction to fail because of a verify error.
 626
 627         The intended use is for incrementing the "next_cfg" column in
 628         the Open_vSwitch table."""
 629         self._idl.txn._increment(self, column_name)
 630
 631
 632 def _uuid_name_from_uuid(uuid):
 633     return "row%s" % str(uuid).replace("-", "_")
 634
 635
 636 def _where_uuid_equals(uuid):
 637     return [["_uuid", "==", ["uuid", str(uuid)]]]
 638
 639
 640 class _InsertedRow(object):
 641     def __init__(self, op_index):
 642         self.op_index = op_index
 643         self.real = None
 644
 645
 646 class Transaction(object):
 647     """A transaction may modify the contents of a database by modifying the
 648     values of columns, deleting rows, inserting rows, or adding checks that
 649     columns in the database have not changed ("verify" operations), through
 650     Row methods.
 651
 652     Reading and writing columns and inserting and deleting rows are all
 653     straightforward.  The reasons to verify columns are less obvious.
 654     Verification is the key to maintaining transactional integrity.  Because
 655     OVSDB handles multiple clients, it can happen that between the time that
 656     OVSDB client A reads a column and writes a new value, OVSDB client B has
 657     written that column.  Client A's write should not ordinarily overwrite
 658     client B's, especially if the column in question is a "map" column that
 659     contains several more or less independent data items.  If client A adds a
 660     "verify" operation before it writes the column, then the transaction fails
 661     in case client B modifies it first.  Client A will then see the new value
 662     of the column and compose a new transaction based on the new contents
 663     written by client B.
 664
 665     When a transaction is complete, which must be before the next call to
 666     Idl.run(), call Transaction.commit() or Transaction.abort().
 667
 668     The life-cycle of a transaction looks like this:
 669
 670     1. Create the transaction and record the initial sequence number:
 671
 672         seqno = idl.change_seqno(idl)
 673         txn = Transaction(idl)
 674
 675     2. Modify the database with Row and Transaction methods.
 676
 677     3. Commit the transaction by calling Transaction.commit().  The first call
 678        to this function probably returns Transaction.INCOMPLETE.  The client
 679        must keep calling again along as this remains true, calling Idl.run() in
 680        between to let the IDL do protocol processing.  (If the client doesn't
 681        have anything else to do in the meantime, it can use
 682        Transaction.commit_block() to avoid having to loop itself.)
 683
 684     4. If the final status is Transaction.TRY_AGAIN, wait for Idl.change_seqno
 685        to change from the saved 'seqno' (it's possible that it's already
 686        changed, in which case the client should not wait at all), then start
 687        over from step 1.  Only a call to Idl.run() will change the return value
 688        of Idl.change_seqno.  (Transaction.commit_block() calls Idl.run().)"""
 689
 690     # Status values that Transaction.commit() can return.
 691     UNCOMMITTED = "uncommitted"  # Not yet committed or aborted.
 692     UNCHANGED = "unchanged"      # Transaction didn't include any changes.
 693     INCOMPLETE = "incomplete"    # Commit in progress, please wait.
 694     ABORTED = "aborted"          # ovsdb_idl_txn_abort() called.
 695     SUCCESS = "success"          # Commit successful.
 696     TRY_AGAIN = "try again"      # Commit failed because a "verify" operation
 697                                  # reported an inconsistency, due to a network
 698                                  # problem, or other transient failure.  Wait
 699                                  # for a change, then try again.
 700     NOT_LOCKED = "not locked"    # Server hasn't given us the lock yet.
 701     ERROR = "error"              # Commit failed due to a hard error.
 702
 703     @staticmethod
 704     def status_to_string(status):
 705         """Converts one of the status values that Transaction.commit() can
 706         return into a human-readable string.
 707
 708         (The status values are in fact such strings already, so
 709         there's nothing to do.)"""
 710         return status
 711
 712     def __init__(self, idl):
 713         """Starts a new transaction on 'idl' (an instance of ovs.db.idl.Idl).
 714         A given Idl may only have a single active transaction at a time.
 715
 716         A Transaction may modify the contents of a database by assigning new
 717         values to columns (attributes of Row), deleting rows (with
 718         Row.delete()), or inserting rows (with Transaction.insert()).  It may
 719         also check that columns in the database have not changed with
 720         Row.verify().
 721
 722         When a transaction is complete (which must be before the next call to
 723         Idl.run()), call Transaction.commit() or Transaction.abort()."""
 724         assert idl.txn is None
 725
 726         idl.txn = self
 727         self._request_id = None
 728         self.idl = idl
 729         self.dry_run = False
 730         self._txn_rows = {}
 731         self._status = Transaction.UNCOMMITTED
 732         self._error = None
 733         self._comments = []
 734         self._commit_seqno = self.idl.change_seqno
 735
 736         self._inc_row = None
 737         self._inc_column = None
 738
 739         self._inserted_rows = {}  # Map from UUID to _InsertedRow
 740
 741     def add_comment(self, comment):
 742         """Appens 'comment' to the comments that will be passed to the OVSDB
 743         server when this transaction is committed.  (The comment will be
 744         committed to the OVSDB log, which "ovsdb-tool show-log" can print in a
 745         relatively human-readable form.)"""
 746         self._comments.append(comment)
 747
 748     def wait(self, poller):
 749         """Causes poll_block() to wake up if this transaction has completed
 750         committing."""
 751         if self._status not in (Transaction.UNCOMMITTED,
 752                                 Transaction.INCOMPLETE):
 753             poller.immediate_wake()
 754
 755     def _substitute_uuids(self, json):
 756         if type(json) in (list, tuple):
 757             if (len(json) == 2
 758                 and json[0] == 'uuid'
 759                 and ovs.ovsuuid.is_valid_string(json[1])):
 760                 uuid = ovs.ovsuuid.from_string(json[1])
 761                 row = self._txn_rows.get(uuid, None)
 762                 if row and row._data is None:
 763                     return ["named-uuid", _uuid_name_from_uuid(uuid)]
 764             else:
 765                 return [self._substitute_uuids(elem) for elem in json]
 766         return json
 767
 768     def __disassemble(self):
 769         self.idl.txn = None
 770
 771         for row in self._txn_rows.itervalues():
 772             if row._changes is None:
 773                 row._table.rows[row.uuid] = row
 774             elif row._data is None:
 775                 del row._table.rows[row.uuid]
 776             row.__dict__["_changes"] = {}
 777             row.__dict__["_prereqs"] = {}
 778         self._txn_rows = {}
 779
 780     def commit(self):
 781         """Attempts to commit 'txn'.  Returns the status of the commit
 782         operation, one of the following constants:
 783
 784           Transaction.INCOMPLETE:
 785
 786               The transaction is in progress, but not yet complete.  The caller
 787               should call again later, after calling Idl.run() to let the
 788               IDL do OVSDB protocol processing.
 789
 790           Transaction.UNCHANGED:
 791
 792               The transaction is complete.  (It didn't actually change the
 793               database, so the IDL didn't send any request to the database
 794               server.)
 795
 796           Transaction.ABORTED:
 797
 798               The caller previously called Transaction.abort().
 799
 800           Transaction.SUCCESS:
 801
 802               The transaction was successful.  The update made by the
 803               transaction (and possibly other changes made by other database
 804               clients) should already be visible in the IDL.
 805
 806           Transaction.TRY_AGAIN:
 807
 808               The transaction failed for some transient reason, e.g. because a
 809               "verify" operation reported an inconsistency or due to a network
 810               problem.  The caller should wait for a change to the database,
 811               then compose a new transaction, and commit the new transaction.
 812
 813               Use Idl.change_seqno to wait for a change in the database.  It is
 814               important to use its value *before* the initial call to
 815               Transaction.commit() as the baseline for this purpose, because
 816               the change that one should wait for can happen after the initial
 817               call but before the call that returns Transaction.TRY_AGAIN, and
 818               using some other baseline value in that situation could cause an
 819               indefinite wait if the database rarely changes.
 820
 821           Transaction.NOT_LOCKED:
 822
 823               The transaction failed because the IDL has been configured to
 824               require a database lock (with Idl.set_lock()) but didn't
 825               get it yet or has already lost it.
 826
 827         Committing a transaction rolls back all of the changes that it made to
 828         the IDL's copy of the database.  If the transaction commits
 829         successfully, then the database server will send an update and, thus,
 830         the IDL will be updated with the committed changes."""
 831         # The status can only change if we're the active transaction.
 832         # (Otherwise, our status will change only in Idl.run().)
 833         if self != self.idl.txn:
 834             return self._status
 835
 836         # If we need a lock but don't have it, give up quickly.
 837         if self.idl.lock_name and not self.idl.has_lock():
 838             self._status = Transaction.NOT_LOCKED
 839             self.__disassemble()
 840             return self._status
 841
 842         operations = [self.idl._db.name]
 843
 844         # Assert that we have the required lock (avoiding a race).
 845         if self.idl.lock_name:
 846             operations.append({"op": "assert",
 847                                "lock": self.idl.lock_name})
 848
 849         # Add prerequisites and declarations of new rows.
 850         for row in self._txn_rows.itervalues():
 851             if row._prereqs:
 852                 rows = {}
 853                 columns = []
 854                 for column_name in row._prereqs:
 855                     columns.append(column_name)
 856                     rows[column_name] = row._data[column_name].to_json()
 857                 operations.append({"op": "wait",
 858                                    "table": row._table.name,
 859                                    "timeout": 0,
 860                                    "where": _where_uuid_equals(row.uuid),
 861                                    "until": "==",
 862                                    "columns": columns,
 863                                    "rows": [rows]})
 864
 865         # Add updates.
 866         any_updates = False
 867         for row in self._txn_rows.itervalues():
 868             if row._changes is None:
 869                 if row._table.is_root:
 870                     operations.append({"op": "delete",
 871                                        "table": row._table.name,
 872                                        "where": _where_uuid_equals(row.uuid)})
 873                     any_updates = True
 874                 else:
 875                     # Let ovsdb-server decide whether to really delete it.
 876                     pass
 877             elif row._changes:
 878                 op = {"table": row._table.name}
 879                 if row._data is None:
 880                     op["op"] = "insert"
 881                     op["uuid-name"] = _uuid_name_from_uuid(row.uuid)
 882                     any_updates = True
 883
 884                     op_index = len(operations) - 1
 885                     self._inserted_rows[row.uuid] = _InsertedRow(op_index)
 886                 else:
 887                     op["op"] = "update"
 888                     op["where"] = _where_uuid_equals(row.uuid)
 889
 890                 row_json = {}
 891                 op["row"] = row_json
 892
 893                 for column_name, datum in row._changes.iteritems():
 894                     if row._data is not None or not datum.is_default():
 895                         row_json[column_name] = (
 896                                 self._substitute_uuids(datum.to_json()))
 897
 898                         # If anything really changed, consider it an update.
 899                         # We can't suppress not-really-changed values earlier
 900                         # or transactions would become nonatomic (see the big
 901                         # comment inside Transaction._write()).
 902                         if (not any_updates and row._data is not None and
 903                             row._data[column_name] != datum):
 904                             any_updates = True
 905
 906                 if row._data is None or row_json:
 907                     operations.append(op)
 908
 909         # Add increment.
 910         if self._inc_row and any_updates:
 911             self._inc_index = len(operations) - 1
 912
 913             operations.append({"op": "mutate",
 914                                "table": self._inc_row._table.name,
 915                                "where": self._substitute_uuids(
 916                                    _where_uuid_equals(self._inc_row.uuid)),
 917                                "mutations": [[self._inc_column, "+=", 1]]})
 918             operations.append({"op": "select",
 919                                "table": self._inc_row._table.name,
 920                                "where": self._substitute_uuids(
 921                                    _where_uuid_equals(self._inc_row.uuid)),
 922                                "columns": [self._inc_column]})
 923
 924         # Add comment.
 925         if self._comments:
 926             operations.append({"op": "comment",
 927                                "comment": "\n".join(self._comments)})
 928
 929         # Dry run?
 930         if self.dry_run:
 931             operations.append({"op": "abort"})
 932
 933         if not any_updates:
 934             self._status = Transaction.UNCHANGED
 935         else:
 936             msg = ovs.jsonrpc.Message.create_request("transact", operations)
 937             self._request_id = msg.id
 938             if not self.idl._session.send(msg):
 939                 self.idl._outstanding_txns[self._request_id] = self
 940                 self._status = Transaction.INCOMPLETE
 941             else:
 942                 self._status = Transaction.TRY_AGAIN
 943
 944         self.__disassemble()
 945         return self._status
 946
 947     def commit_block(self):
 948         """Attempts to commit this transaction, blocking until the commit
 949         either succeeds or fails.  Returns the final commit status, which may
 950         be any Transaction.* value other than Transaction.INCOMPLETE.
 951
 952         This function calls Idl.run() on this transaction'ss IDL, so it may
 953         cause Idl.change_seqno to change."""
 954         while True:
 955             status = self.commit()
 956             if status != Transaction.INCOMPLETE:
 957                 return status
 958
 959             self.idl.run()
 960
 961             poller = ovs.poller.Poller()
 962             self.idl.wait(poller)
 963             self.wait(poller)
 964             poller.block()
 965
 966     def get_increment_new_value(self):
 967         """Returns the final (incremented) value of the column in this
 968         transaction that was set to be incremented by Row.increment.  This
 969         transaction must have committed successfully."""
 970         assert self._status == Transaction.SUCCESS
 971         return self._inc_new_value
 972
 973     def abort(self):
 974         """Aborts this transaction.  If Transaction.commit() has already been
 975         called then the transaction might get committed anyhow."""
 976         self.__disassemble()
 977         if self._status in (Transaction.UNCOMMITTED,
 978                             Transaction.INCOMPLETE):
 979             self._status = Transaction.ABORTED
 980
 981     def get_error(self):
 982         """Returns a string representing this transaction's current status,
 983         suitable for use in log messages."""
 984         if self._status != Transaction.ERROR:
 985             return Transaction.status_to_string(self._status)
 986         elif self._error:
 987             return self._error
 988         else:
 989             return "no error details available"
 990
 991     def __set_error_json(self, json):
 992         if self._error is None:
 993             self._error = ovs.json.to_string(json)
 994
 995     def get_insert_uuid(self, uuid):
 996         """Finds and returns the permanent UUID that the database assigned to a
 997         newly inserted row, given the UUID that Transaction.insert() assigned
 998         locally to that row.
 999
1000         Returns None if 'uuid' is not a UUID assigned by Transaction.insert()
1001         or if it was assigned by that function and then deleted by Row.delete()
1002         within the same transaction.  (Rows that are inserted and then deleted
1003         within a single transaction are never sent to the database server, so
1004         it never assigns them a permanent UUID.)
1005
1006         This transaction must have completed successfully."""
1007         assert self._status in (Transaction.SUCCESS,
1008                                 Transaction.UNCHANGED)
1009         inserted_row = self._inserted_rows.get(uuid)
1010         if inserted_row:
1011             return inserted_row.real
1012         return None
1013
1014     def _increment(self, row, column):
1015         assert not self._inc_row
1016         self._inc_row = row
1017         self._inc_column = column
1018
1019     def _write(self, row, column, datum):
1020         assert row._changes is not None
1021
1022         txn = row._idl.txn
1023
1024         # If this is a write-only column and the datum being written is the
1025         # same as the one already there, just skip the update entirely.  This
1026         # is worth optimizing because we have a lot of columns that get
1027         # periodically refreshed into the database but don't actually change
1028         # that often.
1029         #
1030         # We don't do this for read/write columns because that would break
1031         # atomicity of transactions--some other client might have written a
1032         # different value in that column since we read it.  (But if a whole
1033         # transaction only does writes of existing values, without making any
1034         # real changes, we will drop the whole transaction later in
1035         # ovsdb_idl_txn_commit().)
1036         if not column.alert and row._data.get(column.name) == datum:
1037             new_value = row._changes.get(column.name)
1038             if new_value is None or new_value == datum:
1039                 return
1040
1041         txn._txn_rows[row.uuid] = row
1042         row._changes[column.name] = datum.copy()
1043
1044     def insert(self, table, new_uuid=None):
1045         """Inserts and returns a new row in 'table', which must be one of the
1046         ovs.db.schema.TableSchema objects in the Idl's 'tables' dict.
1047
1048         The new row is assigned a provisional UUID.  If 'uuid' is None then one
1049         is randomly generated; otherwise 'uuid' should specify a randomly
1050         generated uuid.UUID not otherwise in use.  ovsdb-server will assign a
1051         different UUID when 'txn' is committed, but the IDL will replace any
1052         uses of the provisional UUID in the data to be to be committed by the
1053         UUID assigned by ovsdb-server."""
1054         assert self._status == Transaction.UNCOMMITTED
1055         if new_uuid is None:
1056             new_uuid = uuid.uuid4()
1057         row = Row(self.idl, table, new_uuid, None)
1058         table.rows[row.uuid] = row
1059         self._txn_rows[row.uuid] = row
1060         return row
1061
1062     def _process_reply(self, msg):
1063         if msg.type == ovs.jsonrpc.Message.T_ERROR:
1064             self._status = Transaction.ERROR
1065         elif type(msg.result) not in (list, tuple):
1066             # XXX rate-limit
1067             vlog.warn('reply to "transact" is not JSON array')
1068         else:
1069             hard_errors = False
1070             soft_errors = False
1071             lock_errors = False
1072
1073             ops = msg.result
1074             for op in ops:
1075                 if op is None:
1076                     # This isn't an error in itself but indicates that some
1077                     # prior operation failed, so make sure that we know about
1078                     # it.
1079                     soft_errors = True
1080                 elif type(op) == dict:
1081                     error = op.get("error")
1082                     if error is not None:
1083                         if error == "timed out":
1084                             soft_errors = True
1085                         elif error == "not owner":
1086                             lock_errors = True
1087                         elif error == "aborted":
1088                             pass
1089                         else:
1090                             hard_errors = True
1091                             self.__set_error_json(op)
1092                 else:
1093                     hard_errors = True
1094                     self.__set_error_json(op)
1095                     # XXX rate-limit
1096                     vlog.warn("operation reply is not JSON null or object")
1097
1098             if not soft_errors and not hard_errors and not lock_errors:
1099                 if self._inc_row and not self.__process_inc_reply(ops):
1100                     hard_errors = True
1101
1102                 for insert in self._inserted_rows.itervalues():
1103                     if not self.__process_insert_reply(insert, ops):
1104                         hard_errors = True
1105
1106             if hard_errors:
1107                 self._status = Transaction.ERROR
1108             elif lock_errors:
1109                 self._status = Transaction.NOT_LOCKED
1110             elif soft_errors:
1111                 self._status = Transaction.TRY_AGAIN
1112             else:
1113                 self._status = Transaction.SUCCESS
1114
1115     @staticmethod
1116     def __check_json_type(json, types, name):
1117         if not json:
1118             # XXX rate-limit
1119             vlog.warn("%s is missing" % name)
1120             return False
1121         elif type(json) not in types:
1122             # XXX rate-limit
1123             vlog.warn("%s has unexpected type %s" % (name, type(json)))
1124             return False
1125         else:
1126             return True
1127
1128     def __process_inc_reply(self, ops):
1129         if self._inc_index + 2 > len(ops):
1130             # XXX rate-limit
1131             vlog.warn("reply does not contain enough operations for "
1132                       "increment (has %d, needs %d)" %
1133                       (len(ops), self._inc_index + 2))
1134
1135         # We know that this is a JSON object because the loop in
1136         # __process_reply() already checked.
1137         mutate = ops[self._inc_index]
1138         count = mutate.get("count")
1139         if not Transaction.__check_json_type(count, (int, long),
1140                                              '"mutate" reply "count"'):
1141             return False
1142         if count != 1:
1143             # XXX rate-limit
1144             vlog.warn('"mutate" reply "count" is %d instead of 1' % count)
1145             return False
1146
1147         select = ops[self._inc_index + 1]
1148         rows = select.get("rows")
1149         if not Transaction.__check_json_type(rows, (list, tuple),
1150                                              '"select" reply "rows"'):
1151             return False
1152         if len(rows) != 1:
1153             # XXX rate-limit
1154             vlog.warn('"select" reply "rows" has %d elements '
1155                       'instead of 1' % len(rows))
1156             return False
1157         row = rows[0]
1158         if not Transaction.__check_json_type(row, (dict,),
1159                                              '"select" reply row'):
1160             return False
1161         column = row.get(self._inc_column)
1162         if not Transaction.__check_json_type(column, (int, long),
1163                                              '"select" reply inc column'):
1164             return False
1165         self._inc_new_value = column
1166         return True
1167
1168     def __process_insert_reply(self, insert, ops):
1169         if insert.op_index >= len(ops):
1170             # XXX rate-limit
1171             vlog.warn("reply does not contain enough operations "
1172                       "for insert (has %d, needs %d)"
1173                       % (len(ops), insert.op_index))
1174             return False
1175
1176         # We know that this is a JSON object because the loop in
1177         # __process_reply() already checked.
1178         reply = ops[insert.op_index]
1179         json_uuid = reply.get("uuid")
1180         if not Transaction.__check_json_type(json_uuid, (tuple, list),
1181                                              '"insert" reply "uuid"'):
1182             return False
1183
1184         try:
1185             uuid_ = ovs.ovsuuid.from_json(json_uuid)
1186         except error.Error:
1187             # XXX rate-limit
1188             vlog.warn('"insert" reply "uuid" is not a JSON UUID')
1189             return False
1190
1191         insert.real = uuid_
1192         return True
1193
1194
1195 class SchemaHelper(object):
1196     """IDL Schema helper.
1197
1198     This class encapsulates the logic required to generate schemas suitable
1199     for creating 'ovs.db.idl.Idl' objects.  Clients should register columns
1200     they are interested in using register_columns().  When finished, the
1201     get_idl_schema() function may be called.
1202
1203     The location on disk of the schema used may be found in the
1204     'schema_location' variable."""
1205
1206     def __init__(self, location=None, schema_json=None):
1207         """Creates a new Schema object.
1208
1209         'location' file path to ovs schema. None means default location
1210         'schema_json' schema in json preresentation in memory
1211         """
1212
1213         if location and schema_json:
1214             raise ValueError("both location and schema_json can't be "
1215                              "specified. it's ambiguous.")
1216         if schema_json is None:
1217             if location is None:
1218                 location = "%s/vswitch.ovsschema" % ovs.dirs.PKGDATADIR
1219             schema_json = ovs.json.from_file(location)
1220
1221         self.schema_json = schema_json
1222         self._tables = {}
1223         self._all = False
1224
1225     def register_columns(self, table, columns):
1226         """Registers interest in the given 'columns' of 'table'.  Future calls
1227         to get_idl_schema() will include 'table':column for each column in
1228         'columns'. This function automatically avoids adding duplicate entries
1229         to the schema.
1230
1231         'table' must be a string.
1232         'columns' must be a list of strings.
1233         """
1234
1235         assert type(table) is str
1236         assert type(columns) is list
1237
1238         columns = set(columns) | self._tables.get(table, set())
1239         self._tables[table] = columns
1240
1241     def register_table(self, table):
1242         """Registers interest in the given all columns of 'table'. Future calls
1243         to get_idl_schema() will include all columns of 'table'.
1244
1245         'table' must be a string
1246         """
1247         assert type(table) is str
1248         self._tables[table] = set()  # empty set means all columns in the table
1249
1250     def register_all(self):
1251         """Registers interest in every column of every table."""
1252         self._all = True
1253
1254     def get_idl_schema(self):
1255         """Gets a schema appropriate for the creation of an 'ovs.db.id.IDL'
1256         object based on columns registered using the register_columns()
1257         function."""
1258
1259         schema = ovs.db.schema.DbSchema.from_json(self.schema_json)
1260         self.schema_json = None
1261
1262         if not self._all:
1263             schema_tables = {}
1264             for table, columns in self._tables.iteritems():
1265                 schema_tables[table] = (
1266                     self._keep_table_columns(schema, table, columns))
1267
1268             schema.tables = schema_tables
1269         return schema
1270
1271     def _keep_table_columns(self, schema, table_name, columns):
1272         assert table_name in schema.tables
1273         table = schema.tables[table_name]
1274
1275         if not columns:
1276             # empty set means all columns in the table
1277             return table
1278
1279         new_columns = {}
1280         for column_name in columns:
1281             assert type(column_name) is str
1282             assert column_name in table.columns
1283
1284             new_columns[column_name] = table.columns[column_name]
1285
1286         table.columns = new_columns
1287         return table