From dde6954de4aaa219dadf300fa58532a57c6922b6 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 18 Feb 2019 14:35:03 -0600 Subject: [PATCH] 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 | 63 +++++++++++++++++++++---------------------- view_internal_test.go | 53 +++++++++++++++++------------------- 2 files changed, 56 insertions(+), 60 deletions(-) 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 }