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 34ebe6acbbc5e1ab29e0dbdad44a99e52bbd5c30..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 |
@@ -78,6 +84,61 @@ struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *na
        return NULL;
 }
 
+bool ctdb_db_persistent(struct ctdb_db_context *ctdb_db)
+{
+       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)
+{
+       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)
+{
+       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->db_flags |= CTDB_DB_FLAGS_READONLY;
+}
+
+void ctdb_db_reset_readonly(struct ctdb_db_context *ctdb_db)
+{
+       ctdb_db->db_flags &= ~CTDB_DB_FLAGS_READONLY;
+}
+
+bool ctdb_db_sticky(struct ctdb_db_context *ctdb_db)
+{
+       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->db_flags |= CTDB_DB_FLAGS_STICKY;
+}
 
 /*
   return the lmaster given a key
@@ -133,7 +194,8 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
                if (data) {
                        *data = tdb_null;
                }
-               if (ctdb_db->persistent || header->dmaster == ctdb_db->ctdb->pnn) {
+               if (ctdb_db_persistent(ctdb_db) ||
+                   header->dmaster == ctdb_db->ctdb->pnn) {
                        if (ctdb_ltdb_store(ctdb_db, key, header, tdb_null) != 0) {
                                DEBUG(DEBUG_NOTICE,
                                      (__location__ "failed to store initial header\n"));
@@ -159,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
 */
@@ -204,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);
@@ -233,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;
 }
@@ -285,8 +293,10 @@ int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 */
 int ctdb_ltdb_delete(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 {
-       if (ctdb_db->persistent != 0) {
-               DEBUG(DEBUG_ERR,("Trying to delete empty record in persistent database\n"));
+       if (! ctdb_db_volatile(ctdb_db)) {
+               DEBUG(DEBUG_WARNING,
+                     ("Ignored deletion of empty record from "
+                      "non-volatile database\n"));
                return 0;
        }
        if (tdb_delete(ctdb_db->ltdb->tdb, key) != 0) {