From bb04f7f6aca4bba101f0e0aaa60c4be3de383e32 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 19 Mar 2020 14:44:14 -0500 Subject: [PATCH 1/8] 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) From 40a3dce93cdd69fa1cd109ba260ad6989349386a Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 23 Mar 2020 18:04:08 -0500 Subject: [PATCH 2/8] Co-authored-by: Travis Turner --- cluster.go | 3 ++- field.go | 22 +++++++++------------- holder.go | 2 +- server/cluster_test.go | 5 +++-- view.go | 5 +++-- view_internal_test.go | 5 ++--- 6 files changed, 20 insertions(+), 22 deletions(-) diff --git a/cluster.go b/cluster.go index 54f8795ab..c026066a5 100644 --- a/cluster.go +++ b/cluster.go @@ -1443,7 +1443,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { return errors.Wrap(err, "merging cluster status") } - c.logger.Printf("done MergeClusterStatus, start goroutine") + c.logger.Printf("done MergeClusterStatus, start goroutine (%s)", c.Node.ID) // The actual resizing runs in a goroutine because we don't want to block // the distribution of other ResizeInstructions to the rest of the cluster. @@ -1482,6 +1482,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { c.logger.Printf("local field not found: %s/%s", is.Name, fs.Name) continue } + fmt.Printf("field: %+v (%s)\n", fs.AvailableShards.Slice(), c.Node.ID) if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil { return errors.Wrap(err, "adding remote available shards") } diff --git a/field.go b/field.go index 49bd5e6d9..5208debd0 100644 --- a/field.go +++ b/field.go @@ -375,7 +375,6 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) { OpenTranslateStore: OpenInMemTranslateStore, } - f.options.ContainsShard = f.containsShard //used for notification optimization return f, nil } @@ -417,6 +416,8 @@ func (f *Field) AvailableShards() *roaring.Bitmap { func (f *Field) containsShard(shard uint64) bool { f.mu.RLock() defer f.mu.RUnlock() + fmt.Println("CONTAINS SHARD:", shard, f.Name()) + fmt.Println(f.remoteAvailableShards) return f.remoteAvailableShards.Contains(shard) } @@ -608,7 +609,7 @@ 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) { +func blockingWriteAvailableShards(fieldPath string, availableShardBytes []byte) { path := filepath.Join(fieldPath, ".available.shards") // Create a temporary file to save to. tempPath := path + tempExt @@ -624,12 +625,12 @@ func saveIt(fieldPath string, availableShardBytes []byte) { } } -func saveItAsync(fieldPath string, availableShardBytes []byte, done chan bool) { +func nonBlockingWriteAvailableShards(fieldPath string, availableShardBytes []byte, done chan bool) { if len(availableShardBytes) == 0 { return } go func() { - saveIt(fieldPath, availableShardBytes) + blockingWriteAvailableShards(fieldPath, availableShardBytes) done <- true }() } @@ -649,18 +650,18 @@ func (f *Field) writeAvailableShards() { if len(data) > 0 { if !writing { writing = true - saveItAsync(f.path, data, tracker) + nonBlockingWriteAvailableShards(f.path, data, tracker) data = nil } } case <-tracker: writing = false case <-f.doneChan: - if writing { + if writing { //wait to writing is complete <-tracker } if len(data) > 0 { - saveIt(f.path, data) + blockingWriteAvailableShards(f.path, data) } alive = false } @@ -1186,6 +1187,7 @@ func (f *Field) newView(path, name string) *view { view.rowAttrStore = f.rowAttrStore view.stats = f.Stats view.broadcaster = f.broadcaster + view.shardPresent = f.containsShard if f.snapshotQueue != nil { view.snapshotQueue = f.snapshotQueue } @@ -2005,16 +2007,12 @@ 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 { @@ -2043,8 +2041,6 @@ 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/holder.go b/holder.go index 987b8939d..bb4206dfa 100644 --- a/holder.go +++ b/holder.go @@ -1185,7 +1185,7 @@ func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) { // Find appropriate store. f := s.Holder.Field(entry.Index, entry.Field) if f == nil { - s.Holder.Logger.Printf("field not found: %q/%q", entry.Index, entry.Field) + s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field) return } diff --git a/server/cluster_test.go b/server/cluster_test.go index a991c9f7a..9ecb57437 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -138,7 +138,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) { } // Ensure that adding a node correctly resizes the cluster. -func TestClusterResize_AddNode(t *testing.T) { +func TestXClusterResize_AddNode(t *testing.T) { t.Run("NoData", func(t *testing.T) { clus := test.MustRunCluster(t, 2) defer clus.Close() @@ -206,7 +206,6 @@ func TestClusterResize_AddNode(t *testing.T) { `); err != nil { t.Fatal(err) } - // exp is the expected result for the Row queries that follow. exp := `{"results":[{"attrs":{},"columns":[1,1300000]}]}` + "\n" @@ -388,6 +387,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { t.Fatal(err) } else if res != exp { t.Fatalf("unexpected result: %s", res) + } else { + fmt.Println(res) } // Configure node1 diff --git a/view.go b/view.go index b70aed022..e4c34c308 100644 --- a/view.go +++ b/view.go @@ -80,7 +80,7 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view { broadcaster: NopBroadcaster, stats: stats.NopStatsClient, logger: logger.NopLogger, - shardPresent: fieldOptions.ContainsShard, + shardPresent: func(uint64) bool { return false }, } } @@ -283,10 +283,11 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { } func (v *view) notifyIfNew(shard uint64) { + fmt.Println("Present", shard) if v.shardPresent(shard) { return } - + fmt.Println("BROADCAST", shard) broadcastChan := make(chan struct{}) go func() { diff --git a/view_internal_test.go b/view_internal_test.go index f4c79537e..bb50003d2 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -30,9 +30,8 @@ func mustOpenView(index, field, name string) *view { } fo := FieldOptions{ - CacheType: DefaultCacheType, - CacheSize: DefaultCacheSize, - ContainsShard: func(uint64) bool { return false }, + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, } v := newView(path, index, field, name, fo) From af068e08e75dd5f95dcc40f8a633b859ce76b4db Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 24 Mar 2020 11:24:02 -0500 Subject: [PATCH 3/8] cleanup and comments --- field.go | 5 ++--- view.go | 24 +++++++++++------------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/field.go b/field.go index 5208debd0..64b3a886d 100644 --- a/field.go +++ b/field.go @@ -413,11 +413,10 @@ func (f *Field) AvailableShards() *roaring.Bitmap { return b } +// constainsShard is used for limiting unnecessary CreateShard broadcast func (f *Field) containsShard(shard uint64) bool { f.mu.RLock() defer f.mu.RUnlock() - fmt.Println("CONTAINS SHARD:", shard, f.Name()) - fmt.Println(f.remoteAvailableShards) return f.remoteAvailableShards.Contains(shard) } @@ -1187,7 +1186,7 @@ func (f *Field) newView(path, name string) *view { view.rowAttrStore = f.rowAttrStore view.stats = f.Stats view.broadcaster = f.broadcaster - view.shardPresent = f.containsShard + view.remoteShardPresent = f.containsShard if f.snapshotQueue != nil { view.snapshotQueue = f.snapshotQueue } diff --git a/view.go b/view.go index e4c34c308..65965b43e 100644 --- a/view.go +++ b/view.go @@ -55,12 +55,12 @@ type view struct { // Fragments by shard. fragments map[uint64]*fragment - broadcaster broadcaster - stats stats.StatsClient - rowAttrStore AttrStore - logger logger.Logger - snapshotQueue snapshotQueue - shardPresent func(uint64) bool + broadcaster broadcaster + stats stats.StatsClient + rowAttrStore AttrStore + logger logger.Logger + snapshotQueue snapshotQueue + remoteShardPresent func(uint64) bool } // newView returns a new instance of View. @@ -77,10 +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, - shardPresent: func(uint64) bool { return false }, + broadcaster: NopBroadcaster, + stats: stats.NopStatsClient, + logger: logger.NopLogger, + remoteShardPresent: func(uint64) bool { return false }, } } @@ -283,11 +283,9 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { } func (v *view) notifyIfNew(shard uint64) { - fmt.Println("Present", shard) - if v.shardPresent(shard) { + if v.remoteShardPresent(shard) { //checks the fields remoteShards bitmap to see if broadcast needed return } - fmt.Println("BROADCAST", shard) broadcastChan := make(chan struct{}) go func() { From 000ea90877da055852aea97d076ca2d539ff0ea9 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 24 Mar 2020 13:48:29 -0500 Subject: [PATCH 4/8] unbuffer channel --- field.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/field.go b/field.go index 64b3a886d..670071c6d 100644 --- a/field.go +++ b/field.go @@ -595,7 +595,7 @@ func (f *Field) Open() error { return errors.Wrap(err, "checking foreign index") } } - f.availableShardChan = make(chan []byte, 1) + f.availableShardChan = make(chan []byte) f.doneChan = make(chan struct{}) f.wg.Add(1) go f.writeAvailableShards() From 6bd81b87eb309739b146f7483850d68b3bc9a375 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 24 Mar 2020 13:55:11 -0500 Subject: [PATCH 5/8] unexport availableShardFileFlushDuration --- field.go | 4 ++-- field_internal_test.go | 16 ++++++++-------- holder_internal_test.go | 6 +++--- utils_internal_test.go | 2 +- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/field.go b/field.go index 670071c6d..899e6f399 100644 --- a/field.go +++ b/field.go @@ -80,7 +80,7 @@ func (p *protected) Get() time.Duration { return p.duration } -var AvailableShardFileFlushDuration = &protected{ +var availableShardFileFlushDuration = &protected{ duration: 5 * time.Second, } @@ -636,7 +636,7 @@ func nonBlockingWriteAvailableShards(fieldPath string, availableShardBytes []byt func (f *Field) writeAvailableShards() { defer f.wg.Done() - ticker := time.NewTicker(AvailableShardFileFlushDuration.Get()) + ticker := time.NewTicker(availableShardFileFlushDuration.Get()) var data []byte tracker := make(chan bool) writing := false diff --git a/field_internal_test.go b/field_internal_test.go index 688f1cda2..ee13a269c 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -351,7 +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 + availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -360,7 +360,7 @@ func TestField_PersistAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } - time.Sleep(2 * AvailableShardFileFlushDuration.Get()) + time.Sleep(2 * availableShardFileFlushDuration.Get()) // Reload field and verify that shard data is persisted. if err := f.Reopen(); err != nil { @@ -372,7 +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 + availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -381,7 +381,7 @@ func TestField_CorruptAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } - time.Sleep(2 * AvailableShardFileFlushDuration.Get()) + time.Sleep(2 * availableShardFileFlushDuration.Get()) path := filepath.Join(f.path, ".available.shards") @@ -404,7 +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 + availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -413,7 +413,7 @@ func TestField_TruncatedAvailableShards(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } - time.Sleep(2 * AvailableShardFileFlushDuration.Get()) + time.Sleep(2 * availableShardFileFlushDuration.Get()) path := filepath.Join(f.path, ".available.shards") @@ -434,7 +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 + availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) // bm represents remote available shards. @@ -449,7 +449,7 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) { if err := f.AddRemoteAvailableShards(bm); err != nil { t.Fatal(err) } - time.Sleep(2 * AvailableShardFileFlushDuration.Get()) + time.Sleep(2 * availableShardFileFlushDuration.Get()) // Reload field and verify that shard data is persisted. if err := f.Reopen(); err != nil { diff --git a/holder_internal_test.go b/holder_internal_test.go index ecc479be1..ccc52c5b0 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -99,7 +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) + availableShardFileFlushDuration.Set(100 * time.Millisecond) h := newHolder() defer h.Close() @@ -184,7 +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 + availableShardFileFlushDuration.Set(100 * time.Millisecond) //shorten the default time to force a file write cluster := NewTestCluster(2) // Create a local holder. @@ -226,7 +226,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) { if err != nil { t.Fatalf("adding remote shards: %v", err) } - time.Sleep(2 * AvailableShardFileFlushDuration.Get()) + 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 9a52cfcd0..83751f4bf 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -34,7 +34,7 @@ func NewTestCluster(n int) *cluster { panic(err) } - AvailableShardFileFlushDuration.Set(100 * time.Millisecond) + availableShardFileFlushDuration.Set(100 * time.Millisecond) c := newCluster() c.ReplicaN = 1 c.Hasher = NewTestModHasher() From 0e2bb550db9cf4146cc1199b2d2834dbea10031b Mon Sep 17 00:00:00 2001 From: Travis Date: Tue, 24 Mar 2020 21:08:47 -0500 Subject: [PATCH 6/8] add deleted (rebalanced) shards to remoteAvailableShards --- holder.go | 15 +++++++++++++++ server/cluster_test.go | 4 +--- view.go | 4 ++-- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/holder.go b/holder.go index bb4206dfa..3b9631761 100644 --- a/holder.go +++ b/holder.go @@ -1233,6 +1233,15 @@ func (c *holderCleaner) CleanHolder() error { // Get the fragments registered in memory. for _, field := range index.Fields() { + // deletedShards is used to track which shards for the field + // were deleted. Any shards that get deleted from this node + // get added to remoteAvailableShards. This is done because + // the CleanHolder process is cleaning up shards which got + // moved to other nodes. Because those shards still exist + // (just no longer on this particular node), this node still + // needs to consider each of them as an available shard in + // the cluster. + var deletedShards []uint64 for _, view := range field.views() { for _, fragment := range view.allFragments() { fragShard := fragment.shard @@ -1244,6 +1253,12 @@ func (c *holderCleaner) CleanHolder() error { if err := view.deleteFragment(fragShard); err != nil { return errors.Wrap(err, "deleting fragment") } + deletedShards = append(deletedShards, fragShard) + } + } + if len(deletedShards) > 0 { + if err := field.AddRemoteAvailableShards(roaring.NewBitmap(deletedShards...)); err != nil { + return errors.Wrap(err, "adding remote available shards") } } } diff --git a/server/cluster_test.go b/server/cluster_test.go index 9ecb57437..0899b8e97 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -138,7 +138,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) { } // Ensure that adding a node correctly resizes the cluster. -func TestXClusterResize_AddNode(t *testing.T) { +func TestClusterResize_AddNode(t *testing.T) { t.Run("NoData", func(t *testing.T) { clus := test.MustRunCluster(t, 2) defer clus.Close() @@ -387,8 +387,6 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) { t.Fatal(err) } else if res != exp { t.Fatalf("unexpected result: %s", res) - } else { - fmt.Println(res) } // Configure node1 diff --git a/view.go b/view.go index 65965b43e..06d38dce8 100644 --- a/view.go +++ b/view.go @@ -278,11 +278,11 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) { frag.RowAttrStore = v.rowAttrStore v.fragments[shard] = frag - v.notifyIfNew(shard) + v.notifyIfNewShard(shard) return frag, nil } -func (v *view) notifyIfNew(shard uint64) { +func (v *view) notifyIfNewShard(shard uint64) { if v.remoteShardPresent(shard) { //checks the fields remoteShards bitmap to see if broadcast needed return } From 5470753cb57e387c220e79790aa6e7da4cafc53d Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Sun, 5 Apr 2020 18:10:10 -0500 Subject: [PATCH 7/8] fix conflict --- http/client.go | 2 ++ http/handler.go | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/http/client.go b/http/client.go index c688e5565..a8a5106d6 100644 --- a/http/client.go +++ b/http/client.go @@ -1115,6 +1115,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [ req.Header.Set("Content-Type", "application/x-protobuf") req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) req.Header.Set("Accept", "application/json") + req.Header.Set("Connection", "keep-alive") // Execute request. resp, err := c.executeRequest(req.WithContext(ctx)) @@ -1232,6 +1233,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI, // is closed. func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, error) { tracing.GlobalTracer.InjectHTTPHeaders(req) + req.Close = false resp, err := c.httpClient.Do(req) if err != nil { if resp != nil { diff --git a/http/handler.go b/http/handler.go index 584791d75..68bb384ce 100644 --- a/http/handler.go +++ b/http/handler.go @@ -1590,7 +1590,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) return } - + defer r.Body.Close() err := h.api.ClusterMessage(r.Context(), r.Body) if err != nil { // TODO this was the previous behavior, but perhaps not everything is a bad request From 6712f8cf0673c96bb074f54817779ba909dbcf51 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 6 Apr 2020 08:43:30 -0500 Subject: [PATCH 8/8] removed debug log --- cluster.go | 1 - 1 file changed, 1 deletion(-) diff --git a/cluster.go b/cluster.go index c026066a5..794b8935c 100644 --- a/cluster.go +++ b/cluster.go @@ -1482,7 +1482,6 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { c.logger.Printf("local field not found: %s/%s", is.Name, fs.Name) continue } - fmt.Printf("field: %+v (%s)\n", fs.AvailableShards.Slice(), c.Node.ID) if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil { return errors.Wrap(err, "adding remote available shards") }