From ebdce4c0ed29d6bcb2aad2c54486ef30447a1223 Mon Sep 17 00:00:00 2001 From: Sanket Sudake Date: Thu, 30 Oct 2025 14:32:44 +0530 Subject: [PATCH] Use go 1.25 waitgroup Go method instead of Add/Wait (#3265) * Use go 1.25 waitgroup Go method instead of Add/Wait Signed-off-by: Sanket Sudake * Disable java builder test for now Signed-off-by: Sanket Sudake --------- Signed-off-by: Sanket Sudake --- pkg/executor/executor.go | 22 +++--- .../executortype/container/containermgr.go | 7 +- .../executortype/newdeploy/newdeploymgr.go | 7 +- pkg/executor/executortype/poolmgr/gpm.go | 13 ++-- pkg/executor/fscache/poolcache_test.go | 59 +++++++-------- pkg/executor/fscache/queue_test.go | 6 +- pkg/fission-cli/cmd/support/dump.go | 10 +-- .../cmd/support/resources/kubernetes.go | 71 +++++++++---------- pkg/utils/manager/manager.go | 6 +- pkg/utils/metrics/http_metrics_test.go | 6 +- .../test_environments/test_java_builder.sh | 2 + 11 files changed, 96 insertions(+), 113 deletions(-) diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index eb58d15a..dac30cb5 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -154,12 +154,11 @@ func (executor *Executor) serveCreateFuncServices(ctx context.Context) { // create a waitgroup for other requests for // the same function to wait on wg := &sync.WaitGroup{} - wg.Add(1) executor.fsCreateWg.Store(fnkeyUR, wg) // launch a goroutine for each request, to parallelize // the specialization of different functions - go func() { + wg.Go(func() { // Control overall specialization time by setting function // specialization time to context. The reason not to use // context from router requests is because a request maybe @@ -188,8 +187,7 @@ func (executor *Executor) serveCreateFuncServices(ctx context.Context) { err: err, } executor.fsCreateWg.Delete(fnkeyUR) - wg.Done() - }() + }) } else { // There's an existing request for this function, wait for it to finish go func() { @@ -352,14 +350,14 @@ func StartExecutor(ctx context.Context, clientGen crd.ClientGeneratorInterface, wg := &sync.WaitGroup{} for _, et := range executorTypes { - wg.Add(1) - go func(et executortype.ExecutorType) { - defer wg.Done() - if adoptExistingResources { - et.AdoptExistingResources(ctx) - } - et.CleanupOldExecutorObjects(ctx) - }(et) + wg.Go(func() { + func(et executortype.ExecutorType) { + if adoptExistingResources { + et.AdoptExistingResources(ctx) + } + et.CleanupOldExecutorObjects(ctx) + }(et) + }) } // set hard timeout for resource adoption // TODO: use context to control the waiting time once kubernetes client supports it. diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 943dd146..70cb4e75 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -302,17 +302,14 @@ func (caaf *Container) AdoptExistingResources(ctx context.Context) { for i := range fnList.Items { fn := &fnList.Items[i] if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeContainer { - wg.Add(1) - go func() { - defer wg.Done() - + wg.Go(func() { _, err = caaf.fnCreate(ctx, fn) if err != nil { caaf.logger.Warn("failed to adopt resources for function", zap.Error(err)) return } caaf.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name)) - }() + }) } } } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 61167f4c..146564d5 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -322,17 +322,14 @@ func (deploy *NewDeploy) AdoptExistingResources(ctx context.Context) { for i := range fnList.Items { fn := &fnList.Items[i] if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { - wg.Add(1) - go func() { - defer wg.Done() - + wg.Go(func() { _, err = deploy.fnCreate(ctx, fn) if err != nil { deploy.logger.Warn("failed to adopt resources for function", zap.Error(err)) return } deploy.logger.Info("adopt resources for function", zap.String("function", fn.ObjectMeta.Name)) - }() + }) } } } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 8aafa8a4..67b8120c 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -368,9 +368,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { env := envs.Items[i] if getEnvPoolSize(&env) > 0 { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, created, err := gpm.getPool(ctx, &env) if err != nil { gpm.logger.Error("adopt pool failed", zap.Error(err)) @@ -378,7 +376,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { if created { gpm.logger.Info("created pool for the environment", zap.String("env", env.ObjectMeta.Name), zap.String("namespace", gpm.nsResolver.ResolveNamespace(gpm.nsResolver.FunctionNamespace))) } - }() + }) } // create environment map for later use @@ -407,10 +405,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { continue } - wg.Add(1) - go func() { - defer wg.Done() - + wg.Go(func() { // avoid too many requests arrive Kubernetes API server at the same time. time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) @@ -482,7 +477,7 @@ func (gpm *GenericPoolManager) AdoptExistingResources(ctx context.Context) { gpm.logger.Info("adopt function pod", zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations)) - }() + }) } } diff --git a/pkg/executor/fscache/poolcache_test.go b/pkg/executor/fscache/poolcache_test.go index 86bf9dec..3daa6941 100644 --- a/pkg/executor/fscache/poolcache_test.go +++ b/pkg/executor/fscache/poolcache_test.go @@ -224,32 +224,33 @@ func TestPoolCacheRequests(t *testing.T) { } for i := 1; i <= tt.requests; i++ { reqno := i - wg.Add(1) - go func(reqno int) { - defer wg.Done() - svc, err := p.GetSvcValue(context.Background(), key, tt.rpp, tt.concurrency) - if err != nil { - code, _ := ferror.GetHTTPError(err) - if code == http.StatusNotFound { - atomic.AddUint64(&svcCounter, 1) - address := fmt.Sprintf("svc-%d", atomic.LoadUint64(&svcCounter)) - p.SetSvcValue(context.Background(), key, address, &FuncSvc{ - Name: address, - }, resource.MustParse("45m"), tt.rpp, tt.retainPods) + wg.Go(func() { + func(reqno int) { + svc, err := p.GetSvcValue(context.Background(), key, tt.rpp, tt.concurrency) + if err != nil { + code, _ := ferror.GetHTTPError(err) + if code == http.StatusNotFound { + atomic.AddUint64(&svcCounter, 1) + address := fmt.Sprintf("svc-%d", atomic.LoadUint64(&svcCounter)) + p.SetSvcValue(context.Background(), key, address, &FuncSvc{ + Name: address, + }, resource.MustParse("45m"), tt.rpp, tt.retainPods) + } else { + t.Log(reqno, "=>", err) + atomic.AddUint64(&failedRequests, 1) + } } else { - t.Log(reqno, "=>", err) - atomic.AddUint64(&failedRequests, 1) + if svc == nil { + t.Log(reqno, "=>", "svc is nil") + atomic.AddUint64(&failedRequests, 1) + } + // } else { + // t.Log(reqno, "=>", svc.Name) + // } } - } else { - if svc == nil { - t.Log(reqno, "=>", "svc is nil") - atomic.AddUint64(&failedRequests, 1) - } - // } else { - // t.Log(reqno, "=>", svc.Name) - // } - } - }(reqno) + }(reqno) + }) + if reqno%simultaneous == 0 { wg.Wait() } @@ -261,12 +262,12 @@ func TestPoolCacheRequests(t *testing.T) { for i := 0; i < tt.concurrency; i++ { for j := 0; j < tt.rpp; j++ { - wg.Add(1) svcno := i - go func(svcno int) { - defer wg.Done() - p.MarkAvailable(key, fmt.Sprintf("svc-%d", svcno+1)) - }(svcno) + wg.Go(func() { + func(svcno int) { + p.MarkAvailable(key, fmt.Sprintf("svc-%d", svcno+1)) + }(svcno) + }) } } wg.Wait() diff --git a/pkg/executor/fscache/queue_test.go b/pkg/executor/fscache/queue_test.go index fb724f6c..2f92f5a4 100644 --- a/pkg/executor/fscache/queue_test.go +++ b/pkg/executor/fscache/queue_test.go @@ -48,16 +48,14 @@ func TestQueuePushWithConcurrentRequest(t *testing.T) { q := NewQueue() noOfRequests := 20 var wg sync.WaitGroup - wg.Add(noOfRequests) for i := 0; i < noOfRequests; i++ { - go func() { - defer wg.Done() + wg.Go(func() { item := &svcWait{ svcChannel: make(chan *FuncSvc), ctx: nil, } q.Push(item) - }() + }) } wg.Wait() diff --git a/pkg/fission-cli/cmd/support/dump.go b/pkg/fission-cli/cmd/support/dump.go index 5b93c8e9..0128a21f 100644 --- a/pkg/fission-cli/cmd/support/dump.go +++ b/pkg/fission-cli/cmd/support/dump.go @@ -128,11 +128,11 @@ func (opts *DumpSubCommand) do(input cli.Input) error { panic(err) } } - wg.Add(1) - go func(res resources.Resource, dir string) { - defer wg.Done() - res.Dump(input.Context(), dir) - }(res, dir) + wg.Go(func() { + func(res resources.Resource, dir string) { + res.Dump(input.Context(), dir) + }(res, dir) + }) } wg.Wait() diff --git a/pkg/fission-cli/cmd/support/resources/kubernetes.go b/pkg/fission-cli/cmd/support/resources/kubernetes.go index c6cc174a..8bf69275 100644 --- a/pkg/fission-cli/cmd/support/resources/kubernetes.go +++ b/pkg/fission-cli/cmd/support/resources/kubernetes.go @@ -207,49 +207,48 @@ func (res KubernetesPodLogDumper) Dump(ctx context.Context, dumpDir string) { wg := &sync.WaitGroup{} for _, p := range l.Items { - wg.Add(1) + wg.Go(func() { + func(pod corev1.Pod) { + // dump logs from each containers + for _, container := range append(pod.Spec.Containers, pod.Spec.InitContainers...) { + req := res.client.CoreV1().Pods(pod.Namespace). + GetLogs(pod.Name, &corev1.PodLogOptions{Container: container.Name}) - go func(pod corev1.Pod) { - defer wg.Done() - - // dump logs from each containers - for _, container := range append(pod.Spec.Containers, pod.Spec.InitContainers...) { - req := res.client.CoreV1().Pods(pod.Namespace). - GetLogs(pod.Name, &corev1.PodLogOptions{Container: container.Name}) - - stream, err := req.Stream(ctx) - if err != nil { - console.Error(fmt.Sprintf("Error streaming logs for pod %v: %v", pod.Name, err)) - return - } - - reader := bufio.NewReader(stream) - var buffer bytes.Buffer - - for { - line, _, err := reader.ReadLine() + stream, err := req.Stream(ctx) if err != nil { - if err == io.EOF { - stream.Close() - break + console.Error(fmt.Sprintf("Error streaming logs for pod %v: %v", pod.Name, err)) + return + } + + reader := bufio.NewReader(stream) + var buffer bytes.Buffer + + for { + line, _, err := reader.ReadLine() + if err != nil { + if err == io.EOF { + stream.Close() + break + } + console.Error(fmt.Sprintf("Error reading logs from buffer: %v", err)) + return + } + + _, err = buffer.WriteString(string(line) + "\n") + if err != nil { + console.Error(fmt.Sprintf("Error writing bytes to buffer: %v", err)) + return } - console.Error(fmt.Sprintf("Error reading logs from buffer: %v", err)) - return } - _, err = buffer.WriteString(string(line) + "\n") - if err != nil { - console.Error(fmt.Sprintf("Error writing bytes to buffer: %v", err)) - return - } + f := getPodFileName(dumpDir, pod.ObjectMeta, container.Name) + writeToFile(f, buffer.String()) + + stream.Close() } + }(p) + }) - f := getPodFileName(dumpDir, pod.ObjectMeta, container.Name) - writeToFile(f, buffer.String()) - - stream.Close() - } - }(p) } wg.Wait() diff --git a/pkg/utils/manager/manager.go b/pkg/utils/manager/manager.go index bb8cbdb4..ecd5687c 100644 --- a/pkg/utils/manager/manager.go +++ b/pkg/utils/manager/manager.go @@ -36,11 +36,9 @@ func New() Interface { } } func (g *GroupManager) Add(ctx context.Context, f func(context.Context)) { - g.wg.Add(1) - go func() { - defer g.wg.Done() + g.wg.Go(func() { f(ctx) - }() + }) } func (g *GroupManager) AddInformers(ctx context.Context, informers map[string]k8sCache.SharedIndexInformer) { diff --git a/pkg/utils/metrics/http_metrics_test.go b/pkg/utils/metrics/http_metrics_test.go index 61ede88f..a183469a 100644 --- a/pkg/utils/metrics/http_metrics_test.go +++ b/pkg/utils/metrics/http_metrics_test.go @@ -23,9 +23,7 @@ func chunkedHandler(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithCancel(r.Context()) ticker := time.NewTicker(time.Second) // We may set it to 10 secs var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for { select { case <-ticker.C: @@ -37,7 +35,7 @@ func chunkedHandler(w http.ResponseWriter, r *http.Request) { return } } - }() + }) // Emulate some work time.Sleep(5 * time.Second) diff --git a/test/tests/test_environments/test_java_builder.sh b/test/tests/test_environments/test_java_builder.sh index be65857b..305f1620 100755 --- a/test/tests/test_environments/test_java_builder.sh +++ b/test/tests/test_environments/test_java_builder.sh @@ -1,5 +1,7 @@ #!/bin/bash +#test:disabled + set -euo pipefail source $(dirname $0)/../../utils.sh