dgroup

package

v1.3.1 Latest Latest Go to latest Published: Aug 14, 2023 License: Apache-2.0 Imports: 15 Imported by: 30

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/datawire/dlib

Links

Open Source Insights

Documentation ¶

Rendered for

Overview ¶

Package dgroup provides tools for managing groups of goroutines.

The main part of this is Group, but the naming utilities may be useful outside of that.

dgroup should be the goroutine-group-management abstraction used by all new code at Ambassador Labs. The principle remaining limitation (at least when compared to the *other* Ambassador Labs "goroutine group" library that shall-not-be-named) is that dgroup does not have a notion of dependencies. Not having a notion of dependencies implies a few things:

it does not have a notion of "readiness", as it doesn't have any dependents that would block on a goroutine becoming ready
it launches worker goroutines right away when you call .Go(), as it doesn't have any dependencies that would block the worker from starting

So, if you need to enforce ordering requirements during goroutine startup and shutdown, then (for now, anyway) you'll need to implement that separately on top of this library.

Example (SignalHandling1) ¶

This example shows the signal handler triggering a soft shutdown when the user hits Ctrl-C, and what the related logging looks like.

if !ensureProcessGroup() {
	// Needed for signal handling in tests on Windows.
	return
}
exEvents := make(chan struct{})
go func() {
	// This goroutine isn't "part of" the example, but
	// simulates the user doing things, in order to drive
	// the example.
	self, _ := os.FindProcess(os.Getpid())

	<-exEvents // wait for things to get running

	// Simulate the user hitting Ctrl-C: This will log
	// that a signal was received, and trigger a
	// graceful-shutdown, logging that it's triggered
	// because of a signal.
	_ = sigint.SendInterrupt(self)

	// The worker goroutine will then respond to the
	// graceful-shutdown, and Wait() will then log each of
	// the goroutines' "final" statuses and return.
}()

ctx := baseContext()
group := dgroup.NewGroup(ctx, dgroup.GroupConfig{
	EnableSignalHandling: true,
})

group.Go("worker", func(ctx context.Context) error {
	// start up
	exEvents <- struct{}{}

	// wait for shutdown to be triggered
	<-ctx.Done()

	// quit
	return nil
})

if err := group.Wait(); err != nil {
	dlog.Errorln(ctx, "exiting with error:", err)
}

close(exEvents)

Output:

level=error msg="goroutine \":signal_handler:0\" exited with error: received signal interrupt (triggering graceful shutdown)" THREAD=":signal_handler:0"
level=info msg="shutting down (gracefully)..." THREAD=":shutdown_logger"
level=info msg="  final goroutine statuses:" THREAD=":shutdown_status"
level=info msg="    /worker          : exited" THREAD=":shutdown_status"
level=info msg="    :signal_handler:0: exited with error" THREAD=":shutdown_status"
level=error msg="exiting with error: received signal interrupt (triggering graceful shutdown)"

Example (SignalHandling2) ¶

This example shows how the signal handler behaves when a worker is poorly behaved, and doesn't quit during soft-shutdown when the user hits Ctrl-C, but does handle hard-shutdown.

if !ensureProcessGroup() {
	// Needed for signal handling in tests on Windows.
	return
}
exEvents := make(chan struct{})
go func() {
	// This goroutine isn't "part of" the example, but
	// simulates the user doing things, in order to drive
	// the example.
	self, _ := os.FindProcess(os.Getpid())

	<-exEvents // wait for things to get running

	// Simulate the user hitting Ctrl-C: This will log
	// that a signal was received, and trigger a
	// graceful-shutdown, logging that it's triggered
	// because of a signal.  However, the worker goroutine
	// will ignore it and keep running.
	_ = sigint.SendInterrupt(self)

	// wait for the soft-shutdown to be triggered
	<-exEvents

	// Simulate the user hitting Ctrl-C again: This will
	// log that a signal was received, and trigger a
	// not-so-graceful-shutdown, logging that it's
	// triggered because of a second signal; and, because
	// the user being impatient might be a first sign that
	// shutdown might be going wrong, it will log each of
	// the goroutines' statuses, so you can see which task
	// is hanging.
	_ = sigint.SendInterrupt(self)

	// The worker goroutine will then respond to the
	// not-so-graceful-shutdown, and Wait() will then log
	// each of the goroutines' "final" statuses and
	// return.
}()

ctx := baseContext()
group := dgroup.NewGroup(ctx, dgroup.GroupConfig{
	EnableSignalHandling: true,
})

group.Go("worker", func(ctx context.Context) error {
	// start up
	exEvents <- struct{}{}

	// wait for soft-shutdown to be triggered
	<-ctx.Done()

	// respond to hard-shutdown: quit, but don't do a good job of it
	exEvents <- struct{}{}

	// wait for hard-shutdown to be triggered
	<-dcontext.HardContext(ctx).Done()

	// respond to hard-shutdown: quit
	time.Sleep(1 * time.Second)
	return nil
})

if err := group.Wait(); err != nil {
	dlog.Errorln(ctx, "exiting with error:", err)
}

close(exEvents)

Output:

level=error msg="goroutine \":signal_handler:0\" exited with error: received signal interrupt (triggering graceful shutdown)" THREAD=":signal_handler:0"
level=info msg="shutting down (gracefully)..." THREAD=":shutdown_logger"
level=error msg="received signal interrupt (graceful shutdown already triggered; triggering not-so-graceful shutdown)" THREAD=":signal_handler:1"
level=error msg="  goroutine statuses:" THREAD=":signal_handler:1"
level=error msg="    /worker          : running" THREAD=":signal_handler:1"
level=error msg="    :signal_handler:0: exited with error" THREAD=":signal_handler:1"
level=info msg="shutting down (not-so-gracefully)..." THREAD=":shutdown_logger"
level=info msg="  final goroutine statuses:" THREAD=":shutdown_status"
level=info msg="    /worker          : exited" THREAD=":shutdown_status"
level=info msg="    :signal_handler:0: exited with error" THREAD=":shutdown_status"
level=error msg="exiting with error: received signal interrupt (triggering graceful shutdown)"

Example (SignalHandling3) ¶

This example shows how the signal handler behaves when a worker is poorly behaved, and doesn't quit at all.

if !ensureProcessGroup() {
	// Needed for signal handling in tests on Windows.
	return
}
exEvents := make(chan struct{})
exFinished := make(chan struct{})
go func() {
	// This goroutine isn't "part of" the example, but
	// simulates the user doing things, in order to drive
	// the example.
	self, _ := os.FindProcess(os.Getpid())

	<-exEvents // wait for things to get running

	// Simulate the user hitting Ctrl-C: This will log
	// that a signal was received, and trigger a
	// graceful-shutdown, logging that it's triggered
	// because of a signal.  However, the worker goroutine
	// will ignore it and keep running.
	_ = sigint.SendInterrupt(self)

	// wait for the soft-shutdown to be triggered
	<-exEvents

	// Simulate the user hitting Ctrl-C a 2nd time: This
	// will log that a signal was received, and trigger a
	// not-so-graceful-shutdown, logging that it's
	// triggered because of a second signal; and, because
	// the user being impatient might be a first sign that
	// shutdown might be going wrong, it will log each of
	// the goroutines' statuses, so you can see which task
	// is hanging.  However, the worker goroutine will
	// ignore it and keep running.
	_ = sigint.SendInterrupt(self)

	// wait for the hard-shutdown to be triggered
	<-exEvents

	// Simulate the user hitting Ctrl-C a 3rd time: This
	// will log that a 3rd signal was received, and
	// because the user having to hit Ctrl-C this many
	// times before something happens indicates that
	// something is probably wrong, it will not only log
	// each of the goroutines' statuses so you can see
	// which task is hanging, but it will also log a
	// stacktraces of each goroutine.  However, the worker
	// goroutine will ignore it and keep running.
	_ = sigint.SendInterrupt(self)

	// Because the worker goroutine is hanging, and not
	// responding to our shutdown signals, we've set a
	// HardShutdownTimeout that will let Wait() return
	// even though some goroutines are still running.
	// Because something is definitely wrong, once again
	// we log each of the goroutine's statuses (at
	// loglevel info), and stacktraces for each goroutine
	// (at loglevel error).
	close(exFinished)
}()
dgroup.SetStacktraceForTesting(exampleStackTrace)

ctx := baseContext()
group := dgroup.NewGroup(ctx, dgroup.GroupConfig{
	EnableSignalHandling: true,
	HardShutdownTimeout:  1 * time.Second,
})

group.Go("worker", func(ctx context.Context) error {
	// start up
	exEvents <- struct{}{}

	// wait for soft-shutdown to be triggered
	<-ctx.Done()

	// respond to soft-shutdown: hang
	exEvents <- struct{}{}

	// wait for hard-shutdown to be triggered
	<-dcontext.HardContext(ctx).Done()

	// respond to hard-shutdown: hang
	exEvents <- struct{}{}
	select {}
})

if err := group.Wait(); err != nil {
	dlog.Errorln(ctx, "exiting with error:", err)
}

<-exFinished
close(exEvents)

Output:

