Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: copy names from mmapped memory before closing iterator #22040

Merged
merged 3 commits into from
Aug 4, 2021
Merged

fix: copy names from mmapped memory before closing iterator #22040

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5ea7c0d
fix: copy names from mmapped memory before closing iterator
davidby-influx Aug 4, 2021
b6ccacc
chore: formatting
davidby-influx Aug 4, 2021
ed6039e
chore: refactor for clarity
davidby-influx Aug 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
- [#21978](https://github.com/influxdata/influxdb/pull/21978): fix: restore portable backup bug
- [#21992](https://github.com/influxdata/influxdb/pull/21992): fix: systemd-start script should be executable by group and others
- [#22019](https://github.com/influxdata/influxdb/pull/22019): fix: error instead of panic when enterprise tries to restore with OSS
- [#22040](https://github.com/influxdata/influxdb/pull/22040): fix: copy names from mmapped memory before closing iterator

v1.9.2 [unreleased]
- [#21631](https://github.com/influxdata/influxdb/pull/21631): fix: group by returns multiple results per group in some circumstances
Expand Down
190 changes: 160 additions & 30 deletions tsdb/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -938,6 +938,26 @@ type MeasurementIterator interface {
Next() ([]byte, error)
}

func ToSlice(itr MeasurementIterator) ([][]byte, error) {
if itr == nil {
return nil, nil
} else if iterator, ok := itr.(*measurementSliceIterator); ok {
return iterator.ToSlice(), nil
} else if iterator, ok := itr.(*fileMeasurementSliceIterator); ok {
return iterator.measurementSliceIterator.ToSlice(), nil
}
s := make([][]byte, 0)
for {
if value, err := itr.Next(); err != nil {
return nil, err
} else if value == nil {
return nil, nil
} else {
s = append(s, value)
}
}
}

type MeasurementIterators []MeasurementIterator

func (a MeasurementIterators) Close() (err error) {
Expand Down Expand Up @@ -968,6 +988,46 @@ func (itr *measurementSliceIterator) Next() (name []byte, err error) {
return name, nil
}

func (itr *measurementSliceIterator) ToSlice() [][]byte {
return itr.names
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Having both ToSlice and Next on the same object seems prone to accident...

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe if it was named RemainingAsSlice or something that suggests it is part of the iteration interface

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

UnderlyingSlice() is the new name

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, UnderlyingSlice still hints to me that the slice is the whole thing instead of eaten away at by Next, but ok.

}

// fileMeasurementSliceIterator is designed to allow a tag value slice
// iterator to use memory from a memory-mapped file, pinning it
// with the underlying file iterators
type fileMeasurementSliceIterator struct {
measurementSliceIterator
fileIterators MeasurementIterators
}

func (itr *fileMeasurementSliceIterator) Close() error {
e1 := itr.fileIterators.Close()
e2 := itr.measurementSliceIterator.Close()
if e1 != nil {
return e1
} else {
return e2
}
}

func newFileMeasurementSliceIterator(names [][]byte, itrs MeasurementIterators) *fileMeasurementSliceIterator {
return &fileMeasurementSliceIterator{
measurementSliceIterator: measurementSliceIterator{
names: names,
},
fileIterators: itrs,
}
}

func measurementIteratorToSliceIterator(iterator MeasurementIterator) ([][]byte, *fileMeasurementSliceIterator, error) {
s, err := ToSlice(iterator)
if err != nil {
return nil, nil, err
} else {
return s, newFileMeasurementSliceIterator(s, MeasurementIterators{iterator}), nil
}
}

// MergeMeasurementIterators returns an iterator that merges a set of iterators.
// Iterators that are first in the list take precedence and a deletion by those
// early iterators will invalidate elements by later iterators.
Expand Down Expand Up @@ -1320,17 +1380,29 @@ func (is IndexSet) DedupeInmemIndexes() IndexSet {

// MeasurementNamesByExpr returns a slice of measurement names matching the
// provided condition. If no condition is provided then all names are returned.
func (is IndexSet) MeasurementNamesByExpr(auth query.FineAuthorizer, expr influxql.Expr) ([][]byte, error) {
func (is IndexSet) MeasurementNamesByExpr(auth query.FineAuthorizer, expr influxql.Expr) (names [][]byte, err error) {
release := is.SeriesFile.Retain()
defer release()

// Return filtered list if expression exists.
if expr != nil {
names, err := is.measurementNamesByExpr(auth, expr)
itr, err := is.measurementNamesByExpr(auth, expr)
if err != nil {
return nil, err
} else if itr == nil {
return nil, nil
}
defer func() {
if e := itr.Close(); err == nil {
err = e
}
}()

if names, err := ToSlice(itr); err != nil {
return nil, err
} else {
return slices.CopyChunkedByteSlices(names, 1000), nil
}
return slices.CopyChunkedByteSlices(names, 1000), nil
}

itr, err := is.measurementIterator()
Expand All @@ -1339,10 +1411,13 @@ func (is IndexSet) MeasurementNamesByExpr(auth query.FineAuthorizer, expr influx
} else if itr == nil {
return nil, nil
}
defer itr.Close()
defer func() {
if e := itr.Close(); err == nil {
err = e
}
}()

// Iterate over all measurements if no condition exists.
var names [][]byte
for {
e, err := itr.Next()
if err != nil {
Expand All @@ -1360,7 +1435,7 @@ func (is IndexSet) MeasurementNamesByExpr(auth query.FineAuthorizer, expr influx
return slices.CopyChunkedByteSlices(names, 1000), nil
}

func (is IndexSet) measurementNamesByExpr(auth query.FineAuthorizer, expr influxql.Expr) ([][]byte, error) {
func (is IndexSet) measurementNamesByExpr(auth query.FineAuthorizer, expr influxql.Expr) (MeasurementIterator, error) {
if expr == nil {
return nil, nil
}
Expand Down Expand Up @@ -1400,20 +1475,35 @@ func (is IndexSet) measurementNamesByExpr(auth query.FineAuthorizer, expr influx
return is.measurementNamesByTagFilter(auth, e.Op, tag.Val, value, regex)

case influxql.OR, influxql.AND:
lhs, err := is.measurementNamesByExpr(auth, e.LHS)

exprToSlice := func(e influxql.Expr) (slice [][]byte, iterator *fileMeasurementSliceIterator, err error) {
mi, err := is.measurementNamesByExpr(auth, e)
if err != nil {
return nil, nil, err
}
slice, iterator, err = measurementIteratorToSliceIterator(mi)
if err != nil {
mi.Close()
return nil, nil, err
}
return
}

ls, lsfi, err := exprToSlice(e.LHS)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lsfi.ToSlice() gives ls right? Why name both?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because ToSlice() can fail on none-slice-based iterators.

if err != nil {
return nil, err
}

rhs, err := is.measurementNamesByExpr(auth, e.RHS)
rs, rsfi, err := exprToSlice(e.RHS)
if err != nil {
lsfi.Close()
return nil, err
}

mis := MeasurementIterators{lsfi, rsfi}
if e.Op == influxql.OR {
return bytesutil.Union(lhs, rhs), nil
return newFileMeasurementSliceIterator(bytesutil.Union(ls, rs), mis), nil
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similarly it might be clearer if this read like:

newFileMeasurementSliceIterator(bytesutil.Union(lsfi.UnsafeToSlice(), rsfi.UnsafeToSlice), MeasurementIterators{lsfi, rsfi})

Also would it be clearer to name MeasurementIterators as MeasurementIteratorClosers ?

}
return bytesutil.Intersect(lhs, rhs), nil
return newFileMeasurementSliceIterator(bytesutil.Intersect(ls, rs), mis), nil

default:
return nil, fmt.Errorf("invalid tag comparison operator")
Expand All @@ -1427,19 +1517,19 @@ func (is IndexSet) measurementNamesByExpr(auth query.FineAuthorizer, expr influx
}

// measurementNamesByNameFilter returns matching measurement names in sorted order.
func (is IndexSet) measurementNamesByNameFilter(auth query.FineAuthorizer, op influxql.Token, val string, regex *regexp.Regexp) ([][]byte, error) {
func (is IndexSet) measurementNamesByNameFilter(auth query.FineAuthorizer, op influxql.Token, val string, regex *regexp.Regexp) (MeasurementIterator, error) {
itr, err := is.measurementIterator()
if err != nil {
return nil, err
} else if itr == nil {
return nil, nil
}
defer itr.Close()

var names [][]byte
for {
e, err := itr.Next()
if err != nil {
itr.Close()
return nil, err
} else if e == nil {
break
Expand All @@ -1462,23 +1552,33 @@ func (is IndexSet) measurementNamesByNameFilter(auth query.FineAuthorizer, op in
}
}
bytesutil.Sort(names)
return names, nil
return newFileMeasurementSliceIterator(names, MeasurementIterators{itr}), nil
}

// MeasurementNamesByPredicate returns a slice of measurement names matching the
// provided condition. If no condition is provided then all names are returned.
// This behaves differently from MeasurementNamesByExpr because it will
// return measurements using flux predicates.
func (is IndexSet) MeasurementNamesByPredicate(auth query.FineAuthorizer, expr influxql.Expr) ([][]byte, error) {
func (is IndexSet) MeasurementNamesByPredicate(auth query.FineAuthorizer, expr influxql.Expr) (names [][]byte, err error) {
release := is.SeriesFile.Retain()
defer release()

// Return filtered list if expression exists.
if expr != nil {
names, err := is.measurementNamesByPredicate(auth, expr)
itr, err := is.measurementNamesByPredicate(auth, expr)
if err != nil {
return nil, err
}
if itr != nil {
defer func() {
if e := itr.Close(); err == nil {
err = e
}
}()
}
if names, err = ToSlice(itr); err != nil {
return nil, err
}
return slices.CopyChunkedByteSlices(names, 1000), nil
}

Expand All @@ -1488,10 +1588,13 @@ func (is IndexSet) MeasurementNamesByPredicate(auth query.FineAuthorizer, expr i
} else if itr == nil {
return nil, nil
}
defer itr.Close()
defer func() {
if e := itr.Close(); err == nil {
err = e
}
}()

// Iterate over all measurements if no condition exists.
var names [][]byte
for {
e, err := itr.Next()
if err != nil {
Expand All @@ -1509,7 +1612,7 @@ func (is IndexSet) MeasurementNamesByPredicate(auth query.FineAuthorizer, expr i
return slices.CopyChunkedByteSlices(names, 1000), nil
}

func (is IndexSet) measurementNamesByPredicate(auth query.FineAuthorizer, expr influxql.Expr) ([][]byte, error) {
func (is IndexSet) measurementNamesByPredicate(auth query.FineAuthorizer, expr influxql.Expr) (MeasurementIterator, error) {
if expr == nil {
return nil, nil
}
Expand Down Expand Up @@ -1549,20 +1652,35 @@ func (is IndexSet) measurementNamesByPredicate(auth query.FineAuthorizer, expr i
return is.measurementNamesByTagPredicate(auth, e.Op, tag.Val, value, regex)

case influxql.OR, influxql.AND:
lhs, err := is.measurementNamesByPredicate(auth, e.LHS)
predToSlice := func(e influxql.Expr) (slice [][]byte, iterator *fileMeasurementSliceIterator, err error) {
mi, err := is.measurementNamesByPredicate(auth, e)
if err != nil {
return nil, nil, err
}
slice, iterator, err = measurementIteratorToSliceIterator(mi)
if err != nil {
mi.Close()
return nil, nil, err
}
return
}

ls, litr, err := predToSlice(e.LHS)
if err != nil {
return nil, err
}

rhs, err := is.measurementNamesByPredicate(auth, e.RHS)
rs, ritr, err := predToSlice(e.RHS)
if err != nil {
litr.Close()
return nil, err
}

mis := MeasurementIterators{litr, ritr}

if e.Op == influxql.OR {
return bytesutil.Union(lhs, rhs), nil
return newFileMeasurementSliceIterator(bytesutil.Union(ls, rs), mis), nil
}
return bytesutil.Intersect(lhs, rhs), nil
return newFileMeasurementSliceIterator(bytesutil.Intersect(ls, rs), mis), nil

default:
return nil, fmt.Errorf("invalid tag comparison operator")
Expand All @@ -1575,16 +1693,21 @@ func (is IndexSet) measurementNamesByPredicate(auth query.FineAuthorizer, expr i
}
}

func (is IndexSet) measurementNamesByTagFilter(auth query.FineAuthorizer, op influxql.Token, key, val string, regex *regexp.Regexp) ([][]byte, error) {
func (is IndexSet) measurementNamesByTagFilter(auth query.FineAuthorizer, op influxql.Token, key, val string, regex *regexp.Regexp) (MeasurementIterator, error) {
var names [][]byte
failed := true

mitr, err := is.measurementIterator()
if err != nil {
return nil, err
} else if mitr == nil {
return nil, nil
}
defer mitr.Close()
defer func() {
if failed {
mitr.Close()
}
}()

// valEqual determines if the provided []byte is equal to the tag value
// to be filtered on.
Expand Down Expand Up @@ -1699,19 +1822,25 @@ func (is IndexSet) measurementNamesByTagFilter(auth query.FineAuthorizer, op inf
}

bytesutil.Sort(names)
return names, nil
failed = false
return newFileMeasurementSliceIterator(names, MeasurementIterators{mitr}), nil
}

func (is IndexSet) measurementNamesByTagPredicate(auth query.FineAuthorizer, op influxql.Token, key, val string, regex *regexp.Regexp) ([][]byte, error) {
func (is IndexSet) measurementNamesByTagPredicate(auth query.FineAuthorizer, op influxql.Token, key, val string, regex *regexp.Regexp) (MeasurementIterator, error) {
var names [][]byte
failed := true

mitr, err := is.measurementIterator()
if err != nil {
return nil, err
} else if mitr == nil {
return nil, nil
}
defer mitr.Close()
defer func() {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've recently seen a pattern using sync.Once to do this sort of thing, and I like it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

E.g.:

// early:
once sync.Once{}
defer once.Do(mitr.Close)

//later
var toReturn ...
once.Do(func(){
  toReturn = newFileMeasurementSliceIterator(names, MeasurementIterators{mitr}), nil
})
return toReturn

So you're guaranteed just one of either the close or the delegation of the close succeeds.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This comment doesn't need to be taken for the PR, just an observation.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that this pattern won't work here, because the initial defer once.Do(mitr.Close) will be called first, before the close on the toReturn. And that's what is causing the panic; a close on the iterator before the data is read out. But it's a nice pattern for other places.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm - I think it would work, here's a playground example. But not necessary here. https://play.golang.org/p/35qO7UPTOyG

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I see; Once works differently than a similar construct I used in another language; my mistake.

if failed {
mitr.Close()
}
}()

var checkMeasurement func(auth query.FineAuthorizer, me []byte) (bool, error)
switch op {
Expand Down Expand Up @@ -1762,7 +1891,8 @@ func (is IndexSet) measurementNamesByTagPredicate(auth query.FineAuthorizer, op
}

bytesutil.Sort(names)
return names, nil
failed = false
return newFileMeasurementSliceIterator(names, MeasurementIterators{mitr}), nil
}

// measurementAuthorizedSeries determines if the measurement contains a series
Expand Down