-
-
Save kevinclcn/cce6d43f736b6827b724823a3eb8599e to your computer and use it in GitHub Desktop.
Code snippets for my blog post "The X-Files: Avoiding Concurrency Boilerplate with golang.org/x/sync"
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| // An Action performs a single arbitrary task. | |
| type Action interface { | |
| // Execute performs the work of an Action. This method should make a best | |
| // effort to be cancelled if the provided ctx is cancelled. | |
| Execute(ctx context.Context) error | |
| } | |
| // An Executor performs a set of Actions. It is up to the implementing type | |
| // the concurrency and open/closed failure behavior of the actions. | |
| type Executor interface { | |
| // Execute performs all provided actions by calling their Execute method. | |
| // This method should make a best-effort to cancel outstanding actions if the | |
| // provided ctx is cancelled. | |
| Execute(ctx context.Context, actions []Action) error | |
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| type flow struct { | |
| maxActions int64 | |
| actions *semaphore.Weighted | |
| calls *semaphore.Weighted | |
| ex Executor | |
| } | |
| // ControlFlow decorates an Executor, limiting it to a maximum concurrent | |
| // number of calls and actions. | |
| func ControlFlow(e Executor, maxCalls, maxActions int64) Executor { | |
| return &flow{ | |
| maxActions: maxActions, | |
| calls: semaphore.NewWeighted(maxCalls), | |
| actions: semaphore.NewWeighted(maxActions), | |
| } | |
| } | |
| // Execute attempts to acquire the semaphores for the concurrent calls and | |
| // actions before delegating to the decorated Executor. If Execute is called | |
| // with more actions than maxActions, an error is returned. | |
| func (f *flow) Execute(ctx context.Context, actions []Action) error { | |
| qty := int64(len(actions)) | |
| if qty > f.maxActions { | |
| return fmt.Errorf("maximum %d actions allowed", f.maxActions) | |
| } | |
| if err := f.calls.Acquire(ctx, 1); err != nil { | |
| return err | |
| } | |
| defer f.calls.Release(1) | |
| if err := f.actions.Acquire(ctx, qty); err != nil { | |
| return err | |
| } | |
| defer f.calls.Release(qty) | |
| return f.ex.Execute(ctx, actions) | |
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| // Parallel is a concurrent implementation of Executor | |
| type Parallel struct{} | |
| // Execute performs all provided actions in concurrently, failing closed on the | |
| // first error or if ctx is cancelled. | |
| func (p Parallel) Execute(ctx context.Context, actions []Action) error { | |
| grp, ctx := errgroup.WithContext(ctx) | |
| for _, a := range actions { | |
| grp.Go(p.execFn(ctx, a)) | |
| } | |
| return grp.Wait() | |
| } | |
| // execFn binds the Context and Action to the proper function signature for the | |
| // errgroup.Group. | |
| func (p Parallel) execFn(ctx context.Context, a Action) func() error { | |
| return func() error { return a.Execute(ctx) } | |
| } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment