Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use go 1.25 waitgroup Go method instead of Add/Wait (#3265)

Browse Source 
* Use go 1.25 waitgroup Go method instead of Add/Wait

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

* Disable java builder test for now

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

---------

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2025-10-30 14:32:44 +05:30
committed by GitHub
parent 4ebdb077b5
commit ebdce4c0ed
11 changed files with 96 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/executor.go
+10 -12
View File
@@ -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.
pkg/executor/executortype/container/containermgr.go
+2 -5
View File
@@ -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))
}()
})
}
}
}
pkg/executor/executortype/newdeploy/newdeploymgr.go
+2 -5
View File
@@ -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))
}()
})
}
}
}
pkg/executor/executortype/poolmgr/gpm.go
+4 -9
View File
@@ -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))
}()
})
}
}
pkg/executor/fscache/poolcache_test.go
+30 -29
View File
@@ -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()
pkg/executor/fscache/queue_test.go
+2 -4
View File
@@ -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()
pkg/fission-cli/cmd/support/dump.go
+5 -5
View File
@@ -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()
pkg/fission-cli/cmd/support/resources/kubernetes.go
+35 -36
View File
@@ -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()
pkg/utils/manager/manager.go
+2 -4
View File
@@ -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) {
pkg/utils/metrics/http_metrics_test.go
+2 -4
View File
@@ -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)
test/tests/test_environments/test_java_builder.sh
+2
View File
@@ -1,5 +1,7 @@
#!/bin/bash
#test:disabled
set -euo pipefail
source $(dirname $0)/../../utils.sh