level=error msg="goroutine \":signal_handler:0\" exited with error: received signal interrupt (triggering graceful shutdown)" THREAD=":signal_handler:0"
level=info msg="shutting down (gracefully)..." THREAD=":shutdown_logger"
level=error msg="received signal interrupt (graceful shutdown already triggered; triggering not-so-graceful shutdown)" THREAD=":signal_handler:1"
level=error msg="  goroutine statuses:" THREAD=":signal_handler:1"
level=error msg="    /worker          : running" THREAD=":signal_handler:1"
level=error msg="    :signal_handler:0: exited with error" THREAD=":signal_handler:1"
level=info msg="shutting down (not-so-gracefully)..." THREAD=":shutdown_logger"
level=error msg="received signal interrupt (not-so-graceful shutdown already triggered)" THREAD=":signal_handler:2"
level=error msg="  goroutine statuses:" THREAD=":signal_handler:2"
level=error msg="    /worker          : running" THREAD=":signal_handler:2"
level=error msg="    :signal_handler:0: exited with error" THREAD=":signal_handler:2"
level=error msg="  goroutine stack traces:" THREAD=":signal_handler:2"
level=error msg="    goroutine 1405 [running]:" THREAD=":signal_handler:2"
level=error msg="    runtime/pprof.writeGoroutineStacks(0x6575e0, 0xc0003803c0, 0x30, 0x7f56be200788)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:693 +0x9f" THREAD=":signal_handler:2"
level=error msg="    runtime/pprof.writeGoroutine(0x6575e0, 0xc0003803c0, 0x2, 0x203000, 0xc)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:682 +0x45" THREAD=":signal_handler:2"
level=error msg="    runtime/pprof.(*Profile).WriteTo(0x7770c0, 0x6575e0, 0xc0003803c0, 0x2, 0xc000380340, 0xc00038a4d0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:331 +0x3f2" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.logGoroutineTraces(0x659e80, 0xc000392300, 0x61b619, 0x16, 0x629ca0)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:109 +0xba" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func4(0x659e80, 0xc00017c330, 0xc0001586b0, 0xc0000357c0)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:305 +0x4e5" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c330, 0xc0000b7920)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":signal_handler:2"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1 [select]:" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).Wait(0xc000153e30, 0x616a5c, 0x6)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:418 +0x139" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup_test.Example_signalHandling3()" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/example_test.go:233 +0x125" THREAD=":signal_handler:2"
level=error msg="    testing.runExample(0x61ba2a, 0x17, 0x629c90, 0x6245c4, 0x491, 0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:62 +0x209" THREAD=":signal_handler:2"
level=error msg="    testing.runExamples(0xc0001fbe98, 0x77b360, 0x5, 0x5, 0xbfe34159434d384f)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/testing/example.go:44 +0x1af" THREAD=":signal_handler:2"
level=error msg="    testing.(*M).Run(0xc0000fc000, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/testing/testing.go:1346 +0x273" THREAD=":signal_handler:2"
level=error msg="    main.main()" THREAD=":signal_handler:2"
level=error msg="    \t_testmain.go:105 +0x1c5" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1400 [IO wait]:" THREAD=":signal_handler:2"
level=error msg="    internal/poll.runtime_pollWait(0x7f56be1fb738, 0x72, 0x657920)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/netpoll.go:220 +0x55" THREAD=":signal_handler:2"
level=error msg="    internal/poll.(*pollDesc).wait(0xc00012fa58, 0x72, 0x657901, 0x749470, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_poll_runtime.go:87 +0x45" THREAD=":signal_handler:2"
level=error msg="    internal/poll.(*pollDesc).waitRead(...)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_poll_runtime.go:92" THREAD=":signal_handler:2"
level=error msg="    internal/poll.(*FD).Read(0xc00012fa40, 0xc0003b6000, 0x8000, 0x8000, 0x0, 0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_unix.go:159 +0x1a5" THREAD=":signal_handler:2"
level=error msg="    os.(*File).read(...)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/os/file_posix.go:31" THREAD=":signal_handler:2"
level=error msg="    os.(*File).Read(0xc0000a8028, 0xc0003b6000, 0x8000, 0x8000, 0x56, 0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/os/file.go:116 +0x71" THREAD=":signal_handler:2"
level=error msg="    io.copyBuffer(0x6575e0, 0xc000380140, 0x657500, 0xc0000a8028, 0xc0003b6000, 0x8000, 0x8000, 0x491, 0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/io/io.go:409 +0x12c" THREAD=":signal_handler:2"
level=error msg="    io.Copy(...)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/io/io.go:368" THREAD=":signal_handler:2"
level=error msg="    testing.runExample.func1(0xc0000a8028, 0xc00015e660)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:37 +0x85" THREAD=":signal_handler:2"
level=error msg="    created by testing.runExample" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:35 +0x176" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1404 [chan receive]:" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func3(0x659e80, 0xc00017c270, 0xc0000a5d70, 0xd)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:271 +0x4a" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c270, 0xc0000b7900)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":signal_handler:2"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1403 [select]:" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func2(0x659e80, 0xc00017c1b0, 0xc0000a5d80, 0x9)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:255 +0x254" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c1b0, 0xc000158870)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":signal_handler:2"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1443 [syscall]:" THREAD=":signal_handler:2"
level=error msg="    os/signal.signal_recv(0x6585e0)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/sigqueue.go:147 +0x9d" THREAD=":signal_handler:2"
level=error msg="    os/signal.loop()" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/os/signal/signal_unix.go:23 +0x25" THREAD=":signal_handler:2"
level=error msg="    created by os/signal.Notify.func1.1" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/os/signal/signal.go:150 +0x45" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1406 [select (no cases)]:" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup_test.Example_signalHandling3.func2(0x659e80, 0xc00017c3f0, 0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/example_test.go:230 +0x2ab" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goWorkerCtx.func1(0x0, 0x0)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:354 +0x9f" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/derrgroup.(*Group).Go.func2(0xc00017c420, 0xc00012fbc0, 0xc0000a5de5, 0x7)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:131 +0x2b" THREAD=":signal_handler:2"
level=error msg="    created by github.com/datawire/dlib/derrgroup.(*Group).Go" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:129 +0x12d" THREAD=":signal_handler:2"
level=error msg="    " THREAD=":signal_handler:2"
level=error msg="    goroutine 1407 [semacquire]:" THREAD=":signal_handler:2"
level=error msg="    sync.runtime_Semacquire(0xc00012fbcc)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/runtime/sema.go:56 +0x45" THREAD=":signal_handler:2"
level=error msg="    sync.(*WaitGroup).Wait(0xc00012fbcc)" THREAD=":signal_handler:2"
level=error msg="    \t/usr/lib/go/src/sync/waitgroup.go:130 +0x65" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/derrgroup.(*Group).Wait(...)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:99" THREAD=":signal_handler:2"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).Wait.func1(0xc00015e840, 0xc000153e30)" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:413 +0x48" THREAD=":signal_handler:2"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).Wait" THREAD=":signal_handler:2"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:412 +0x85" THREAD=":signal_handler:2"
level=info msg="  final goroutine statuses:" THREAD=":shutdown_status"
level=info msg="    /worker          : running" THREAD=":shutdown_status"
level=info msg="    :signal_handler:0: exited with error" THREAD=":shutdown_status"
level=error msg="  final goroutine stack traces:" THREAD=":shutdown_status"
level=error msg="    goroutine 1405 [running]:" THREAD=":shutdown_status"
level=error msg="    runtime/pprof.writeGoroutineStacks(0x6575e0, 0xc0003803c0, 0x30, 0x7f56be200788)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:693 +0x9f" THREAD=":shutdown_status"
level=error msg="    runtime/pprof.writeGoroutine(0x6575e0, 0xc0003803c0, 0x2, 0x203000, 0xc)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:682 +0x45" THREAD=":shutdown_status"
level=error msg="    runtime/pprof.(*Profile).WriteTo(0x7770c0, 0x6575e0, 0xc0003803c0, 0x2, 0xc000380340, 0xc00038a4d0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/pprof/pprof.go:331 +0x3f2" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.logGoroutineTraces(0x659e80, 0xc000392300, 0x61b619, 0x16, 0x629ca0)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:109 +0xba" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func4(0x659e80, 0xc00017c330, 0xc0001586b0, 0xc0000357c0)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:305 +0x4e5" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c330, 0xc0000b7920)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":shutdown_status"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1 [select]:" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).Wait(0xc000153e30, 0x616a5c, 0x6)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:418 +0x139" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup_test.Example_signalHandling3()" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/example_test.go:233 +0x125" THREAD=":shutdown_status"
level=error msg="    testing.runExample(0x61ba2a, 0x17, 0x629c90, 0x6245c4, 0x491, 0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:62 +0x209" THREAD=":shutdown_status"
level=error msg="    testing.runExamples(0xc0001fbe98, 0x77b360, 0x5, 0x5, 0xbfe34159434d384f)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/testing/example.go:44 +0x1af" THREAD=":shutdown_status"
level=error msg="    testing.(*M).Run(0xc0000fc000, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/testing/testing.go:1346 +0x273" THREAD=":shutdown_status"
level=error msg="    main.main()" THREAD=":shutdown_status"
level=error msg="    \t_testmain.go:105 +0x1c5" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1400 [IO wait]:" THREAD=":shutdown_status"
level=error msg="    internal/poll.runtime_pollWait(0x7f56be1fb738, 0x72, 0x657920)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/netpoll.go:220 +0x55" THREAD=":shutdown_status"
level=error msg="    internal/poll.(*pollDesc).wait(0xc00012fa58, 0x72, 0x657901, 0x749470, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_poll_runtime.go:87 +0x45" THREAD=":shutdown_status"
level=error msg="    internal/poll.(*pollDesc).waitRead(...)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_poll_runtime.go:92" THREAD=":shutdown_status"
level=error msg="    internal/poll.(*FD).Read(0xc00012fa40, 0xc0003b6000, 0x8000, 0x8000, 0x0, 0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/internal/poll/fd_unix.go:159 +0x1a5" THREAD=":shutdown_status"
level=error msg="    os.(*File).read(...)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/os/file_posix.go:31" THREAD=":shutdown_status"
level=error msg="    os.(*File).Read(0xc0000a8028, 0xc0003b6000, 0x8000, 0x8000, 0x56, 0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/os/file.go:116 +0x71" THREAD=":shutdown_status"
level=error msg="    io.copyBuffer(0x6575e0, 0xc000380140, 0x657500, 0xc0000a8028, 0xc0003b6000, 0x8000, 0x8000, 0x491, 0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/io/io.go:409 +0x12c" THREAD=":shutdown_status"
level=error msg="    io.Copy(...)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/io/io.go:368" THREAD=":shutdown_status"
level=error msg="    testing.runExample.func1(0xc0000a8028, 0xc00015e660)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:37 +0x85" THREAD=":shutdown_status"
level=error msg="    created by testing.runExample" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/testing/run_example.go:35 +0x176" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1404 [chan receive]:" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func3(0x659e80, 0xc00017c270, 0xc0000a5d70, 0xd)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:271 +0x4a" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c270, 0xc0000b7900)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":shutdown_status"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1403 [select]:" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).launchSupervisors.func2(0x659e80, 0xc00017c1b0, 0xc0000a5d80, 0x9)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:255 +0x254" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx.func1(0xc000153e30, 0x659e80, 0xc00017c1b0, 0xc000158870)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:396 +0xb0" THREAD=":shutdown_status"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).goSupervisorCtx" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:384 +0x88" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1443 [syscall]:" THREAD=":shutdown_status"
level=error msg="    os/signal.signal_recv(0x6585e0)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/sigqueue.go:147 +0x9d" THREAD=":shutdown_status"
level=error msg="    os/signal.loop()" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/os/signal/signal_unix.go:23 +0x25" THREAD=":shutdown_status"
level=error msg="    created by os/signal.Notify.func1.1" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/os/signal/signal.go:150 +0x45" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1406 [select (no cases)]:" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup_test.Example_signalHandling3.func2(0x659e80, 0xc00017c3f0, 0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/example_test.go:230 +0x2ab" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).goWorkerCtx.func1(0x0, 0x0)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:354 +0x9f" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/derrgroup.(*Group).Go.func2(0xc00017c420, 0xc00012fbc0, 0xc0000a5de5, 0x7)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:131 +0x2b" THREAD=":shutdown_status"
level=error msg="    created by github.com/datawire/dlib/derrgroup.(*Group).Go" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:129 +0x12d" THREAD=":shutdown_status"
level=error msg="    " THREAD=":shutdown_status"
level=error msg="    goroutine 1407 [semacquire]:" THREAD=":shutdown_status"
level=error msg="    sync.runtime_Semacquire(0xc00012fbcc)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/runtime/sema.go:56 +0x45" THREAD=":shutdown_status"
level=error msg="    sync.(*WaitGroup).Wait(0xc00012fbcc)" THREAD=":shutdown_status"
level=error msg="    \t/usr/lib/go/src/sync/waitgroup.go:130 +0x65" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/derrgroup.(*Group).Wait(...)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/derrgroup/errgroup.go:99" THREAD=":shutdown_status"
level=error msg="    github.com/datawire/dlib/dgroup.(*Group).Wait.func1(0xc00015e840, 0xc000153e30)" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:413 +0x48" THREAD=":shutdown_status"
level=error msg="    created by github.com/datawire/dlib/dgroup.(*Group).Wait" THREAD=":shutdown_status"
level=error msg="    \t/home/lukeshu/src/github.com/datawire/apro/ambassador/pkg/dgroup/group.go:412 +0x85" THREAD=":shutdown_status"
level=error msg="exiting with error: failed to shut down within the 1s shutdown timeout; some goroutines are left running"

Example (Timeout) ¶

package main

import (
	"context"
	"os"
	"time"

	"github.com/sirupsen/logrus"

	"github.com/datawire/dlib/dcontext"
	"github.com/datawire/dlib/dgroup"
	"github.com/datawire/dlib/dlog"
)

func baseContext() context.Context {
	logger := logrus.New()
	logger.SetOutput(os.Stdout)
	logger.SetFormatter(&logrus.TextFormatter{
		DisableTimestamp: true,
	})
	return dlog.WithLogger(context.Background(), dlog.WrapLogrus(logger))
}

func main() {
	ctx := baseContext()

	group := dgroup.NewGroup(ctx, dgroup.GroupConfig{
		// Once one of the workers quits, the others should quit too.
		ShutdownOnNonError: true,
		// Give those others 1 second to quit after the first one quits.
		SoftShutdownTimeout: 1 * time.Second,
	})

	group.Go("a", func(ctx context.Context) error {
		dlog.Infoln(ctx, "I'm running")
		// now I'm done
		return nil
	})

	group.Go("b", func(ctx context.Context) error {
		dlog.Infoln(ctx, "I'm running")

		<-ctx.Done() // graceful shutdown
		dlog.Infoln(ctx, "I should shutdown now... but give me just a moment more")

		<-dcontext.HardContext(ctx).Done() // not-so-graceful shutdown
		dlog.Infoln(ctx, "oops, I've been a bad boy, I really do need to shut down now")
		return nil
	})

	if err := group.Wait(); err != nil {
		dlog.Errorln(ctx, "exiting with error:", err)
	}

}

Output:

level=info msg="I'm running" THREAD=/a
level=info msg="I'm running" THREAD=/b
level=info msg="I should shutdown now... but give me just a moment more" THREAD=/b
level=info msg="shutting down (gracefully)..." THREAD=":shutdown_logger"
level=info msg="oops, I've been a bad boy, I really do need to shut down now" THREAD=/b
level=info msg="shutting down (not-so-gracefully)..." THREAD=":shutdown_logger"

Index ¶

func WithGoroutineName(ctx context.Context, newName string) context.Context
type Group
- func NewGroup(ctx context.Context, cfg GroupConfig) *Group
- func ParentGroup(ctx context.Context) *Group
type GroupConfig

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func WithGoroutineName ¶

func WithGoroutineName(ctx context.Context, newName string) context.Context

WithGoroutineName associates a name with the context, which gets logged by dlog as the "THREAD" field.

If the context already has a name, then the new name is appended to it. This allows a "tree" to be formed. There are no delimiters added between names; you must include the delimiter as part of the name passed to WithGoroutineName.

Group.Go calls this for you (using "/" as a delimiter); you shouldn't need to call WithGoroutineName for goroutines managed by a Group.

Types ¶

type Group ¶

type Group struct {
	// contains filtered or unexported fields
}

A Group is a collection of goroutines working on subtasks that are part of the same overall task. Compared to a minimum-viable goroutine-group abstraction (such as stdlib "sync.WaitGroup" or the bog-standard "golang.org/x/sync/errgroup.Group"), the things that make dgroup attractive are:

(optionally) handles SIGINT and SIGTERM
(configurable) manages Context for you
(optionally) adds hard/soft cancellation
(optionally) does panic recovery
(optionally) does some minimal logging
(optionally) adds configurable shutdown timeouts
a concept of goroutine names
adds a way to call to the parent group, making it possible to launch a "sibling" goroutine

A zero Group is NOT valid; a Group must be created with NewGroup.

A Group is suitable for use at the top-level of a program (especially if signal handling is enabled for the Group), and is also suitable to be nested somewhere deep inside of an application (but signal handling should probably be disabled for that use).

Example (JustErrors) ¶

JustErrors illustrates the use of a Group in place of a sync.WaitGroup to simplify goroutine counting and error handling. This example is derived from the sync.WaitGroup example at https://golang.org/pkg/sync/#example_WaitGroup.

package main

import (
	"context"
	"fmt"
	"net/http"

	errgroup "github.com/datawire/dlib/dgroup"
)

func main() {
	g := errgroup.NewGroup(context.Background(), errgroup.GroupConfig{
		DisableLogging: true,
	})
	var urls = []string{
		"http://www.golang.org/",
		"http://www.google.com/",
		"http://www.somestupidname.com/",
	}
	for _, url := range urls {
		// Launch a goroutine to fetch the URL.
		url := url // https://golang.org/doc/faq#closures_and_goroutines
		g.Go(url, func(ctx context.Context) error {
			// Fetch the URL.
			resp, err := http.Get(url)
			if err == nil {
				resp.Body.Close()
			}
			return err
		})
	}
	// Wait for all HTTP fetches to complete.
	if err := g.Wait(); err == nil {
		fmt.Println("Successfully fetched all URLs.")
	}
}

Output:

Example (Parallel) ¶

Parallel illustrates the use of a Group for synchronizing a simple parallel task: the "Google Search 2.0" function from https://talks.golang.org/2012/concurrency.slide#46, augmented with a Context and error-handling.

package main

import (
	"context"
	"fmt"
	"os"

	errgroup "github.com/datawire/dlib/dgroup"
)

var (
	Web   = fakeSearch("web")
	Image = fakeSearch("image")
	Video = fakeSearch("video")
)

type Result string
type Search func(ctx context.Context, query string) (Result, error)

func fakeSearch(kind string) Search {
	return func(_ context.Context, query string) (Result, error) {
		return Result(fmt.Sprintf("%s result for %q", kind, query)), nil
	}
}

func main() {
	Google := func(ctx context.Context, query string) ([]Result, error) {
		g := errgroup.NewGroup(ctx, errgroup.GroupConfig{
			DisableLogging: true,
		})

		searches := []Search{Web, Image, Video}
		results := make([]Result, len(searches))
		for i, search := range searches {
			i, search := i, search // https://golang.org/doc/faq#closures_and_goroutines
			g.Go(fmt.Sprintf("search-%d", i), func(ctx context.Context) error {
				result, err := search(ctx, query)
				if err == nil {
					results[i] = result
				}
				return err
			})
		}
		if err := g.Wait(); err != nil {
			return nil, err
		}
		return results, nil
	}

	results, err := Google(context.Background(), "golang")
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		return
	}
	for _, result := range results {
		fmt.Println(result)
	}

}

