github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-11-21 14:44:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/persistentqueue: check that readerOffset doesnt exceed writerOffset after each readerOffset increase

Browse source 
This should help detecting the source of the panic from https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1981
This commit is contained in:
Aliaksandr Valialkin 2021-12-20 17:25:08 +02:00
parent ad388ecd78
commit 5a36e241f4
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
1 changed files with 14 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
lib/persistentqueue/persistentqueue.go
View file

@ -113,6 +113,9 @@ func (q *queue) mustResetFiles() {
// GetPendingBytes returns the number of pending bytes in the queue. // GetPendingBytes returns the number of pending bytes in the queue.
func (q *queue) GetPendingBytes() uint64 { func (q *queue) GetPendingBytes() uint64 {
if q.readerOffset > q.writerOffset {
logger.Panicf("BUG: readerOffset=%d cannot exceed writerOffset=%d", q.readerOffset, q.writerOffset)
}
n := q.writerOffset - q.readerOffset n := q.writerOffset - q.readerOffset
return n return n
} }
@ -555,6 +558,9 @@ func (q *queue) nextChunkFileForRead() error {
if n := q.readerOffset % q.chunkFileSize; n > 0 { if n := q.readerOffset % q.chunkFileSize; n > 0 {
q.readerOffset += q.chunkFileSize - n q.readerOffset += q.chunkFileSize - n
} }
if err := q.checkReaderWriterOffsets(); err != nil {
return err
}
q.readerLocalOffset = 0 q.readerLocalOffset = 0
q.readerPath = q.chunkFilePath(q.readerOffset) q.readerPath = q.chunkFilePath(q.readerOffset)
r, err := filestream.Open(q.readerPath, true) r, err := filestream.Open(q.readerPath, true)
@ -598,6 +604,14 @@ func (q *queue) readFull(buf []byte) error {
} }
q.readerLocalOffset += bufLen q.readerLocalOffset += bufLen
q.readerOffset += bufLen q.readerOffset += bufLen
return q.checkReaderWriterOffsets()
}
func (q *queue) checkReaderWriterOffsets() error {
if q.readerOffset > q.writerOffset {
return fmt.Errorf("readerOffset=%d cannot exceed writerOffset=%d; it is likely persistent queue files were corrupted on unclean shutdown",
q.readerOffset, q.writerOffset)
}
return nil return nil
} }