wal: report when wal is ready

in this way the wal instance will be used only after it's ready (initialized
etcd when needed)
This commit is contained in:
Simone Gotti 2019-03-28 15:46:24 +01:00
parent d13fe43e19
commit 65c425b22b
4 changed files with 27 additions and 12 deletions

View File

@ -95,8 +95,13 @@ func NewConfigStore(ctx context.Context, c *config.ConfigStore) (*ConfigStore, e
func (s *ConfigStore) Run(ctx context.Context) error {
errCh := make(chan error)
walReadyCh := make(chan struct{})
go func() { errCh <- s.wal.Run(ctx, walReadyCh) }()
// wait for wal to be ready
<-walReadyCh
go func() { errCh <- s.wal.Run(ctx) }()
go func() { errCh <- s.readDB.Run(ctx) }()
// noop coors handler

View File

@ -1193,8 +1193,14 @@ func NewScheduler(ctx context.Context, c *config.RunServiceScheduler) (*Schedule
func (s *Scheduler) Run(ctx context.Context) error {
errCh := make(chan error)
walReadyCh := make(chan struct{})
go func() { errCh <- s.wal.Run(ctx, walReadyCh) }()
// wait for wal to be ready
<-walReadyCh
go func() { errCh <- s.wal.Run(ctx) }()
go s.readDB.Run(ctx)
ch := make(chan *types.ExecutorTask)

View File

@ -1241,8 +1241,7 @@ func NewWalManager(ctx context.Context, logger *zap.Logger, conf *WalManagerConf
return w, nil
}
func (w *WalManager) Run(ctx context.Context) error {
func (w *WalManager) Run(ctx context.Context, readyCh chan struct{}) error {
for {
err := w.InitEtcd(ctx)
if err == nil {
@ -1252,6 +1251,8 @@ func (w *WalManager) Run(ctx context.Context) error {
time.Sleep(1 * time.Second)
}
readyCh <- struct{}{}
go w.watcherLoop(ctx)
go w.syncLoop(ctx)
go w.checkpointLoop(ctx)

View File

@ -89,8 +89,9 @@ func TestEtcdReset(t *testing.T) {
EtcdWalsKeepNum: 10,
}
wal, err := NewWalManager(ctx, logger, walConfig)
go wal.Run(ctx)
time.Sleep(1 * time.Second)
walReadyCh := make(chan struct{})
go wal.Run(ctx, walReadyCh)
<-walReadyCh
actions := []*Action{
{
@ -123,8 +124,8 @@ func TestEtcdReset(t *testing.T) {
cancel()
ctx = context.Background()
go wal.Run(ctx)
time.Sleep(5 * time.Second)
go wal.Run(ctx, walReadyCh)
<-walReadyCh
curObjects := []string{}
doneCh := make(chan struct{})
@ -178,8 +179,9 @@ func TestConcurrentUpdate(t *testing.T) {
},
}
go wal.Run(ctx)
time.Sleep(1 * time.Second)
walReadyCh := make(chan struct{})
go wal.Run(ctx, walReadyCh)
<-walReadyCh
cgNames := []string{"changegroup01", "changegroup02"}
cgt := wal.GetChangeGroupsUpdateToken(cgNames)
@ -253,8 +255,9 @@ func TestWalCleaner(t *testing.T) {
},
}
go wal.Run(ctx)
time.Sleep(1 * time.Second)
walReadyCh := make(chan struct{})
go wal.Run(ctx, walReadyCh)
<-walReadyCh
for i := 0; i < 20; i++ {
if _, err := wal.WriteWal(ctx, actions, nil); err != nil {