Output:

web result for "golang"
image result for "golang"
video result for "golang"

Example (Pipeline) ¶

Pipeline demonstrates the use of a Group to implement a multi-stage pipeline: a version of the MD5All function with bounded parallelism from https://blog.golang.org/pipelines.

package main

import (
	"context"
	"crypto/md5"
	"fmt"
	"log"
	"os"
	"path/filepath"

	errgroup "github.com/datawire/dlib/dgroup"
)

// Pipeline demonstrates the use of a Group to implement a multi-stage
// pipeline: a version of the MD5All function with bounded parallelism from
// https://blog.golang.org/pipelines.
func main() {
	m, err := MD5All(context.Background(), ".")
	if err != nil {
		log.Fatal(err)
	}

	for k, sum := range m {
		fmt.Printf("%s:\t%x\n", k, sum)
	}
}

type result struct {
	path string
	sum  [md5.Size]byte
}

// MD5All reads all the files in the file tree rooted at root and returns a map
// from file path to the MD5 sum of the file's contents. If the directory walk
// fails or any read operation fails, MD5All returns an error.
func MD5All(ctx context.Context, root string) (map[string][md5.Size]byte, error) {
	// ctx is canceled when g.Wait() returns. When this version of MD5All returns
	// - even in case of error! - we know that all of the goroutines have finished
	// and the memory they were using can be garbage-collected.
	g := errgroup.NewGroup(ctx, errgroup.GroupConfig{
		DisableLogging: true,
	})
	paths := make(chan string)

	g.Go("walk", func(ctx context.Context) error {
		defer close(paths)
		return filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
			if err != nil {
				return err
			}
			if !info.Mode().IsRegular() {
				return nil
			}
			select {
			case paths <- path:
			case <-ctx.Done():
				return ctx.Err()
			}
			return nil
		})
	})

	// Start a fixed number of goroutines to read and digest files.
	c := make(chan result)
	const numDigesters = 20
	for i := 0; i < numDigesters; i++ {
		g.Go(fmt.Sprintf("digestor-%d", i), func(ctx context.Context) error {
			for path := range paths {
				data, err := os.ReadFile(path)
				if err != nil {
					return err
				}
				select {
				case c <- result{path, md5.Sum(data)}:
				case <-ctx.Done():
					return ctx.Err()
				}
			}
			return nil
		})
	}
	go func() {
		_ = g.Wait()
		close(c)
	}()

	m := make(map[string][md5.Size]byte)
	for r := range c {
		m[r.path] = r.sum
	}
	// Check whether any of the goroutines failed. Since g is accumulating the
	// errors, we don't need to send them (or check for them) in the individual
	// results sent on the channel.
	if err := g.Wait(); err != nil {
		return nil, err
	}
	return m, nil
}

