✨ [parallelisation] Create a store for closing functions so they can be done concurrently (#653)

<!--
Copyright (C) 2020-2022 Arm Limited or its affiliates and Contributors.
All rights reserved.
SPDX-License-Identifier: Apache-2.0
-->
### Description

- Close functions concurrently


### Test Coverage

<!--
Please put an `x` in the correct box e.g. `[x]` to indicate the testing
coverage of this change.
-->

- [x]  This change is covered by existing or additional automated tests.
- [ ] Manual testing has been performed (and evidence provided) as
automated testing was not feasible.
- [ ] Additional tests are not required for this change (e.g.
documentation update).

Author: acabarbaye

changes/20250715185008.feature

@@ -0,0 +1 @@
+:sparkles: `[parallelisation]` Create a store for closing functions so they can be done concurrently

utils/parallelisation/cancel_functions.go

@@ -9,36 +9,86 @@ import (
 	"context"
 
 	"github.com/sasha-s/go-deadlock"
+	"golang.org/x/sync/errgroup"
 )
 
-type CancelFunctionStore struct {
-	mu              deadlock.RWMutex
-	cancelFunctions []context.CancelFunc
+func newFunctionStore[T any](clearOnExecution, stopOnFirstError bool, executeFunc func(context.Context, T) error) *store[T] {
+	return &store[T]{
+		mu:               deadlock.RWMutex{},
+		functions:        make([]T, 0),
+		executeFunc:      executeFunc,
+		clearOnExecution: clearOnExecution,
+		stopOnFirstError: stopOnFirstError,
+	}
 }
 
-func (s *CancelFunctionStore) RegisterCancelFunction(cancel ...context.CancelFunc) {
+type store[T any] struct {
+	mu               deadlock.RWMutex
+	functions        []T
+	executeFunc      func(ctx context.Context, element T) error
+	clearOnExecution bool
+	stopOnFirstError bool
+}
+
+func (s *store[T]) RegisterFunction(function ...T) {
 	defer s.mu.Unlock()
 	s.mu.Lock()
-	s.cancelFunctions = append(s.cancelFunctions, cancel...)
+	s.functions = append(s.functions, function...)
 }
 
-func (s *CancelFunctionStore) Cancel() {
+func (s *store[T]) Len() int {
+	defer s.mu.RUnlock()
+	s.mu.RLock()
+	return len(s.functions)
+}
+
+func (s *store[T]) Execute(ctx context.Context) error {
 	defer s.mu.Unlock()
 	s.mu.Lock()
-	for _, c := range s.cancelFunctions {
-		c()
+	g, gCtx := errgroup.WithContext(ctx)
+	if !s.stopOnFirstError {
+		gCtx = ctx
+	}
+	g.SetLimit(len(s.functions))
+	for i := range s.functions {
+		g.Go(func() error {
+			err := DetermineContextError(gCtx)
+			if err != nil {
+				return err
+			}
+			return s.executeFunc(gCtx, s.functions[i])
+		})
+	}
+
+	err := g.Wait()
+	if err == nil && s.clearOnExecution {
+		s.functions = make([]T, 0, len(s.functions))
 	}
-	s.cancelFunctions = []context.CancelFunc{}
+	return err
+}
+
+type CancelFunctionStore struct {
+	store[context.CancelFunc]
+}
+
+func (s *CancelFunctionStore) RegisterCancelFunction(cancel ...context.CancelFunc) {
+	s.store.RegisterFunction(cancel...)
+}
+
+func (s *CancelFunctionStore) Cancel() {
+	_ = s.Execute(context.Background())
 }
 
 func (s *CancelFunctionStore) Len() int {
-	defer s.mu.RUnlock()
-	s.mu.RLock()
-	return len(s.cancelFunctions)
+	return s.store.Len()
 }
 
+// NewCancelFunctionsStore creates a store for cancel functions.
 func NewCancelFunctionsStore() *CancelFunctionStore {
 	return &CancelFunctionStore{
-		cancelFunctions: []context.CancelFunc{},
+		store: *newFunctionStore[context.CancelFunc](true, false, func(_ context.Context, cancelFunc context.CancelFunc) error {
+			cancelFunc()
+			return nil
+		}),
 	}
 }

utils/parallelisation/mocks/mock_parallelisation.go

@@ -0,0 +1,84 @@
+package parallelisation
+
+import (
+	"context"
+	"io"
+
+	"github.com/ARM-software/golang-utils/utils/commonerrors"
+)
+
+type CloserStore struct {
+	store[io.Closer]
+}
+
+func (s *CloserStore) RegisterCloser(closerObj ...io.Closer) {
+	s.store.RegisterFunction(closerObj...)
+}
+
+func (s *CloserStore) Close() error {
+	return s.Execute(context.Background())
+}
+
+func (s *CloserStore) Len() int {
+	return s.store.Len()
+}
+
+// NewCloserStore returns a store of io.Closer object which will all be closed concurrently on Close(). The first error received will be returned
+func NewCloserStore(stopOnFirstError bool) *CloserStore {
+	return &CloserStore{
+		store: *newFunctionStore[io.Closer](false, stopOnFirstError, func(_ context.Context, closerObj io.Closer) error {
+			if closerObj == nil {
+				return commonerrors.UndefinedVariable("closer object")
+			}
+			return closerObj.Close()
+		}),
+	}
+}
+
+// CloseAll calls concurrently Close on all io.Closer implementations passed as arguments and returns the first error encountered
+func CloseAll(cs ...io.Closer) error {
+	group := NewCloserStore(false)
+	group.RegisterFunction(cs...)
+	return group.Close()
+}
+
+// CloseAllWithContext is similar to CloseAll but can be controlled using a context.
+func CloseAllWithContext(ctx context.Context, cs ...io.Closer) error {
+	group := NewCloserStore(false)
+	group.RegisterFunction(cs...)
+	return group.Execute(ctx)
+}
+
+// CloseAllFunc calls concurrently all Close functions passed as arguments and returns the first error encountered
+func CloseAllFunc(cs ...CloseFunc) error {
+	group := NewCloseFunctionStoreStore(false)
+	group.RegisterFunction(cs...)
+	return group.Close()
+}
+
+type CloseFunc func() error
+
+type CloseFunctionStore struct {
+	store[CloseFunc]
+}
+
+func (s *CloseFunctionStore) RegisterCloseFunction(closerObj ...CloseFunc) {
+	s.store.RegisterFunction(closerObj...)
+}
+
+func (s *CloseFunctionStore) Close() error {
+	return s.Execute(context.Background())
+}
+
+func (s *CloseFunctionStore) Len() int {
+	return s.store.Len()
+}
+
+// NewCloseFunctionStoreStore returns a store closing functions which will all be called concurrently on Close(). The first error received will be returned.
+func NewCloseFunctionStoreStore(stopOnFirstError bool) *CloseFunctionStore {
+	return &CloseFunctionStore{
+		store: *newFunctionStore[CloseFunc](false, stopOnFirstError, func(_ context.Context, closerObj CloseFunc) error {
+			return closerObj()
+		}),
+	}
+}

utils/parallelisation/onclose.go

@@ -0,0 +1,43 @@
+package parallelisation
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	"github.com/ARM-software/golang-utils/utils/commonerrors"
+	"github.com/ARM-software/golang-utils/utils/commonerrors/errortest"
+	"github.com/ARM-software/golang-utils/utils/parallelisation/mocks"
+)
+
+//go:generate go tool mockgen -destination=./mocks/mock_$GOPACKAGE.go -package=mocks io Closer
+func TestCloseAll(t *testing.T) {
+	t.Run("close", func(t *testing.T) {
+		ctlr := gomock.NewController(t)
+		defer ctlr.Finish()
+
+		closerMock := mocks.NewMockCloser(ctlr)
+		closerMock.EXPECT().Close().Return(nil).MinTimes(1)
+
+		require.NoError(t, CloseAll(closerMock, closerMock, closerMock))
+	})
+
+	t.Run("close with error", func(t *testing.T) {
+		ctlr := gomock.NewController(t)
+		defer ctlr.Finish()
+		closeError := commonerrors.ErrUnexpected
+
+		closerMock := mocks.NewMockCloser(ctlr)
+		closerMock.EXPECT().Close().Return(closeError).MinTimes(1)
+
+		errortest.AssertError(t, CloseAll(closerMock, closerMock, closerMock), closeError)
+	})
+
+	t.Run("close with 1 error", func(t *testing.T) {
+		closeError := commonerrors.ErrUnexpected
+
+		errortest.AssertError(t, CloseAllFunc(func() error { return nil }, func() error { return nil }, func() error { return closeError }, func() error { return nil }), closeError)
+	})
+
+}

utils/parallelisation/onclose_test.go

@@ -28,7 +28,7 @@ func WaitWithContextAndError(ctx context.Context, wg IErrorWaiter) (err error) {
 	}
 }
 
-// IError can be used to wait on sync WaitGroups and similar types where Wait() does not return an error
+// IWaiter can be used to wait on sync WaitGroups and similar types where Wait() does not return an error
 // This is used to support use in the WaitWithContext function to wait but listen to contexts
 type IWaiter interface {
 	Wait()