Rev 730: more efficient traversal in pulldb control in http://samba.org/~tridge/ctdb

[tridge at samba.org]

------------------------------------------------------------
revno: 730
revision-id:tridge at samba.org-20080107030701-l8awuh0wr4lcpbgt
parent: tridge at samba.org-20080107030530-cv6i3j5ue7z3sk3t
committer: Andrew Tridgell <tridge at samba.org>
branch nick: tridge.stable
timestamp: Mon 2008-01-07 14:07:01 +1100
message:
  more efficient traversal in pulldb control
modified:
  server/ctdb_recover.c          ctdb_recover.c-20070503002147-admmfgt1oj6gexfo-1
=== modified file 'server/ctdb_recover.c'
--- a/server/ctdb_recover.c	2008-01-06 01:38:01 +0000
+++ b/server/ctdb_recover.c	2008-01-07 03:07:01 +0000
@@ -170,36 +170,38 @@
 	return 0;
 }
 
-struct getkeys_params {
+/* 
+   a traverse function for pulling all relevent records from pulldb
+ */
+struct pulldb_data {
 	struct ctdb_context *ctdb;
-	uint32_t lmaster;
-	uint32_t rec_count;
-	struct getkeys_rec {
-		TDB_DATA key;
-		TDB_DATA data;
-	} *recs;
+	struct ctdb_control_pulldb_reply *pulldata;
+	uint32_t len;
+	bool failed;
 };
 
-static int traverse_getkeys(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
+static int traverse_pulldb(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
 {
-	struct getkeys_params *params = (struct getkeys_params *)p;
-	uint32_t lmaster;
-
-	lmaster = ctdb_lmaster(params->ctdb, &key);
-
-	/* only include this record if the lmaster matches or if
-	   the wildcard lmaster (-1) was specified.
-	*/
-	if ((params->lmaster != CTDB_LMASTER_ANY) && (params->lmaster != lmaster)) {
-		return 0;
-	}
-
-	params->recs = talloc_realloc(NULL, params->recs, struct getkeys_rec, params->rec_count+1);
-	key.dptr = talloc_memdup(params->recs, key.dptr, key.dsize);
-	data.dptr = talloc_memdup(params->recs, data.dptr, data.dsize);
-	params->recs[params->rec_count].key = key;
-	params->recs[params->rec_count].data = data;
-	params->rec_count++;
+	struct pulldb_data *params = (struct pulldb_data *)p;
+	struct ctdb_rec_data *rec;
+
+	/* add the record to the blob */
+	rec = ctdb_marshall_record(params->pulldata, 0, key, NULL, data);
+	if (rec == NULL) {
+		params->failed = true;
+		return -1;
+	}
+	params->pulldata = talloc_realloc_size(NULL, params->pulldata, rec->length + params->len);
+	if (params->pulldata == NULL) {
+		DEBUG(0,(__location__ " Failed to expand pulldb_data to %u (%u records)\n", 
+			 rec->length + params->len, params->pulldata->count));
+		params->failed = true;
+		return -1;
+	}
+	params->pulldata->count++;
+	memcpy(params->len+(uint8_t *)params->pulldata, rec, rec->length);
+	params->len += rec->length;
+	talloc_free(rec);
 
 	return 0;
 }
@@ -211,10 +213,8 @@
 {
 	struct ctdb_control_pulldb *pull;
 	struct ctdb_db_context *ctdb_db;
-	struct getkeys_params params;
+	struct pulldb_data params;
 	struct ctdb_control_pulldb_reply *reply;
-	int i;
-	size_t len = 0;
 
 	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
 		DEBUG(0,("rejecting ctdb_control_pull_db when not frozen\n"));
@@ -229,43 +229,32 @@
 		return -1;
 	}
 
-	params.ctdb = ctdb;
-	params.lmaster = pull->lmaster;
-
-	params.rec_count = 0;
-	params.recs = talloc_array(outdata, struct getkeys_rec, 0);
-	CTDB_NO_MEMORY(ctdb, params.recs);
-
-	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
-		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
-		return -1;
-	}
-
-	tdb_traverse_read(ctdb_db->ltdb->tdb, traverse_getkeys, &params);
-
-	ctdb_lock_all_databases_unmark(ctdb);
-
-	reply = talloc(outdata, struct ctdb_control_pulldb_reply);
+	reply = talloc_zero(outdata, struct ctdb_control_pulldb_reply);
 	CTDB_NO_MEMORY(ctdb, reply);
 
 	reply->db_id = pull->db_id;
-	reply->count = params.rec_count;
-
-	len = offsetof(struct ctdb_control_pulldb_reply, data);
-
-	for (i=0;i<reply->count;i++) {
-		struct ctdb_rec_data *rec;
-		rec = ctdb_marshall_record(outdata, 0, params.recs[i].key, NULL, params.recs[i].data);
-		reply = talloc_realloc_size(outdata, reply, rec->length + len);
-		memcpy(len+(uint8_t *)reply, rec, rec->length);
-		len += rec->length;
-		talloc_free(rec);
-	}
-
-	talloc_free(params.recs);
-
-	outdata->dptr = (uint8_t *)reply;
-	outdata->dsize = len;
+
+	params.ctdb = ctdb;
+	params.pulldata = reply;
+	params.len = offsetof(struct ctdb_control_pulldb_reply, data);
+	params.failed = false;
+
+	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
+		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
+		return -1;
+	}
+
+	if (tdb_traverse_read(ctdb_db->ltdb->tdb, traverse_pulldb, &params) == -1) {
+		DEBUG(0,(__location__ " Failed to get traverse db '%s'\n", ctdb_db->db_name));
+		ctdb_lock_all_databases_unmark(ctdb);
+		talloc_free(params.pulldata);
+		return -1;
+	}
+
+	ctdb_lock_all_databases_unmark(ctdb);
+
+	outdata->dptr = (uint8_t *)params.pulldata;
+	outdata->dsize = params.len;
 
 	return 0;
 }