From 3fbf6993d0111709578fc036dd07f6f49208a521 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 22 Feb 2021 09:46:24 -0500 Subject: [PATCH] defer cluster messages until startup --- field.go | 2 +- holder.go | 28 ++++++++++++++++- index.go | 2 +- server.go | 92 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ view.go | 13 +++----- 5 files changed, 126 insertions(+), 11 deletions(-) diff --git a/field.go b/field.go index d2ecd0f0a..a0243c458 100644 --- a/field.go +++ b/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") } diff --git a/holder.go b/holder.go index 8400f1811..87875d982 100644 --- a/holder.go +++ b/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") } diff --git a/index.go b/index.go index 93de521ce..080fa6ba1 100644 --- a/index.go +++ b/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") } } diff --git a/server.go b/server.go index 94569bb13..bd08550f0 100644 --- a/server.go +++ b/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 } diff --git a/view.go b/view.go index 8e9f8e2b9..df44c219d 100644 --- a/view.go +++ b/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") } }