ctdb-daemon: Add accessors for CTDB_DB_FLAGS_PERSISTENT flag

This allows to differentiate between the two database models.

ctdb_db_persistent() - replicated and permanent
ctdb_db_volatile() - distributed and temporary

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

10 files changed, 47 insertions, 26 deletions

diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
index cc23b050825..b1087c498f9 100644
--- a/ctdb/client/ctdb_client.c
+++ b/ctdb/client/ctdb_client.c
@@ -933,8 +933,9 @@ again:
 */
 int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data)
 {
-	if (h->ctdb_db->persistent) {
-		DEBUG(DEBUG_ERR, (__location__ " ctdb_record_store prohibited for persistent dbs\n"));
+	if (! ctdb_db_volatile(h->ctdb_db)) {
+		DEBUG(DEBUG_ERR,
+		      ("ctdb_record_store prohibited for non-volatile dbs\n"));
 		return -1;
 	}
 
diff --git a/ctdb/common/common.h b/ctdb/common/common.h
index 0423c5e2538..5c823a14e6f 100644
--- a/ctdb/common/common.h
+++ b/ctdb/common/common.h
@@ -44,6 +44,9 @@ int ctdb_db_tdb_flags(uint8_t db_flags, bool with_valgrind, bool with_mutex);
 struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb,
 				       const char *name);
 
+bool ctdb_db_persistent(struct ctdb_db_context *ctdb_db);
+bool ctdb_db_volatile(struct ctdb_db_context *ctdb_db);
+
 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key);
 
 int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index 34ebe6acbbc..37ffda8f070 100644
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -78,6 +78,15 @@ 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)
+{
+	return ctdb_db->persistent;
+}
+
+bool ctdb_db_volatile(struct ctdb_db_context *ctdb_db)
+{
+	return !ctdb_db->persistent;
+}
 
 /*
   return the lmaster given a key
@@ -133,7 +142,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"));
@@ -285,8 +295,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) {
diff --git a/ctdb/server/ctdb_call.c b/ctdb/server/ctdb_call.c
index c6bff4954ca..320a13aceac 100644
--- a/ctdb/server/ctdb_call.c
+++ b/ctdb/server/ctdb_call.c
@@ -1974,7 +1974,7 @@ int ctdb_migration_init(struct ctdb_db_context *ctdb_db)
 	struct tevent_timer *te;
 	int ret;
 
-	if (ctdb_db->persistent) {
+	if (! ctdb_db_volatile(ctdb_db)) {
 		return 0;
 	}
 
diff --git a/ctdb/server/ctdb_freeze.c b/ctdb/server/ctdb_freeze.c
index 2666013744b..d92f707b25d 100644
--- a/ctdb/server/ctdb_freeze.c
+++ b/ctdb/server/ctdb_freeze.c
@@ -845,7 +845,7 @@ int32_t ctdb_control_wipe_database(struct ctdb_context *ctdb, TDB_DATA indata)
 		return -1;
 	}
 
-	if (!ctdb_db->persistent) {
+	if (ctdb_db_volatile(ctdb_db)) {
 		talloc_free(ctdb_db->delete_queue);
 		ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
 		if (ctdb_db->delete_queue == NULL) {
diff --git a/ctdb/server/ctdb_ltdb_server.c b/ctdb/server/ctdb_ltdb_server.c
index b016537a312..5c994c90694 100644
--- a/ctdb/server/ctdb_ltdb_server.c
+++ b/ctdb/server/ctdb_ltdb_server.c
@@ -143,7 +143,7 @@ static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
 	}
 
 	if (keep) {
-		if (!ctdb_db->persistent &&
+		if (ctdb_db_volatile(ctdb_db) &&
 		    (ctdb_db->ctdb->pnn == header->dmaster) &&
 		    !(header->flags & CTDB_REC_RO_FLAGS))
 		{
@@ -607,7 +607,7 @@ int ctdb_recheck_persistent_health(struct ctdb_context *ctdb)
 	int fail = 0;
 
 	for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
-		if (!ctdb_db->persistent) {
+		if (!ctdb_db_persistent(ctdb_db)) {
 			continue;
 		}
 
@@ -722,8 +722,9 @@ int ctdb_set_db_readonly(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb
 		return 0;
 	}
 
-	if (ctdb_db->persistent) {
-		DEBUG(DEBUG_ERR,("Persistent databases do not support readonly property\n"));
+	if (! ctdb_db_volatile(ctdb_db)) {
+		DEBUG(DEBUG_ERR,
+		      ("Non-volatile databases do not support readonly flag\n"));
 		return -1;
 	}
 
@@ -779,7 +780,7 @@ static int ctdb_local_attach(struct ctdb_context *ctdb, const char *db_name,
 	ctdb_db->db_id = ctdb_hash(&key);
 	ctdb_db->persistent = persistent;
 
-	if (!ctdb_db->persistent) {
+	if (ctdb_db_volatile(ctdb_db)) {
 		ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
 		if (ctdb_db->delete_queue == NULL) {
 			CTDB_NO_MEMORY(ctdb, ctdb_db->delete_queue);
@@ -1170,10 +1171,10 @@ int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 	/* see if we already have this name */
 	db = ctdb_db_handle(ctdb, db_name);
 	if (db) {
-		if (db->persistent != persistent) {
+		if (ctdb_db_persistent(db) != persistent) {
 			DEBUG(DEBUG_ERR, ("ERROR: DB Attach %spersistent to %spersistent "
 					  "database %s\n", persistent ? "" : "non-",
-					  db-> persistent ? "" : "non-", db_name));
+					  ctdb_db_persistent(db) ? "" : "non-", db_name));
 			return -1;
 		}
 		outdata->dptr  = (uint8_t *)&db->db_id;
