From bb04f7f6aca4bba101f0e0aaa60c4be3de383e32 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 19 Mar 2020 14:44:14 -0500 Subject: [PATCH] limit frequency of writes for available shards --- field.go | 143 +++++++++++++++++++++++++++++++++------- field_internal_test.go | 11 +++- holder_internal_test.go | 4 ++ utils_internal_test.go | 1 + view.go | 19 ++++-- view_internal_test.go | 5 +- 6 files changed, 150 insertions(+), 33 deletions(-) diff --git a/field.go b/field.go index 811786def..49bd5e6d9 100644 --- a/field.go +++ b/field.go @@ -15,11 +15,12 @@ package pilosa import ( - "bufio" + "bytes" "context" "encoding/json" "fmt" "io/ioutil" + "log" "math" "os" "path/filepath" @@ -63,6 +64,26 @@ const ( FieldTypeDecimal = "decimal" ) +type protected struct { + mu sync.Mutex + duration time.Duration +} + +func (p *protected) Set(d time.Duration) { + p.mu.Lock() + defer p.mu.Unlock() + p.duration = d +} +func (p *protected) Get() time.Duration { + p.mu.Lock() + defer p.mu.Unlock() + return p.duration +} + +var AvailableShardFileFlushDuration = &protected{ + duration: 5 * time.Second, +} + // Field represents a container for views. type Field struct { mu sync.RWMutex @@ -112,6 +133,12 @@ type Field struct { // based on a foreign index; this prevents having to // call holder.index.Keys() every time. usesKeys bool + + // Synchronization primitives needed for async writing of + // the remoteAvailableShards + availableShardChan chan []byte + doneChan chan struct{} + wg sync.WaitGroup } // FieldOption is a functional option type for pilosa.fieldOptions. @@ -348,6 +375,8 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) { OpenTranslateStore: OpenInMemTranslateStore, } + f.options.ContainsShard = f.containsShard //used for notification optimization + return f, nil } @@ -385,6 +414,12 @@ func (f *Field) AvailableShards() *roaring.Bitmap { return b } +func (f *Field) containsShard(shard uint64) bool { + f.mu.RLock() + defer f.mu.RUnlock() + return f.remoteAvailableShards.Contains(shard) +} + // AddRemoteAvailableShards merges the set of available shards into the current known set // and saves the set to a file. func (f *Field) AddRemoteAvailableShards(b *roaring.Bitmap) error { @@ -441,29 +476,11 @@ func (f *Field) saveAvailableShards() error { } func (f *Field) unprotectedSaveAvailableShards() error { - path := filepath.Join(f.path, ".available.shards") - // Create a temporary file to save to. - tempPath := path + tempExt - - // Open or create file. - file, err := os.OpenFile(tempPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666) - if err != nil { - return errors.Wrap(err, "opening temporary available shards file") + var buf bytes.Buffer + if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil { + return errors.Wrap(err, "rendering available shards ") } - defer file.Close() - - // Write available shards to file. - bw := bufio.NewWriter(file) - if _, err = f.remoteAvailableShards.WriteTo(bw); err != nil { - return errors.Wrap(err, "writing bitmap to buffer") - } - bw.Flush() - - // Move snapshot to data file location. - if err := os.Rename(tempPath, path); err != nil { - return fmt.Errorf("rename snapshot: %s", err) - } - + f.availableShardChan <- buf.Bytes() return nil } @@ -578,7 +595,10 @@ func (f *Field) Open() error { return errors.Wrap(err, "checking foreign index") } } - + f.availableShardChan = make(chan []byte, 1) + f.doneChan = make(chan struct{}) + f.wg.Add(1) + go f.writeAvailableShards() return nil }(); err != nil { f.Close() @@ -588,6 +608,65 @@ func (f *Field) Open() error { f.logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name) return nil } +func saveIt(fieldPath string, availableShardBytes []byte) { + path := filepath.Join(fieldPath, ".available.shards") + // Create a temporary file to save to. + tempPath := path + tempExt + err := ioutil.WriteFile(tempPath, availableShardBytes, 0666) + if err != nil { + log.Println("failed to write ", tempPath) + return + } + + // Move snapshot to data file location. + if err := os.Rename(tempPath, path); err != nil { + log.Printf("rename snapshot: %s", err) + } + +} +func saveItAsync(fieldPath string, availableShardBytes []byte, done chan bool) { + if len(availableShardBytes) == 0 { + return + } + go func() { + saveIt(fieldPath, availableShardBytes) + done <- true + }() +} + +func (f *Field) writeAvailableShards() { + defer f.wg.Done() + ticker := time.NewTicker(AvailableShardFileFlushDuration.Get()) + var data []byte + tracker := make(chan bool) + writing := false + + for alive := true; alive; { + select { + case newdata := <-f.availableShardChan: + data = newdata + case <-ticker.C: + if len(data) > 0 { + if !writing { + writing = true + saveItAsync(f.path, data, tracker) + data = nil + } + } + case <-tracker: + writing = false + case <-f.doneChan: + if writing { + <-tracker + } + if len(data) > 0 { + saveIt(f.path, data) + } + alive = false + } + } + ticker.Stop() +} // applyTranslateStore opens the configured translate store. func (f *Field) applyTranslateStore() error { @@ -887,7 +966,15 @@ func (f *Field) applyOptions(opt FieldOptions) error { func (f *Field) Close() error { f.mu.Lock() defer f.mu.Unlock() - + // Shutdown the available shards writer + if f.doneChan != nil { + f.doneChan <- struct{}{} + f.wg.Wait() + close(f.availableShardChan) + close(f.doneChan) + f.availableShardChan = nil + f.doneChan = nil + } // Close the attribute store. if f.rowAttrStore != nil { _ = f.rowAttrStore.Close() @@ -1918,12 +2005,16 @@ type FieldOptions struct { Type string `json:"type,omitempty"` TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"` ForeignIndex string `json:"foreignIndex"` + ContainsShard func(uint64) bool } // newFieldOptions returns a new instance of FieldOptions // with applied and validated functional options. func newFieldOptions(opts ...FieldOption) (*FieldOptions, error) { fo := FieldOptions{} + fo.ContainsShard = func(uint64) bool { + return false + } for _, opt := range opts { err := opt(&fo) if err != nil { @@ -1952,6 +2043,8 @@ func applyDefaultOptions(o *FieldOptions) *FieldOptions { o.CacheType = DefaultCacheType o.CacheSize = DefaultCacheSize } + o.ContainsShard = func(uint64) bool { return false } //used for shardnotify optimization + return o } diff --git a/field_internal_test.go b/field_internal_test.go index 507feac8b..688f1cda2 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -351,6 +351,7 @@ 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()) // bm represents remote available shards. @@ -359,6 +360,7 @@ func TestField_PersistAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } + time.Sleep(2 * AvailableShardFileFlushDuration.Get()) // Reload field and verify that shard data is persisted. if err := f.Reopen(); err != nil { @@ -370,6 +372,7 @@ func TestField_PersistAvailableShards(t *testing.T) { } func TestField_CorruptAvailableShards(t *testing.T) { + AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -378,6 +381,7 @@ func TestField_CorruptAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } + time.Sleep(2 * AvailableShardFileFlushDuration.Get()) path := filepath.Join(f.path, ".available.shards") @@ -400,6 +404,7 @@ func TestField_CorruptAvailableShards(t *testing.T) { } func TestField_TruncatedAvailableShards(t *testing.T) { + AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -408,6 +413,7 @@ func TestField_TruncatedAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } + time.Sleep(2 * AvailableShardFileFlushDuration.Get()) path := filepath.Join(f.path, ".available.shards") @@ -428,6 +434,7 @@ func TestField_TruncatedAvailableShards(t *testing.T) { // Ensure that persisting available shards having a smaller footprint (for example, // when going from a bitmap to a smaller, RLE representation) succeeds. func TestField_PersistAvailableShardsFootprint(t *testing.T) { + AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -442,12 +449,14 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } + time.Sleep(2 * AvailableShardFileFlushDuration.Get()) // Reload field and verify that shard data is persisted. if err := f.Reopen(); err != nil { t.Fatal(err) } else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), bm.Slice()) { - t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice()) + t.Fatalf("unexpected available shards (reopen). expected: %v, \n but got: %v", bm.Slice(), f.remoteAvailableShards.Slice()) + } bm1 := roaring.NewBitmap() diff --git a/holder_internal_test.go b/holder_internal_test.go index afee39c6c..ecc479be1 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -21,6 +21,7 @@ import ( "reflect" "strings" "testing" + "time" "github.com/pilosa/pilosa/v2/roaring" ) @@ -98,6 +99,7 @@ func TestHolder_Optn(t *testing.T) { if os.Geteuid() == 0 { t.Skip("Skipping permissions test since user is root.") } + AvailableShardFileFlushDuration.Set(100 * time.Millisecond) h := newHolder() defer h.Close() @@ -182,6 +184,7 @@ func TestHolder_Optn(t *testing.T) { // Ensure holder can clean up orphaned fragments. func TestHolderCleaner_CleanHolder(t *testing.T) { + AvailableShardFileFlushDuration.Set(100 * time.Millisecond) //shorten the default time to force a file write cluster := NewTestCluster(2) // Create a local holder. @@ -223,6 +226,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { if err != nil { t.Fatalf("adding remote shards: %v", err) } + time.Sleep(2 * AvailableShardFileFlushDuration.Get()) // Keep replication the same and ensure we get the expected results. cluster.ReplicaN = 2 diff --git a/utils_internal_test.go b/utils_internal_test.go index 8abfdeb91..9a52cfcd0 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -34,6 +34,7 @@ func NewTestCluster(n int) *cluster { panic(err) } + AvailableShardFileFlushDuration.Set(100 * time.Millisecond) c := newCluster() c.ReplicaN = 1 c.Hasher = NewTestModHasher() diff --git a/view.go b/view.go index 815e2a71e..b70aed022 100644 --- a/view.go +++ b/view.go @@ -60,6 +60,7 @@ type view struct { rowAttrStore AttrStore logger logger.Logger snapshotQueue snapshotQueue + shardPresent func(uint64) bool } // newView returns a new instance of View. @@ -76,9 +77,10 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { fragments: make(map[uint64]*fragment), - broadcaster: NopBroadcaster, - stats: stats.NopStatsClient, - logger: logger.NopLogger, + broadcaster: NopBroadcaster, + stats: stats.NopStatsClient, + logger: logger.NopLogger, + shardPresent: fieldOptions.ContainsShard, } } @@ -276,6 +278,15 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { frag.RowAttrStore = v.rowAttrStore v.fragments[shard] = frag + v.notifyIfNew(shard) + return frag, nil +} + +func (v *view) notifyIfNew(shard uint64) { + if v.shardPresent(shard) { + return + } + broadcastChan := make(chan struct{}) go func() { @@ -299,8 +310,6 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { case <-time.After(50 * time.Millisecond): v.logger.Debugf("broadcasting create shard took >50ms") } - - return frag, nil } func (v *view) newFragment(path string, shard uint64) *fragment { diff --git a/view_internal_test.go b/view_internal_test.go index bb50003d2..f4c79537e 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -30,8 +30,9 @@ func mustOpenView(index, field, name string) *view { } fo := FieldOptions{ - CacheType: DefaultCacheType, - CacheSize: DefaultCacheSize, + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, + ContainsShard: func(uint64) bool { return false }, } v := newView(path, index, field, name, fo)