Skip to content

Commit

Permalink
fix(tsi1/partition/test): fix data race in test code (#25298)
Browse files Browse the repository at this point in the history 
* feat: add hook for optimizing series reads based on authorizer (#25207)

* fix(edge): Backporting #24613 in to 1.11

* fix(test): fixes typo from merge in test

* fix(testing): adds error checking to cleanups

---------

Co-authored-by: Geoffrey Wossum <[email protected]>
  • Loading branch information
@devanbenz @gwossum
devanbenz and gwossum authored Sep 10, 2024
1 parent 1e710f8 commit ee51e5c
Show file tree
Hide file tree
Showing 3 changed files with 83 additions and 62 deletions.
3 changes: 2 additions & 1 deletion flux/stdlib/universe/bare_aggregate_test.flux
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,8 @@ testcase bare_last {
)
result = csv.from(csv: input)
|> testing.load()
|> range(start: -100y)
|> range(start: 2015-01-01T00:00:00Z)
|> filter(fn: (r) => r._measurement == "pge_bill" and r._field == "bank")
|> last()
|> keep(columns: ["_time", "_value", "_field", "_measurement"])

Expand Down
107 changes: 54 additions & 53 deletions tsdb/index/tsi1/partition.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@ const ManifestFileName = "MANIFEST"

// Partition represents a collection of layered index files and WAL.
type Partition struct {
mu sync.RWMutex
// exported for tests
Mu sync.RWMutex
opened bool

sfile *tsdb.SeriesFile // series lookup file
Expand Down Expand Up @@ -150,8 +151,8 @@ var ErrIncompatibleVersion = errors.New("incompatible tsi1 index MANIFEST")

// Open opens the partition.
func (p *Partition) Open() (rErr error) {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()

p.closing = make(chan struct{})

Expand Down Expand Up @@ -339,8 +340,8 @@ func (p *Partition) buildSeriesSet() error {

// CurrentCompactionN returns the number of compactions currently running.
func (p *Partition) CurrentCompactionN() int {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return p.currentCompactionN
}

Expand All @@ -367,8 +368,8 @@ func (p *Partition) Close() error {
p.Wait()

// Lock index and close remaining
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()

var err error

Expand Down Expand Up @@ -402,8 +403,8 @@ func (p *Partition) SeriesFile() *tsdb.SeriesFile { return p.sfile }

// NextSequence returns the next file identifier.
func (p *Partition) NextSequence() int {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()
return p.nextSequence()
}

Expand Down Expand Up @@ -445,8 +446,8 @@ func (p *Partition) manifest(newFileSet *FileSet) *Manifest {

// SetManifestPathForTest is only to force a bad path in testing
func (p *Partition) SetManifestPathForTest(path string) {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()
p.manifestPathFn = func() string { return path }
}

Expand All @@ -457,16 +458,16 @@ func (p *Partition) WithLogger(logger *zap.Logger) {

// SetFieldSet sets a shared field set from the engine.
func (p *Partition) SetFieldSet(fs *tsdb.MeasurementFieldSet) {
p.mu.Lock()
p.Mu.Lock()
p.fieldset = fs
p.mu.Unlock()
p.Mu.Unlock()
}

// FieldSet returns the fieldset.
func (p *Partition) FieldSet() *tsdb.MeasurementFieldSet {
p.mu.Lock()
p.Mu.Lock()
fs := p.fieldset
p.mu.Unlock()
p.Mu.Unlock()
return fs
}

Expand All @@ -476,8 +477,8 @@ func (p *Partition) RetainFileSet() (*FileSet, error) {
case <-p.closing:
return nil, tsdb.ErrIndexClosing
default:
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return p.retainFileSet(), nil
}
}
Expand All @@ -490,8 +491,8 @@ func (p *Partition) retainFileSet() *FileSet {

// FileN returns the active files in the file set.
func (p *Partition) FileN() int {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return len(p.fileSet.files)
}

Expand Down Expand Up @@ -671,12 +672,12 @@ func (p *Partition) DropMeasurement(name []byte) error {
// Mark measurement as deleted.
entries = append(entries, LogEntry{Flag: LogEntryMeasurementTombstoneFlag, Name: name})

p.mu.RLock()
p.Mu.RLock()
if err := p.activeLogFile.Writes(entries); err != nil {
p.mu.RUnlock()
p.Mu.RUnlock()
return err
}
p.mu.RUnlock()
p.Mu.RUnlock()

// Check if the log file needs to be swapped.
if err := p.CheckLogFile(); err != nil {
Expand Down Expand Up @@ -704,14 +705,14 @@ func (p *Partition) createSeriesListIfNotExists(names [][]byte, tagsSlice []mode
defer fs.Release()

// Ensure fileset cannot change during insert.
p.mu.RLock()
p.Mu.RLock()
// Insert series into log file.
ids, err := p.activeLogFile.AddSeriesList(p.seriesIDSet, names, tagsSlice, tracker)
if err != nil {
p.mu.RUnlock()
p.Mu.RUnlock()
return nil, err
}
p.mu.RUnlock()
p.Mu.RUnlock()

if err := p.CheckLogFile(); err != nil {
return nil, err
Expand All @@ -722,8 +723,8 @@ func (p *Partition) createSeriesListIfNotExists(names [][]byte, tagsSlice []mode
func (p *Partition) DropSeries(seriesID uint64) error {
// Delete series from index.
if err := func() error {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return p.activeLogFile.DeleteSeriesID(seriesID)
}(); err != nil {
return err
Expand All @@ -742,8 +743,8 @@ func (p *Partition) DropSeriesList(seriesIDs []uint64) error {

// Delete series from index.
if err := func() error {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return p.activeLogFile.DeleteSeriesIDList(seriesIDs)
}(); err != nil {
return err
Expand Down Expand Up @@ -910,14 +911,14 @@ func (p *Partition) AssignShard(k string, shardID uint64) {}

// Compact requests a compaction of log files.
func (p *Partition) Compact() {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()
p.compact()
}

func (p *Partition) DisableCompactions() {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()
p.compactionsDisabled++

select {
Expand All @@ -933,8 +934,8 @@ func (p *Partition) DisableCompactions() {
}

func (p *Partition) EnableCompactions() {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()

// Already enabled?
if p.compactionsEnabled() {
Expand All @@ -952,9 +953,9 @@ func (p *Partition) runPeriodicCompaction() {
p.Compact()

// Avoid a race when using Reopen in tests
p.mu.RLock()
p.Mu.RLock()
closing := p.closing
p.mu.RUnlock()
p.Mu.RUnlock()

// check for compactions once an hour (usually not necessary but a nice safety check)
t := time.NewTicker(1 * time.Hour)
Expand All @@ -977,8 +978,8 @@ func (p *Partition) runPeriodicCompaction() {
// If checkRunning = true, only count as needing a compaction if there is not a compaction already
// in progress for the level that would be compacted
func (p *Partition) NeedsCompaction(checkRunning bool) bool {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
if p.needsLogCompaction() {
return true
}
Expand Down Expand Up @@ -1031,10 +1032,10 @@ func (p *Partition) compact() {
p.currentCompactionN++
go func() {
p.compactLogFile(logFile)
p.mu.Lock()
p.Mu.Lock()
p.currentCompactionN--
p.levelCompacting[0] = false
p.mu.Unlock()
p.Mu.Unlock()
p.Compact()
}()
}
Expand Down Expand Up @@ -1074,10 +1075,10 @@ func (p *Partition) compact() {
p.compactToLevel(files, level+1, interrupt)

// Ensure compaction lock for the level is released.
p.mu.Lock()
p.Mu.Lock()
p.levelCompacting[level] = false
p.currentCompactionN--
p.mu.Unlock()
p.Mu.Unlock()

// Check for new compactions
p.Compact()
Expand Down Expand Up @@ -1155,8 +1156,8 @@ func (p *Partition) compactToLevel(files []*IndexFile, level int, interrupt <-ch

// Obtain lock to swap in index file and write manifest.
if err := func() (rErr error) {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()

// Replace previous files with new index file.
newFileSet := p.fileSet.MustReplace(IndexFiles(files).Files(), file)
Expand Down Expand Up @@ -1220,17 +1221,17 @@ func (p *Partition) needsLogCompaction() bool {
func (p *Partition) CheckLogFile() error {
// Check log file under read lock.
needsCompaction := func() bool {
p.mu.RLock()
defer p.mu.RUnlock()
p.Mu.RLock()
defer p.Mu.RUnlock()
return p.needsLogCompaction()
}()
if !needsCompaction {
return nil
}

// If file size exceeded then recheck under write lock and swap files.
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()
return p.checkLogFile()
}

Expand Down Expand Up @@ -1260,9 +1261,9 @@ func (p *Partition) compactLogFile(logFile *LogFile) {
return
}

p.mu.Lock()
p.Mu.Lock()
interrupt := p.compactionInterrupt
p.mu.Unlock()
p.Mu.Unlock()

start := time.Now()

Expand Down Expand Up @@ -1312,8 +1313,8 @@ func (p *Partition) compactLogFile(logFile *LogFile) {

// Obtain lock to swap in index file and write manifest.
if err := func() (rErr error) {
p.mu.Lock()
defer p.mu.Unlock()
p.Mu.Lock()
defer p.Mu.Unlock()

// Replace previous log file with index file.
newFileSet := p.fileSet.MustReplace([]File{logFile}, file)
Expand Down
35 changes: 27 additions & 8 deletions tsdb/index/tsi1/partition_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,19 @@ func TestPartition_Open(t *testing.T) {
func TestPartition_Manifest(t *testing.T) {
t.Run("current MANIFEST", func(t *testing.T) {
sfile := MustOpenSeriesFile()
defer sfile.Close()
t.Cleanup(func() {
if err := sfile.Close(); err != nil {
t.Fatalf("error closing series file %v", err)
}
})

p := MustOpenPartition(sfile.SeriesFile)
t.Cleanup(func() {
if err := p.Close(); err != nil {
t.Fatalf("error closing partition %v", err)
}
})

if got, exp := p.Manifest().Version, tsi1.Version; got != exp {
t.Fatalf("got MANIFEST version %d, expected %d", got, exp)
}
Expand All @@ -98,14 +108,17 @@ func TestPartition_Manifest_Write_Fail(t *testing.T) {
func TestPartition_PrependLogFile_Write_Fail(t *testing.T) {
t.Run("write MANIFEST", func(t *testing.T) {
sfile := MustOpenSeriesFile()
defer sfile.Close()

t.Cleanup(func() {
if err := sfile.Close(); err != nil {
t.Fatalf("error closing series file %v", err)
}
})
p := MustOpenPartition(sfile.SeriesFile)
defer func() {
t.Cleanup(func() {
if err := p.Close(); err != nil {
t.Fatalf("error closing partition: %v", err)
}
}()
})
p.Partition.MaxLogFileSize = -1
fileN := p.FileN()
p.CheckLogFile()
Expand All @@ -124,15 +137,21 @@ func TestPartition_PrependLogFile_Write_Fail(t *testing.T) {
func TestPartition_Compact_Write_Fail(t *testing.T) {
t.Run("write MANIFEST", func(t *testing.T) {
sfile := MustOpenSeriesFile()
defer sfile.Close()
t.Cleanup(func() {
if err := sfile.Close(); err != nil {
t.Fatalf("error closing series file %v", err)
}
})

p := MustOpenPartition(sfile.SeriesFile)
defer func() {
t.Cleanup(func() {
if err := p.Close(); err != nil {
t.Fatalf("error closing partition: %v", err)
}
}()
})
p.Partition.Mu.Lock()
p.Partition.MaxLogFileSize = -1
p.Partition.Mu.Unlock()
fileN := p.FileN()
p.Compact()
if (1 + fileN) != p.FileN() {
Expand Down

0 comments on commit ee51e5c

Please sign in to comment.