Improve readability of the code and update code from nsqio#29 issue 29

Author: kev1n80

diskqueue.go

@@ -553,7 +553,7 @@ func (d *diskQueue) getAllBadFileInfo() ([]os.FileInfo, error) {
 }
 
 // get the accurate total non-"bad" file size
-func (d *diskQueue) updateTotalDiskSpaceUsed() error {
+func (d *diskQueue) updateTotalDiskSpaceUsed() {
 	d.totalDiskSpaceUsed = maxMetaDataFileSize
 
 	updateTotalDiskSpaceUsed := func(fileInfo os.FileInfo) error {
@@ -565,7 +565,10 @@ func (d *diskQueue) updateTotalDiskSpaceUsed() error {
 		return nil
 	}
 
-	return d.walkDiskQueueDir(updateTotalDiskSpaceUsed)
+	err := d.walkDiskQueueDir(updateTotalDiskSpaceUsed)
+	if err != nil {
+		d.logf(ERROR, "DISKQUEUE(%s) failed to update write bytes - %s", d.name, err)
+	}
 }
 
 func (d *diskQueue) freeDiskSpace(expectedBytesIncrease int64) error {
@@ -936,6 +939,11 @@ func (d *diskQueue) handleReadError() {
 		}
 		d.writeFileNum++
 		d.writePos = 0
+
+		if d.enableDiskLimitation {
+			d.totalDiskSpaceUsed = 0
+			d.writeMessages = 0
+		}
 	}
 
 	badFn := d.fileName(d.readFileNum)
@@ -952,27 +960,18 @@ func (d *diskQueue) handleReadError() {
 			d.name, badFn, badRenameFn)
 	}
 
-	if d.enableDiskLimitation {
-		if d.readFileNum == d.writeFileNum {
-			// we moved on to the next writeFile
-			d.totalDiskSpaceUsed = 0
-			d.writeMessages = 0
-		} else {
-			if err != nil {
-				d.logf(ERROR, "DISKQUEUE(%s) failed to update write bytes - %s", d.name, err)
-			}
-		}
-
-		d.readMessages = 0
-	}
-
 	d.readFileNum++
 	d.readPos = 0
 	d.nextReadFileNum = d.readFileNum
 	d.nextReadPos = 0
+	if d.enableDiskLimitation {
+		d.readMessages = 0
+	}
 
 	// significant state change, schedule a sync on the next iteration
 	d.needSync = true
+
+	d.checkTailCorruption(d.depth)
 }
 
 // ioLoop provides the backend for exposing a go channel (via ReadChan())