dbwrap_ctdb: handle read-only records.

The new read-only record flags make determining if we can use a record
a bit more complex, so extract it into its own function.

The OLD logic was:
1) If the record doesn't exist, we can't use it.
2) If we are the dmaster for the record, we can use it.

The new logic is:
1) If the record doesn't exist, we can't use it.
2) If we are the dmaster for the record, we can use it IF we only
   want read-only access, OR there are no read-only delegations.
3) If we are not dmaster, we can only use it if we want read-only
   access and it is marked as a read-only copy.

This logic is unused until the next patches which begin to ask
for read-only copies of records.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

diff --git a/source3/lib/dbwrap/dbwrap_ctdb.c b/source3/lib/dbwrap/dbwrap_ctdb.c
index 3c1ab44d4e0351e6c08ab5449496b07f179be3ec..0f4efa359cca7d26e393ccc464aada28e08ef2c3 100644 (file)
--- a/source3/lib/dbwrap/dbwrap_ctdb.c
+++ b/source3/lib/dbwrap/dbwrap_ctdb.c
@@ -1001,6 +1001,27 @@ static int db_ctdb_record_destr(struct db_record* data)
        return 0;
 }
 
+/* Do I own this record? */
+static bool db_ctdb_own_record(TDB_DATA ctdb_data, bool read_only)
+{
+       struct ctdb_ltdb_header *hdr;
+
+       if (ctdb_data.dptr == NULL)
+               return false;
+
+       if (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header))
+               return false;
+
+       hdr = (struct ctdb_ltdb_header *)ctdb_data.dptr;
+       if (hdr->dmaster != get_my_vnn()) {
+               /* If we're not dmaster, it must be r/o copy. */
+               return read_only && (hdr->flags & CTDB_REC_RO_HAVE_READONLY);
+       }
+
+       /* If we want write access, noone can have r/o copies. */
+       return read_only || !(hdr->flags & CTDB_REC_RO_HAVE_DELEGATIONS);
+}
+
 static struct db_record *fetch_locked_internal(struct db_ctdb_ctx *ctx,
                                               TALLOC_CTX *mem_ctx,
                                               TDB_DATA key)
@@ -1064,9 +1085,7 @@ again:
         * take the shortcut and just return it.
         */
 
-       if ((ctdb_data.dptr == NULL) ||
-           (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header)) ||
-           ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster != get_my_vnn()
+       if (!db_ctdb_own_record(ctdb_data, false)
 #if 0
            || (random() % 2 != 0)
 #endif
@@ -1077,10 +1096,11 @@ again:
 
                migrate_attempts += 1;
 
-               DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %u (%u)\n",
+               DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %u (%u) %u\n",
                           ctdb_data.dptr, ctdb_data.dptr ?
                           ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster : -1,
-                          get_my_vnn()));
+                          get_my_vnn(),
+                          ((struct ctdb_ltdb_header *)ctdb_data.dptr)->flags));
 
                status = ctdbd_migrate(messaging_ctdbd_connection(), ctx->db_id,
                                       key);
@@ -1164,10 +1184,7 @@ static NTSTATUS db_ctdb_fetch(struct db_context *db, TALLOC_CTX *mem_ctx,
         * take the shortcut and just return it.
         * we bypass the dmaster check for persistent databases
         */
-       if ((ctdb_data.dptr != NULL) &&
-           (ctdb_data.dsize >= sizeof(struct ctdb_ltdb_header)) &&
-           ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster == get_my_vnn())
-       {
+       if (db_ctdb_own_record(ctdb_data, true)) {
                /* we are the dmaster - avoid the ctdb protocol op */
 
                data->dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);