Disk size limit

[kev1n80]

Implementation

• Get metaData file size
• Get accurate count of writeBytes whenever a readFile is deleted or a readFile is converted into a .bad file
• Get all bad file info
• Remove read file
• Check to see if there is enough space when writing data to file
  • If there is not, delete all .bad files before deleting readFile
• Abstract code from moveForward() to create moveToNextReadFile() which is also used in removeReadFile()

Testing

• Test that files are deleted when a new write to file will surpass disk size limit
• Test that if data size is greater than file size and the new write to file will surpass disk size limit, DiskQueue will delete multiple files
• Test that DiskQueue will delete .bad files before deleting readFile
• Test that DiskQueue is able to account for preexisting .bad files associated with DiskQueue object
• Test that DiskQueue is able to remove corrupted files when making disk space
• Test that DiskQueue is able to account for corrupted files when reading one and converting the corrupted read file into a bad file

Note

Half of the PR is code for testing

[CatherineF-dev]

Hi Kevin,
I also need a function d.TotalFileBytes() to calculate how many bytes under buffer directory folderPath. It also would be a metrics for disk queue, which we would monitor. Thanks a
lot!

(You could choose a better function name)

[CatherineF-dev]

If d.writeFileNum++ >= 1000000 or >= MAX_INT64, it might be an issue.
If the retention_size is not big enough, d.writeFileNum would increase quickly.
Since the d.writeFileNum is always increasing, it would be over MAX_INT64 in the future.

func (d *diskQueue) fileName(fileNum int64) string {
	return fmt.Sprintf(path.Join(d.dataPath, "%s.diskqueue.%06d.dat"), d.name, fileNum)
}

[kev1n80]

If d.writeFileNum++ >= 1000000 or >= MAX_INT64, it might be an issue.
If the retention_size is not big enough, d.writeFileNum would increase quickly.
Since the d.writeFileNum is always increasing, it would be over MAX_INT64 in the future.

func (d *diskQueue) fileName(fileNum int64) string {
	return fmt.Sprintf(path.Join(d.dataPath, "%s.diskqueue.%06d.dat"), d.name, fileNum)
}

Unfortunately, this seems like a limitation in NSQIO's design, which relies on writeFileNum to be greater than or equal to readFileNum. I was wondering if we could overcome this limitation by "resetting" DiskQueue once all of the data in the queue has been read.

[CatherineF-dev]

If d.readMessages is not sync well, d.readMessages is small and d.depth might be negative.
Could we add a check here?

    d.depth -= totalMessages - d.readMessages
	if d.depth < 0 {
		d.logf(ERROR, "DISKQUEUE(%s) d.depth = %d is negative", d.name, d.depth)
		d.depth = 0
	}

[kev1n80]

In that case, what if we only do that subtraction if totalMessages is greater than d.readMessages and less than d.depth.

if totalMessages > d.readMessages && totalMessages < d.depth {
	// update depth with the remaining number of messages
	d.depth -= totalMessages - d.readMessages
}

This way, we still have a rough estimate of what the total depth is. In the situation where the depth is 1,000 and a file has 100 messages. I think it would be better to have a rough estimate as to what the depth is rather than set it to 0. Plus, this is what happens when we encounter a bad file or a file that may have been truncated: we do not subtract depth by anything and proceed with an estimated number.

[kev1n80]

We have decided that there are two problems: 1) when the current depth is a negative number and 2) when totalMessages is a negative number. I think we should approach the second problem with my suggestion
and the first problem by adding your check when we create a new DiskQueue object.

However, we will hold off on making these changes until we find the real issue of the negative number.

[CatherineF-dev]

Good job!! Thanks Kevin!

I like the "moveToNextReadFile", which is shared by readOne() and removeReadFile().
Indeed, removeReadFile() is like readOne() without putting data to readChan.

[CatherineF-dev]

fmt.Println(fmt.Sprintf( "%s.diskqueue.%06d.dat", "1", 10000000000)) = 10000000000

Need to change how to read here https://github.com/kev1n80/go-diskqueue/blob/master/diskqueue.go#L189