second try for safe transaction stores into persistend tdb databases

for stores into persistent databases, ALWAYS use a lockwait child take out the lock for the record and never the daemon itself.

(This used to be ctdb commit 7fb6cf549de1b5e9ac5a3e4483c7591850ea2464)

4 files changed, 62 insertions, 18 deletions

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index a3df65ebd4..266489ca05 100644
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -121,9 +121,7 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
 
 
 /*
-  fetch a record from the ltdb, separating out the header information
-  and returning the body of the record. A valid (initial) header is
-  returned if the record is not present
+  write a record to a normal database
 */
 int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
 		    struct ctdb_ltdb_header *header, TDB_DATA data)
@@ -150,25 +148,64 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
 	memcpy(rec.dptr, header, sizeof(*header));
 	memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 
+	ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, (__location__ " Failed to store dynamic data\n"));
+	}
+
+	talloc_free(rec.dptr);
+
+	return ret;
+}
+
+/*
+  write a record to a persistent database
+  at this stage the the record is locked by a lockwait child.
+*/
+int ctdb_ltdb_persistent_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
+		    struct ctdb_ltdb_header *header, TDB_DATA data)
+{
+	struct ctdb_context *ctdb = ctdb_db->ctdb;
+	TDB_DATA rec;
+	int ret;
+
+	if (ctdb->flags & CTDB_FLAG_TORTURE) {
+		struct ctdb_ltdb_header *h2;
+		rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
+		h2 = (struct ctdb_ltdb_header *)rec.dptr;
+		if (rec.dptr && rec.dsize >= sizeof(h2) && h2->rsn > header->rsn) {
+			DEBUG(DEBUG_CRIT,("RSN regression! %llu %llu\n",
+				 (unsigned long long)h2->rsn, (unsigned long long)header->rsn));
+		}
+		if (rec.dptr) free(rec.dptr);
+	}
+
+	rec.dsize = sizeof(*header) + data.dsize;
+	rec.dptr = talloc_size(ctdb, rec.dsize);
+	CTDB_NO_MEMORY(ctdb, rec.dptr);
+
+	memcpy(rec.dptr, header, sizeof(*header));
+	memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
+
 	/* if this is a persistent database without NOSYNC then we
 	   will do this via a transaction */
 	if (ctdb_db->persistent && !(ctdb_db->client_tdb_flags & TDB_NOSYNC)) {
-		bool transaction_started = true;
-
 		ret = tdb_transaction_start(ctdb_db->ltdb->tdb);
 		if (ret != 0) {
-			transaction_started = false;
-			DEBUG(DEBUG_NOTICE, ("Failed to start local transaction\n"));
+			DEBUG(DEBUG_ERR, (__location__ " Failed to start local transaction\n"));
+			goto failed;
 		}
 		ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
 		if (ret != 0) {
-			if (transaction_started) {
-				tdb_transaction_cancel(ctdb_db->ltdb->tdb);
-			}
+			DEBUG(DEBUG_ERR, (__location__ " Failed to store persistent data\n"));
+			tdb_transaction_cancel(ctdb_db->ltdb->tdb);
 			goto failed;
 		}
-		if (transaction_started) {
-			ret = tdb_transaction_commit(ctdb_db->ltdb->tdb);
+		ret = tdb_transaction_commit(ctdb_db->ltdb->tdb);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR, (__location__ " Failed to commit persistent store transaction.\n"));
+			tdb_transaction_cancel(ctdb_db->ltdb->tdb);
+			goto failed;
 		}
 	} else {
 		ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
@@ -180,7 +217,6 @@ failed:
 	return ret;
 }
 
-
 /*
   lock a record in the ltdb, given a key
  */
diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 4eccc84198..758b506c65 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -790,6 +790,8 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
 		    TALLOC_CTX *mem_ctx, TDB_DATA *data);
 int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
 		    struct ctdb_ltdb_header *header, TDB_DATA data);
+int ctdb_ltdb_persistent_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
+		    struct ctdb_ltdb_header *header, TDB_DATA data);
 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db, 
 			   TDB_DATA key, struct ctdb_req_header *hdr,
diff --git a/ctdb/lib/tdb/common/transaction.c b/ctdb/lib/tdb/common/transaction.c
index 5e5260b80a..4e2127be64 100644
--- a/ctdb/lib/tdb/common/transaction.c
+++ b/ctdb/lib/tdb/common/transaction.c
@@ -419,7 +419,7 @@ int tdb_transaction_start(struct tdb_context *tdb)
 		/* the caller must not have any locks when starting a
 		   transaction as otherwise we'll be screwed by lack
 		   of nested locks in posix */
-//		TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: cannot start a transaction with locks held\n"));
+		TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction with locks held\n"));
 		tdb->ecode = TDB_ERR_LOCK;
 		return -1;
 	}
diff --git a/ctdb/server/ctdb_persistent.c b/ctdb/server/ctdb_persistent.c
index 053bd400ae..ce58b52cd3 100644
--- a/ctdb/server/ctdb_persistent.c
+++ b/ctdb/server/ctdb_persistent.c
@@ -147,7 +147,7 @@ struct ctdb_persistent_lock_state {
 
 
 /*
-  called with a lock held in the current process
+  called with a lock held by a lockwait child
  */
 static int ctdb_persistent_store(struct ctdb_persistent_lock_state *state)
 {
@@ -169,7 +169,7 @@ static int ctdb_persistent_store(struct ctdb_persistent_lock_state *state)
 		return -1;
 	}
 
-	ret = ctdb_ltdb_store(state->ctdb_db, state->key, state->header, state->data);
+	ret = ctdb_ltdb_persistent_store(state->ctdb_db, state->key, state->header, state->data);
 	if (ret != 0) {
 		DEBUG(DEBUG_CRIT,("Failed to store record for db_id 0x%08x in ctdb_persistent_store\n", 
 			 state->ctdb_db->db_id));
@@ -224,7 +224,6 @@ int32_t ctdb_control_update_record(struct ctdb_context *ctdb,
 				   bool *async_reply)
 {
 	struct ctdb_rec_data *rec = (struct ctdb_rec_data *)&recdata.dptr[0];
-	int ret;
 	struct ctdb_db_context *ctdb_db;
 	uint32_t db_id = rec->reqid;
 	struct lockwait_handle *handle;
@@ -262,13 +261,20 @@ int32_t ctdb_control_update_record(struct ctdb_context *ctdb,
 	state->data.dptr  += sizeof(struct ctdb_ltdb_header);
 	state->data.dsize -= sizeof(struct ctdb_ltdb_header);
 
-	/* try and do it without a lockwait */
+#if 0
+	/* We can not take out a lock here pourself since if this persistent
+	   database needs safe transaction writes we can not be holding
+	   a lock on the database.
+	   Therefore we always create a lock wait child to take out and hold
+	   the lock for us.
+	*/
 	ret = tdb_chainlock_nonblock(state->tdb, state->key);
 	if (ret == 0) {
 		ret = ctdb_persistent_store(state);
 		tdb_chainunlock(state->tdb, state->key);
 		return ret;
 	}
+#endif
 
 	/* wait until we have a lock on this record */
 	handle = ctdb_lockwait(ctdb_db, state->key, ctdb_persistent_lock_callback, state);