Merge pull request #145 from travisturner/translate-partition-better-fix

add translationSyncer interface

cluster.go

@@ -224,6 +224,8 @@ type cluster struct { // nolint: maligned
 
 	abortAntiEntropyCh chan struct{}
 
+	translationSyncer translationSyncer
+
 	mu         sync.RWMutex
 	jobs       map[int64]*resizeJob
 	currentJob *resizeJob
@@ -251,6 +253,8 @@ func newCluster() *cluster {
 		closing:             make(chan struct{}),
 		joining:             make(chan struct{}),
 
+		translationSyncer: NopTranslationSyncer,
+
 		InternalClient: newNopInternalClient(),
 
 		logger: logger.NopLogger,
@@ -463,7 +467,7 @@ func (c *cluster) unprotectedSetState(state string) {
 
 	switch state {
 	case ClusterStateNormal, ClusterStateDegraded:
-		// If state is RESIZING -> NORMAL then run cleanup.
+		// If state is RESIZING -> [NORMAL, DEGRADED] then run cleanup.
 		if c.state == ClusterStateResizing {
 			doCleanup = true
 		}
@@ -471,7 +475,17 @@ func (c *cluster) unprotectedSetState(state string) {
 
 	c.state = state
 
-	if state == ClusterStateResizing {
+	switch state {
+	case ClusterStateNormal:
+		// Because the cluster state is changing to NORMAL,
+		// we [potentially] need to reset the translation sync.
+		// If, for example, the cluster has changed size and is
+		// now settling to NORMAL, the partition ownership may
+		// have changed, and this will force that to be recalculated.
+		if err := c.translationSyncer.Reset(); err != nil {
+			c.logger.Printf("error resetting translation syncer: %s", err)
+		}
+	case ClusterStateResizing:
 		c.abortAntiEntropy()
 	}
 

go.sum

@@ -105,6 +105,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU=
 github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
+github.com/pilosa/pilosa v1.4.0 h1:nqHNIK4nDslFnem3yDp9R+6TgLdlkY9WdJD88Z83T8U=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

holder.go

@@ -85,6 +85,8 @@ type Holder struct {
 	OpenTranslateStore  OpenTranslateStoreFunc
 	OpenTranslateReader OpenTranslateReaderFunc
 
+	translationSyncer translationSyncer
+
 	// Queue of fields (having a foreign index) which have
 	// opened before their foreign index has opened.
 	foreignIndexFields []*Field
@@ -140,6 +142,8 @@ func NewHolder(partitionN int) *Holder {
 
 		OpenTranslateStore: OpenInMemTranslateStore,
 
+		translationSyncer: NopTranslationSyncer,
+
 		Logger: logger.NopLogger,
 	}
 }
@@ -228,7 +232,10 @@ func (h *Holder) Open() error {
 	h.snapshotQueue.ScanHolder(h)
 
 	h.opened.Close()
-	return nil
+
+	// Since the holder is just opening, this is not so much a reset,
+	// as it is an initial setting of the transation sync process.
+	return h.translationSyncer.Reset()
 }
 
 // checkForeignIndex is a check before applying a foreign
@@ -484,6 +491,12 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
 	// Update options.
 	h.indexes[index.Name()] = index
 
+	// Since this is a new index, we need to kick off
+	// its translation sync.
+	if err := h.translationSyncer.Reset(); err != nil {
+		return nil, errors.Wrap(err, "resetting translation sync")
+	}
+
 	return index, nil
 }
 
@@ -499,6 +512,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
 	index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
 	index.snapshotQueue = h.snapshotQueue
 	index.OpenTranslateStore = h.OpenTranslateStore
+	index.translationSyncer = h.translationSyncer
 	index.holder = h
 	return index, nil
 }
@@ -527,7 +541,10 @@ func (h *Holder) DeleteIndex(name string) error {
 	// Remove reference.
 	delete(h.indexes, name)
 
-	return nil
+	// I'm not sure if calling Reset() here is necessary
+	// since closing the index stops its translation
+	// sync processes.
+	return h.translationSyncer.Reset()
 }
 
 // Field returns the field for an index and name.
@@ -922,8 +939,8 @@ func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) err
 	return nil
 }
 
-// ResetTranslationSync reinitializes streaming sync of translation data.
-func (s *holderSyncer) ResetTranslationSync() error {
+// resetTranslationSync reinitializes streaming sync of translation data.
+func (s *holderSyncer) resetTranslationSync() error {
 	// Stop existing streams.
 	if err := s.stopTranslationSync(); err != nil {
 		return errors.Wrap(err, "stop translation sync")
@@ -944,6 +961,53 @@ func (s *holderSyncer) ResetTranslationSync() error {
 	return nil
 }
 
+////////////////////////////////////////////////////////////
+
+// translationSyncer provides an interface allowing a function
+// to notify the server that an action has occurred which requires
+// the translation sync process to be reset. In general, this
+// includes anything which modifies schema (add/remove index, etc),
+// or anything that changes the cluster topology (add/remove node).
+// I originally considered leveraging the broadcaster since that was
+// already in place and provides similar event messages, but the
+// broadcaster is really meant for notifiying other nodes, while
+// this is more akin to an internal message bus. In fact, I think
+// a future iteration on this may be to make it more generic so
+// it can act as an internal message bus where one of the messages
+// being published is "translationSyncReset".
+type translationSyncer interface {
+	Reset() error
+}
+
+// NopTranslationSyncer represents a translationSyncer that doesn't do anything.
+var NopTranslationSyncer translationSyncer = &nopTranslationSyncer{}
+
+type nopTranslationSyncer struct{}
+
+// Reset is a no-op implementation of translationSyncer Reset method.
+func (nopTranslationSyncer) Reset() error { return nil }
+
+// activeTranslationSyncer represents a translationSyncer that resets
+// the server's translation syncer.
+type activeTranslationSyncer struct {
+	ch chan struct{}
+}
+
+// newActiveTranslationSyncer returns a new instance of activeTranslationSyncer.
+func newActiveTranslationSyncer(ch chan struct{}) *activeTranslationSyncer {
+	return &activeTranslationSyncer{
+		ch: ch,
+	}
+}
+
+// Reset resets the server's translation syncer.
+func (a *activeTranslationSyncer) Reset() error {
+	a.ch <- struct{}{}
+	return nil
+}
+
+////////////////////////////////////////////////////////////
+
 // stopTranslationSync closes and waits for all outstanding translation readers
 // to complete. This should be called before reconnecting to the cluster in case
 // of a cluster resize or schema change.
@@ -958,17 +1022,36 @@ func (s *holderSyncer) stopTranslationSync() error {
 	return g.Wait()
 }
 
-// setTranslateReadOnlyFlags updates all translation stores to enabled or disable
+// setTranslateReadOnlyFlags updates all translation stores to enable or disable
 // writing new translation keys. Index stores are writable if the node owns the
 // partition. Field stores are writable if the node is the coordinator.
 func (s *holderSyncer) setTranslateReadOnlyFlags() {
 	isCoordinator := s.Cluster.isCoordinator()
 
 	for _, index := range s.Holder.Indexes() {
+		// There is a race condition here:
+		// if Indexes() returns idx1, and then in another
+		// process, holder.DeleteIndex(idx1) is called,
+		// then the next step trying to get TranslateStore(partitionID)
+		// for an index that is closed (and therefore its transateStores
+		// no longer exist) will fail with a nil pointer error.
+		// For now, I just checked that the translateStore hasn't been
+		// set to nil before trying to use it, but another option may
+		// be to prevent the translateStores from being zeroed out
+		// while this process is active. Checking for nil as we do
+		// really obviates the need for the RLock around the for loop.
+
+		// Obtain a read lock on index to prevent Index.Close() from
+		// destroying the Index.translateStores map before this is
+		// done using it.
+		index.mu.RLock()
 		for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ {
 			ownsPartition := s.Cluster.ownsPartition(s.Node.ID, partitionID)
-			index.TranslateStore(partitionID).SetReadOnly(!ownsPartition)
+			if ts := index.TranslateStore(partitionID); ts != nil {
+				ts.SetReadOnly(!ownsPartition)
+			}
 		}
+		index.mu.RUnlock()
 
 		for _, field := range index.Fields() {
 			field.TranslateStore().SetReadOnly(!isCoordinator)
@@ -1030,16 +1113,13 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error {
 // initializeFieldTranslateReplication connects the coordinator to stream field data.
 func (s *holderSyncer) initializeFieldTranslateReplication() error {
 	// Skip if coordinator.
-	if s.Cluster.Node.IsCoordinator {
+	if s.Cluster.isCoordinator() {
 		return nil
 	}
 
 	// Build a map of partition offsets to stream from.
 	m := make(TranslateOffsetMap)
 	for _, index := range s.Holder.Indexes() {
-		if !index.Keys() {
-			continue
-		}
 		for _, field := range index.Fields() {
 			store := field.TranslateStore()
 			offset, err := store.MaxID()

index.go

@@ -68,6 +68,8 @@ type Index struct {
 	// Per-partition translation stores
 	translateStores map[int]TranslateStore
 
+	translationSyncer translationSyncer
+
 	// Instantiates new translation stores
 	OpenTranslateStore OpenTranslateStoreFunc
 }
@@ -95,6 +97,8 @@ func NewIndex(path, name string, partitionN int) (*Index, error) {
 
 		translateStores: make(map[int]TranslateStore),
 
+		translationSyncer: NopTranslationSyncer,
+
 		OpenTranslateStore: OpenInMemTranslateStore,
 	}, nil
 }
@@ -478,6 +482,11 @@ func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) {
 	// Add to index's field lookup.
 	i.fields[name] = f
 
+	// Kick off the field's translation sync process.
+	if err := i.translationSyncer.Reset(); err != nil {
+		return nil, errors.Wrap(err, "resetting translation syncer")
+	}
+
 	return f, nil
 }
 
@@ -533,7 +542,7 @@ func (i *Index) DeleteField(name string) error {
 	// Remove reference.
 	delete(i.fields, name)
 
-	return nil
+	return i.translationSyncer.Reset()
 }
 
 type indexSlice []*Index

server.go

@@ -77,6 +77,9 @@ type Server struct { // nolint: maligned
 	isCoordinator       bool
 	syncer              holderSyncer
 
+	translationSyncer      translationSyncer
+	resetTranslationSyncCh chan struct{}
+
 	defaultClient InternalClient
 	dataDir       string
 }
@@ -316,10 +319,16 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 		metricInterval:      0,
 		diagnosticInterval:  0,
 
+		resetTranslationSyncCh: make(chan struct{}),
+
 		logger: logger.NopLogger,
 	}
 	s.cluster.InternalClient = s.defaultClient
 
+	s.translationSyncer = newActiveTranslationSyncer(s.resetTranslationSyncCh)
+	s.holder.translationSyncer = s.translationSyncer
+	s.cluster.translationSyncer = s.translationSyncer
+
 	s.diagnostics.server = s
 
 	for _, opt := range opts {
@@ -503,6 +512,18 @@ func (s *Server) Open() error {
 		log.Println(errors.Wrap(err, "logging startup"))
 	}
 
+	// Set up the holderSyncer.
+	s.syncer.Holder = s.holder
+	s.syncer.Node = s.cluster.Node
+	s.syncer.Cluster = s.cluster
+	s.syncer.Closing = s.closing
+	s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
+
+	// Start background process listening for translation
+	// sync resets.
+	s.wg.Add(1)
+	go func() { defer s.wg.Done(); s.monitorResetTranslationSync() }()
+
 	// Open Cluster management.
 	if err := s.cluster.waitForStarted(); err != nil {
 		return errors.Wrap(err, "opening Cluster")
@@ -523,16 +544,6 @@ func (s *Server) Open() error {
 	// buffered channel.
 	s.cluster.listenForJoins()
 
-	s.syncer.Holder = s.holder
-	s.syncer.Node = s.cluster.Node
-	s.syncer.Cluster = s.cluster
-	s.syncer.Closing = s.closing
-	s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
-
-	if err := s.syncer.ResetTranslationSync(); err != nil {
-		return err
-	}
-
 	// Start background monitoring.
 	s.wg.Add(3)
 	go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
@@ -595,6 +606,34 @@ func (s *Server) SyncData() error {
 	return errors.Wrap(s.syncer.SyncHolder(), "syncing holder")
 }
 
+// monitorResetTranslationSync is a background process which
+// listens for events indicating the need to reset the translation
+// sync processes.
+func (s *Server) monitorResetTranslationSync() {
+	s.logger.Printf("holder translation sync monitor initializing")
+	for {
+		// Wait for a reset or a close.
+		select {
+		case <-s.closing:
+			return
+		case <-s.resetTranslationSyncCh:
+			s.logger.Printf("holder translation sync beginning")
+			s.wg.Add(1)
+			go func() {
+				// Obtaining this lock ensures that there is only
+				// once instance of resetTranslationSync() running
+				// at once.
+				s.syncer.mu.Lock()
+				defer s.syncer.mu.Unlock()
+				defer s.wg.Done()
+				if err := s.syncer.resetTranslationSync(); err != nil {
+					s.logger.Printf("holder translation sync error: err=%s", err)
+				}
+			}()
+		}
+	}
+}
+
 func (s *Server) monitorAntiEntropy() {
 	if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 {
 		return // anti entropy disabled
@@ -666,16 +705,10 @@ func (s *Server) receiveMessage(m Message) error {
 		if err != nil {
 			return err
 		}
-		if err := s.syncer.ResetTranslationSync(); err != nil {
-			return err
-		}
 	case *DeleteIndexMessage:
 		if err := s.holder.DeleteIndex(obj.Index); err != nil {
 			return err
 		}
-		if err := s.syncer.ResetTranslationSync(); err != nil {
-			return err
-		}
 	case *CreateFieldMessage:
 		idx := s.holder.Index(obj.Index)
 		if idx == nil {
@@ -686,17 +719,11 @@ func (s *Server) receiveMessage(m Message) error {
 		if err != nil {
 			return err
 		}
-		if err := s.syncer.ResetTranslationSync(); err != nil {
-			return err
-		}
 	case *DeleteFieldMessage:
 		idx := s.holder.Index(obj.Index)
 		if err := idx.DeleteField(obj.Field); err != nil {
 			return err
 		}
-		if err := s.syncer.ResetTranslationSync(); err != nil {
-			return err
-		}
 	case *DeleteAvailableShardMessage:
 		f := s.holder.Field(obj.Index, obj.Field)
 		if err := f.RemoveAvailableShard(obj.ShardID); err != nil {
@@ -725,11 +752,6 @@ func (s *Server) receiveMessage(m Message) error {
 		if err != nil {
 			return err
 		}
-		if s.syncer.Cluster != nil {
-			if err := s.syncer.ResetTranslationSync(); err != nil {
-				return err
-			}
-		}
 	case *ResizeInstruction:
 		err := s.cluster.followResizeInstruction(obj)
 		if err != nil {