X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=ustd%2Flowlevel.c;h=ff3875931a36c464c345c7c2a8823ee94e61f11d;hb=b7619a864595a5511a498345635affe2547b91a0;hp=f386089e21b54f9c1a0b98cc351d26084deed454;hpb=0b0cd93782698872ea4778d29cd68091663ca367;p=lttng-ust.git

diff --git a/ustd/lowlevel.c b/ustd/lowlevel.c
index f386089e..ff387593 100644
--- a/ustd/lowlevel.c
+++ b/ustd/lowlevel.c
@@ -1,7 +1,10 @@
+#include <assert.h>
+
 #include "tracer.h"
 #include "ustd.h"
 #include "localerr.h"
 
+/* This truncates to an offset in the buffer. */
 #define USTD_BUFFER_TRUNC(offset, bufinfo) \
 	((offset) & (~(((bufinfo)->subbuf_size*(bufinfo)->n_subbufs)-1)))
 
@@ -18,32 +21,70 @@ void finish_consuming_dead_subbuffer(struct buffer_info *buf)
 	DBG("consumed offset is %ld", consumed_offset);
 	DBG("write offset is %ld", write_offset);
 
-	long first_subbuf = write_offset / buf->subbuf_size;
-	long last_subbuf = consumed_offset / buf->subbuf_size;
+	/* First subbuf that we need to consume now. It is not modulo'd.
+	 * Consumed_offset is the next byte to consume.  */
+	long first_subbuf = consumed_offset / buf->subbuf_size;
+	/* Last subbuf that we need to consume now. It is not modulo'd. 
+	 * Write_offset is the next place to write so write_offset-1 is the
+	 * last place written. */
+	long last_subbuf = (write_offset - 1) / buf->subbuf_size;
+
+	DBG("first_subbuf=%d", first_subbuf);
+	DBG("last_subbuf=%d", last_subbuf);
 
-	if(last_subbuf - first_subbuf > buf->n_subbufs) {
+	if(last_subbuf - first_subbuf >= buf->n_subbufs) {
 		DBG("an overflow has occurred, nothing can be recovered");
 		return;
 	}
 
+	/* Iterate on subbuffers to recover. */
 	for(i_subbuf=first_subbuf; ; i_subbuf++, i_subbuf %= buf->n_subbufs) {
-		long commit_count = local_read(&ltt_buf->commit_count[i_subbuf]);
+		void *tmp;
+		/* commit_seq is the offset in the buffer of the end of the last sequential commit.
+		 * Bytes beyond this limit cannot be recovered. This is a free-running counter. */
+		long commit_seq = local_read(&ltt_buf->commit_seq[i_subbuf]);
 
 		unsigned long valid_length = buf->subbuf_size;
 		long n_subbufs_order = get_count_order(buf->n_subbufs);
-		long commit_count_mask = (~0UL >> n_subbufs_order);
+		long commit_seq_mask = (~0UL >> n_subbufs_order);
+
+		struct ltt_subbuffer_header *header = (struct ltt_subbuffer_header *)((char *)buf->mem+i_subbuf*buf->subbuf_size);
+
+		if((commit_seq & commit_seq_mask) == 0) {
+			/* There is nothing to do. */
+			/* FIXME: is this needed? */
+			break;
+		}
 
-		/* check if subbuf was fully written */
-		if (((commit_count - buf->subbuf_size) & commit_count_mask)
+		/* Check if subbuf was fully written. This is from Mathieu's algorithm/paper. */
+		if (((commit_seq - buf->subbuf_size) & commit_seq_mask)
 		    - (USTD_BUFFER_TRUNC(consumed_offset, buf) >> n_subbufs_order)
-		    != 0) {
-			struct ltt_subbuffer_header *header = (struct ltt_subbuffer_header *)((char *)buf->mem)+i_subbuf*buf->subbuf_size;
-			valid_length = buf->subbuf_size - header->lost_size;
+		    == 0) {
+			/* If it was, we only check the lost_size. This is the lost padding at the end of
+ 			 * the subbuffer. */
+			valid_length = (unsigned long)buf->subbuf_size - header->lost_size;
+		}
+		else {
+			/* If the subbuffer was not fully written, then we don't check lost_size because
+			 * it hasn't been written yet. Instead we check commit_seq and use it to choose
+			 * a value for lost_size. The viewer will need this value when parsing.
+			 */
+
+			valid_length = commit_seq & (buf->subbuf_size-1);
+			header->lost_size = buf->subbuf_size-valid_length;
+			assert(i_subbuf == (last_subbuf % buf->n_subbufs));
 		}
 
-		patient_write(buf->file_fd, buf->mem + i_subbuf * buf->subbuf_size, buf->subbuf_size);
 
-		if(i_subbuf == last_subbuf)
+		patient_write(buf->file_fd, buf->mem + i_subbuf * buf->subbuf_size, valid_length);
+
+		/* pad with empty bytes */
+		tmp = malloc(buf->subbuf_size-valid_length);
+		memset(tmp, 0, buf->subbuf_size-valid_length);
+		patient_write(buf->file_fd, tmp, buf->subbuf_size-valid_length);
+		free(tmp);
+
+		if(i_subbuf == last_subbuf % buf->n_subbufs)
 			break;
 	}
 }