trans_logger: add CRC checksumming of data

Off by default, since it may hurt performance.

lib_log.c

@@ -307,6 +307,7 @@ bool log_finalize(struct log_status *logst, int len, void (*preio)(void *private
 	int offset;
 	int restlen;
 	int nr_cb;
+	int crc;
 	bool ok = false;
 
 	CHECK_PTR(mref, err);
@@ -327,6 +328,13 @@ bool log_finalize(struct log_status *logst, int len, void (*preio)(void *private
 
 	data = mref->ref_data;
 
+	crc = 0;
+	if (logst->do_crc) {
+		unsigned char checksum[mars_digest_size];
+		mars_digest(checksum, data + logst->payload_offset, len);
+		crc = *(int*)checksum;
+	}
+
 	/* Correct the length in the header.
 	 */
 	offset = logst->reallen_offset;
@@ -336,7 +344,7 @@ bool log_finalize(struct log_status *logst, int len, void (*preio)(void *private
 	 */
 	offset = logst->payload_offset + len;
 	DATA_PUT(data, offset, END_MAGIC);
-	DATA_PUT(data, offset, (int)0);   // crc
+	DATA_PUT(data, offset, crc);
 	DATA_PUT(data, offset, (char)1);  // valid_flag copy
 	DATA_PUT(data, offset, (char)0);  // spare
 	DATA_PUT(data, offset, (short)0); // spare
@@ -461,6 +469,15 @@ int log_scan(void *buf, int len, loff_t file_pos, int file_offset, struct log_he
 		DATA_GET(buf, offset, lh->l_written.tv_sec);
 		DATA_GET(buf, offset, lh->l_written.tv_nsec);
 
+		if (lh->l_crc) {
+			unsigned char checksum[mars_digest_size];
+			mars_digest(checksum, buf + found_offset, lh->l_len);
+			if (unlikely(*(int*)checksum != lh->l_crc)) {
+				MARS_ERR("data checksumming mismatch, length = %d\n", lh->l_len);
+				return -EBADMSG;
+			}
+		}
+
 		// last check
 		if (total_len != offset - i) {
 			MARS_WRN(SCAN_TXT "size mismatch: %d != %d\n", SCAN_PAR, total_len, offset - i);

lib_log.h

@@ -29,7 +29,7 @@ struct log_header_v1 {
 	int    l_extra_len;
 	short  l_code;
 	short  l_extra;
-	int    l_crc; // NYI
+	int    l_crc;
 };
 
 #define FORMAT_VERSION   1 // version of disk format, currently there is no other one
@@ -94,6 +94,7 @@ struct log_status {
 	int align_size;   // alignment between requests
 	int chunk_size;   // must be at least 8K (better 64k)
 	int io_prio;
+	bool do_crc;
 	// informational
 	atomic_t mref_flying;
 	int count;

mars_trans_logger.c

@@ -46,6 +46,14 @@
 
 ///////////////////////// global tuning ////////////////////////
 
+int trans_logger_do_crc =
+#ifdef CONFIG_MARS_DEBUG
+	true;
+#else
+	false;
+#endif
+EXPORT_SYMBOL_GPL(trans_logger_do_crc);
+
 int trans_logger_mem_usage; // in KB
 EXPORT_SYMBOL_GPL(trans_logger_mem_usage);
 
@@ -1470,6 +1478,7 @@ bool phase0_startio(struct trans_logger_mref_aspect *orig_mref_a)
 	CHECK_PTR(input, err);
 	orig_mref_a->log_input = input;
 	logst = &input->logst;
+	logst->do_crc = trans_logger_do_crc;
 
 	{
 		struct log_header l = {
@@ -1783,6 +1792,7 @@ bool _phase2_startio(struct trans_logger_mref_aspect *sub_mref_a)
 	input = sub_mref_a->log_input;
 	CHECK_PTR(input, err);
 	logst = &input->logst;
+	logst->do_crc = trans_logger_do_crc;
 
 	{
 		struct log_header l = {

mars_trans_logger.h

@@ -17,6 +17,7 @@
 
 ///////////////////////// global tuning ////////////////////////
 
+extern int trans_logger_do_crc;
 extern int trans_logger_mem_usage; // in KB
 extern atomic_t   global_mshadow_count;
 extern atomic64_t global_mshadow_used;

sy_old/mars_proc.c

@@ -198,6 +198,7 @@ ctl_table mars_table[] = {
 		.mode		= 0400,
 		.proc_handler	= &lamport_sysctl_handler,
 	},
+	INT_ENTRY("logger_do_crc",        trans_logger_do_crc,    0600),
 	INT_ENTRY("syslog_min_class",     brick_say_syslog_min,   0600),
 	INT_ENTRY("syslog_max_class",     brick_say_syslog_max,   0600),
 	INT_ENTRY("delay_say_on_overflow",delay_say_on_overflow,  0600),