Rework how Network background jobs are managed a bit

This commit is contained in:
Brian Picciano 2024-11-05 11:59:45 +01:00
parent efdab29ae6
commit 8e264cf028

View File

@ -23,6 +23,7 @@ import (
"sync"
"time"
"dev.mediocregopher.com/mediocre-go-lib.git/mctx"
"dev.mediocregopher.com/mediocre-go-lib.git/mlog"
"golang.org/x/exp/maps"
)
@ -181,13 +182,15 @@ type network struct {
networkConfig daecommon.NetworkConfig
currBootstrap bootstrap.Bootstrap
shutdownCh chan struct{}
wg sync.WaitGroup
workerCtx context.Context
workerCancel context.CancelFunc
wg sync.WaitGroup
}
// instatiateNetwork returns an instantiated *network instance which has not yet
// been initialized.
func instatiateNetwork(
ctx context.Context,
logger *mlog.Logger,
networkConfig daecommon.NetworkConfig,
envBinDirPath string,
@ -195,6 +198,8 @@ func instatiateNetwork(
runtimeDir toolkit.Dir,
opts *Opts,
) *network {
ctx = context.WithoutCancel(ctx)
ctx, cancel := context.WithCancel(ctx)
return &network{
logger: logger,
networkConfig: networkConfig,
@ -202,7 +207,8 @@ func instatiateNetwork(
stateDir: stateDir,
runtimeDir: runtimeDir,
opts: opts.withDefaults(),
shutdownCh: make(chan struct{}),
workerCtx: ctx,
workerCancel: cancel,
}
}
@ -244,6 +250,7 @@ func Load(
Network, error,
) {
n := instatiateNetwork(
ctx,
logger,
networkConfig,
envBinDirPath,
@ -289,6 +296,7 @@ func Join(
Network, error,
) {
n := instatiateNetwork(
ctx,
logger,
networkConfig,
envBinDirPath,
@ -355,6 +363,7 @@ func Create(
garageRPCSecret := toolkit.RandStr(32)
n := instatiateNetwork(
ctx,
logger,
networkConfig,
envBinDirPath,
@ -407,6 +416,38 @@ func (n *network) initializeDirs(mayExist bool) error {
return nil
}
func (n *network) periodically(
logger *mlog.Logger,
fn func(context.Context) error,
period time.Duration,
) {
n.wg.Add(1)
go func() {
defer n.wg.Done()
ctx := mctx.Annotate(n.workerCtx, "period", period)
ticker := time.NewTicker(period)
defer ticker.Stop()
logger.Info(ctx, "Starting background job runner")
defer logger.Info(ctx, "Stopping background job runner")
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
logger.Info(ctx, "Background job running")
if err := fn(ctx); err != nil {
logger.Error(ctx, "Background job failed", err)
}
}
}
}()
}
func (n *network) initialize(
ctx context.Context, prevBootstrap bootstrap.Bootstrap,
) error {
@ -463,21 +504,11 @@ func (n *network) initialize(
return fmt.Errorf("Reloading network bootstrap: %w", err)
}
ctx = context.WithoutCancel(ctx)
ctx, cancel := context.WithCancel(ctx)
n.wg.Add(1)
go func() {
defer n.wg.Done()
<-n.shutdownCh
cancel()
}()
n.wg.Add(1)
go func() {
defer n.wg.Done()
n.reloadLoop(ctx)
n.logger.Debug(ctx, "Daemon reload loop stopped")
}()
n.periodically(
n.logger.WithNamespace("reloadHosts"),
n.reloadHosts,
3*time.Minute,
)
return nil
}
@ -575,25 +606,6 @@ func (n *network) reloadHosts(ctx context.Context) error {
return nil
}
func (n *network) reloadLoop(ctx context.Context) {
const period = 3 * time.Minute
ticker := time.NewTicker(period)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := n.reloadHosts(ctx); err != nil {
n.logger.Error(ctx, "Attempting to reload", err)
continue
}
}
}
}
// returns the bootstrap prior to the reload being applied.
func (n *network) reload(
ctx context.Context,
@ -966,10 +978,11 @@ func (n *network) GetNetworkCreationParams(
}
func (n *network) Shutdown() error {
close(n.shutdownCh)
n.workerCancel()
n.wg.Wait()
if n.children != nil {
n.logger.Info(context.Background(), "Shutting down children")
n.children.Shutdown()
}