@@ -1234,9 +1235,10 @@ int32_t ctdb_control_db_detach(struct ctdb_context *ctdb, TDB_DATA indata,
 		return -1;
 	}
 
-	if (ctdb_db->persistent) {
-		DEBUG(DEBUG_ERR, ("DB detach from persistent database %s "
-				  "denied\n", ctdb_db->db_name));
+	if (! ctdb_db_volatile(ctdb_db)) {
+		DEBUG(DEBUG_ERR,
+		      ("Detaching non-volatile database %s denied\n",
+		       ctdb_db->db_name));
 		return -1;
 	}
 
@@ -1572,8 +1574,9 @@ int ctdb_set_db_sticky(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_d
 		return 0;
 	}
 
-	if (ctdb_db->persistent) {
-		DEBUG(DEBUG_ERR,("Trying to set persistent database with sticky property\n"));
+	if (! ctdb_db_volatile(ctdb_db)) {
+		DEBUG(DEBUG_ERR,
+		      ("Non-volatile databases do not support sticky flag\n"));
 		return -1;
 	}
 
diff --git a/ctdb/server/ctdb_recover.c b/ctdb/server/ctdb_recover.c
index 813a1adbb54..b9fdaa7ed74 100644
--- a/ctdb/server/ctdb_recover.c
+++ b/ctdb/server/ctdb_recover.c
@@ -115,7 +115,7 @@ ctdb_control_getdbmap(struct ctdb_context *ctdb, uint32_t opcode, TDB_DATA indat
 	dbid_map->num = len;
 	for (i=0,ctdb_db=ctdb->db_list;ctdb_db;i++,ctdb_db=ctdb_db->next){
 		dbid_map->dbs[i].db_id       = ctdb_db->db_id;
-		if (ctdb_db->persistent != 0) {
+		if (ctdb_db_persistent(ctdb_db)) {
 			dbid_map->dbs[i].flags |= CTDB_DB_FLAGS_PERSISTENT;
 		}
 		if (ctdb_db->readonly != 0) {
diff --git a/ctdb/server/ctdb_traverse.c b/ctdb/server/ctdb_traverse.c
index 7fba1295909..04a41138a72 100644
--- a/ctdb/server/ctdb_traverse.c
+++ b/ctdb/server/ctdb_traverse.c
@@ -120,7 +120,7 @@ static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DAT
 
 	hdr = (struct ctdb_ltdb_header *)data.dptr;
 
-	if (h->ctdb_db->persistent == 0) {
+	if (ctdb_db_volatile(h->ctdb_db)) {
 		/* filter out zero-length records */
 		if (!h->withemptyrecords &&
 		    data.dsize <= sizeof(struct ctdb_ltdb_header))
@@ -386,7 +386,7 @@ static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_
 		data.dsize = sizeof(r);
 	}
 
-	if (ctdb_db->persistent == 0) {
+	if (ctdb_db_volatile(ctdb_db)) {
 		/* normal database, traverse all nodes */	  
 		destination = CTDB_BROADCAST_VNNMAP;
 	} else {
@@ -587,7 +587,7 @@ int32_t ctdb_control_traverse_data(struct ctdb_context *ctdb, TDB_DATA data, TDB
 		/* Persistent databases are only scanned on one node (the local
 		 * node)
 		 */
-		if (state->ctdb_db->persistent == 0) {
+		if (ctdb_db_volatile(state->ctdb_db)) {
 			if (state->null_count != ctdb_get_num_active_nodes(ctdb)) {
 				return 0;
 			}
diff --git a/ctdb/server/ctdb_update_record.c b/ctdb/server/ctdb_update_record.c
index 046e42f38a9..04b6901ce55 100644
--- a/ctdb/server/ctdb_update_record.c
+++ b/ctdb/server/ctdb_update_record.c
@@ -339,7 +339,7 @@ int32_t ctdb_control_update_record(struct ctdb_context *ctdb,
 	state->c       = c;
 	state->m       = m;
 	state->flags   = 0;
-	if (!ctdb_db->persistent) {
+	if (ctdb_db_volatile(ctdb_db)) {
 		state->flags   = UPDATE_FLAGS_REPLACE_ONLY;
 	}
 
diff --git a/ctdb/server/ctdb_vacuum.c b/ctdb/server/ctdb_vacuum.c
index 240b8dcc9a7..e7491164262 100644
--- a/ctdb/server/ctdb_vacuum.c
+++ b/ctdb/server/ctdb_vacuum.c
@@ -1574,8 +1574,10 @@ void ctdb_stop_vacuuming(struct ctdb_context *ctdb)
  */
 int ctdb_vacuum_init(struct ctdb_db_context *ctdb_db)
 {
-	if (ctdb_db->persistent != 0) {
-		DEBUG(DEBUG_ERR,("Vacuuming is disabled for persistent database %s\n", ctdb_db->db_name));
+	if (! ctdb_db_volatile(ctdb_db)) {
+		DEBUG(DEBUG_ERR,
+		      ("Vacuuming is disabled for non-volatile database %s\n",
+		       ctdb_db->db_name));
 		return 0;
 	}