fix race condition when resetting translation

Previously, we never waited for translation sync goroutines to stop.
That issue should be mostly harmless in the normal path.
Additionally, this waits for the translation sync to shut down when stopping the server.
This commit is contained in:
Nia Weiss 2020-09-29 13:54:16 -04:00
parent 3d07d13a7c
commit 1b210080c6
No known key found for this signature in database
GPG key ID: 895E83409BFDA1BB
3 changed files with 43 additions and 3 deletions

View file

@ -20,6 +20,7 @@ import (
"io/ioutil"
"os"
"reflect"
"strconv"
"testing"
"time"
@ -216,6 +217,30 @@ func TestTranslateStore_TranslateIDs(t *testing.T) {
}
}
func TestTranslateStore_MaxID(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
// Generate a bunch of keys.
var lastk uint64
for i := 0; i < 1026; i++ {
k, err := s.TranslateKey(strconv.Itoa(i), true)
if err != nil {
t.Fatalf("translating %d: %v", i, err)
}
lastk = k
}
// Verify the max ID.
max, err := s.MaxID()
if err != nil {
t.Fatalf("checking max ID: %v", err)
}
if max != lastk {
t.Fatalf("last key is %d but max is %d", lastk, max)
}
}
func TestTranslateStore_EntryReader(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenNewTranslateStore()

View file

@ -1275,6 +1275,8 @@ type holderSyncer struct {
// Translation sync handling.
readers []TranslateEntryReader
syncers errgroup.Group
// Stats
Stats stats.StatsClient
@ -1569,6 +1571,7 @@ func (s *holderSyncer) stopTranslationSync() error {
return rd.Close()
})
}
g.Go(s.syncers.Wait)
return g.Wait()
}
@ -1658,7 +1661,11 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error {
}
s.readers = append(s.readers, rd)
go func() { defer rd.Close(); s.readIndexTranslateReader(rd) }()
s.syncers.Go(func() error {
defer rd.Close()
s.readIndexTranslateReader(rd)
return nil
})
}
return nil
@ -1697,8 +1704,11 @@ func (s *holderSyncer) initializeFieldTranslateReplication() error {
}
s.readers = append(s.readers, rd)
go func() { defer rd.Close(); s.readFieldTranslateReader(rd) }()
s.syncers.Go(func() error {
defer rd.Close()
s.readFieldTranslateReader(rd)
return nil
})
return nil
}

View file

@ -534,10 +534,12 @@ func (s *Server) Close() error {
s.wg.Wait()
var errh error
var errhs error
var errc error
if s.cluster != nil {
errc = s.cluster.close()
}
errhs = s.syncer.stopTranslationSync()
if s.holder != nil {
errh = s.holder.Close()
}
@ -553,6 +555,9 @@ func (s *Server) Close() error {
if errh != nil {
return errors.Wrap(errh, "closing holder")
}
if errhs != nil {
return errors.Wrap(errhs, "terminating holder translation sync")
}
if errc != nil {
return errors.Wrap(errc, "closing cluster")
}