ctdb: Make TDB_SEQNUM work synchronously with ctdb

[samba.git] / ctdb / common / ctdb_ltdb.c

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index aa1db9de6eaf2915662148cd63471ab993ca321a..1fc9ce28c1b436e259ab4a1b53ad258d45922923 100644 (file)
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -43,6 +43,12 @@ int ctdb_db_tdb_flags(uint8_t db_flags, bool with_valgrind, bool with_mutex)
 
        if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
                tdb_flags = TDB_DEFAULT;
+
+       } else if (db_flags & CTDB_DB_FLAGS_REPLICATED) {
+               tdb_flags = TDB_NOSYNC |
+                           TDB_CLEAR_IF_FIRST |
+                           TDB_INCOMPATIBLE_HASH;
+
        } else {
                tdb_flags = TDB_NOSYNC |
                            TDB_CLEAR_IF_FIRST |
@@ -80,37 +86,58 @@ struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *na
 
 bool ctdb_db_persistent(struct ctdb_db_context *ctdb_db)
 {
-       return ctdb_db->persistent;
+       if (ctdb_db->db_flags & CTDB_DB_FLAGS_PERSISTENT) {
+               return true;
+       }
+       return false;
+}
+
+bool ctdb_db_replicated(struct ctdb_db_context *ctdb_db)
+{
+       if (ctdb_db->db_flags & CTDB_DB_FLAGS_REPLICATED) {
+               return true;
+       }
+       return false;
 }
 
 bool ctdb_db_volatile(struct ctdb_db_context *ctdb_db)
 {
-       return !ctdb_db->persistent;
+       if ((ctdb_db->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
+           (ctdb_db->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
+               return false;
+       }
+       return true;
 }
 
 bool ctdb_db_readonly(struct ctdb_db_context *ctdb_db)
 {
-       return ctdb_db->readonly;
+       if (ctdb_db->db_flags & CTDB_DB_FLAGS_READONLY) {
+               return true;
+       }
+       return false;
 }
 
 void ctdb_db_set_readonly(struct ctdb_db_context *ctdb_db)
 {
-       ctdb_db->readonly = true;
+       ctdb_db->db_flags |= CTDB_DB_FLAGS_READONLY;
 }
 
 void ctdb_db_reset_readonly(struct ctdb_db_context *ctdb_db)
 {
-       ctdb_db->readonly = false;
+       ctdb_db->db_flags &= ~CTDB_DB_FLAGS_READONLY;
 }
 
 bool ctdb_db_sticky(struct ctdb_db_context *ctdb_db)
 {
-       return ctdb_db->sticky;
+       if (ctdb_db->db_flags & CTDB_DB_FLAGS_STICKY) {
+               return true;
+       }
+       return false;
 }
 
 void ctdb_db_set_sticky(struct ctdb_db_context *ctdb_db)
 {
-       ctdb_db->sticky = true;
+       ctdb_db->db_flags |= CTDB_DB_FLAGS_STICKY;
 }
 
 /*
@@ -194,41 +221,6 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
        return 0;
 }
 
-/*
-  fetch a record from the ltdb, separating out the header information
-  and returning the body of the record.
-  if the record does not exist, *header will be NULL
-  and data = {0, NULL}
-*/
-int ctdb_ltdb_fetch_with_header(struct ctdb_db_context *ctdb_db, 
-                   TDB_DATA key, struct ctdb_ltdb_header *header, 
-                   TALLOC_CTX *mem_ctx, TDB_DATA *data)
-{
-       TDB_DATA rec;
-
-       rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
-       if (rec.dsize < sizeof(*header)) {
-               free(rec.dptr);
-
-               data->dsize = 0;
-               data->dptr = NULL;
-               return -1;
-       }
-
-       *header = *(struct ctdb_ltdb_header *)rec.dptr;
-       if (data) {
-               data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
-               data->dptr = talloc_memdup(mem_ctx, 
-                                          sizeof(struct ctdb_ltdb_header)+rec.dptr,
-                                          data->dsize);
-       }
-
-       free(rec.dptr);
-
-       return 0;
-}
-
-
 /*
   write a record to a normal database
 */
@@ -239,7 +231,6 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
        TDB_DATA rec[2];
        uint32_t hsize = sizeof(struct ctdb_ltdb_header);
        int ret;
-       bool seqnum_suppressed = false;
 
        if (ctdb_db->ctdb_ltdb_store_fn) {
                return ctdb_db->ctdb_ltdb_store_fn(ctdb_db, key, header, data);
@@ -268,28 +259,10 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
        rec[1].dsize = data.dsize;
        rec[1].dptr = data.dptr;
 
-       /* Databases with seqnum updates enabled only get their seqnum
-          changes when/if we modify the data */
-       if (ctdb_db->seqnum_update != NULL) {
-               TDB_DATA old;
-               old = tdb_fetch(ctdb_db->ltdb->tdb, key);
-
-               if ((old.dsize == hsize + data.dsize) &&
-                   memcmp(old.dptr+hsize, data.dptr, data.dsize) == 0) {
-                       tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
-                       seqnum_suppressed = true;
-               }
-               if (old.dptr != NULL) {
-                       free(old.dptr);
-               }
-       }
        ret = tdb_storev(ctdb_db->ltdb->tdb, key, rec, 2, TDB_REPLACE);
        if (ret != 0) {
                DEBUG(DEBUG_ERR, (__location__ " Failed to store dynamic data\n"));
        }
-       if (seqnum_suppressed) {
-               tdb_add_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
-       }
 
        return ret;
 }