Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

call plugin in parallel #613

Merged
merged 2 commits into from
Sep 21, 2023
+27 −12
Merged

call plugin in parallel #613

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
92c9529
call plugin in parallel
yuyang0 Sep 21, 2023
002c510
remove lock
yuyang0 Sep 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 19 additions & 9 deletions resource/cobalt/call.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,26 +2,36 @@ package cobalt

import (
"context"
"sync"

"github.com/cockroachdb/errors"
"github.com/projecteru2/core/log"
"github.com/projecteru2/core/resource/plugins"
)

func call[T any](ctx context.Context, ps []plugins.Plugin, f func(plugins.Plugin) (T, error)) (map[plugins.Plugin]T, error) {
// TODO 并行化，意义不大
var mu sync.Mutex
var wg sync.WaitGroup
var combinedErr error
results := map[plugins.Plugin]T{}

for _, p := range ps {
result, err := f(p)
if err != nil {
log.WithFunc("resource.cobalt.call").Errorf(ctx, err, "failed to call plugin %+v", p.Name())
combinedErr = errors.CombineErrors(combinedErr, errors.Wrap(err, p.Name()))
continue
}
results[p] = result
}
wg.Add(1)
go func(p plugins.Plugin) {
defer wg.Done()

result, err := f(p)
mu.Lock()
defer mu.Unlock()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

我觉得不如用个 buffer chan 传错误出来……反正长度是可控的

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

results还是需要锁保护

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

results 可以无锁 map，我记得 core 依赖了个haxmap 吧

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

可以，这个UT又很奇怪，完全不相干的地方，而且本地跑好像没问题

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

重跑一次好了

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

mutex删了，haxmap没法用，plugin不能作为它的key，感觉这么写更丑了

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

我看了下既然都用上了 sync.Map, 不如更进一步 sync.Map 存 error，看上去没这么丑？

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

应该好了吧


if err != nil {
log.WithFunc("resource.cobalt.call").Errorf(ctx, err, "failed to call plugin %+v", p.Name())
combinedErr = errors.CombineErrors(combinedErr, errors.Wrap(err, p.Name()))
return
}
results[p] = result
}(p)
}
wg.Wait()
return results, combinedErr
}