From 59051e2f2b951493c72a3162f7823e2d51ec35f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juan=20Pablo=20Villaf=C3=A1=C3=B1ez?= Date: Fri, 19 Apr 2024 14:37:11 +0200 Subject: [PATCH] feat: make the interrupt duration optional and with a default --- ocis-pkg/runner/grouprunner_test.go | 18 +++++++++--------- ocis-pkg/runner/option.go | 27 +++++++++++++++++++++++++++ ocis-pkg/runner/runner.go | 18 ++++++++++++++---- ocis-pkg/runner/runner_test.go | 28 ++++++++++++++-------------- 4 files changed, 64 insertions(+), 27 deletions(-) create mode 100644 ocis-pkg/runner/option.go diff --git a/ocis-pkg/runner/grouprunner_test.go b/ocis-pkg/runner/grouprunner_test.go index 3d20ad7850..110960a5bf 100644 --- a/ocis-pkg/runner/grouprunner_test.go +++ b/ocis-pkg/runner/grouprunner_test.go @@ -19,14 +19,14 @@ var _ = Describe("GroupRunner", func() { task1Ch := make(chan error) task1 := TimedTask(task1Ch, 30*time.Second) - gr.Add(runner.New("task1", 30*time.Second, task1, func() { + gr.Add(runner.New("task1", task1, func() { task1Ch <- nil close(task1Ch) })) task2Ch := make(chan error) task2 := TimedTask(task2Ch, 20*time.Second) - gr.Add(runner.New("task2", 30*time.Second, task2, func() { + gr.Add(runner.New("task2", task2, func() { task2Ch <- nil close(task2Ch) })) @@ -35,7 +35,7 @@ var _ = Describe("GroupRunner", func() { Describe("Add", func() { It("Duplicated runner id panics", func() { Expect(func() { - gr.Add(runner.New("task1", 30*time.Second, func() error { + gr.Add(runner.New("task1", func() error { time.Sleep(6 * time.Second) return nil }, func() { @@ -64,7 +64,7 @@ var _ = Describe("GroupRunner", func() { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 15*time.Second) Expect(func() { - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) @@ -78,7 +78,7 @@ var _ = Describe("GroupRunner", func() { Expect(func() { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 15*time.Second) - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) @@ -90,7 +90,7 @@ var _ = Describe("GroupRunner", func() { It("Context is done", func(ctx SpecContext) { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 15*time.Second) - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) @@ -117,7 +117,7 @@ var _ = Describe("GroupRunner", func() { It("One task finishes early", func(ctx SpecContext) { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 1*time.Second) - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) @@ -157,7 +157,7 @@ var _ = Describe("GroupRunner", func() { It("Wait in channel", func(ctx SpecContext) { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 1*time.Second) - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) @@ -182,7 +182,7 @@ var _ = Describe("GroupRunner", func() { It("Interrupt async", func(ctx SpecContext) { task3Ch := make(chan error) task3 := TimedTask(task3Ch, 15*time.Second) - gr.Add(runner.New("task3", 30*time.Second, task3, func() { + gr.Add(runner.New("task3", task3, func() { task3Ch <- nil close(task3Ch) })) diff --git a/ocis-pkg/runner/option.go b/ocis-pkg/runner/option.go new file mode 100644 index 0000000000..2bef541379 --- /dev/null +++ b/ocis-pkg/runner/option.go @@ -0,0 +1,27 @@ +package runner + +import ( + "time" +) + +var ( + // DefaultInterruptDuration is the default value for the `WithInterruptDuration` + // This global value can be adjusted if needed. + DefaultInterruptDuration = 10 * time.Second +) + +// Option defines a single option function. +type Option func(o *Options) + +// Options defines the available options for this package. +type Options struct { + InterruptDuration time.Duration +} + +// WithInterruptDuration provides a function to set the interrupt +// duration option. +func WithInterruptDuration(val time.Duration) Option { + return func(o *Options) { + o.InterruptDuration = val + } +} diff --git a/ocis-pkg/runner/runner.go b/ocis-pkg/runner/runner.go index d720a45b8e..79b93cc54b 100644 --- a/ocis-pkg/runner/runner.go +++ b/ocis-pkg/runner/runner.go @@ -31,8 +31,10 @@ type Runner struct { // otherwise undefined behavior might occur), and will run the provided // runable task, using the "interrupt" function to stop that task if needed. // -// The interrupt duration will be used to ensure the runner doesn't block -// forever. The interrupt duration will be used to start a timeout when the +// The interrupt duration, which can be set through the `WithInterruptDuration` +// option, will be used to ensure the runner doesn't block forever. If the +// option isn't suplied, the default value will be used. +// The interrupt duration will be used to start a timeout when the // runner gets interrupted (either the context of the `Run` method is done // or this runner's `Interrupt` method is called). If the timeout is reached, // a timeout result will be returned instead of whatever result the task should @@ -41,10 +43,18 @@ type Runner struct { // Note that it's your responsibility to provide a proper stopper for the task. // The runner will just call that method assuming it will be enough to // eventually stop the task at some point. -func New(id string, interruptDur time.Duration, fn Runable, interrupt Stopper) *Runner { +func New(id string, fn Runable, interrupt Stopper, opts ...Option) *Runner { + options := Options{ + InterruptDuration: DefaultInterruptDuration, + } + + for _, o := range opts { + o(&options) + } + return &Runner{ ID: id, - interruptDur: interruptDur, + interruptDur: options.InterruptDuration, fn: fn, interrupt: interrupt, interruptedCh: make(chan time.Duration), diff --git a/ocis-pkg/runner/runner_test.go b/ocis-pkg/runner/runner_test.go index 8ee2a4f899..7198e83d01 100644 --- a/ocis-pkg/runner/runner_test.go +++ b/ocis-pkg/runner/runner_test.go @@ -44,7 +44,7 @@ var _ = Describe("Runner", func() { // channel, so the task can finish // Worst case, the task will finish after 15 secs ch := make(chan error) - r := runner.New("run001", 30*time.Second, TimedTask(ch, 15*time.Second), func() { + r := runner.New("run001", TimedTask(ch, 15*time.Second), func() { ch <- nil close(ch) }) @@ -76,7 +76,7 @@ var _ = Describe("Runner", func() { // channel, so the task can finish // Worst case, the task will finish after 15 secs ch := make(chan error) - r := runner.New("run001", 30*time.Second, TimedTask(ch, 15*time.Second), func() { + r := runner.New("run001", TimedTask(ch, 15*time.Second), func() { ch <- nil close(ch) }) @@ -106,7 +106,7 @@ var _ = Describe("Runner", func() { It("Task finishes naturally", func(ctx SpecContext) { e := errors.New("overslept!") - r := runner.New("run002", 30*time.Second, func() error { + r := runner.New("run002", func() error { time.Sleep(50 * time.Millisecond) return e }, func() { @@ -134,7 +134,7 @@ var _ = Describe("Runner", func() { }, SpecTimeout(5*time.Second)) It("Task doesn't finish", func(ctx SpecContext) { - r := runner.New("run003", 30*time.Second, func() error { + r := runner.New("run003", func() error { time.Sleep(20 * time.Second) return nil }, func() { @@ -157,11 +157,11 @@ var _ = Describe("Runner", func() { }, SpecTimeout(5*time.Second)) It("Task doesn't finish and times out", func(ctx SpecContext) { - r := runner.New("run003", 3*time.Second, func() error { + r := runner.New("run003", func() error { time.Sleep(20 * time.Second) return nil }, func() { - }) + }, runner.WithInterruptDuration(3*time.Second)) // context will be done in 1 second myCtx, cancel := context.WithTimeout(ctx, 1*time.Second) @@ -186,7 +186,7 @@ var _ = Describe("Runner", func() { It("Run mutiple times panics", func(ctx SpecContext) { e := errors.New("overslept!") - r := runner.New("run002", 30*time.Second, func() error { + r := runner.New("run002", func() error { time.Sleep(50 * time.Millisecond) return e }, func() { @@ -208,7 +208,7 @@ var _ = Describe("Runner", func() { ch := make(chan *runner.Result) e := errors.New("Task has finished") - r := runner.New("run004", 30*time.Second, func() error { + r := runner.New("run004", func() error { time.Sleep(50 * time.Millisecond) return e }, func() { @@ -227,7 +227,7 @@ var _ = Describe("Runner", func() { ch := make(chan *runner.Result) e := errors.New("Task has finished") - r := runner.New("run004", 30*time.Second, func() error { + r := runner.New("run004", func() error { time.Sleep(50 * time.Millisecond) return e }, func() { @@ -245,7 +245,7 @@ var _ = Describe("Runner", func() { e := errors.New("Task interrupted") taskCh := make(chan error) - r := runner.New("run005", 30*time.Second, TimedTask(taskCh, 20*time.Second), func() { + r := runner.New("run005", TimedTask(taskCh, 20*time.Second), func() { taskCh <- e close(taskCh) }) @@ -265,11 +265,11 @@ var _ = Describe("Runner", func() { ch := make(chan *runner.Result) e := errors.New("Task interrupted") - r := runner.New("run005", 3*time.Second, func() error { + r := runner.New("run005", func() error { time.Sleep(30 * time.Second) return e }, func() { - }) + }, runner.WithInterruptDuration(3*time.Second)) r.RunAsync(ch) r.Interrupt() @@ -287,7 +287,7 @@ var _ = Describe("Runner", func() { e := errors.New("Task interrupted") taskCh := make(chan error) - r := runner.New("run005", 30*time.Second, TimedTask(taskCh, 20*time.Second), func() { + r := runner.New("run005", TimedTask(taskCh, 20*time.Second), func() { taskCh <- e close(taskCh) }) @@ -309,7 +309,7 @@ var _ = Describe("Runner", func() { Describe("Finished", func() { It("Finish channel closes", func(ctx SpecContext) { - r := runner.New("run006", 30*time.Second, func() error { + r := runner.New("run006", func() error { time.Sleep(50 * time.Millisecond) return nil }, func() {