From 7b36f417d8a5af17a122bffc89b48ff83833eef1 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 29 Apr 2020 14:59:37 -0500 Subject: [PATCH 1/2] avoid deadlock on translationSync.Reset during startup --- api.go | 2 +- cluster.go | 38 +++++++++++++++++++++++++++++++------- server/server.go | 9 +-------- 3 files changed, 33 insertions(+), 16 deletions(-) diff --git a/api.go b/api.go index 7c2da4893..13fad3e77 100644 --- a/api.go +++ b/api.go @@ -769,7 +769,7 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error { return errors.Wrap(err, "deserializing cluster message") } - // Forward the error message. + // Forward the message. if err := api.server.receiveMessage(msg); err != nil { return errors.Wrap(err, "receiving message") } diff --git a/cluster.go b/cluster.go index c453aa2ef..6fce679fb 100644 --- a/cluster.go +++ b/cluster.go @@ -482,9 +482,21 @@ func (c *cluster) unprotectedSetState(state string) { // 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) - } + // + // We can't call Reset() if Server.Open() hasn't run yet, + // because that's where we start monitorResetTranslationSync() + // which reads the reset channel. If we get here before + // Server.Open(), this will deadlock on that channel read. + // In order to address this, we call Reset() in a goroutine + // so even if it blocks waiting for monitorResetTranslationSync() + // to start, it doesn't cause a deadlock, and once Server.Open() + // is called, then the sync reset (or in the STARTING case, the + // initial sync start) will happen. + go func() { + if err := c.translationSyncer.Reset(); err != nil { + c.logger.Printf("error resetting translation syncer: %s", err) + } + }() } // TODO: consider NOT running cleanup on an active node that has @@ -573,6 +585,19 @@ func (c *cluster) determineClusterState() (clusterState string) { if c.haveTopologyAgreement() && c.allNodesReady() { return ClusterStateNormal } + // TODO: + // If the cluster is still STARTING, there's no need to put it into + // state DEGRADED. It's possible to force a starting cluster to go + // into state DEGRADED by, for example, restarting a 2-node cluster + // with replica=3. In that case, the coordinator would come up and + // it would immediately trigger this condition. Checking for + // state != STARTING here would prevent that. Unfortunately, based + // on test TestClusteringNodesReplica2, we expect a DEGRADED cluster + // to go back into state STARTING if it loses more replicas than + // can support queries. In that case, we might actually want it to + // go from STARTING back to DEGRADED. Leaving it as is for now, but + // noting that it's a little confusing that a cluster starting up + // could possibly go into state DEGRADED. if len(c.Topology.nodeIDs)-len(c.nodeIDs()) < c.ReplicaN && c.allNodesReady() { return ClusterStateDegraded } @@ -1100,7 +1125,7 @@ func (c *cluster) waitForStarted() error { // (and now in a state of STARTING) so that it can be put to the correct // cluster state. // TODO: Because the normal code path already sends a NodeJoin event (via - // memberlist), this it a bit redundant in most cases. Perhaps determine + // memberlist), this is a bit redundant in most cases. Perhaps determine // that the node has been restarted and don't do this step. msg := &NodeEvent{ Event: NodeJoin, @@ -1225,7 +1250,6 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error { // Broadcast cluster status changes to the cluster. status := c.unprotectedStatus() return c.unprotectedSendSync(status) // TODO fix c.Status - } func (c *cluster) sendTo(node *Node, m Message) error { @@ -2067,8 +2091,8 @@ func (c *cluster) nodeLeave(nodeID string) error { } if c.state != ClusterStateNormal && c.state != ClusterStateDegraded { - return fmt.Errorf("cluster must be '%s' to remove a node but is '%s'", - ClusterStateNormal, c.state) + return fmt.Errorf("cluster must be '%s' or '%s' to remove a node but is '%s'", + ClusterStateNormal, ClusterStateDegraded, c.state) } // Ensure that node is in the cluster. diff --git a/server/server.go b/server/server.go index 478e5cfe9..5d12a745c 100644 --- a/server/server.go +++ b/server/server.go @@ -158,15 +158,8 @@ func (m *Command) Start() (err error) { if err != nil { return errors.Wrap(err, "setting up server") } - if !m.API.Node().IsCoordinator { - // hack to give coordinator a head start - // TODO https://github.com/molecula/pilosa/issues/266 - if len(m.Config.Gossip.Seeds) > 0 && !m.noSleep { - time.Sleep(5 * time.Second) - } - } - // SetupNetworking + // Set up networking (i.e. gossip) err = m.setupNetworking() if err != nil { return errors.Wrap(err, "setting up networking") From 92a94c9ec1910ee15e150531983caac3bd26bce7 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 29 Apr 2020 15:34:04 -0500 Subject: [PATCH 2/2] remove noSleep option --- server.go | 2 +- server/server.go | 12 ------------ test/pilosa.go | 1 - 3 files changed, 1 insertion(+), 14 deletions(-) diff --git a/server.go b/server.go index 7e0690c30..d3926ab23 100644 --- a/server.go +++ b/server.go @@ -634,7 +634,7 @@ func (s *Server) monitorResetTranslationSync() { s.wg.Add(1) go func() { // Obtaining this lock ensures that there is only - // once instance of resetTranslationSync() running + // one instance of resetTranslationSync() running // at once. s.syncer.mu.Lock() defer s.syncer.mu.Unlock() diff --git a/server/server.go b/server/server.go index 5d12a745c..5ff688096 100644 --- a/server/server.go +++ b/server/server.go @@ -87,7 +87,6 @@ type Command struct { listenURI *pilosa.URI tlsConfig *tls.Config closeTimeout time.Duration - noSleep bool serverOptions []pilosa.ServerOption } @@ -115,17 +114,6 @@ func OptCommandConfig(config *Config) CommandOption { } } -// OptCommandNoSleep disables the 5 second sleep for non-coordinator -// nodes on startup. See https://github.com/molecula/pilosa/issues/266 -// This option should only be used by tests, and expect it to be -// deprecated. -func OptCommandNoSleep() CommandOption { - return func(c *Command) error { - c.noSleep = true - return nil - } -} - // NewCommand returns a new instance of Main. func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command { c := &Command{ diff --git a/test/pilosa.go b/test/pilosa.go index 6903a2010..6963f635f 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -65,7 +65,6 @@ func newCommand(opts ...server.CommandOption) *Command { // does not fail on 32-bit systems. opts = append([]server.CommandOption{ server.OptCommandCloseTimeout(time.Millisecond * 2), - server.OptCommandNoSleep(), }, opts...) m := &Command{commandOptions: opts} m.Command = server.NewCommand(bytes.NewReader(nil), ioutil.Discard, ioutil.Discard, opts...)