Reverted switch to next file before maxBytesPerFile is reached

Author: Mikhail Faraponov

diskqueue.go

@@ -641,35 +641,6 @@ func (d *diskQueue) checkDiskSpace(expectedBytesIncrease int64) error {
 func (d *diskQueue) writeOne(data []byte) error {
 	var err error
 
-	dataLen := int32(len(data))
-	totalBytes := int64(4 + dataLen)
-	reachedFileSizeLimit := false
-
-	if dataLen < d.minMsgSize || dataLen > d.maxMsgSize {
-		return fmt.Errorf("invalid message write size (%d) minMsgSize=%d maxMsgSize=%d", dataLen, d.minMsgSize, d.maxMsgSize)
-	}
-
-	// will not wrap-around if maxBytesPerFile + maxMsgSize < Int64Max
-	if d.writePos > 0 && d.writePos+totalBytes > d.maxBytesPerFile {
-		if d.readFileNum == d.writeFileNum {
-			d.maxBytesPerFileRead = d.writePos
-		}
-
-		d.writeFileNum++
-		d.writePos = 0
-
-		// sync every time we start writing to a new file
-		err = d.sync()
-		if err != nil {
-			d.logf(ERROR, "DISKQUEUE(%s) failed to sync - %s", d.name, err)
-		}
-
-		if d.writeFile != nil {
-			d.writeFile.Close()
-			d.writeFile = nil
-		}
-	}
-
 	if d.writeFile == nil {
 		curFileName := d.fileName(d.writeFileNum)
 		d.writeFile, err = os.OpenFile(curFileName, os.O_RDWR|os.O_CREATE, 0600)
@@ -689,6 +660,15 @@ func (d *diskQueue) writeOne(data []byte) error {
 		}
 	}
 
+	dataLen := int32(len(data))
+
+	if dataLen < d.minMsgSize || dataLen > d.maxMsgSize {
+		return fmt.Errorf("invalid message write size (%d) minMsgSize=%d maxMsgSize=%d", dataLen, d.minMsgSize, d.maxMsgSize)
+	}
+
+	totalBytes := int64(4 + dataLen)
+	reachedFileSizeLimit := false
+
 	if d.enableDiskLimitation {
 		expectedBytesIncrease := totalBytes
 		// check if we will reach or surpass file size limit
@@ -706,6 +686,8 @@ func (d *diskQueue) writeOne(data []byte) error {
 		reachedFileSizeLimit = true
 	}
 
+	// add all data to writeBuf before writing to file
+	// this causes everything to be written to file or nothing
 	d.writeBuf.Reset()
 	err = binary.Write(&d.writeBuf, binary.BigEndian, dataLen)
 	if err != nil {

diskqueue_test.go

@@ -295,11 +295,7 @@ func TestDiskQueueCorruption(t *testing.T) {
 	dq := New(dqName, tmpDir, 1000, 10, 1<<10, 5, 2*time.Second, l)
 	defer dq.Close()
 
-	msg := make([]byte, 120) // 124 bytes per message, 8 messages (992 bytes) per file
-	msg[0] = 91
-	msg[62] = 4
-	msg[119] = 211
-
+	msg := make([]byte, 123) // 127 bytes per message, 8 (1016 bytes) messages per file
 	for i := 0; i < 25; i++ {
 		dq.Put(msg)
 	}
@@ -308,7 +304,7 @@ func TestDiskQueueCorruption(t *testing.T) {
 
 	// corrupt the 2nd file
 	dqFn := dq.(*diskQueue).fileName(1)
-	os.Truncate(dqFn, 400) // 3 valid messages, 5 corrupted
+	os.Truncate(dqFn, 500) // 3 valid messages, 5 corrupted
 
 	for i := 0; i < 19; i++ { // 1 message leftover in 4th file
 		Equal(t, msg, <-dq.ReadChan())
@@ -328,7 +324,7 @@ func TestDiskQueueCorruption(t *testing.T) {
 	Equal(t, msg, <-dq.ReadChan())
 	badFilesCount = numberOfBadFiles(dqName, tmpDir)
 	if badFilesCount != 2 {
-		panic(badFilesCount)
+		panic("fail")
 	}
 
 	// write a corrupt (len 0) message at the 5th (current) file
@@ -339,30 +335,10 @@ func TestDiskQueueCorruption(t *testing.T) {
 	dq.Put(msg)
 
 	Equal(t, msg, <-dq.ReadChan())
-
-	// conflict was here
 	badFilesCount = numberOfBadFiles(dqName, tmpDir)
 	if badFilesCount != 3 {
 		panic("fail")
 	}
-	//
-
-	dq.Put(msg)
-	dq.Put(msg)
-	// corrupt the last file
-	dqFn = dq.(*diskQueue).fileName(5)
-	os.Truncate(dqFn, 100)
-
-	Equal(t, int64(2), dq.Depth())
-
-	// return one message and try reading again from corrupted file
-	<-dq.ReadChan()
-
-	// give diskqueue time to handle read error
-	time.Sleep(50 * time.Millisecond)
-
-	// the last log file is now considered corrupted leaving no more log messages
-	Equal(t, int64(0), dq.Depth())
 }
 
 type md struct {
@@ -476,8 +452,9 @@ func TestDiskQueueSyncAfterReadWithDiskSizeImplementation(t *testing.T) {
 		panic("fail")
 	}
 
-	for i := 0; i < 10; i++ {
-		d := readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
+	var d md
+	for i := 0; i < 20; i++ {
+		d = readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
 		if d.depth == 1 &&
 			d.readFileNum == 0 &&
 			d.writeFileNum == 0 &&
@@ -599,7 +576,7 @@ completeWriteFileAgain:
 	for i := 0; i < 10; i++ {
 		// test that write position and messages reset when a file is completely read
 		// test the writeFileNum correctly increments
-		d := readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
+		d = readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
 		if d.depth == 7 &&
 			d.readFileNum == 1 &&
 			d.writeFileNum == 3 &&
@@ -613,7 +590,7 @@ completeWriteFileAgain:
 		}
 		time.Sleep(100 * time.Millisecond)
 	}
-	panic("fail")
+	panic(fmt.Sprintf("%+v", d))
 
 completeReadFileAgain:
 	<-dq.ReadChan()
@@ -844,6 +821,7 @@ func numberOfBadFiles(diskQueueName string, dataPath string) int64 {
 
 func TestDiskSizeImplementationWithBadFiles(t *testing.T) {
 	// write three files
+	t.SkipNow()
 
 	l := NewTestLogger(t)
 	dqName := "test_disk_queue_implementation_with_bad_files" + strconv.Itoa(int(time.Now().Unix()))