From faa89e57c1d871c6f5980cb9cd5daae6416e5852 Mon Sep 17 00:00:00 2001 From: jmills Date: Thu, 4 Aug 2022 23:14:56 +0000 Subject: [PATCH 1/9] Implements ShutdownCode option and ShutdownSignal os.Signal wrapper --- app.go | 34 ++++++++++++++++-- shutdown.go | 66 +++++++++++++++++++++++++++++++++-- shutdown_code_example_test.go | 55 +++++++++++++++++++++++++++++ shutdown_test.go | 58 ++++++++++++++++++++++++++++++ 4 files changed, 207 insertions(+), 6 deletions(-) create mode 100644 shutdown_code_example_test.go diff --git a/app.go b/app.go index 0f51531ea..988dd8542 100644 --- a/app.go +++ b/app.go @@ -287,9 +287,12 @@ type App struct { errorHooks []ErrorHandler validate bool // Used to signal shutdowns. - donesMu sync.Mutex // guards dones and shutdownSig - dones []chan os.Signal - shutdownSig os.Signal + donesMu sync.Mutex // guards dones and shutdownSig + dones []chan os.Signal + shutdownSig os.Signal + waitsMu sync.Mutex // guards waits and shutdownCode + waits []chan ShutdownSignal + shutdownSignal *ShutdownSignal osExit func(code int) // os.Exit override; used for testing only } @@ -683,6 +686,31 @@ func (app *App) Done() <-chan os.Signal { return c } +func (app *App) wait() <-chan ShutdownSignal { + c := make(chan ShutdownSignal, 1) + + app.waitsMu.Lock() + defer app.waitsMu.Unlock() + + if app.shutdownSignal != nil { + c <- *app.shutdownSignal + return c + } + + app.waits = append(app.waits, c) + return c +} + +func (app *App) Wait(ctx context.Context) (ShutdownSignal, error) { + c := app.wait() + select { + case s := <-c: + return s, nil + case <-ctx.Done(): + return ShutdownSignal{}, ctx.Err() + } +} + // StartTimeout returns the configured startup timeout. Apps default to using // DefaultTimeout, but users can configure this behavior using the // StartTimeout option. diff --git a/shutdown.go b/shutdown.go index d5b8488c0..d5f125d14 100644 --- a/shutdown.go +++ b/shutdown.go @@ -23,6 +23,8 @@ package fx import ( "fmt" "os" + + "go.uber.org/multierr" ) // Shutdowner provides a method that can manually trigger the shutdown of the @@ -39,8 +41,26 @@ type ShutdownOption interface { apply(*shutdowner) } +type shutdownCode int + +func (c shutdownCode) apply(s *shutdowner) { + s.exitCode = int(c) +} + +// ShutdownCode implements a shutdown option that allows a user specify the +// os.Exit code that an application should exit with. +func ShutdownCode(code int) ShutdownOption { + return shutdownCode(code) +} + type shutdowner struct { - app *App + exitCode int + app *App +} + +type ShutdownSignal struct { + Signal os.Signal + ExitCode int } // Shutdown broadcasts a signal to all of the application's Done channels @@ -49,14 +69,25 @@ type shutdowner struct { // In practice this means Shutdowner.Shutdown should not be called from an // fx.Invoke, but from a fx.Lifecycle.OnStart hook. func (s *shutdowner) Shutdown(opts ...ShutdownOption) error { - return s.app.broadcastSignal(_sigTERM) + for _, opt := range opts { + opt.apply(s) + } + + return s.app.broadcastSignal(_sigTERM, s.exitCode) } func (app *App) shutdowner() Shutdowner { return &shutdowner{app: app} } -func (app *App) broadcastSignal(signal os.Signal) error { +func (app *App) broadcastSignal(signal os.Signal, code int) error { + return multierr.Combine( + app.broadcastDoneSignal(signal), + app.broadcastWaitSignal(signal, code), + ) +} + +func (app *App) broadcastDoneSignal(signal os.Signal) error { app.donesMu.Lock() defer app.donesMu.Unlock() @@ -81,3 +112,32 @@ func (app *App) broadcastSignal(signal os.Signal) error { return nil } + +func (app *App) broadcastWaitSignal(signal os.Signal, code int) error { + app.waitsMu.Lock() + defer app.waitsMu.Unlock() + + app.shutdownSignal = &ShutdownSignal{ + Signal: signal, + ExitCode: code, + } + + var unsent int + for _, wait := range app.waits { + select { + case wait <- *app.shutdownSignal: + default: + // shutdown called when wait channel has already received a + // termination signal that has not been cleared + unsent++ + } + } + + if unsent != 0 { + return fmt.Errorf("failed to send %v codes to %v out of %v channels", + signal, unsent, len(app.waits), + ) + } + + return nil +} diff --git a/shutdown_code_example_test.go b/shutdown_code_example_test.go new file mode 100644 index 000000000..9ac69001a --- /dev/null +++ b/shutdown_code_example_test.go @@ -0,0 +1,55 @@ +// Copyright (c) 2022 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package fx_test + +import ( + "context" + "fmt" + "time" + + "go.uber.org/fx" +) + +func ExampleShutdownCode() { + app := fx.New( + fx.Invoke(func(shutdowner fx.Shutdowner) { + // Call the shutdowner Shutdown method with a shutdown code + // option + shutdowner.Shutdown(fx.ShutdownCode(1)) + }), + ) + + app.Run() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + shutdown, err := app.Wait(ctx) + + if err != nil { + panic(err) + } + + fmt.Printf("os.Exit(%v)\n", shutdown.ExitCode) + + // Output: + // os.Exit(1) +} diff --git a/shutdown_test.go b/shutdown_test.go index b6af93f13..6f96547e8 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -22,8 +22,10 @@ package fx_test import ( "context" + "fmt" "sync" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -87,6 +89,62 @@ func TestShutdown(t *testing.T) { assert.NotNil(t, <-done1, "done channel 1 did not receive signal") assert.NotNil(t, <-done2, "done channel 2 did not receive signal") }) + + t.Run("shutdown app with exit code(s)", func(t *testing.T) { + t.Parallel() + + t.Run("default", func(t *testing.T) { + t.Parallel() + var s fx.Shutdowner + app := fxtest.New(t, fx.Populate(&s)) + + done := app.Done() + defer app.RequireStart().RequireStop() + + assert.NoError(t, s.Shutdown(), "error returned from first shutdown call") + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + signal, err := app.Wait(ctx) + assert.NoError(t, err, "error in app wait") + assert.NotEmpty(t, signal, "no shutdown signal") + assert.NotNil(t, signal.Signal) + assert.Zero(t, signal.ExitCode) + assert.Equal(t, signal.Signal, <-done) + assert.NoError(t, ctx.Err()) + }) + + for expected := 0; expected <= 3; expected++ { + expected := expected + t.Run(fmt.Sprintf("with exit code %v", expected), func(t *testing.T) { + t.Parallel() + var s fx.Shutdowner + app := fxtest.New( + t, + fx.Populate(&s), + ) + + done := app.Done() + defer app.RequireStart().RequireStop() + + assert.NoError( + t, + s.Shutdown(fx.ShutdownCode(expected)), + "error in app shutdown", + ) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + signal, err := app.Wait(ctx) + assert.NoError(t, err, "error in app wait") + assert.NotEmpty(t, signal, "no shutdown signal") + assert.NotNil(t, signal.Signal) + assert.Equal(t, expected, signal.ExitCode) + assert.Equal(t, signal.Signal, <-done) + }) + } + }) } func TestDataRace(t *testing.T) { From 995e6721aff9cc0f21954af041d35a9aef578041 Mon Sep 17 00:00:00 2001 From: jmills Date: Wed, 28 Sep 2022 05:41:42 +0000 Subject: [PATCH 2/9] change Wait sig to return channel + Add error type for unsent signals --- app.go | 12 +------ shutdown.go | 34 ++++++++++++++++---- shutdown_code_example_test.go | 14 ++------- shutdown_test.go | 59 ++++++++++++++++++++++++----------- 4 files changed, 72 insertions(+), 47 deletions(-) diff --git a/app.go b/app.go index 988dd8542..c84383185 100644 --- a/app.go +++ b/app.go @@ -686,7 +686,7 @@ func (app *App) Done() <-chan os.Signal { return c } -func (app *App) wait() <-chan ShutdownSignal { +func (app *App) Wait() <-chan ShutdownSignal { c := make(chan ShutdownSignal, 1) app.waitsMu.Lock() @@ -701,16 +701,6 @@ func (app *App) wait() <-chan ShutdownSignal { return c } -func (app *App) Wait(ctx context.Context) (ShutdownSignal, error) { - c := app.wait() - select { - case s := <-c: - return s, nil - case <-ctx.Done(): - return ShutdownSignal{}, ctx.Err() - } -} - // StartTimeout returns the configured startup timeout. Apps default to using // DefaultTimeout, but users can configure this behavior using the // StartTimeout option. diff --git a/shutdown.go b/shutdown.go index d5f125d14..dcec1b23a 100644 --- a/shutdown.go +++ b/shutdown.go @@ -105,9 +105,11 @@ func (app *App) broadcastDoneSignal(signal os.Signal) error { } if unsent != 0 { - return fmt.Errorf("failed to send %v signal to %v out of %v channels", - signal, unsent, len(app.dones), - ) + return ErrOnUnsentSignal{ + Signal: signal, + Unsent: unsent, + Channels: len(app.dones), + } } return nil @@ -134,10 +136,30 @@ func (app *App) broadcastWaitSignal(signal os.Signal, code int) error { } if unsent != 0 { - return fmt.Errorf("failed to send %v codes to %v out of %v channels", - signal, unsent, len(app.waits), - ) + return ErrOnUnsentSignal{ + Signal: signal, + Unsent: unsent, + Code: code, + Channels: len(app.waits), + } } return nil } + +// ErrOnUnsentSignal ... TBD +type ErrOnUnsentSignal struct { + Signal os.Signal + Unsent int + Code int + Channels int +} + +func (err ErrOnUnsentSignal) Error() string { + return fmt.Sprintf( + "failed to send %v signal to %v out of %v channels", + err.Signal, + err.Unsent, + err.Channels, + ) +} diff --git a/shutdown_code_example_test.go b/shutdown_code_example_test.go index 9ac69001a..a96d1982d 100644 --- a/shutdown_code_example_test.go +++ b/shutdown_code_example_test.go @@ -21,10 +21,7 @@ package fx_test import ( - "context" "fmt" - "time" - "go.uber.org/fx" ) @@ -39,16 +36,11 @@ func ExampleShutdownCode() { app.Run() - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - - shutdown, err := app.Wait(ctx) + wait := app.Wait() - if err != nil { - panic(err) - } + signal := <-wait - fmt.Printf("os.Exit(%v)\n", shutdown.ExitCode) + fmt.Printf("os.Exit(%v)\n", signal.ExitCode) // Output: // os.Exit(1) diff --git a/shutdown_test.go b/shutdown_test.go index 6f96547e8..517bae739 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -22,10 +22,10 @@ package fx_test import ( "context" + "errors" "fmt" "sync" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -98,20 +98,45 @@ func TestShutdown(t *testing.T) { var s fx.Shutdowner app := fxtest.New(t, fx.Populate(&s)) - done := app.Done() defer app.RequireStart().RequireStop() + waits := append( + []<-chan fx.ShutdownSignal{}, + app.Wait(), + app.Wait(), + ) + assert.NoError(t, s.Shutdown(), "error returned from first shutdown call") - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - - signal, err := app.Wait(ctx) - assert.NoError(t, err, "error in app wait") - assert.NotEmpty(t, signal, "no shutdown signal") - assert.NotNil(t, signal.Signal) - assert.Zero(t, signal.ExitCode) - assert.Equal(t, signal.Signal, <-done) - assert.NoError(t, ctx.Err()) + + for _, ch := range waits { + signal := <-ch + assert.NotEmpty(t, signal, "no shutdown signal") + assert.NotNil(t, signal.Signal) + assert.Zero(t, signal.ExitCode) + } + }) + + t.Run("unsent", func(t *testing.T) { + t.Parallel() + + var s fx.Shutdowner + app := fxtest.New( + t, + fx.Populate(&s), + ) + + wait := app.Wait() + defer app.RequireStart().RequireStop() + assert.NoError(t, s.Shutdown(), "error returned from first shutdown call") + + err := s.Shutdown() + assert.Error(t, err) + var o fx.ErrOnUnsentSignal + assert.True(t, errors.As(err, &o)) + + assert.Equal(t, 1, o.Unsent) + assert.Equal(t, 1, o.Channels) + assert.NotNil(t, <-wait) }) for expected := 0; expected <= 3; expected++ { @@ -124,24 +149,20 @@ func TestShutdown(t *testing.T) { fx.Populate(&s), ) - done := app.Done() defer app.RequireStart().RequireStop() + wait := app.Wait() + assert.NoError( t, s.Shutdown(fx.ShutdownCode(expected)), "error in app shutdown", ) - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - - signal, err := app.Wait(ctx) - assert.NoError(t, err, "error in app wait") + signal := <-wait assert.NotEmpty(t, signal, "no shutdown signal") assert.NotNil(t, signal.Signal) assert.Equal(t, expected, signal.ExitCode) - assert.Equal(t, signal.Signal, <-done) }) } }) From c9972ddc90b52c2a79024bdb7c4dd9f31d488153 Mon Sep 17 00:00:00 2001 From: Abhinav Gupta Date: Wed, 28 Sep 2022 16:37:35 -0400 Subject: [PATCH 3/9] App.Done/App.Wait: Share internals This is a proposed change to #912 by @jasonmills that DRYs up internal state management by unifying `chan os.Signal` and `chan ShutdownSignal` into a single interface as suggested in this comment: https://github.com/uber-go/fx/pull/912#discussion_r982790228 This change isn't quite right because mapping os.Signal to a ShutdownSignal currently relies on a goroutine which isn't reliably shut down -- so we have leaking tests. Note that this also fixes a behavioral bug in #912: `Wait()` channels would not resolve if a plain signal was received. --- app.go | 84 ++++++++++++++++++++++------------ shutdown.go | 117 ++++++++++++++++++++++++++--------------------- shutdown_test.go | 2 +- 3 files changed, 121 insertions(+), 82 deletions(-) diff --git a/app.go b/app.go index c84383185..40d879e3a 100644 --- a/app.go +++ b/app.go @@ -275,6 +275,7 @@ type App struct { err error clock fxclock.Clock lifecycle *lifecycleWrapper + stopch chan struct{} // closed when Stop is called container *dig.Container root *module @@ -286,6 +287,7 @@ type App struct { // Decides how we react to errors when building the graph. errorHooks []ErrorHandler validate bool + // Used to signal shutdowns. donesMu sync.Mutex // guards dones and shutdownSig dones []chan os.Signal @@ -294,6 +296,10 @@ type App struct { waits []chan ShutdownSignal shutdownSignal *ShutdownSignal + // Used to make sure Start/Stop is called only once. + runStart sync.Once + runStop sync.Once + osExit func(code int) // os.Exit override; used for testing only } @@ -411,7 +417,11 @@ func New(opts ...Option) *App { // user gave us. For the last case, however, we need to fall // back to what was provided to fx.Logger if fx.WithLogger // fails. - log: logger, + log: logger, + clock: fxclock.System, + startTimeout: DefaultTimeout, + stopTimeout: DefaultTimeout, + stopch: make(chan struct{}), } app.modules = append(app.modules, app.root) @@ -649,9 +659,12 @@ func (app *App) start(ctx context.Context) error { // called are executed. However, all those hooks are executed, even if some // fail. func (app *App) Stop(ctx context.Context) (err error) { - defer func() { - app.log().LogEvent(&fxevent.Stopped{Err: err}) - }() + app.runStop.Do(func() { + // Protect the Stop hooks from being called multiple times. + defer func() { + app.log.LogEvent(&fxevent.Stopped{Err: err}) + close(app.stopch) + }() return withTimeout(ctx, &withTimeoutParams{ hook: _onStopHook, @@ -669,36 +682,49 @@ func (app *App) Stop(ctx context.Context) (err error) { // Alternatively, a signal can be broadcast to all done channels manually by // using the Shutdown functionality (see the Shutdowner documentation for details). func (app *App) Done() <-chan os.Signal { - c := make(chan os.Signal, 1) - - app.donesMu.Lock() - defer app.donesMu.Unlock() - // If shutdown signal has been received already - // send it and return. If not, wait for user to send a termination - // signal. - if app.shutdownSig != nil { - c <- app.shutdownSig - return c - } - - signal.Notify(c, os.Interrupt, _sigINT, _sigTERM) - app.dones = append(app.dones, c) - return c + rcv, ch := newOSSignalReceiver() + app.appendSignalReceiver(rcv) + return ch } func (app *App) Wait() <-chan ShutdownSignal { - c := make(chan ShutdownSignal, 1) - - app.waitsMu.Lock() - defer app.waitsMu.Unlock() + rcv, ch := newShutdownSignalReceiver() + app.appendSignalReceiver(rcv) + return ch +} - if app.shutdownSignal != nil { - c <- *app.shutdownSignal - return c - } +func (app *App) appendSignalReceiver(r signalReceiver) { + app.shutdownMu.Lock() + defer app.shutdownMu.Unlock() - app.waits = append(app.waits, c) - return c + // If shutdown signal has been received already + // send it and return. + // If not, wait for user to send a termination signal. + if sig := app.shutdownSig; sig != nil { + // Ignore the error from ReceiveSignal. + // This is a newly created channel and can't possibly be + // blocked. + _ = r.ReceiveShutdownSignal(*sig) + return + } + + app.sigReceivers = append(app.sigReceivers, r) + + // The first time either Wait or Done is called, + // register an OS signal handler + // and make that broadcast the signal to all sigReceivers + // regardless of whether they're Wait or Done based. + app.signalOnce.Do(func() { + sigch := make(chan os.Signal, 1) + signal.Notify(sigch, os.Interrupt, _sigINT, _sigTERM) + go func() { + select { + case sig := <-sigch: + app.broadcastSignal(sig, 1) + case <-app.stopch: + } + }() + }) } // StartTimeout returns the configured startup timeout. Apps default to using diff --git a/shutdown.go b/shutdown.go index dcec1b23a..3ac7cea5c 100644 --- a/shutdown.go +++ b/shutdown.go @@ -21,12 +21,55 @@ package fx import ( + "errors" "fmt" "os" "go.uber.org/multierr" ) +var errReceiverBlocked = errors.New("receiver is blocked") + +type signalReceiver interface { + ReceiveShutdownSignal(ShutdownSignal) error +} + +type osSignalReceiver struct{ ch chan<- os.Signal } + +var _ signalReceiver = (*osSignalReceiver)(nil) + +func newOSSignalReceiver() (*osSignalReceiver, <-chan os.Signal) { + ch := make(chan os.Signal, 1) + return &osSignalReceiver{ch: ch}, ch +} + +func (r *osSignalReceiver) ReceiveShutdownSignal(sig ShutdownSignal) error { + select { + case r.ch <- sig.Signal: + return nil + default: + return errReceiverBlocked + } +} + +type shutdownSignalReceiver struct{ ch chan<- ShutdownSignal } + +var _ signalReceiver = (*shutdownSignalReceiver)(nil) + +func newShutdownSignalReceiver() (*shutdownSignalReceiver, <-chan ShutdownSignal) { + ch := make(chan ShutdownSignal, 1) + return &shutdownSignalReceiver{ch: ch}, ch +} + +func (r *shutdownSignalReceiver) ReceiveShutdownSignal(sig ShutdownSignal) error { + select { + case r.ch <- sig: + return nil + default: + return errReceiverBlocked + } +} + // Shutdowner provides a method that can manually trigger the shutdown of the // application by sending a signal to all open Done channels. Shutdowner works // on applications using Run as well as Start, Done, and Stop. The Shutdowner is @@ -81,70 +124,40 @@ func (app *App) shutdowner() Shutdowner { } func (app *App) broadcastSignal(signal os.Signal, code int) error { - return multierr.Combine( - app.broadcastDoneSignal(signal), - app.broadcastWaitSignal(signal, code), - ) -} - -func (app *App) broadcastDoneSignal(signal os.Signal) error { - app.donesMu.Lock() - defer app.donesMu.Unlock() - - app.shutdownSig = signal + app.shutdownMu.Lock() + defer app.shutdownMu.Unlock() - var unsent int - for _, done := range app.dones { - select { - case done <- signal: - default: - // shutdown called when done channel has already received a - // termination signal that has not been cleared - unsent++ - } - } - - if unsent != 0 { - return ErrOnUnsentSignal{ - Signal: signal, - Unsent: unsent, - Channels: len(app.dones), - } - } - - return nil -} - -func (app *App) broadcastWaitSignal(signal os.Signal, code int) error { - app.waitsMu.Lock() - defer app.waitsMu.Unlock() - - app.shutdownSignal = &ShutdownSignal{ + sig := ShutdownSignal{ Signal: signal, ExitCode: code, } + app.shutdownSig = &sig - var unsent int - for _, wait := range app.waits { - select { - case wait <- *app.shutdownSignal: - default: - // shutdown called when wait channel has already received a - // termination signal that has not been cleared - unsent++ + var ( + unsent int + resultErr error + ) + for _, rcv := range app.sigReceivers { + // shutdown called when done channel has already received a + // termination signal that has not been cleared + if err := rcv.ReceiveShutdownSignal(sig); err != nil { + if errors.Is(err, errReceiverBlocked) { + unsent++ + } else { + resultErr = multierr.Append(resultErr, err) + } } } if unsent != 0 { - return ErrOnUnsentSignal{ + resultErr = multierr.Append(resultErr, &ErrOnUnsentSignal{ Signal: signal, Unsent: unsent, - Code: code, - Channels: len(app.waits), - } + Channels: len(app.sigReceivers), + }) } - return nil + return resultErr } // ErrOnUnsentSignal ... TBD @@ -155,7 +168,7 @@ type ErrOnUnsentSignal struct { Channels int } -func (err ErrOnUnsentSignal) Error() string { +func (err *ErrOnUnsentSignal) Error() string { return fmt.Sprintf( "failed to send %v signal to %v out of %v channels", err.Signal, diff --git a/shutdown_test.go b/shutdown_test.go index 517bae739..20b48de51 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -131,7 +131,7 @@ func TestShutdown(t *testing.T) { err := s.Shutdown() assert.Error(t, err) - var o fx.ErrOnUnsentSignal + var o *fx.ErrOnUnsentSignal assert.True(t, errors.As(err, &o)) assert.Equal(t, 1, o.Unsent) From 638daf4f6859db6d102caf63cc577daee1ef111a Mon Sep 17 00:00:00 2001 From: Abhinav Gupta Date: Wed, 28 Sep 2022 16:42:04 -0400 Subject: [PATCH 4/9] app.Run: Respect exit code specified by ShutdownSignal --- app.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/app.go b/app.go index 40d879e3a..bf65cd07c 100644 --- a/app.go +++ b/app.go @@ -557,12 +557,12 @@ func (app *App) Run() { // Historically, we do not os.Exit(0) even though most applications // cede control to Fx with they call app.Run. To avoid a breaking // change, never os.Exit for success. - if code := app.run(app.Done()); code != 0 { + if code := app.run(app.Wait()); code != 0 { app.exit(code) } } -func (app *App) run(done <-chan os.Signal) (exitCode int) { +func (app *App) run(done <-chan ShutdownSignal) (exitCode int) { startCtx, cancel := app.clock.WithTimeout(context.Background(), app.StartTimeout()) defer cancel() @@ -571,13 +571,13 @@ func (app *App) run(done <-chan os.Signal) (exitCode int) { } sig := <-done - app.log().LogEvent(&fxevent.Stopping{Signal: sig}) + app.log.LogEvent(&fxevent.Stopping{Signal: sig.Signal}) stopCtx, cancel := app.clock.WithTimeout(context.Background(), app.StopTimeout()) defer cancel() if err := app.Stop(stopCtx); err != nil { - return 1 + return sig.ExitCode } return 0 From 7ebd1301413313a18608f55c7a2312b3f5302ede Mon Sep 17 00:00:00 2001 From: jmills Date: Mon, 10 Oct 2022 22:57:29 +0000 Subject: [PATCH 5/9] fix failing test after rebase --- app.go | 30 +++++++++++++++++++----------- app_internal_test.go | 5 ++--- app_test.go | 2 +- shutdown.go | 7 +++---- shutdown_test.go | 6 ------ 5 files changed, 25 insertions(+), 25 deletions(-) diff --git a/app.go b/app.go index bf65cd07c..814c7580f 100644 --- a/app.go +++ b/app.go @@ -289,12 +289,10 @@ type App struct { validate bool // Used to signal shutdowns. - donesMu sync.Mutex // guards dones and shutdownSig - dones []chan os.Signal - shutdownSig os.Signal - waitsMu sync.Mutex // guards waits and shutdownCode - waits []chan ShutdownSignal - shutdownSignal *ShutdownSignal + shutdownMu sync.Mutex + shutdownSig *ShutdownSignal + sigReceivers []signalReceiver + signalOnce sync.Once // Used to make sure Start/Stop is called only once. runStart sync.Once @@ -577,6 +575,10 @@ func (app *App) run(done <-chan ShutdownSignal) (exitCode int) { defer cancel() if err := app.Stop(stopCtx); err != nil { + // if we encounter a timeout during stop, force exit code 1 + if errors.Is(err, context.DeadlineExceeded) { + return 1 + } return sig.ExitCode } @@ -618,7 +620,9 @@ var ( // encountered any errors in application initialization. func (app *App) Start(ctx context.Context) (err error) { defer func() { - app.log().LogEvent(&fxevent.Started{Err: err}) + app.runStart.Do(func() { + app.log.LogEvent(&fxevent.Started{Err: err}) + }) }() if app.err != nil { @@ -659,19 +663,23 @@ func (app *App) start(ctx context.Context) error { // called are executed. However, all those hooks are executed, even if some // fail. func (app *App) Stop(ctx context.Context) (err error) { - app.runStop.Do(func() { + + defer func() { // Protect the Stop hooks from being called multiple times. - defer func() { + app.runStop.Do(func() { app.log.LogEvent(&fxevent.Stopped{Err: err}) close(app.stopch) - }() + }) + }() - return withTimeout(ctx, &withTimeoutParams{ + err = withTimeout(ctx, &withTimeoutParams{ hook: _onStopHook, callback: app.lifecycle.Stop, lifecycle: app.lifecycle, log: app.log(), }) + + return } // Done returns a channel of signals to block on after starting the diff --git a/app_internal_test.go b/app_internal_test.go index 82de85e2c..65263e1fd 100644 --- a/app_internal_test.go +++ b/app_internal_test.go @@ -22,7 +22,6 @@ package fx import ( "fmt" - "os" "sync" "testing" @@ -41,7 +40,7 @@ func TestAppRun(t *testing.T) { app := New( WithLogger(func() fxevent.Logger { return spy }), ) - done := make(chan os.Signal) + done := make(chan ShutdownSignal) var wg sync.WaitGroup wg.Add(1) @@ -50,7 +49,7 @@ func TestAppRun(t *testing.T) { app.run(done) }() - done <- _sigINT + done <- ShutdownSignal{Signal: _sigINT} wg.Wait() assert.Equal(t, []string{ diff --git a/app_test.go b/app_test.go index d99a58afd..a887fbb58 100644 --- a/app_test.go +++ b/app_test.go @@ -917,7 +917,7 @@ func TestAppRunTimeout(t *testing.T) { err, _ := errv.Interface().(error) assert.ErrorIs(t, err, context.DeadlineExceeded, - "should fail because of a timeout") + "should fail because of a timeout: %v", err) }) } } diff --git a/shutdown.go b/shutdown.go index 3ac7cea5c..b915824c8 100644 --- a/shutdown.go +++ b/shutdown.go @@ -150,7 +150,7 @@ func (app *App) broadcastSignal(signal os.Signal, code int) error { } if unsent != 0 { - resultErr = multierr.Append(resultErr, &ErrOnUnsentSignal{ + resultErr = multierr.Append(resultErr, &errOnUnsentSignal{ Signal: signal, Unsent: unsent, Channels: len(app.sigReceivers), @@ -160,15 +160,14 @@ func (app *App) broadcastSignal(signal os.Signal, code int) error { return resultErr } -// ErrOnUnsentSignal ... TBD -type ErrOnUnsentSignal struct { +type errOnUnsentSignal struct { Signal os.Signal Unsent int Code int Channels int } -func (err *ErrOnUnsentSignal) Error() string { +func (err *errOnUnsentSignal) Error() string { return fmt.Sprintf( "failed to send %v signal to %v out of %v channels", err.Signal, diff --git a/shutdown_test.go b/shutdown_test.go index 20b48de51..31dece916 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -22,7 +22,6 @@ package fx_test import ( "context" - "errors" "fmt" "sync" "testing" @@ -131,11 +130,6 @@ func TestShutdown(t *testing.T) { err := s.Shutdown() assert.Error(t, err) - var o *fx.ErrOnUnsentSignal - assert.True(t, errors.As(err, &o)) - - assert.Equal(t, 1, o.Unsent) - assert.Equal(t, 1, o.Channels) assert.NotNil(t, <-wait) }) From b4e5c2a13cf585464d619f17e80062c0c22eb67c Mon Sep 17 00:00:00 2001 From: jmills Date: Thu, 27 Oct 2022 20:33:21 +0000 Subject: [PATCH 6/9] rebase from upstream --- app.go | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/app.go b/app.go index 814c7580f..096ff74a1 100644 --- a/app.go +++ b/app.go @@ -400,7 +400,9 @@ func New(opts ...Option) *App { clock: fxclock.System, startTimeout: DefaultTimeout, stopTimeout: DefaultTimeout, + stopch: make(chan struct{}), } + app.root = &module{ app: app, // We start with a logger that writes to stderr. One of the @@ -415,11 +417,7 @@ func New(opts ...Option) *App { // user gave us. For the last case, however, we need to fall // back to what was provided to fx.Logger if fx.WithLogger // fails. - log: logger, - clock: fxclock.System, - startTimeout: DefaultTimeout, - stopTimeout: DefaultTimeout, - stopch: make(chan struct{}), + log: logger, } app.modules = append(app.modules, app.root) @@ -569,7 +567,7 @@ func (app *App) run(done <-chan ShutdownSignal) (exitCode int) { } sig := <-done - app.log.LogEvent(&fxevent.Stopping{Signal: sig.Signal}) + app.log().LogEvent(&fxevent.Stopping{Signal: sig.Signal}) stopCtx, cancel := app.clock.WithTimeout(context.Background(), app.StopTimeout()) defer cancel() @@ -621,7 +619,7 @@ var ( func (app *App) Start(ctx context.Context) (err error) { defer func() { app.runStart.Do(func() { - app.log.LogEvent(&fxevent.Started{Err: err}) + app.log().LogEvent(&fxevent.Started{Err: err}) }) }() @@ -667,7 +665,7 @@ func (app *App) Stop(ctx context.Context) (err error) { defer func() { // Protect the Stop hooks from being called multiple times. app.runStop.Do(func() { - app.log.LogEvent(&fxevent.Stopped{Err: err}) + app.log().LogEvent(&fxevent.Stopped{Err: err}) close(app.stopch) }) }() From 848742ab2220333456e2877082c7d1dbb449f6b1 Mon Sep 17 00:00:00 2001 From: jmills Date: Thu, 27 Oct 2022 22:22:12 +0000 Subject: [PATCH 7/9] fix goleak and data races around stop channels for signal broadcast by using mutex --- app.go | 46 +++++++++++++++++++++++++++++++++++++++------- shutdown_test.go | 1 + 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/app.go b/app.go index 096ff74a1..958a20e96 100644 --- a/app.go +++ b/app.go @@ -275,7 +275,9 @@ type App struct { err error clock fxclock.Clock lifecycle *lifecycleWrapper - stopch chan struct{} // closed when Stop is called + + stopch chan struct{} // closed when Stop is called + stopChLock sync.RWMutex // mutex for init and closing of stopch container *dig.Container root *module @@ -400,7 +402,6 @@ func New(opts ...Option) *App { clock: fxclock.System, startTimeout: DefaultTimeout, stopTimeout: DefaultTimeout, - stopch: make(chan struct{}), } app.root = &module{ @@ -563,6 +564,7 @@ func (app *App) run(done <-chan ShutdownSignal) (exitCode int) { defer cancel() if err := app.Start(startCtx); err != nil { + app.closeStopChannel() return 1 } @@ -628,6 +630,8 @@ func (app *App) Start(ctx context.Context) (err error) { return app.err } + app.initStopChannel() + return withTimeout(ctx, &withTimeoutParams{ hook: _onStartHook, callback: app.start, @@ -653,6 +657,30 @@ func (app *App) start(ctx context.Context) error { return nil } +func (app *App) initStopChannel() { + app.stopChLock.Lock() + defer app.stopChLock.Unlock() + if app.stopch == nil { + app.stopch = make(chan struct{}) + } +} + +func (app *App) stopChannel() chan struct{} { + app.stopChLock.RLock() + defer app.stopChLock.RUnlock() + ch := app.stopch + return ch +} + +func (app *App) closeStopChannel() { + app.stopChLock.Lock() + defer app.stopChLock.Unlock() + if app.stopch != nil { + close(app.stopch) + app.stopch = nil + } +} + // Stop gracefully stops the application. It executes any registered OnStop // hooks in reverse order, so that each constructor's stop hooks are called // before its dependencies' stop hooks. @@ -666,7 +694,7 @@ func (app *App) Stop(ctx context.Context) (err error) { // Protect the Stop hooks from being called multiple times. app.runStop.Do(func() { app.log().LogEvent(&fxevent.Stopped{Err: err}) - close(app.stopch) + app.closeStopChannel() }) }() @@ -724,10 +752,14 @@ func (app *App) appendSignalReceiver(r signalReceiver) { sigch := make(chan os.Signal, 1) signal.Notify(sigch, os.Interrupt, _sigINT, _sigTERM) go func() { - select { - case sig := <-sigch: - app.broadcastSignal(sig, 1) - case <-app.stopch: + // if the stop channel is nil; that means that the app was never started + // thus, do not broadcast any signals + if stopch := app.stopChannel(); stopch != nil { + select { + case sig := <-sigch: + app.broadcastSignal(sig, 1) + case <-stopch: + } } }() }) diff --git a/shutdown_test.go b/shutdown_test.go index 31dece916..86b1592b9 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -171,6 +171,7 @@ func TestDataRace(t *testing.T) { fx.Populate(&s), ) require.NoError(t, app.Start(context.Background()), "error starting app") + defer require.NoError(t, app.Stop(context.Background()), "error stopping app") const N = 50 ready := make(chan struct{}) // used to orchestrate goroutines for Done() and ShutdownOption() From a69d369244bea8f961a3a76286c481d67556c586 Mon Sep 17 00:00:00 2001 From: jmills Date: Thu, 27 Oct 2022 22:26:42 +0000 Subject: [PATCH 8/9] update unsent error string --- shutdown.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/shutdown.go b/shutdown.go index b915824c8..9b6b992bf 100644 --- a/shutdown.go +++ b/shutdown.go @@ -169,7 +169,7 @@ type errOnUnsentSignal struct { func (err *errOnUnsentSignal) Error() string { return fmt.Sprintf( - "failed to send %v signal to %v out of %v channels", + "send %v signal: %v/%v channels are blocked", err.Signal, err.Unsent, err.Channels, From 196f6997951e626aaf17a7af1b1a9fe12da85164 Mon Sep 17 00:00:00 2001 From: jmills Date: Fri, 28 Oct 2022 21:11:52 +0000 Subject: [PATCH 9/9] fix unit test for error string --- shutdown_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/shutdown_test.go b/shutdown_test.go index 86b1592b9..c74367478 100644 --- a/shutdown_test.go +++ b/shutdown_test.go @@ -65,7 +65,7 @@ func TestShutdown(t *testing.T) { defer app.RequireStart().RequireStop() assert.NoError(t, s.Shutdown(), "error returned from first shutdown call") - assert.EqualError(t, s.Shutdown(), "failed to send terminated signal to 1 out of 1 channels", + assert.EqualError(t, s.Shutdown(), "send terminated signal: 1/1 channels are blocked", "unexpected error returned when shutdown is called with a blocked channel") assert.NotNil(t, <-done, "done channel did not receive signal") })