From 32fec70816cc3942e93dffe2bea6a1fbff21f325 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 17 Mar 2022 12:35:37 -0500 Subject: [PATCH] track closing status for index/field/view, shut down cache flush early 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. --- cluster_internal_test.go | 1 - executor_internal_test.go | 1 - executor_test.go | 7 ++++--- field.go | 43 +++++++++++++++++++++++++++++++++++---- field_internal_test.go | 12 ----------- field_test.go | 7 ------- fragment_internal_test.go | 1 - holder.go | 30 +++++++++++++-------------- holder_test.go | 3 ++- index.go | 39 ++++++++++++++++++++++++++++++----- index_test.go | 17 +--------------- planner_test.go | 10 ++------- view.go | 20 ++++++++++++++++++ 13 files changed, 117 insertions(+), 74 deletions(-) diff --git a/cluster_internal_test.go b/cluster_internal_test.go index e2261a41f..f6231ae3b 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -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 { diff --git a/executor_internal_test.go b/executor_internal_test.go index 0fa10f44e..719c7f3d5 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -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) diff --git a/executor_test.go b/executor_test.go index 8064a2e5d..6f0e8839d 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) diff --git a/field.go b/field.go index 228f18c8a..c61093569 100644 --- a/field.go +++ b/field.go @@ -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") diff --git a/field_internal_test.go b/field_internal_test.go index 5f0f7d013..fb2f01060 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -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<