view.go: deal with races in fragment creation

There existed a case where two goroutines would try to
CreateIfNotExists the same fragment, and the first would
create it, but not put it in the fragments table, then
drop the lock, try to broadcast a message, and if it
succeeded then populate the fragments table. The second
would come along during the broadcast, not find an
entry, try to create one, and fail because the file was
already locked.

Basic problem: At least one test in server/ will fail
if we don't delay to send out broadcast messages. Everything
will lock up if we can wait forever (or even just a very
long time) for the message broadcast. We don't ever want
to have an inconsistent state -- so we don't want to either
fail to get a fragment when one's been created, or get one
that's about to be deleted if the broadcast fails.

So, creation and stashing in the fragments table is
atomic and immediate. After that, we optimistically attempt
to broadcast. If we fail, we fail. We delay up to about
50ms for the broadcast to be done, but after that return
anyway. This way, if things are going well everything
works, and if there's unexpected delays, things work except
some nodes in a cluster may not know about available
shards on other nodes sometimes. But that would have
happened anyway. A proper fix is beyond the scope of this
patch.

view.go

@@ -21,6 +21,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pilosa/pilosa/logger"
 	"github.com/pilosa/pilosa/pql"
@@ -180,11 +181,9 @@ func (v *view) fragmentPath(shard uint64) string {
 func (v *view) Fragment(shard uint64) *fragment {
 	v.mu.RLock()
 	defer v.mu.RUnlock()
-	return v.fragment(shard)
+	return v.fragments[shard]
 }
 
-func (v *view) fragment(shard uint64) *fragment { return v.fragments[shard] }
-
 // allFragments returns a list of all fragments in the view.
 func (v *view) allFragments() []*fragment {
 	v.mu.Lock()
@@ -206,45 +205,46 @@ func (v *view) recalculateCaches() {
 
 // CreateFragmentIfNotExists returns a fragment in the view by shard.
 func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
-	frag, msg, err := v.createFragmentIfNotExists(shard)
-
-	// if msg is not nil, then a new shard was created
-	if err == nil && msg != nil {
-		// Broadcast a message that a new max shard was just created.
-		if err = v.broadcaster.SendSync(msg); err != nil {
-			frag.close()
-			return nil, errors.Wrap(err, "sending createshard message")
-		}
-		v.mu.Lock()
-		v.fragments[shard] = frag
-		v.mu.Unlock()
-	}
-
-	return frag, err
-}
-
-func (v *view) createFragmentIfNotExists(shard uint64) (*fragment, *CreateShardMessage, error) {
 	v.mu.Lock()
 	defer v.mu.Unlock()
 	// Find fragment in cache first.
 	if frag := v.fragments[shard]; frag != nil {
-		return frag, nil, nil
+		return frag, nil
 	}
 
 	// Initialize and open fragment.
 	frag := v.newFragment(v.fragmentPath(shard), shard)
 	if err := frag.Open(); err != nil {
-		return nil, nil, errors.Wrap(err, "opening fragment")
+		return nil, errors.Wrap(err, "opening fragment")
 	}
 	frag.RowAttrStore = v.rowAttrStore
 
-	msg := &CreateShardMessage{
-		Index: v.index,
-		Field: v.field,
-		Shard: shard,
+	v.fragments[shard] = frag
+	broadcastChan := make(chan struct{})
+
+	go func() {
+		msg := &CreateShardMessage{
+			Index: v.index,
+			Field: v.field,
+			Shard: shard,
+		}
+		// Broadcast a message that a new max shard was just created.
+		err := v.broadcaster.SendSync(msg)
+		if err != nil {
+			v.logger.Printf("broadcasting create shard: %v", err)
+		}
+		close(broadcastChan)
+	}()
+
+	// We want to wait until the broadcast is complete, but what if it
+	// takes a really long time? So we time out.
+	select {
+	case <-broadcastChan:
+	case <-time.After(50 * time.Millisecond):
+		v.logger.Debugf("broadcasting create shard took >50ms")
 	}
 
-	return frag, msg, nil
+	return frag, nil
 }
 
 func (v *view) newFragment(path string, shard uint64) *fragment {
@@ -263,8 +263,7 @@ func (v *view) newFragment(path string, shard uint64) *fragment {
 
 // deleteFragment removes the fragment from the view.
 func (v *view) deleteFragment(shard uint64) error {
-
-	fragment := v.fragments[shard]
+	fragment := v.Fragment(shard)
 	if fragment == nil {
 		return ErrFragmentNotFound
 	}
@@ -318,8 +317,8 @@ func (v *view) setBit(rowID, columnID uint64) (changed bool, err error) {
 // clearBit clears a bit within the view.
 func (v *view) clearBit(rowID, columnID uint64) (changed bool, err error) {
 	shard := columnID / ShardWidth
-	frag, found := v.fragments[shard]
-	if !found {
+	frag := v.Fragment(shard)
+	if frag == nil {
 		return false, nil
 	}
 	return frag.clearBit(rowID, columnID)

view_internal_test.go

@@ -16,10 +16,10 @@ package pilosa
 
 import (
 	"io/ioutil"
-	"strings"
 	"testing"
+	"time"
 
-	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // mustOpenView returns a new instance of View with a temporary path.
@@ -77,44 +77,41 @@ func TestView_DeleteFragment(t *testing.T) {
 	}
 }
 
-// Ensure view closes fragment after failed shard broadcast.
-func TestView_CreateFragmentError(t *testing.T) {
+// Ensure that simultaneous attempts to grab a new fragment don't clash even
+// if the broadcast operation takes a bit of time.
+func TestView_CreateFragmentRace(t *testing.T) {
+	var creates errgroup.Group
 	v := mustOpenView("i", "f", "v")
 	defer v.close()
 
 	// Use a broadcaster which intentionally fails.
-	v.broadcaster = errorBroadcaster{}
+	v.broadcaster = delayBroadcaster{delay: 10 * time.Millisecond}
 
 	shard := uint64(0)
 
-	// Create fragment (with error on broadcast).
-	fragment, err := v.CreateFragmentIfNotExists(shard)
-	if !strings.Contains(err.Error(), "intentional error") {
-		if err != nil {
-			t.Fatal(err)
-		} else if fragment == nil {
-			t.Fatal("expected fragment")
-		} else {
-			t.Fatal("expected intentional error")
-		}
-	}
-
-	// Set the broadcaster back to no-op.
-	v.broadcaster = nopBroadcaster{}
-
-	// Try to create the fragment again.
-	_, err = v.CreateFragmentIfNotExists(shard)
+	creates.Go(func() error {
+		_, err := v.CreateFragmentIfNotExists(shard)
+		return err
+	})
+	creates.Go(func() error {
+		_, err := v.CreateFragmentIfNotExists(shard)
+		return err
+	})
+	err := creates.Wait()
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("unexpected error: %v", err)
 	}
 }
 
-// errorBroadcaster is a broadcaster which always returns an error.
-type errorBroadcaster struct {
+// delayBroadcaster is a nopBroadcaster with a configurable delay.
+type delayBroadcaster struct {
 	nopBroadcaster
+	delay time.Duration
 }
 
-// SendSync is an implementation of Broadcaster SendSync which always returns an error.
-func (errorBroadcaster) SendSync(Message) error {
-	return errors.New("intentional error")
+// SendSync is an implementation of Broadcaster SendSync which delays for a
+// specified interval before succeeding.
+func (d delayBroadcaster) SendSync(Message) error {
+	time.Sleep(d.delay)
+	return nil
 }