Merge pull request #1750 from jaffee/more-races

More races

cluster.go

@@ -1249,7 +1249,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 	go func() {
 
 		// Make sure the holder has opened.
-		<-c.holder.opened
+		c.holder.opened.Recv()
 
 		// Prepare the return message.
 		complete := &ResizeInstructionComplete{

executor.go

@@ -1664,7 +1664,7 @@ func (e *executor) executeSetRowShard(ctx context.Context, index string, c *pql.
 		if err != nil {
 			return false, errors.Wrap(err, "creating view")
 		}
-		fragment, err = view.createFragmentIfNotExists(shard)
+		fragment, err = view.CreateFragmentIfNotExists(shard)
 		if err != nil {
 			return false, errors.Wrapf(err, "creating fragment: %d", shard)
 		}
@@ -2055,7 +2055,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
 	if !opt.Remote {
 		nodes = Nodes(e.Cluster.nodes).Clone()
 	} else {
-		nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)}
+		nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)}
 	}
 
 	// Start mapping across all primary owners.

holder.go

@@ -57,7 +57,7 @@ type Holder struct {
 	NewPrimaryTranslateStore func(interface{}) TranslateStore
 
 	// opened channel is closed once Open() completes.
-	opened chan struct{}
+	opened lockedChan
 
 	broadcaster broadcaster
 
@@ -79,13 +79,39 @@ type Holder struct {
 	Logger logger.Logger
 }
 
+// lockedChan looks a little ridiculous admittedly, but exists for good reason.
+// The channel within is used (for example) to signal to other goroutines when
+// the Holder has finished opening (via closing the channel). However, it is
+// possible for the holder to be closed and then reopened, but a channel which
+// is closed cannot be re-opened. We must create a new channel - this creates a
+// data race with any goroutine which might be accessing the channel. To ensure
+// that there is no data race on the value of the channel itself, we wrap any
+// operation on it with an RWMutex so that we can guarantee that nothing is
+// trying to listen on it when it gets swapped.
+type lockedChan struct {
+	ch chan struct{}
+	mu sync.RWMutex
+}
+
+func (lc *lockedChan) Close() {
+	lc.mu.RLock()
+	close(lc.ch)
+	lc.mu.RUnlock()
+}
+
+func (lc *lockedChan) Recv() {
+	lc.mu.RLock()
+	<-lc.ch
+	lc.mu.RUnlock()
+}
+
 // NewHolder returns a new instance of Holder.
 func NewHolder() *Holder {
 	return &Holder{
 		indexes: make(map[string]*Index),
 		closing: make(chan struct{}),
 
-		opened: make(chan struct{}),
+		opened: lockedChan{ch: make(chan struct{})},
 
 		translateFile:            NewTranslateFile(),
 		NewPrimaryTranslateStore: newNopTranslateStore,
@@ -159,7 +185,7 @@ func (h *Holder) Open() error {
 
 	h.Stats.Open()
 
-	close(h.opened)
+	h.opened.Close()
 	return nil
 }
 
@@ -184,7 +210,9 @@ func (h *Holder) Close() error {
 	}
 
 	// Reset opened in case Holder needs to be reopened.
-	h.opened = make(chan struct{})
+	h.opened.mu.Lock()
+	h.opened.ch = make(chan struct{})
+	h.opened.mu.Unlock()
 
 	return nil
 }

server.go

@@ -628,7 +628,7 @@ func (s *Server) handleRemoteStatus(pb Message) {
 
 	go func() {
 		// Make sure the holder has opened.
-		<-s.holder.opened
+		s.holder.opened.Recv()
 
 		err := s.mergeRemoteStatus(pb.(*NodeStatus))
 		if err != nil {