scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

track closing status for index/field/view, shut down cache flush early

Browse source 
This is a lot more complex than it sounds like it will be.

We shut down the cache flush when a holder is closed, but if you're
deleting an index, we don't check for that, and can have a cache flush
still creating cache files in an index which could conceivably result
in os.RemoteAll() failing. This shouldn't happen often, but it's happened
at least once.

To address this, first, we make sure that every tier of this operation
bails as quickly as it can after the thing it's working on closes. Second,
we retry RemoveAll.

Unfortunately, some things get reopened, so we have to handle that,
have mutexes covering the access to the channel, and so on. Also, some
things were getting double-closed, which was previously harmless but
could now fail. So, first, catch all the existing double-closes and
remove them, second, make the double-close fail with an error. Note
that virtually none of the tests check for errors on close.

This passes tests and should be unable to hit the original problem.
Unfortunately, it's unreasonably hard to check that, because it
requires an incredible coincidence of timing on the delete aligning
with a cache flush.
This commit is contained in:
Seebs 2022-03-17 12:35:37 -05:00
parent f7820e45f9
commit 32fec70816
13 changed files with 117 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cluster_internal_test.go
View file

@ -162,7 +162,6 @@ func TestFragSources(t *testing.T) {
c5.addNodeBasicSorted(node3)
idx := newIndexWithTempPath(t, "i")
defer idx.Close()
field, err := idx.CreateFieldIfNotExists("f", OptFieldTypeDefault())
if err != nil {

1
executor_internal_test.go
View file

@ -492,7 +492,6 @@ func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) {
func TestGetScaledInt(t *testing.T) {
f := OpenField(t, OptFieldTypeTimestamp(time.Now(), "ms"))
defer f.Close()
// check that fields with type timestamp return the int64 passed in to getScaledInt with nil err
v := time.Now().Unix()
res, err := getScaledInt(f.Field, v)

7
executor_test.go
View file

@ -59,8 +59,10 @@ func getTempDirString() (td *string) {
func TestExecutor(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
defer func() {
t.Logf("TestExecutor: closing cluster")
c.Close()
}()
// Ensure a row query can be executed.
t.Run("ExecuteRow", func(t *testing.T) {
t.Run("RowIDColumnID", func(t *testing.T) {
@ -1084,7 +1086,6 @@ func runCallTest(c *test.Cluster, t *testing.T, writeQuery string, readQueries [
if err != nil {
t.Fatal(err)
}
defer index.Close()
_, err = index.CreateField("f", fieldOption...)
if err != nil {
t.Fatal(err)

43
field.go
View file

@ -114,6 +114,9 @@ type Field struct {
// the remoteAvailableShards
availableShardChan chan struct{}
wg sync.WaitGroup
// track whether we're shutting down
closing chan struct{}
}
// FieldOption is a functional option type for pilosa.fieldOptions.
@ -528,6 +531,8 @@ func (f *Field) Options() FieldOptions {
// Open opens and initializes the field.
func (f *Field) Open() error {
f.mu.Lock()
defer f.mu.Unlock()
if err := func() (err error) {
// Ensure the field's path exists.
f.holder.Logger.Debugf("ensure field path exists: %s", f.path)
@ -570,9 +575,10 @@ func (f *Field) Open() error {
go f.writeAvailableShards()
return nil
}(); err != nil {
f.Close()
f.unprotectedClose()
return err
}
f.closing = make(chan struct{})
_ = testhook.Opened(f.holder.Auditor, f, nil)
f.holder.Logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name)
@ -846,6 +852,21 @@ func (f *Field) applyOptions(opt FieldOptions) error {
func (f *Field) Close() error {
f.mu.Lock()
defer f.mu.Unlock()
return f.unprotectedClose()
}
// unprotectedClose is the actual closing part of the operation, without the
// locking.
func (f *Field) unprotectedClose() error {
if f.closing != nil {
select {
case <-f.closing:
// already closed. prevent double-close
return errors.New("double close of field")
default:
}
close(f.closing)
}
defer func() {
_ = testhook.Closed(f.holder.Auditor, f, nil)
}()
@ -874,6 +895,23 @@ func (f *Field) Close() error {
return nil
}
func (f *Field) flushCaches() {
// look up the close channel so if we somehow end up living until the
// field gets reopened, we don't have a data race, but correctly detect
// that the old one is closed.
f.mu.RLock()
closing := f.closing
f.mu.RUnlock()
for _, v := range f.views() {
select {
case <-closing:
return
default:
v.flushCaches()
}
}
}
// Keys returns true if the field uses string keys.
func (f *Field) Keys() bool {
f.mu.RLock()
@ -905,9 +943,6 @@ func (f *Field) hasBSIGroup(name string) bool {
// createBSIGroup creates a new bsiGroup on the field.
func (f *Field) createBSIGroup(bsig *bsiGroup) error {
f.mu.Lock()
defer f.mu.Unlock()
// Append bsiGroup.
if err := bsig.validate(); err != nil {
return errors.Wrap(err, "validating bsigroup")

12
field_internal_test.go
View file

@ -184,7 +184,6 @@ func TestBSIGroup_BaseValue(t *testing.T) {
func TestField_ValCountize(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// check that you get an empty val count and err
// BSIGroupNotFound on nil bsig from
// f.bsiGroup(f.name)
@ -202,7 +201,6 @@ func TestField_ValCountize(t *testing.T) {
// Ensure field can open and retrieve a view.
func TestField_DeleteView(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
viewName := viewStandard + "_v"
@ -319,7 +317,6 @@ func (f *TestField) MustSetBit(tx Tx, row, col uint64, ts ...time.Time) {
// Ensure field can open and retrieve a view.
func TestField_CreateViewIfNotExists(t *testing.T) {
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// Create view.
view, err := f.createViewIfNotExists("v")
@ -344,7 +341,6 @@ func TestField_CreateViewIfNotExists(t *testing.T) {
func TestField_SetTimeQuantum(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
defer f.Close()
// Retrieve time quantum.
if q := f.TimeQuantum(); q != TimeQuantum("YMDH") {
@ -361,7 +357,6 @@ func TestField_SetTimeQuantum(t *testing.T) {
func TestField_RowTime(t *testing.T) {
f := OpenField(t, OptFieldTypeTime(TimeQuantum("YMDH"), "0"))
defer f.Close()
// Obtain transaction.
tx := f.idx.holder.txf.NewTx(Txo{Write: writable, Index: f.idx, Field: f.Field, Shard: 0})
@ -414,7 +409,6 @@ func TestField_RowTime(t *testing.T) {
func TestField_PersistAvailableShards(t *testing.T) {
availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// bm represents remote available shards.
bm := roaring.NewBitmap(1, 2, 3)
@ -501,7 +495,6 @@ func TestField_ApplyOptions(t *testing.T) {
// to result in a value of 9 instead of 1.
func TestBSIGroup_importValue(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -566,7 +559,6 @@ func BenchmarkField_ImportValue(b *testing.B) {
for _, bitDepth := range depths {
f := OpenField(b, OptFieldTypeInt(0, 1<<bitDepth))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -582,7 +574,6 @@ func BenchmarkField_ImportValue(b *testing.B) {
func TestIntField_MinMaxForShard(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -742,7 +733,6 @@ func TestDecimalField_MinMaxBoundaries(t *testing.T) {
func TestDecimalField_MinMaxForShard(t *testing.T) {
f := OpenField(t, OptFieldTypeDecimal(3))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -819,7 +809,6 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
func TestBSIGroup_TxReopenDB(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
@ -872,7 +861,6 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
// Ensure that an integer field has the same BitDepth after reopening.
func TestField_SaveMeta(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-10, 1000))
defer f.Close()
colID := uint64(1)
val := int64(88)

7
field_test.go
View file

@ -16,7 +16,6 @@ import (
func TestField_SetValue(t *testing.T) {
t.Run("OK", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
@ -52,7 +51,6 @@ func TestField_SetValue(t *testing.T) {
t.Run("Overwrite", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {
@ -88,7 +86,6 @@ func TestField_SetValue(t *testing.T) {
t.Run("ErrBSIGroupNotFound", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeDefault())
if err != nil {
@ -106,7 +103,6 @@ func TestField_SetValue(t *testing.T) {
t.Run("ErrBSIGroupValueTooLow", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(20, 30))
if err != nil {
@ -124,7 +120,6 @@ func TestField_SetValue(t *testing.T) {
t.Run("ErrBSIGroupValueTooHigh", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(20, 30))
if err != nil {
@ -197,7 +192,6 @@ const includeRemote = false // for calls to Index.AvailableShards(localOnly bool
// Ensure can update and delete available shards.
func TestField_AvailableShards(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("fld-shards", pilosa.OptFieldTypeDefault())
if err != nil {
@ -239,7 +233,6 @@ func TestField_AvailableShards(t *testing.T) {
func TestField_ClearValue(t *testing.T) {
t.Run("OK", func(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64))
if err != nil {

1
fragment_internal_test.go
View file

@ -1328,7 +1328,6 @@ func TestFragment_TopN_CacheSize(t *testing.T) {
// Create Index.
index := mustOpenIndex(t, IndexOptions{})
defer index.Close()
// Create field.
field, err := index.CreateFieldIfNotExists("f", OptFieldTypeSet(CacheTypeRanked, cacheSize))

30
holder.go
View file

@ -1068,7 +1068,16 @@ func (h *Holder) DeleteIndex(name string) error {
// Delete index directory.
if err := os.RemoveAll(h.IndexPath(name)); err != nil {
return errors.Wrap(err, "removing directory")
// There is a rare edge case here: If a cache flush was happening, RemoveAll
// can fail because a file gets created, say in a fragment directory, after
// RemoveAll has deleted everything it found in the directory, but before
// the actual directory is unlinked. In theory, though, this can't happen
// twice; by the time we get here, everything was closed, so at most one
// more file should get created.
err = os.RemoveAll(h.IndexPath(name))
if err != nil {
return errors.Wrap(err, "removing directory")
}
}
// Remove reference.
@ -1136,20 +1145,11 @@ func (h *Holder) monitorCacheFlush() {
func (h *Holder) flushCaches() {
for _, index := range h.Indexes() {
for _, field := range index.Fields() {
for _, view := range field.views() {
for _, fragment := range view.allFragments() {
select {
case <-h.closing:
return
default:
}
if err := fragment.FlushCache(); err != nil {
h.Logger.Errorf("flushing cache: err=%s, path=%s", err, fragment.cachePath())
}
}
}
select {
case <-h.closing:
return
default:
index.flushCaches()
}
}
}

3
holder_test.go
View file

@ -34,7 +34,8 @@ func TestHolder_Open(t *testing.T) {
t.Skip("Skipping permissions test since user is root.")
}
h := test.MustOpenHolder(t)
defer h.Close()
// no automatic close here, because we manually close this, and then
// *fail* to reopen it.
if _, err := h.CreateIndex("test", pilosa.IndexOptions{}); err != nil {
t.Fatal(err)

39
index.go
View file

@ -53,6 +53,9 @@ type Index struct {
// track the subset of shards available to our views
fieldView2shard *FieldView2Shards
// indicate that we're closing and should wrap up and not allow new actions
closing chan struct{}
}
// NewIndex returns an existing (but possibly empty) instance of
@ -180,11 +183,15 @@ func (i *Index) OpenWithSchema(idx *disco.Index) error {
// metadata for the index is not changed from its existing value, and fields are
// not validated against the schema as they are opened.
func (i *Index) open(idx *disco.Index) (err error) {
i.mu.Lock()
defer i.mu.Unlock()
// Ensure the path exists.
i.holder.Logger.Debugf("ensure index path exists: %s", i.FieldsPath())
if err := os.MkdirAll(i.FieldsPath(), 0777); err != nil {
return errors.Wrap(err, "creating directory")
}
i.closing = make(chan struct{})
// fmt.Printf("new channel %p for index %p\n", i.closing, i)
// we don't want to open *all* the views for each shard, since
// most are empty when we are doing time quantums. It slows
@ -239,9 +246,6 @@ func (i *Index) open(idx *disco.Index) (err error) {
mu.Lock()
defer mu.Unlock()
i.mu.Lock()
defer i.mu.Unlock()
i.translateStores[partitionID] = store
return nil
})
@ -337,9 +341,7 @@ func (i *Index) openField(mu *sync.Mutex, cfm *CreateFieldMessage, file string)
}
i.holder.Logger.Debugf("add field to index.fields: %s", file)
i.mu.Lock()
i.fields[fld.Name()] = fld
i.mu.Unlock()
return fld, nil
}
@ -399,6 +401,16 @@ func (i *Index) setFieldBitDepths() error {
func (i *Index) Close() error {
i.mu.Lock()
defer i.mu.Unlock()
// flag that we're trying to shut down
if i.closing != nil {
select {
case <-i.closing:
// already closed. prevent double-close
return errors.New("double close of index")
default:
}
close(i.closing)
}
defer func() {
_ = testhook.Closed(i.holder.Auditor, i, nil)
}()
@ -427,6 +439,23 @@ func (i *Index) Close() error {
return nil
}
func (i *Index) flushCaches() {
// look up the close channel so if we somehow end up living until the
// index gets reopened, we don't have a data race, but correctly detect
// that the old one is closed.
i.mu.RLock()
closing := i.closing
i.mu.RUnlock()
for _, field := range i.Fields() {
select {
case <-closing:
return
default:
field.flushCaches()
}
}
}
// make it clear what the Index.AvailableShards() calls are trying to obtain.
const includeRemote = false

17
index_test.go
View file

@ -24,7 +24,6 @@ const ShardWidth = pilosa.ShardWidth
// Ensure index can open and retrieve a field.
func TestIndex_CreateFieldIfNotExists(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field.
f, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeDefault())
@ -52,7 +51,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("TimeQuantum", func(t *testing.T) {
t.Run("Explicit", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field with explicit quantum.
f, err := index.CreateField("f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0"))
@ -68,7 +66,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("TimeQuantumNoStandardView", func(t *testing.T) {
t.Run("Explicit", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field with explicit quantum with no standard view
f, err := index.CreateField("f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"), "0", true))
@ -84,7 +81,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("BSIFields", func(t *testing.T) {
t.Run("Int", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field with schema and verify it exists.
if f, err := index.CreateField("f", pilosa.OptFieldTypeInt(-990, 1000)); err != nil {
@ -103,7 +99,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("Timestamp", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field with schema and verify it exists.
if f, err := index.CreateField("f", pilosa.OptFieldTypeTimestamp(pilosa.DefaultEpoch, pilosa.TimeUnitSeconds)); err != nil {
@ -125,7 +120,6 @@ func TestIndex_CreateField(t *testing.T) {
/*
t.Run("ErrRangeCacheAllowed", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
CacheType: pilosa.CacheTypeRanked,
@ -136,7 +130,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("BSIFieldsWithCacheTypeNone", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
CacheType: pilosa.CacheTypeNone,
CacheSize: uint32(5),
@ -147,7 +140,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("ErrFieldFieldsAllowed", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -160,7 +152,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("ErrFieldNameRequired", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -173,7 +164,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("ErrInvalidFieldType", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -186,7 +176,6 @@ func TestIndex_CreateField(t *testing.T) {
t.Run("ErrInvalidBSIGroupRange", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
if _, err := index.CreateField("f", pilosa.FieldOptions{
Fields: []*pilosa.Field{
@ -203,7 +192,6 @@ func TestIndex_CreateField(t *testing.T) {
// Don't allow an int field to be created with keys=true
t.Run("IntField", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
_, err := index.CreateField("f", pilosa.OptFieldTypeInt(-1, 1), pilosa.OptFieldKeys())
if errors.Cause(err) != pilosa.ErrIntFieldWithKeys {
@ -214,7 +202,6 @@ func TestIndex_CreateField(t *testing.T) {
// Don't allow a decimal field to be created with keys=true
t.Run("DecimalField", func(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
_, err := index.CreateField("f", pilosa.OptFieldTypeDecimal(1, pql.Decimal{Value: -1}, pql.Decimal{Value: 1}), pilosa.OptFieldKeys())
if errors.Cause(err) != pilosa.ErrDecimalFieldWithKeys {
@ -227,7 +214,6 @@ func TestIndex_CreateField(t *testing.T) {
// Ensure index can delete a field.
func TestIndex_DeleteField(t *testing.T) {
index := test.MustOpenIndex(t)
defer index.Close()
// Create field.
if _, err := index.CreateFieldIfNotExists("f", pilosa.OptFieldTypeDefault()); err != nil {
@ -280,13 +266,12 @@ func TestIndex_RecreateFieldOnRestart(t *testing.T) {
// create index
indexName := fmt.Sprintf("idx_%d", rand.Uint64())
holder := c.GetHolder(0)
index, err := holder.CreateIndex(indexName, pilosa.IndexOptions{
_, err := holder.CreateIndex(indexName, pilosa.IndexOptions{
Keys: false,
})
if err != nil {
t.Fatal(err)
}
defer index.Close()
// create field
fieldName := fmt.Sprintf("field_%d", rand.Uint64())

10
planner_test.go
View file

@ -19,7 +19,6 @@ func TestPlanner_Count(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer index.Close()
if _, err := index.CreateField("f", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
t.Fatal(err)
@ -164,7 +163,6 @@ func TestPlanner_Select(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer i0.Close()
if _, err := i0.CreateField("a", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
t.Fatal(err)
@ -176,7 +174,6 @@ func TestPlanner_Select(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer i1.Close()
if _, err := i1.CreateField("x", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
t.Fatal(err)
@ -301,7 +298,6 @@ func TestPlanner_GroupBy(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer i0.Close()
if _, err := i0.CreateField("x"); err != nil {
t.Fatal(err)
@ -423,7 +419,6 @@ func TestPlanner_InnerJoin(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer i0.Close()
if _, err := i0.CreateField("a", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
t.Fatal(err)
@ -433,7 +428,6 @@ func TestPlanner_InnerJoin(t *testing.T) {
if err != nil {
t.Fatal(err)
}
defer i1.Close()
if _, err := i1.CreateField("parentid", pilosa.OptFieldTypeInt(0, 1000)); err != nil {
t.Fatal(err)
@ -457,10 +451,10 @@ func TestPlanner_InnerJoin(t *testing.T) {
Query: `
Set(1, parentid=1)
Set(1, x=100)
Set(2, parentid=1)
Set(2, x=200)
Set(3, parentid=2)
Set(3, x=300)
`}); err != nil {

20
view.go
View file

@ -54,6 +54,8 @@ type view struct {
knownShards *roaring.Bitmap
knownShardsCopied uint32
closing chan struct{}
}
// newView returns a new instance of View.
@ -78,6 +80,8 @@ func newView(holder *Holder, path, index, field, name string, fieldOptions Field
broadcaster: NopBroadcaster,
stats: stats.NopStatsClient,
knownShards: roaring.NewSliceBitmap(),
closing: make(chan struct{}),
}
}
@ -227,6 +231,7 @@ var workQueue = make(chan struct{}, runtime.NumCPU()*2)
func (v *view) close() error {
v.mu.Lock()
defer v.mu.Unlock()
close(v.closing)
defer func() {
_ = testhook.Closed(v.holder.Auditor, v, nil)
}()
@ -259,6 +264,21 @@ fragLoop:
return err
}
func (v *view) flushCaches() {
// we don't have a lock/cache of the closing mutex here, because
// individual view objects never get reopened, just discarded and recreated.
for _, f := range v.allFragments() {
select {
case <-v.closing:
return
default:
if err := f.FlushCache(); err != nil {
v.holder.Logger.Errorf("flushing cache: err=%s, path=%s", err, f.cachePath())
}
}
}
}
// flags returns a set of flags for the underlying fragments.
func (v *view) flags() byte {
var flag byte