Merge pull request #329 from travisturner/cluster-startup

avoid deadlock on translationSync.Reset during startup

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")
 	}

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.

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()

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{
@@ -158,15 +146,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")

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...)