Output:

func NewGroup ¶

func NewGroup(ctx context.Context, cfg GroupConfig) *Group

NewGroup returns a new Group.

func ParentGroup ¶

func ParentGroup(ctx context.Context) *Group

ParentGroup returns the Group that manages this goroutine/Context. If the Context is not managed by a Group, then nil is returned. The principle use of ParentGroup is to launch a sibling goroutine in the group.

Example ¶

package main

import (
	"context"
	"os"

	"github.com/sirupsen/logrus"

	"github.com/datawire/dlib/dgroup"
	"github.com/datawire/dlib/dlog"
)

func baseContext() context.Context {
	logger := logrus.New()
	logger.SetOutput(os.Stdout)
	logger.SetFormatter(&logrus.TextFormatter{
		DisableTimestamp: true,
	})
	return dlog.WithLogger(context.Background(), dlog.WrapLogrus(logger))
}

func main() {
	ctx := dgroup.WithGoroutineName(baseContext(), "parent")
	group := dgroup.NewGroup(ctx, dgroup.GroupConfig{})

	group.Go("a", func(ctx context.Context) error {
		dlog.Infoln(ctx, `I am goroutine "parent/a"`)

		// Use dgroup.ParentGroup to create a sibling goroutine
		dgroup.ParentGroup(ctx).Go("b", func(ctx context.Context) error {
			dlog.Infoln(ctx, `I am goroutine "parent/b"`)
			return nil
		})

		// Use dgroup.NewGroup to create a child goroutine.
		subgroup := dgroup.NewGroup(ctx, dgroup.GroupConfig{})
		subgroup.Go("c", func(ctx context.Context) error {
			dlog.Infoln(ctx, `I am goroutine "parent/a/c"`)
			return nil
		})

		// If you create a sub-group, be sure to wait
		return subgroup.Wait()
	})

	if err := group.Wait(); err != nil {
		dlog.Errorln(ctx, "exiting with error:", err)
	}

}

