diff --git a/cluster.go b/cluster.go index 32f1dee12..204e39c26 100644 --- a/cluster.go +++ b/cluster.go @@ -546,10 +546,6 @@ func (c *Cluster) Open() error { return fmt.Errorf("opening MemberSet: %v", err) } - // Listen for cluster-resize events. - c.wg.Add(1) - go func() { defer c.wg.Done(); c.listenForJoins() }() - return nil } @@ -604,6 +600,12 @@ func (c *Cluster) setStateAndBroadcast(state string) error { return c.Broadcaster.SendSync(c.Status()) } +// ListenForJoins handles cluster-resize events. +func (c *Cluster) ListenForJoins() { + c.wg.Add(1) + go func() { defer c.wg.Done(); c.listenForJoins() }() +} + func (c *Cluster) listenForJoins() { var uriJoined bool @@ -634,8 +636,8 @@ func (c *Cluster) listenForJoins() { select { case <-c.closing: return - case host := <-c.joiningURIs: - err := c.handleJoiningHost(host) + case uri := <-c.joiningURIs: + err := c.handleJoiningHost(uri) if err != nil { c.logger().Printf("handleJoiningHost error: err=%s", err) continue diff --git a/holder.go b/holder.go index 07c030135..bbda143d6 100644 --- a/holder.go +++ b/holder.go @@ -44,6 +44,7 @@ type Holder struct { // Indexes by name. indexes map[string]*Index + hasData bool Broadcaster Broadcaster // Close management @@ -77,6 +78,36 @@ func NewHolder() *Holder { } } +// Peek reads the root data directory for the holder +// without actually loading any data into memory. +func (h *Holder) Peek() error { + if err := os.MkdirAll(h.Path, 0777); err != nil { + return err + } + + // Open path to read all index directories. + f, err := os.Open(h.Path) + if err != nil { + return err + } + defer f.Close() + + fis, err := f.Readdir(0) + if err != nil { + return err + } + + for _, fi := range fis { + if !fi.IsDir() { + continue + } + h.hasData = true + break + } + + return nil +} + // Open initializes the root data directory for the holder. func (h *Holder) Open() error { h.setFileLimit() @@ -149,7 +180,7 @@ func (h *Holder) Close() error { // This is used to determine if the rebalancing of data is necessary // when a node joins the cluster. func (h *Holder) HasData() bool { - return len(h.indexes) > 0 + return h.hasData || len(h.indexes) > 0 } // MaxSlices returns MaxSlice map for all indexes. diff --git a/server.go b/server.go index ac5c5d06a..06c0a81e7 100644 --- a/server.go +++ b/server.go @@ -145,10 +145,12 @@ func (s *Server) Open() error { } } - // Open holder. + // Peek at the holder to determine if there is data on disk. + // Don't actually load the data until after the Cluster + // management starts. s.Holder.LogOutput = s.LogOutput - if err := s.Holder.Open(); err != nil { - return fmt.Errorf("opening Holder: %v", err) + if err := s.Holder.Peek(); err != nil { + return fmt.Errorf("peeking at the Holder: %v", err) } // Start the BroadcastReceiver. @@ -161,6 +163,18 @@ func (s *Server) Open() error { return fmt.Errorf("opening Cluster: %v", err) } + // Open holder. + if err := s.Holder.Open(); err != nil { + return fmt.Errorf("opening Holder: %v", err) + } + + // Listen for joining nodes. + // This needs to start after the Holder has opened so that nodes can join + // the cluster without waiting for data to load on the coordinator. Before + // this starts, the joins are queued up in the Cluster.joiningURIs buffered + // channel. + s.Cluster.ListenForJoins() + // Create default HTTP client s.createDefaultClient()