make the ringbuffer logging more efficient and marshall the data by writing to a tmpfile instead of continously talloc resizing a blob

(This used to be ctdb commit 6427f0b68d60b556a023f64e15e156000ba6f943)

ctdb/common/ctdb_logging.c

@@ -20,6 +20,7 @@
 #include "includes.h"
 #include "lib/events/events.h"
 #include "lib/tdb/include/tdb.h"
+#include "system/time.h"
 #include "../include/ctdb_private.h"
 #include "../include/ctdb.h"
 
@@ -82,49 +83,59 @@ void log_ringbuffer(const char *format, ...)
 
 static void ctdb_collect_log(struct ctdb_context *ctdb, struct ctdb_get_log_addr *log_addr)
 {
-	char *buf = talloc_size(NULL, 0);
-	struct ctdb_log_entry_wire *log_entry;
-	uint32_t old_size, len;
 	TDB_DATA data;
+	FILE *f;
+	long fsize;
+	int tmp_entry;
+	int count = 0;
+	DEBUG(DEBUG_ERR,("Marshalling log entries  first:%d last:%d\n", first_entry, last_entry));
 
-	DEBUG(DEBUG_INFO,("Marshalling log entries\n"));
+	/* dump to a file, then send the file as a blob */
-	while (first_entry != last_entry) {
+	f = tmpfile();
-		int slen = strlen(log_entries[first_entry].message);
+	if (f == NULL) {
+		DEBUG(DEBUG_ERR,(__location__ " Unable to open tmpfile - %s\n", strerror(errno)));
+		return;
+	}
 
-		if (log_entries[first_entry].level > log_addr->level) {
+	tmp_entry = first_entry;
-			first_entry++;
+	while (tmp_entry != last_entry) {
-			if (first_entry >= MAX_LOG_ENTRIES) {
+		struct tm *tm;
-				first_entry = 0;
+		char tbuf[100];
+
+		if (log_entries[tmp_entry].level > log_addr->level) {
+			tmp_entry++;
+			if (tmp_entry >= MAX_LOG_ENTRIES) {
+				tmp_entry = 0;
 			}
-			continue;
+		 	continue;
 		}
 
-		len = offsetof(struct ctdb_log_entry_wire, message) + slen + 1;
+		tm = localtime(&log_entries[tmp_entry].t.tv_sec);
-		/* pad it to uint42 */
+		strftime(tbuf, sizeof(tbuf)-1,"%Y/%m/%d %H:%M:%S", tm);
-		len = (len+3)&0xfffffffc;
 
-		old_size = talloc_get_size(buf);
+		if (log_entries[tmp_entry].message) {
-		buf = talloc_realloc_size(NULL, buf, old_size + len);
+			count += fprintf(f, "%s:%s %s", tbuf, get_debug_by_level(log_entries[tmp_entry].level), log_entries[tmp_entry].message);
+		}
 
-		log_entry = (struct ctdb_log_entry_wire *)&buf[old_size];
+		tmp_entry++;
-		log_entry->level       = log_entries[first_entry].level;
+		if (tmp_entry >= MAX_LOG_ENTRIES) {
-		log_entry->t           = log_entries[first_entry].t;
+			tmp_entry = 0;
-		log_entry->message_len = slen;
-		memcpy(log_entry->message, log_entries[first_entry].message, slen);
-		log_entry->message[slen] = 0;
-
-		first_entry++;
-		if (first_entry >= MAX_LOG_ENTRIES) {
-			first_entry = 0;
 		}
 	}
 
-	data.dptr  = (uint8_t *)buf;
+	fsize = ftell(f);
-	data.dsize = talloc_get_size(buf);
+	rewind(f);
-	DEBUG(DEBUG_INFO,("Marshalling log entries into a blob of %d bytes\n", (int)data.dsize));
+	data.dptr = talloc_size(NULL, fsize);
+	CTDB_NO_MEMORY_VOID(ctdb, data.dptr);
+	data.dsize = fread(data.dptr, 1, fsize, f);
+	fclose(f);
 
-	DEBUG(DEBUG_INFO,("Send log to %d:%d\n", (int)log_addr->pnn, (int)log_addr->srvid));
+	DEBUG(DEBUG_ERR,("Marshalling log entries into a blob of %d bytes\n", (int)data.dsize));
+
+	DEBUG(DEBUG_ERR,("Send log to %d:%d\n", (int)log_addr->pnn, (int)log_addr->srvid));
 	ctdb_send_message(ctdb, log_addr->pnn, log_addr->srvid, data);
+
+	talloc_free(data.dptr);
 }
 
 int32_t ctdb_control_get_log(struct ctdb_context *ctdb, TDB_DATA addr)

ctdb/include/ctdb_private.h

@@ -1529,14 +1529,6 @@ struct ctdb_get_log_addr {
 	int32_t level;
 };
 
-/* wire data for log entries, padded to uint32 */
-struct ctdb_log_entry_wire {
-	int32_t level;
-	struct timeval t;
-	int32_t message_len;
-	char message[1];
-};
-
 int32_t ctdb_control_get_log(struct ctdb_context *ctdb, TDB_DATA addr);
 int32_t ctdb_control_clear_log(struct ctdb_context *ctdb);
 

ctdb/tools/ctdb.c

@@ -2357,26 +2357,11 @@ static int control_catdb(struct ctdb_context *ctdb, int argc, const char **argv)
 static void log_handler(struct ctdb_context *ctdb, uint64_t srvid, 
 			     TDB_DATA data, void *private_data)
 {
-	struct ctdb_log_entry_wire *entry;
-	int size;
-	struct tm *tm;
-	char tbuf[100];
-
 	DEBUG(DEBUG_ERR,("Log data received\n"));
-	while (data.dsize > 0) {
+	if (data.dsize > 0) {
-		entry = (struct ctdb_log_entry_wire *)data.dptr;
+		printf("%s", data.dptr);
-		size = offsetof(struct ctdb_log_entry_wire, message) + entry->message_len + 1;
-		size = (size+3)&0xfffffffc;
-
-		tm = localtime(&entry->t.tv_sec);
-
-		strftime(tbuf,sizeof(tbuf)-1,"%Y/%m/%d %H:%M:%S", tm);
-
-		printf("%s:%s %s", tbuf, get_debug_by_level(entry->level), entry->message);
-
-		data.dsize -= size;
-		data.dptr  += size;
 	}
+
 	exit(0);
 }