Output:

level=info msg="I am goroutine \"parent/a\"" THREAD=parent/a
level=info msg="I am goroutine \"parent/b\"" THREAD=parent/b
level=info msg="I am goroutine \"parent/a/c\"" THREAD=parent/a/c

func (*Group) Go ¶

func (g *Group) Go(name string, fn func(ctx context.Context) error)

Go calls the given function in a new named-worker-goroutine.

Cancellation of the Context should trigger a graceful shutdown. Cancellation of the dcontext.HardContext(ctx) of it should trigger a not-so-graceful shutdown.

A worker may access its parent group by calling ParentGroup on its Context.

func (*Group) List ¶

func (g *Group) List() map[string]derrgroup.GoroutineState

List returns a listing of all goroutines launched with .Go().

func (*Group) Wait ¶

func (g *Group) Wait() error

Wait for all goroutines in the group to finish, and return returns an error if any of the workers errored or timed out.

Once the group has initiated hard-shutdown (either soft-shutdown was initiated then timed out, a 2nd shutdown signal was received, or the parent context is <-Done()), Wait will return within the HardShutdownTimeout passed to NewGroup. If a poorly-behaved goroutine is still running at the end of that time, it is left running, and an error is returned.

type GroupConfig ¶

type GroupConfig struct {
	// EnableWithSoftness says whether it should call
	// dcontext.WithSoftness() on the Context passed to NewGroup.
	// This should probably NOT be set for a Context that is
	// already soft.  However, this must be set for features that
	// require separate hard/soft cancellation, such as signal
	// handling.  If any of those features are enabled, then it
	// will force EnableWithSoftness to be set.
	EnableWithSoftness   bool
	EnableSignalHandling bool // implies EnableWithSoftness

	// Normally a worker exiting with an error triggers other
	// goroutines to shutdown.  Setting ShutdownOnNonError causes
	// a shutdown to be triggered whenever a goroutine exits, even
	// if it exits without error.
	ShutdownOnNonError bool

	// SoftShutdownTimeout is how long after a soft shutdown is
	// triggered to wait before triggering a hard shutdown.  A
	// zero value means to not trigger a hard shutdown after a
	// soft shutdown.
	//
	// SoftShutdownTimeout implies EnableWithSoftness because
	// otherwise there would be no way of triggering the
	// subsequent hard shutdown.
	SoftShutdownTimeout time.Duration
	// HardShutdownTimeout is how long after a hard shutdown is
	// triggered to wait before forcing Wait() to return early.  A
	// zero value means to not force Wait() to return early.
	HardShutdownTimeout time.Duration

	DisablePanicRecovery bool
	DisableLogging       bool

	WorkerContext func(ctx context.Context, name string) context.Context
}

GroupConfig is a readable way of setting the configuration options for NewGroup.

A zero GroupConfig (`dgroup.GroupConfig{}`) should be sane defaults. Because signal handling should only be enabled for the outermost group, it is off by default.

TODO(lukeshu): Consider enabling timeouts by default?

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL