ReadOnly: We can not use ctdb_ltdb_store from a client/child context since

it sometimes (for empty records) needs to be able to initiate traffic unde rhte daemon context.

This should furhter updated later to allow the use also from non-daemon context.

(This used to be ctdb commit 11dee7f3f881494cf5089d6c69fd40e74f07e670)

1 files changed, 23 insertions, 10 deletions

diff --git a/ctdb/server/ctdb_persistent.c b/ctdb/server/ctdb_persistent.c
index dd8d4793fb7..0f4f4da52d0 100644
--- a/ctdb/server/ctdb_persistent.c
+++ b/ctdb/server/ctdb_persistent.c
@@ -456,12 +456,12 @@ static int ctdb_persistent_store(struct ctdb_persistent_write_state *state)
 	}
 
 	for (i=0;i<m->count;i++) {
-		struct ctdb_ltdb_header oldheader;
-		struct ctdb_ltdb_header header;
+		struct ctdb_ltdb_header *oldheader;
+		struct ctdb_ltdb_header *header;
 		TDB_DATA key, data, olddata;
 		TALLOC_CTX *tmp_ctx = talloc_new(state);
 
-		rec = ctdb_marshall_loop_next(m, rec, NULL, &header, &key, &data);
+		rec = ctdb_marshall_loop_next(m, rec, NULL, NULL, &key, &data);
 		
 		if (rec == NULL) {
 			DEBUG(DEBUG_ERR,("Failed to get next record %d for db_id 0x%08x in ctdb_persistent_store\n",
@@ -469,29 +469,42 @@ static int ctdb_persistent_store(struct ctdb_persistent_write_state *state)
 			talloc_free(tmp_ctx);
 			goto failed;			
 		}
+		header = (struct ctdb_ltdb_header *)&data.dptr[0];
 
 		/* fetch the old header and ensure the rsn is less than the new rsn */
-		ret = ctdb_ltdb_fetch(state->ctdb_db, key, &oldheader, tmp_ctx, &olddata);
-		if (ret != 0) {
+		olddata = tdb_fetch(state->ctdb_db->ltdb->tdb, key);
+		if (olddata.dptr == NULL) {
 			DEBUG(DEBUG_ERR,("Failed to fetch old record for db_id 0x%08x in ctdb_persistent_store\n",
 					 state->ctdb_db->db_id));
 			talloc_free(tmp_ctx);
 			goto failed;
 		}
+		if (olddata.dsize < sizeof(struct ctdb_ltdb_header)) {
+			DEBUG(DEBUG_ERR,("Not enough header for record for db_id 0x%08x in ctdb_persistent_store\n",
+					state->ctdb_db->db_id));
+			talloc_free(tmp_ctx);
+			free(olddata.dptr);
+			goto failed;
+		}
+		oldheader = (struct ctdb_ltdb_header *)&olddata.dptr[0];
 
-		if (oldheader.rsn >= header.rsn &&
-		    (olddata.dsize != data.dsize || 
-		     memcmp(olddata.dptr, data.dptr, data.dsize) != 0)) {
+		if (oldheader->rsn >= header->rsn &&
+		   (olddata.dsize != data.dsize || 
+		   memcmp(&olddata.dptr[sizeof(struct ctdb_ltdb_header)],
+		     &data.dptr[sizeof(struct ctdb_ltdb_header)],
+		     data.dsize - sizeof(struct ctdb_ltdb_header)) != 0)) {
 			DEBUG(DEBUG_CRIT,("existing header for db_id 0x%08x has larger RSN %llu than new RSN %llu in ctdb_persistent_store\n",
 					  state->ctdb_db->db_id, 
-					  (unsigned long long)oldheader.rsn, (unsigned long long)header.rsn));
+					  (unsigned long long)oldheader->rsn, (unsigned long long)header->rsn));
 			talloc_free(tmp_ctx);
+			free(olddata.dptr);
 			goto failed;
 		}
 
 		talloc_free(tmp_ctx);
+		free(olddata.dptr);
 
-		ret = ctdb_ltdb_store(state->ctdb_db, key, &header, data);
+		ret = tdb_store(state->ctdb_db->ltdb->tdb, key, data, TDB_REPLACE);
 		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));