Merge pull request #1449 from niaow/deferredcreateshard

Defer cluster messages until startup

field.go

@@ -1004,7 +1004,7 @@ func (f *Field) createViewIfNotExists(name string) (*view, error) {
 
 	if created {
 		// Broadcast view creation to the cluster.
-		err := f.broadcaster.SendSync(cvm)
+		err := f.holder.sendOrSpool(cvm)
 		if err != nil {
 			return nil, errors.Wrap(err, "sending CreateView message")
 		}

holder.go

@@ -123,6 +123,11 @@ type Holder struct {
 	// opened before their foreign index has opened.
 	foreignIndexFields []*Field
 
+	// Queue of messages to broadcast in bulk when the cluster comes up.
+	// This is wrong, but. . . yeah.
+	startMsgs   []Message
+	startMsgsMu sync.Mutex
+
 	// opening is set to true while Holder is opening.
 	// It's used to determine if foreign index application
 	// needs to be queued and completed after all indexes
@@ -718,6 +723,27 @@ func (h *Holder) Open() error {
 
 }
 
+func (h *Holder) sendOrSpool(msg Message) error {
+	if h.maybeSpool(msg) {
+		return nil
+	}
+
+	return h.broadcaster.SendSync(msg)
+}
+
+func (h *Holder) maybeSpool(msg Message) bool {
+	h.startMsgsMu.Lock()
+	defer h.startMsgsMu.Unlock()
+
+	if h.startMsgs == nil {
+		// Startup is done.
+		return false
+	}
+
+	h.startMsgs = append(h.startMsgs, msg)
+	return true
+}
+
 // Activate runs the background tasks relevant to keeping a holder in a stable
 // state, such as scanning it for needed snapshots, or flushing caches. This
 // is separate from opening because, while a server would nearly always want
@@ -954,7 +980,7 @@ func (h *Holder) applySchema(schema *Schema) error {
 	}
 
 	// Send the load schema message to all nodes.
-	if err := h.broadcaster.SendSync(&LoadSchemaMessage{}); err != nil {
+	if err := h.sendOrSpool(&LoadSchemaMessage{}); err != nil {
 		return errors.Wrap(err, "sending LoadSchemaMessage")
 	}
 

index.go

@@ -767,7 +767,7 @@ func (i *Index) createField(cfm *CreateFieldMessage, broadcast bool) (*Field, er
 
 	if broadcast {
 		// Send the create field message to all nodes.
-		if err := i.broadcaster.SendSync(cfm); err != nil {
+		if err := i.holder.sendOrSpool(cfm); err != nil {
 			return nil, errors.Wrap(err, "sending CreateField message")
 		}
 	}

server.go

@@ -589,6 +589,12 @@ func (s *Server) Open() error {
 	s.syncer.Stats = s.holder.Stats.WithTags("component:HolderSyncer")
 
 	// Open holder.
+	func() {
+		s.holder.startMsgsMu.Lock()
+		defer s.holder.startMsgsMu.Unlock()
+
+		s.holder.startMsgs = []Message{}
+	}()
 	if err := s.holder.Open(); err != nil {
 		return errors.Wrap(err, "opening Holder")
 	}
@@ -612,6 +618,92 @@ func (s *Server) Open() error {
 	go func() { defer s.wg.Done(); s.monitorRuntime() }()
 	go func() { defer s.wg.Done(); s.monitorDiagnostics() }()
 
+	toSend := func() []Message {
+		s.holder.startMsgsMu.Lock()
+		defer s.holder.startMsgsMu.Unlock()
+
+		toSend := s.holder.startMsgs
+		s.holder.startMsgs = nil
+		return toSend
+	}()
+
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		s.wg.Add(1)
+		go func() {
+			defer s.wg.Done()
+			defer cancel()
+			select {
+			case <-s.closing:
+			case <-ctx.Done():
+			}
+		}()
+
+		timer := time.NewTimer(0)
+		defer timer.Stop()
+		if !timer.Stop() {
+			<-timer.C
+		}
+		for {
+			state, err := s.stator.ClusterState(ctx)
+			if err != nil {
+				s.logger.Printf("failed to check cluster state: %v", err)
+				timer.Reset(time.Second)
+				select {
+				case <-s.closing:
+					return
+				case <-timer.C:
+					continue
+				}
+			}
+			switch state {
+			case disco.ClusterStateStarting, disco.ClusterStateUnknown, disco.ClusterStateDown:
+				timer.Reset(time.Second)
+				select {
+				case <-s.closing:
+					return
+				case <-timer.C:
+					continue
+				}
+			}
+			break
+		}
+
+		start := time.Now()
+		prevMsg := start
+		s.logger.Printf("start initial cluster state sync")
+		for i := range toSend {
+			for {
+				err := s.holder.broadcaster.SendSync(&toSend[i])
+				if err != nil {
+					s.logger.Printf("failed to broadcast startup cluster message (trying again in a bit): %v", err)
+					timer.Reset(time.Second)
+					select {
+					case <-s.closing:
+						return
+					case <-timer.C:
+						continue
+					}
+				}
+				break
+			}
+
+			if now := time.Now(); now.Sub(prevMsg) > time.Second {
+				progressRatio := float64(i+1) / float64(len(toSend))
+				remainingRatio := 1 - progressRatio
+				timeRemaining := time.Duration(float64(now.Sub(prevMsg)) * (remainingRatio / progressRatio))
+				s.logger.Printf("synced %d/%d messages (%.2f%% complete; %s remaining)", i+1, len(toSend), 100*progressRatio, timeRemaining)
+				prevMsg = now
+			}
+		}
+		s.logger.Printf("completed initial cluster state sync in %s", time.Since(start).String())
+	}()
+
 	return nil
 }
 

view.go

@@ -340,7 +340,6 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
 }
 
 func (v *view) notifyIfNewShard(shard uint64) {
-
 	// if single node, don't bother serializing only to drop it b/c
 	// we won't send to ourselves.
 	srv, ok := v.broadcaster.(*Server)
@@ -355,24 +354,22 @@ func (v *view) notifyIfNewShard(shard uint64) {
 	broadcastChan := make(chan struct{})
 
 	go func() {
-		msg := &CreateShardMessage{
+		err := v.holder.sendOrSpool(&CreateShardMessage{
 			Index: v.index,
 			Field: v.field,
 			Shard: shard,
-		}
-		// Broadcast a message that a new max shard was just created.
-		err := v.broadcaster.SendSync(msg)
+		})
 		if err != nil {
 			v.holder.Logger.Printf("broadcasting create shard: %v", err)
 		}
 		close(broadcastChan)
 	}()
 
-	// We want to wait until the broadcast is complete, but what if it
-	// takes a really long time? So we time out.
+	timer := time.NewTimer(50 * time.Millisecond)
 	select {
 	case <-broadcastChan:
-	case <-time.After(50 * time.Millisecond):
+		timer.Stop()
+	case <-timer.C:
 		v.holder.Logger.Debugf("broadcasting create shard took >50ms")
 	}
 }