public_git
/
helmfile
mirror of https://github.com/helmfile/helmfile.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix: concurrency issue in diff and sync (#309) 

fixes #301
This commit is contained in:
KUOKA Yusuke 2018-09-06 10:14:10 +09:00 committed by GitHub
parent 2e44da1b54
commit a205f0d586
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
state/state.go
View File

@ -156,11 +156,12 @@ type syncPrepareResult struct {
// SyncReleases wrapper for executing helm upgrade on the releases // SyncReleases wrapper for executing helm upgrade on the releases
func (state *HelmState) prepareSyncReleases(helm helmexec.Interface, additionalValues []string, concurrency int) ([]syncPrepareResult, []error) { func (state *HelmState) prepareSyncReleases(helm helmexec.Interface, additionalValues []string, concurrency int) ([]syncPrepareResult, []error) {
jobs := make(chan *ReleaseSpec) numReleases := len(state.Releases)
results := make(chan syncPrepareResult) jobs := make(chan *ReleaseSpec, numReleases)
results := make(chan syncPrepareResult, numReleases)
if concurrency < 1 { if concurrency < 1 {
concurrency = len(state.Releases) concurrency = numReleases
} }
for w := 1; w <= concurrency; w++ { for w := 1; w <= concurrency; w++ {
go func() { go func() {
@ -195,16 +196,14 @@ func (state *HelmState) prepareSyncReleases(helm helmexec.Interface, additionalV
}() }()
} }
go func() { for i := 0; i < numReleases; i++ {
for i := 0; i < len(state.Releases); i++ { jobs <- &state.Releases[i]
jobs <- &state.Releases[i] }
} close(jobs)
close(jobs)
}()
res := []syncPrepareResult{} res := []syncPrepareResult{}
errs := []error{} errs := []error{}
for i := 0; i < len(state.Releases); { for i := 0; i < numReleases; {
select { select {
case r := <-results: case r := <-results:
for _, e := range r.errors { for _, e := range r.errors {
@ -225,8 +224,8 @@ func (state *HelmState) SyncReleases(helm helmexec.Interface, additionalValues [
return prepErrs return prepErrs
} }
jobQueue := make(chan *syncPrepareResult) jobQueue := make(chan *syncPrepareResult, len(preps))
results := make(chan syncResult) results := make(chan syncResult, len(preps))
if workerLimit < 1 { if workerLimit < 1 {
workerLimit = len(state.Releases) workerLimit = len(state.Releases)
@ -452,11 +451,12 @@ type diffPrepareResult struct {
} }
func (state *HelmState) prepareDiffReleases(helm helmexec.Interface, additionalValues []string, concurrency int, detailedExitCode, suppressSecrets bool) ([]diffPrepareResult, []error) { func (state *HelmState) prepareDiffReleases(helm helmexec.Interface, additionalValues []string, concurrency int, detailedExitCode, suppressSecrets bool) ([]diffPrepareResult, []error) {
jobs := make(chan *ReleaseSpec, len(state.Releases)) numReleases := len(state.Releases)
results := make(chan diffPrepareResult) jobs := make(chan *ReleaseSpec, numReleases)
results := make(chan diffPrepareResult, numReleases)
if concurrency < 1 { if concurrency < 1 {
concurrency = len(state.Releases) concurrency = numReleases
} }
for w := 1; w <= concurrency; w++ { for w := 1; w <= concurrency; w++ {
@ -504,14 +504,14 @@ func (state *HelmState) prepareDiffReleases(helm helmexec.Interface, additionalV
}() }()
} }
for i := 0; i < len(state.Releases); i++ { for i := 0; i < numReleases; i++ {
jobs <- &state.Releases[i] jobs <- &state.Releases[i]
} }
close(jobs) close(jobs)
rs := []diffPrepareResult{} rs := []diffPrepareResult{}
errs := []error{} errs := []error{}
for i := 0; i < len(state.Releases); { for i := 0; i < numReleases; {
select { select {
case res := <-results: case res := <-results:
if res.errors != nil && len(res.errors) > 0 { if res.errors != nil && len(res.errors) > 0 {