diff --git a/view.go b/view.go index 00dcf02c6..658261037 100644 --- a/view.go +++ b/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) diff --git a/view_internal_test.go b/view_internal_test.go index 6867b03ed..bb50003d2 100644 --- a/view_internal_test.go +++ b/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 }