WIP: don't block joining nodes while coordinator loads data.

Implements a Holder.Peek() function, and breaks out
Cluster.ListenForJoins() into a separate method that can be started
after the Holder finishes loading data.

TODO:
- [ ] test Holder.Peek()

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

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.

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