From 40a3dce93cdd69fa1cd109ba260ad6989349386a Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 23 Mar 2020 18:04:08 -0500 Subject: [PATCH] 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)