diff --git a/pkg/debouncer/action/action.go b/pkg/debouncer/action/action.go new file mode 100644 index 00000000000..b2b4c020e15 --- /dev/null +++ b/pkg/debouncer/action/action.go @@ -0,0 +1,39 @@ +package action + +import ( + "context" + "errors" + + "github.com/owncloud/reva/v2/pkg/debouncer/tasklist" +) + +var ( + ErrEmptyList = errors.New("Task list is empty") +) + +// Action represents the action that the debouncer needs to perform +// over the provided task list. +// This action usually involves picking the "right" task, run it and +// return the error if any. +// Choosing the "right" task can be as easy as picking the first or last +// task of the list and run it, or more complex such as gathering info from +// all the tasks and create a new task based on the aggregated info. +// The action isn't limited to choosing just one task, and it can choose +// and run multiple tasks if needed (although not recommended). +type Action interface { + // RunTasks perform an action over the provided task list. This usually + // involves picking one task of the list and run it. + // It's expected that RunTasks will be executed in its own goroutine, + // using a new context. + RunTasks(ctx context.Context, tasks []tasklist.InternalTask) error + // GetId returns the ID of this instance. It must be unique, so multiple + // instances from the same action type must return different IDs. + // The recommendation is to use the instance type followed by a random + // number, such as "ChooseLast_123987" + GetId() string + // GetTracingData returns additional data that will be used for tracing. + // Consider this data as public information. This data is intended to + // be use purely for informational purposes. You can return nil if + // there isn't any data to be published. + GetTracingData() map[string]string +} diff --git a/pkg/debouncer/action/action_suite_test.go b/pkg/debouncer/action/action_suite_test.go new file mode 100644 index 00000000000..cf968dfac26 --- /dev/null +++ b/pkg/debouncer/action/action_suite_test.go @@ -0,0 +1,13 @@ +package action_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestAction(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Debouncer action suite") +} diff --git a/pkg/debouncer/action/chooselast.go b/pkg/debouncer/action/chooselast.go new file mode 100644 index 00000000000..299e1f26f14 --- /dev/null +++ b/pkg/debouncer/action/chooselast.go @@ -0,0 +1,44 @@ +package action + +import ( + "context" + "math/rand" + "strconv" + + "github.com/owncloud/reva/v2/pkg/debouncer/tasklist" +) + +// ChooseLast implements a debouncer action that will always run the last +// task of the list, which is expected to be the most recent one. +type ChooseLast struct { + id string +} + +// NewChooseLast will return a new instance +func NewChooseLast() *ChooseLast { + return &ChooseLast{ + id: "ChooseLast_" + strconv.FormatUint(rand.Uint64(), 10), + } +} + +// RunTasks will run the last task of the provided list. If the list is empty +// a ErrEmptyList will be returned. +func (cl *ChooseLast) RunTasks(ctx context.Context, tasks []tasklist.InternalTask) error { + if len(tasks) <= 0 { + return ErrEmptyList + } + + chosenTask := tasks[len(tasks)-1] + return chosenTask.OriginalTask.Execute(ctx) +} + +// GetId will return the id of this instance. It will return "ChooseLast_" +// followed by a random number. +func (cl *ChooseLast) GetId() string { + return cl.id +} + +// GetTracingData will return nil +func (cl *ChooseLast) GetTracingData() map[string]string { + return nil +} diff --git a/pkg/debouncer/action/chooselast_test.go b/pkg/debouncer/action/chooselast_test.go new file mode 100644 index 00000000000..6bf94651daa --- /dev/null +++ b/pkg/debouncer/action/chooselast_test.go @@ -0,0 +1,80 @@ +package action_test + +import ( + "context" + "errors" + + "github.com/owncloud/reva/v2/pkg/debouncer/action" + "github.com/owncloud/reva/v2/pkg/debouncer/tasklist" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type DummyTask struct { + Err error + Done bool +} + +func (d *DummyTask) ExposeData() map[string]string { + return nil +} + +func (d *DummyTask) Execute(ctx context.Context) error { + d.Done = true + return d.Err +} + +var _ = Describe("ChooseLast", func() { + var cl action.Action + + BeforeEach(func() { + cl = action.NewChooseLast() + }) + + Describe("RunTasks", func() { + It("Choose last task", func() { + ctx := context.Background() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task3 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task4 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + taskList := []tasklist.InternalTask{task1, task2, task3, task4} + + err := cl.RunTasks(ctx, taskList) + Expect(err).To(Succeed()) + Expect(task1.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task2.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task3.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task4.OriginalTask.(*DummyTask).Done).To(Equal(true)) + }) + + It("Last task fails", func() { + ctx := context.Background() + terr := errors.New("oopsie!!") + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task3 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task4 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{Err: terr}) + taskList := []tasklist.InternalTask{task1, task2, task3, task4} + + err := cl.RunTasks(ctx, taskList) + Expect(err).To(Equal(terr)) + Expect(task1.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task2.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task3.OriginalTask.(*DummyTask).Done).To(Equal(false)) + Expect(task4.OriginalTask.(*DummyTask).Done).To(Equal(true)) + }) + + It("Empty task list fails", func() { + taskList := []tasklist.InternalTask{} + err := cl.RunTasks(context.Background(), taskList) + Expect(err).To(Equal(action.ErrEmptyList)) + }) + + It("Nil task list fails", func() { + err := cl.RunTasks(context.Background(), nil) + Expect(err).To(Equal(action.ErrEmptyList)) + }) + }) +}) diff --git a/pkg/debouncer/debouncer.go b/pkg/debouncer/debouncer.go new file mode 100644 index 00000000000..25338e3afce --- /dev/null +++ b/pkg/debouncer/debouncer.go @@ -0,0 +1,233 @@ +package debouncer + +import ( + "context" + "errors" + "sync" + "time" + + "github.com/owncloud/reva/v2/pkg/debouncer/action" + "github.com/owncloud/reva/v2/pkg/debouncer/policy" + "github.com/owncloud/reva/v2/pkg/debouncer/tasklist" + "github.com/rs/zerolog" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +const ( + TracerName = "github.com/owncloud/reva/v2/pkg/debouncer" +) + +var ( + ErrQueueAlreadyCreated = errors.New("queue already created") + ErrQueueFailedAdd = errors.New("task failed to be added to the queue") +) + +// PolicyFactory is a function to create Policies. The function is expected +// to return the same policy each time it's called, but it MUST return +// different instances of that policy (returning the same cached instance +// will cause problems) +type PolicyFactory func() policy.Policy + +// TaskListFactory is a function to create TaskLists. The function is expected +// to return the same task list each time it's called, but it MUST return +// different instances of that task list (returning the same cached instance +// will cause problems) +type TaskListFactory func() tasklist.TaskList + +// ActionFactory is a function to create Actions. The function is expected +// to return the same action each time it's called, but it MUST return +// different instances of that action (returning the same cached instance +// will cause problems) +type ActionFactory func() action.Action + +// queue represents a particular task queue in the debouncer +type queue struct { + id string + policy policy.Policy + tasks tasklist.TaskList + action action.Action +} + +// Debouncer will hold different queues, each one holding a set of tasks. +// An action (set in the queue) will run on those tasks when the policy +// (also set in the queue) triggers. +// A default policy, action and task list can be set. Those will be used +// if no policy, action or task list is provided when a new queue is created. +type Debouncer struct { + queues *sync.Map + defaultPolicyFactory PolicyFactory + defaultTaskListFactory TaskListFactory + defaultActionFactory ActionFactory +} + +// NewDebouncer will create a new instance of the debouncer, using the +// provided factories in the options. +func NewDebouncer(opts ...Option) *Debouncer { + // default options + options := Options{ + DefaultPolicyFactory: func() policy.Policy { return policy.NewTimedWithReset(15 * time.Second) }, + DefaultTaskListFactory: func() tasklist.TaskList { return tasklist.NewSliceWithLocks() }, + DefaultActionFactory: func() action.Action { return action.NewChooseLast() }, + } + // overwrite defaults + for _, o := range opts { + o(&options) + } + + return &Debouncer{ + queues: &sync.Map{}, + defaultPolicyFactory: options.DefaultPolicyFactory, + defaultTaskListFactory: options.DefaultTaskListFactory, + defaultActionFactory: options.DefaultActionFactory, + } +} + +func (d *Debouncer) createAndReturnQueue(ctx context.Context, id string, policy policy.Policy, tasks tasklist.TaskList, action action.Action) (*queue, error) { + q := &queue{ + id: id, + policy: policy, + tasks: tasks, + action: action, + } + + loadedQueue, loaded := d.queues.LoadOrStore(id, q) + if loaded { + // New queue wasn't added because there is already an existing queue. + // There is nothing to do, just return the queue and a proper error. + return loadedQueue.(*queue), ErrQueueAlreadyCreated + } + + currentSpan := trace.SpanFromContext(ctx) + tracer := currentSpan.TracerProvider().Tracer(TracerName) + // spawn a goroutine to monitor the new queue and run the action when + // the policy triggers + go func(qq *queue, tracer trace.Tracer) { + // wait until the policy triggers + <-qq.policy.WaitForTrigger() + + value, loaded := d.queues.LoadAndDelete(qq.id) + if loaded { + // transfer the logger to the new context + logger := zerolog.Ctx(ctx) + queueCtx := logger.WithContext(context.Background()) + + // key was present -> run through the "value" queue + finalQueue := value.(*queue) + finalQueue.tasks.Freeze() + internalTasks := finalQueue.tasks.ToSlice() + + newCtx, newSpan := tracer.Start( + queueCtx, + "Debounce RunQueue", + trace.WithNewRoot(), + trace.WithSpanKind(trace.SpanKindConsumer), + trace.WithAttributes(d.prepareTracingData(finalQueue)...), + trace.WithLinks(d.prepareTracingLinks(internalTasks)...), + ) + defer newSpan.End() + finalQueue.action.RunTasks(newCtx, internalTasks) + } + }(q, tracer) + + return q, nil +} + +// CreateQueue will explicitly create a new queue, identified by the provided +// id, using the provided policy and action. The task list used will always be +// the default one for the debouncer. +// You can use nil as policy and action in order to use the default ones. +func (d *Debouncer) CreateQueue(ctx context.Context, id string, policy policy.Policy, action action.Action) error { + realPolicy := d.defaultPolicyFactory() + if policy != nil { + realPolicy = policy + } + + realAction := d.defaultActionFactory() + if action != nil { + realAction = action + } + + _, err := d.createAndReturnQueue(ctx, id, realPolicy, d.defaultTaskListFactory(), realAction) + return err +} + +// AddToQueue will add the specified task to the queue identified with the id. +// If no queue exists with that id, a new one will be created and make it +// available. +// If the task can't be added because the task list is frozen (which means +// that the queue processing has started), a new queue will be created and +// the task will be added to the new queue. Note that this retry will happen +// only once because it's expected that the policies give enough time for the +// task to be added before triggering. +func (d *Debouncer) AddToQueue(ctx context.Context, id string, task tasklist.Task) error { + currentSpan := trace.SpanFromContext(ctx) + tracer := currentSpan.TracerProvider().Tracer(TracerName) + newCtx, newSpan := tracer.Start(ctx, "Debounce AddToQueue", trace.WithSpanKind(trace.SpanKindProducer)) + defer newSpan.End() + + q, err := d.createAndReturnQueue(newCtx, id, d.defaultPolicyFactory(), d.defaultTaskListFactory(), d.defaultActionFactory()) + if err != nil && !errors.Is(err, ErrQueueAlreadyCreated) { + // it doesn't matter if the queue has been created (returned + // already), but we can't do anything if is different + return err + } + + internalTask := tasklist.NewInternalTaskFromTask(newCtx, task) + + if !q.tasks.AddToList(internalTask) { + if q.tasks.IsFrozen() { + // if the task isn't added because the task list is frozen, + // we got the list right before the action runs, but the action + // froze the task list faster than us. + // We'll retry once to avoid failing the operation. Any trigger + // policy that the queue could have should be tolerant enough + // not to fail adding a task right away, otherwise return the + // error + q, err = d.createAndReturnQueue(newCtx, id, d.defaultPolicyFactory(), d.defaultTaskListFactory(), d.defaultActionFactory()) + if err != nil && !errors.Is(err, ErrQueueAlreadyCreated) { + return err + } + if !q.tasks.AddToList(internalTask) && q.tasks.IsFrozen() { + return ErrQueueFailedAdd + } + } else { + return ErrQueueFailedAdd + } + } + + newSpan.SetAttributes(d.prepareTracingData(q)...) + return nil +} + +func (d *Debouncer) prepareTracingLinks(itasks []tasklist.InternalTask) []trace.Link { + links := make([]trace.Link, len(itasks)) + for i, itask := range itasks { + spanLink := trace.Link{ + SpanContext: itask.SpanContext, + } + links[i] = spanLink + } + return links +} + +func (d *Debouncer) prepareTracingData(q *queue) []attribute.KeyValue { + attrs := []attribute.KeyValue{ + attribute.String("ocis.debouncer.queue.id", q.id), + attribute.String("ocis.debouncer.queue.policy.id", q.policy.GetId()), + attribute.String("ocis.debouncer.queue.tasklist.id", q.tasks.GetId()), + attribute.String("ocis.debouncer.queue.action.id", q.action.GetId()), + } + + for key, value := range q.policy.GetTracingData() { + attrs = append(attrs, attribute.String("ocis.debouncer.queue.policy."+key, value)) + } + for key, value := range q.tasks.GetTracingData() { + attrs = append(attrs, attribute.String("ocis.debouncer.queue.tasklist."+key, value)) + } + for key, value := range q.action.GetTracingData() { + attrs = append(attrs, attribute.String("ocis.debouncer.queue.action."+key, value)) + } + + return attrs +} diff --git a/pkg/debouncer/option.go b/pkg/debouncer/option.go new file mode 100644 index 00000000000..199b861dcec --- /dev/null +++ b/pkg/debouncer/option.go @@ -0,0 +1,32 @@ +package debouncer + +// Option defines a single option function +type Option func(o *Options) + +// Options represents the available options for the debouncer +type Options struct { + DefaultPolicyFactory PolicyFactory + DefaultTaskListFactory TaskListFactory + DefaultActionFactory ActionFactory +} + +// WithDefaultPolicy provides an option to set the default policy +func WithDefaultPolicy(p PolicyFactory) Option { + return func(o *Options) { + o.DefaultPolicyFactory = p + } +} + +// WithDefaultAction provides an option to set the default action +func WithDefaultAction(a ActionFactory) Option { + return func(o *Options) { + o.DefaultActionFactory = a + } +} + +// WithDefaultTaskList provides an option to set the default task list +func WithDefaultTaskList(t TaskListFactory) Option { + return func(o *Options) { + o.DefaultTaskListFactory = t + } +} diff --git a/pkg/debouncer/policy/policy.go b/pkg/debouncer/policy/policy.go new file mode 100644 index 00000000000..a96ab9836b4 --- /dev/null +++ b/pkg/debouncer/policy/policy.go @@ -0,0 +1,49 @@ +package policy + +import ( + "time" +) + +// Policy represents a trigger policy for the debouncer to use. +// The debouncer should queue tasks until the policy triggers, then the +// debouncer can run through the tasks. +// +// The MarkTaskAdded and MarkTaskRemoved methods are used to help the +// policy to decide whether to send the trigger now or wait a bit longer. +// For example, a time-based policy can queue tasks for 2 minutes before +// triggering, and can extend or reset that time period if tasks are being +// added to the queue +// +// The WaitForTrigger will return a channel so other goroutines can wait on +// it. Note that the expectation is the policy will trigger only once. +// Once the policy triggers, consider it unusable for the rest of the +// execution. Create a new instance if you need it. +type Policy interface { + // MarkTaskAdded will notify the policy that a new task has been added + // to the queue. + // Return true if it's acknowledged, false otherwise + MarkTaskAdded() bool + // MarkTaskRemoved will notify the policy that a new task has been + // removed from the queue. + // Return true if it's acknowledged, false otherwise + MarkTaskRemoved() bool + // WaitForTrigger will return a receiver channel to wait for the policy + // to trigger. The channel will send the time when the policy triggered, + // and then the channel will be closed. + // Note that multiple goroutines might be waiting here. + // The recommendation is to use a shared channel, so all the goroutines + // wait on the same channel. Once the channel is closed, all the + // goroutines can proceed. Note that only one goroutine will receive + // the time. + WaitForTrigger() <-chan time.Time + // GetId returns the ID of this instance. It must be unique, so multiple + // instances from the same policy type must return different IDs. + // The recommendation is to use the instance type followed by a random + // number, such as "TimedWithReset_123987" + GetId() string + // GetTracingData returns additional data that will be used for tracing. + // Consider this data as public information. This data is intended to + // be use purely for informational purposes. You can return nil if + // there isn't any data to be published. + GetTracingData() map[string]string +} diff --git a/pkg/debouncer/policy/policy_suite_test.go b/pkg/debouncer/policy/policy_suite_test.go new file mode 100644 index 00000000000..f7f20f0d22f --- /dev/null +++ b/pkg/debouncer/policy/policy_suite_test.go @@ -0,0 +1,13 @@ +package policy_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestPolicy(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Debouncer policy suite") +} diff --git a/pkg/debouncer/policy/timedwithreset.go b/pkg/debouncer/policy/timedwithreset.go new file mode 100644 index 00000000000..558bbdef652 --- /dev/null +++ b/pkg/debouncer/policy/timedwithreset.go @@ -0,0 +1,90 @@ +package policy + +import ( + "math/rand" + "strconv" + "sync/atomic" + "time" +) + +// TimedWithReset implements a debouncer policy. +// The policy will trigger after the specified duration passes. If the policy +// hasn't been triggered yet, the duration will be reset each time the +// MarkTaskAdded is called. +type TimedWithReset struct { + id string + timer *time.Timer + triggerChan chan time.Time + duration time.Duration + done *atomic.Bool +} + +// NewTimedWithReset creates a new instance of TimedWithReset. +// The timer will start immediately. +// Note that this implementation won't provide extreme accurate timing, +// so some minor delays (milliseconds at most) are expected. The expected +// usage should have at least a 10 seconds duration for the timer (tests use +// a 1 second timer), which should be enough to interact with this policy +// before the timer goes off. +func NewTimedWithReset(dur time.Duration) *TimedWithReset { + timer := time.NewTimer(dur) + triggerChan := make(chan time.Time) + done := &atomic.Bool{} + id := "TimedWithReset_" + strconv.FormatUint(rand.Uint64(), 10) + + go func() { + defer timer.Stop() + t := <-timer.C + done.Store(true) + triggerChan <- t + close(triggerChan) + }() + + return &TimedWithReset{ + id: id, + timer: timer, + triggerChan: triggerChan, + duration: dur, + done: done, + } +} + +// MarkTaskAdded will return true if the policy hasn't triggered yet +// and the timer has been reset, false otherwise. +func (tr *TimedWithReset) MarkTaskAdded() bool { + if tr.done.Load() == false { + return tr.timer.Reset(tr.duration) + } + return false +} + +// MarkTaskRemoved has no effect and it will always return true +func (tr *TimedWithReset) MarkTaskRemoved() bool { + return true +} + +// WaitForTrigger will return a channel in order to wait for the policy +// to trigger. The channel will be the same for all the calls to this method +// (the channel will be shared among all the callers) +// The channel will send the time when the trigger happens, however, since +// the channel is shared among all the waiting goroutines, only one of them +// will receive the time. +// The channel will be closed once the policy triggers, so all the waiting +// goroutines can continue afterwards. +func (tr *TimedWithReset) WaitForTrigger() <-chan time.Time { + return tr.triggerChan +} + +// GetId returns the ID of this instance. It will return "TimedWithReset_" +// followed by a random number. +func (tr *TimedWithReset) GetId() string { + return tr.id +} + +// GetTracingData will return the duration used by this instance, under +// the key "duration", using the "Duration.String()" method. +func (tr *TimedWithReset) GetTracingData() map[string]string { + return map[string]string{ + "duration": tr.duration.String(), + } +} diff --git a/pkg/debouncer/policy/timedwithreset_test.go b/pkg/debouncer/policy/timedwithreset_test.go new file mode 100644 index 00000000000..562b1210948 --- /dev/null +++ b/pkg/debouncer/policy/timedwithreset_test.go @@ -0,0 +1,99 @@ +package policy_test + +import ( + "sync" + "time" + + "github.com/owncloud/reva/v2/pkg/debouncer/policy" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + . "github.com/onsi/gomega/gleak" +) + +var _ = Describe("TimedWithReset", func() { + var tr policy.Policy + + BeforeEach(func() { + tr = policy.NewTimedWithReset(1 * time.Second) + }) + + AfterEach(func() { + Eventually(Goroutines).ShouldNot(HaveLeaked()) + }) + + Describe("Only one goroutine", func() { + It("Waiting channel closed after trigger", func() { + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + }) + + It("Immediately returns after channel closes", func() { + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + + t1 := time.Now() + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + Expect(time.Now()).To(BeTemporally("~", t1)) // within 1ms of difference + }) + + It("MarkTaskAdded extends duration", func() { + time.Sleep(700 * time.Millisecond) + Expect(tr.MarkTaskAdded()).To(Equal(true)) + Consistently(tr.WaitForTrigger(), "1s").ShouldNot(BeClosed()) + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + }) + + It("MarkTaskAdded fails if channel closed", func() { + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + Expect(tr.MarkTaskAdded()).To(Equal(false)) + }) + }) + + Describe("Multiple goroutines", func() { + It("Two goroutines wait until channel closes", func() { + // No goroutine gets stuck waiting indefinitely after the policy has triggered + var wg sync.WaitGroup + wg.Go(func() { + defer GinkgoRecover() + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + }) + wg.Go(func() { + defer GinkgoRecover() + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + }) + wg.Wait() + }) + + It("Second goroutine doesn't wait after channel close", func() { + var wg sync.WaitGroup + wg.Go(func() { + defer GinkgoRecover() + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + }) + wg.Wait() + + t1 := time.Now() + Eventually(tr.WaitForTrigger()).WithTimeout(3 * time.Second).Should(BeClosed()) + Expect(time.Now()).To(BeTemporally("~", t1)) // within 1ms of difference + }) + + It("Goroutine keeps adding tasks", func() { + var wg sync.WaitGroup + wg.Go(func() { + defer GinkgoRecover() + // each 500ms add a task, up to 4 + for i := 0; i < 4; i++ { + time.Sleep(500 * time.Millisecond) + Expect(tr.MarkTaskAdded()).To(Equal(true)) + } + }) + wg.Go(func() { + defer GinkgoRecover() + t1 := time.Now() + Eventually(tr.WaitForTrigger()).WithTimeout(5 * time.Second).Should(BeClosed()) + // timer should reset each 0.5s until the 2s, no further resets so final time should be 3s + Expect(time.Now()).To(BeTemporally("~", t1.Add(3*time.Second), 100*time.Millisecond)) // within 100ms of difference + }) + wg.Wait() + }) + }) +}) diff --git a/pkg/debouncer/tasklist/slicewithlocks.go b/pkg/debouncer/tasklist/slicewithlocks.go new file mode 100644 index 00000000000..9351386d9bc --- /dev/null +++ b/pkg/debouncer/tasklist/slicewithlocks.go @@ -0,0 +1,87 @@ +package tasklist + +import ( + "math/rand" + "strconv" + "sync" + "sync/atomic" +) + +// SliceWithLocks implements a TaskList using a slice of tasks and locks +type SliceWithLocks struct { + id string + list []InternalTask + rwmutex *sync.RWMutex + frozen *atomic.Bool +} + +// NewSliceWithLocks creates a new SliceWithLocks instance +func NewSliceWithLocks() TaskList { + rw := &sync.RWMutex{} + frozen := &atomic.Bool{} + + return &SliceWithLocks{ + id: "SliceWithLocks_" + strconv.FormatUint(rand.Uint64(), 10), + list: make([]InternalTask, 0), + rwmutex: rw, + frozen: frozen, + } +} + +// AddToList adds the task to the list. Returns true if added, or false if +// the list is frozen +func (swl *SliceWithLocks) AddToList(task InternalTask) bool { + swl.rwmutex.Lock() + defer swl.rwmutex.Unlock() + + if swl.frozen.Load() == true { + return false + } + + swl.list = append(swl.list, task) + return true +} + +// Freeze will freeze the list so no new task is added. +// You can freeze the list multiple times, but you can't revert (melt) it +func (swl *SliceWithLocks) Freeze() { + // Use a read lock because we don't want to change the state while + // adding a new task to the list. + swl.rwmutex.RLock() + defer swl.rwmutex.RUnlock() + + _ = swl.frozen.CompareAndSwap(false, true) +} + +// IsFrozen returns whether the list has been frozen or not +func (swl *SliceWithLocks) IsFrozen() bool { + return swl.frozen.Load() +} + +// ToSlice will return a slice of tasks. It will return a copy of the backed +// slice, so both the original and the copy can be modified independently +// (although you shouldn't modify the returned list) +// This method will return a shallow copy. The tasks are expected to be +// pointers, so modifying the tasks in any of the lists (either the original +// or the returned copy) will affect both. +// It's recommended to use this method after the freeze to ensure no new task +// is added later and we're operating over an old list. +func (swl *SliceWithLocks) ToSlice() []InternalTask { + swl.rwmutex.RLock() + defer swl.rwmutex.RUnlock() + + out := make([]InternalTask, len(swl.list)) + _ = copy(out, swl.list) + return out +} + +// GetId will return the id of this instance. This will return +// "SliceWithLocks_" followed by a random number. +func (swl *SliceWithLocks) GetId() string { + return swl.id +} + +// GetTracingData will return nil +func (swl *SliceWithLocks) GetTracingData() map[string]string { + return nil +} diff --git a/pkg/debouncer/tasklist/slicewithlocks_test.go b/pkg/debouncer/tasklist/slicewithlocks_test.go new file mode 100644 index 00000000000..1eb65f6b31e --- /dev/null +++ b/pkg/debouncer/tasklist/slicewithlocks_test.go @@ -0,0 +1,145 @@ +package tasklist_test + +import ( + "context" + "sync" + + "github.com/owncloud/reva/v2/pkg/debouncer/tasklist" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type DummyTask struct { + Err error +} + +func (t *DummyTask) ExposeData() map[string]string { + return nil +} + +func (t *DummyTask) Execute(ctx context.Context) error { + return t.Err +} + +var _ = Describe("SliceWithLocks", func() { + var swl tasklist.TaskList + + BeforeEach(func() { + swl = tasklist.NewSliceWithLocks() + }) + + Describe("Only one goroutine", func() { + It("Add to list successful", func() { + task1 := tasklist.NewInternalTaskFromTask(context.Background(), &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeTrue()) + }) + + It("Add to list fails if frozen", func() { + swl.Freeze() + task1 := tasklist.NewInternalTaskFromTask(context.Background(), &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeFalse()) + }) + + It("Check not frozen initially", func() { + Expect(swl.IsFrozen()).To(BeFalse()) + }) + + It("Check frozen state after freeze", func() { + swl.Freeze() + Expect(swl.IsFrozen()).To(BeTrue()) + }) + + It("ToSlice returns added tasks", func() { + ctx := context.Background() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + swl.AddToList(task1) + swl.AddToList(task2) + + swl.Freeze() // not needed, but recommended + returnedList := swl.ToSlice() + Expect(returnedList).To(HaveLen(2)) + Expect(returnedList[0]).To(Equal(task1)) + Expect(returnedList[1]).To(Equal(task2)) + }) + + It("ToSlice won't be modified if not frozen", func() { + ctx := context.Background() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + swl.AddToList(task1) + swl.AddToList(task2) + + list1 := swl.ToSlice() + task3 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + Expect(swl.AddToList(task3)).To(BeTrue()) // not frozen -> adding is allowed + Expect(list1).To(HaveLen(2)) + Expect(list1[0]).To(Equal(task1)) + Expect(list1[1]).To(Equal(task2)) + + list2 := swl.ToSlice() + Expect(list2).To(HaveLen(3)) + Expect(list2[0]).To(Equal(task1)) + Expect(list2[1]).To(Equal(task2)) + Expect(list2[2]).To(Equal(task3)) + }) + }) + + Describe("Multiple goroutines", func() { + It("AddToList can be used from multiple goroutines", func() { + ctx := context.Background() + var wg sync.WaitGroup + wg.Go(func() { + defer GinkgoRecover() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeTrue()) + }) + wg.Go(func() { + defer GinkgoRecover() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeTrue()) + }) + wg.Go(func() { + defer GinkgoRecover() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeTrue()) + }) + wg.Wait() + Expect(swl.IsFrozen()).To(BeFalse()) + Expect(swl.ToSlice()).To(HaveLen(3)) + }) + + It("ToSlice won't modify the backed list", func() { + ctx := context.Background() + task1 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + Expect(swl.AddToList(task1)).To(BeTrue()) + + var wg sync.WaitGroup + wg.Go(func() { + defer GinkgoRecover() + + list := swl.ToSlice() + Expect(list).To(HaveLen(1)) + + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + list = append(list, task2) + Expect(list).To(HaveLen(2)) + }) + wg.Go(func() { + defer GinkgoRecover() + + list := swl.ToSlice() + Expect(list).To(HaveLen(1)) + + task2 := tasklist.NewInternalTaskFromTask(ctx, &DummyTask{}) + list = append(list, task2) + Expect(list).To(HaveLen(2)) + }) + wg.Wait() + + list := swl.ToSlice() + Expect(list).To(HaveLen(1)) + }) + }) +}) diff --git a/pkg/debouncer/tasklist/task.go b/pkg/debouncer/tasklist/task.go new file mode 100644 index 00000000000..3b4b1ffb3ba --- /dev/null +++ b/pkg/debouncer/tasklist/task.go @@ -0,0 +1,39 @@ +package tasklist + +import ( + "context" + + "go.opentelemetry.io/otel/trace" +) + +// Task represents a task that has been queued in the debouncer. +// The task can be anything, from printing a message to overwriting data +// in external services or indexing a space. +// The general expectation is that the same or similar tasks are queued +// together, and only one of them will be executed. This can be changed +// based on the configured policies and actions of the debouncer. +type Task interface { + // ExposeData exposes some task data as key-value pairs. This is + // tightly coupled with the Action in the debouncer, and it's expected + // that the Action uses this information to decide whether to run + // this Task or not. Some basic Actions might not need this info. + ExposeData() map[string]string + // Execute will run the task (such as overwriting a value in an + // external service) in the provided context. Note that the context + // will be new and it might not have some required context + // (authentication info, for example), so you might need to add the + // required information. + Execute(ctx context.Context) error +} + +type InternalTask struct { + SpanContext trace.SpanContext + OriginalTask Task +} + +func NewInternalTaskFromTask(ctx context.Context, t Task) InternalTask { + return InternalTask{ + SpanContext: trace.SpanContextFromContext(ctx), + OriginalTask: t, + } +} diff --git a/pkg/debouncer/tasklist/tasklist.go b/pkg/debouncer/tasklist/tasklist.go new file mode 100644 index 00000000000..3e0c43cf1ca --- /dev/null +++ b/pkg/debouncer/tasklist/tasklist.go @@ -0,0 +1,32 @@ +package tasklist + +// TaskList represents a list of tasks. The basic idea is a []Task BUT all the +// operations MUST be thread-safe. +// Different implementations can be provided, backed by different data +// structures, as long as the implementation is thread-safe. +type TaskList interface { + // AddToList adds the task to the list. Returns true if the task is + // added, false if not. + AddToList(task InternalTask) bool + // Freeze the task list so no new task can be added. After this method + // returns, the AddToList method MUST always return false. + Freeze() + // IsFrozen returns the frozen status of the task list + IsFrozen() bool + // ToSlice returns the tasks as a slice. It's recommended to Freeze + // the task list implementation before this method so all the tasks + // are present in the slice. If the implementation isn't frozen, a + // snapshot of the list is expected (new tasks might be added to the + // list while this method is running) + ToSlice() []InternalTask + // GetId returns the ID of this instance. It must be unique, so multiple + // instances from the same task list type must return different IDs. + // The recommendation is to use the instance type followed by a random + // number, such as "SliceWithLocks_123987" + GetId() string + // GetTracingData returns additional data that will be used for tracing. + // Consider this data as public information. This data is intended to + // be use purely for informational purposes. You can return nil if + // there isn't any data to be published. + GetTracingData() map[string]string +} diff --git a/pkg/debouncer/tasklist/tasklist_suite_test.go b/pkg/debouncer/tasklist/tasklist_suite_test.go new file mode 100644 index 00000000000..d1f5d753606 --- /dev/null +++ b/pkg/debouncer/tasklist/tasklist_suite_test.go @@ -0,0 +1,13 @@ +package tasklist_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestTaskList(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Debouncer task list suite") +}