Skip to content

Commit

Permalink
state: Introduce DependsOn for N-to-1 job dependencies (#1021)
Browse files Browse the repository at this point in the history 
* state: Introduce DependsOn for N-to-N job dependencies

* indexer: Use DependsOn for (most) job dependencies

* fix: check module existence & add in the same txn

This is to prevent a race condition

* fix: account for empty job IDs & wrong ID in cleanup

* Update internal/indexer/module_calls.go

Co-authored-by: Daniel Banck <dbanck@users.noreply.github.com>

Co-authored-by: Daniel Banck <dbanck@users.noreply.github.com>
  • Loading branch information
@radeksimko @dbanck
radeksimko and dbanck authored Aug 4, 2022
1 parent 1996d96 commit 48ab782
Show file tree
Hide file tree
Showing 13 changed files with 564 additions and 234 deletions.
127 changes: 64 additions & 63 deletions internal/indexer/document_change.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,107 +12,108 @@ import (
func (idx *Indexer) DocumentChanged(modHandle document.DirHandle) (job.IDs, error) {
ids := make(job.IDs, 0)

id, err := idx.jobStore.EnqueueJob(job.Job{
parseId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.ParseModuleConfiguration(idx.fs, idx.modStore, modHandle.Path())
},
Type: op.OpTypeParseModuleConfiguration.String(),
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
return idx.decodeModule(ctx, modHandle)
},
})
if err != nil {
return ids, err
}
ids = append(ids, id)
ids = append(ids, parseId)

modIds, err := idx.decodeModule(modHandle, job.IDs{parseId})
if err != nil {
return ids, err
}
ids = append(ids, modIds...)

id, err = idx.jobStore.EnqueueJob(job.Job{
parseVarsId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.ParseVariables(idx.fs, idx.modStore, modHandle.Path())
},
Type: op.OpTypeParseVariables.String(),
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
ids := make(job.IDs, 0)
id, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeVarsReferences(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeVarsReferences.String(),
})
if err != nil {
return ids, err
}
ids = append(ids, id)
return ids, nil
})
if err != nil {
return ids, err
}
ids = append(ids, parseVarsId)

varsRefsId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeVarsReferences(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeVarsReferences.String(),
DependsOn: job.IDs{parseVarsId},
})
if err != nil {
return ids, err
}
ids = append(ids, id)
ids = append(ids, varsRefsId)

return ids, nil
}

func (idx *Indexer) decodeModule(ctx context.Context, modHandle document.DirHandle) (job.IDs, error) {
func (idx *Indexer) decodeModule(modHandle document.DirHandle, dependsOn job.IDs) (job.IDs, error) {
ids := make(job.IDs, 0)

id, err := idx.jobStore.EnqueueJob(job.Job{
metaId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.LoadModuleMetadata(idx.modStore, modHandle.Path())
},
Type: op.OpTypeLoadModuleMetadata.String(),
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
ids := make(job.IDs, 0)
id, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeReferenceTargets(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceTargets.String(),
})
if err != nil {
return ids, err
}
ids = append(ids, id)
Type: op.OpTypeLoadModuleMetadata.String(),
DependsOn: dependsOn,
})
if err != nil {
return ids, err
}
ids = append(ids, metaId)

id, err = idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeReferenceOrigins(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceOrigins.String(),
})
if err != nil {
return ids, err
}
ids = append(ids, id)
refTargetsId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeReferenceTargets(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceTargets.String(),
DependsOn: job.IDs{metaId},
})
if err != nil {
return ids, err
}
ids = append(ids, refTargetsId)

id, err = idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.GetModuleDataFromRegistry(ctx, idx.registryClient,
idx.modStore, idx.registryModStore, modHandle.Path())
},
Priority: job.LowPriority,
Type: op.OpTypeGetModuleDataFromRegistry.String(),
})
if err != nil {
return ids, err
}
refOriginsId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.DecodeReferenceOrigins(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceOrigins.String(),
DependsOn: job.IDs{metaId},
})
if err != nil {
return ids, err
}
ids = append(ids, refOriginsId)

ids = append(ids, id)
return ids, nil
registryId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: modHandle,
Func: func(ctx context.Context) error {
return module.GetModuleDataFromRegistry(ctx, idx.registryClient,
idx.modStore, idx.registryModStore, modHandle.Path())
},
Priority: job.LowPriority,
Type: op.OpTypeGetModuleDataFromRegistry.String(),
})
if err != nil {
return ids, err
}
ids = append(ids, id)

ids = append(ids, registryId)

return ids, nil
}
158 changes: 83 additions & 75 deletions internal/indexer/module_calls.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,104 +11,110 @@ import (
op "github.com/hashicorp/terraform-ls/internal/terraform/module/operation"
)

func (idx *Indexer) decodeInstalledModuleCalls(modHandle document.DirHandle) job.DeferFunc {
return func(ctx context.Context, opErr error) (job.IDs, error) {
jobIds := make(job.IDs, 0)
if opErr != nil {
return jobIds, opErr
}
func (idx *Indexer) decodeInstalledModuleCalls(modHandle document.DirHandle) (job.IDs, error) {
jobIds := make(job.IDs, 0)

moduleCalls, err := idx.modStore.ModuleCalls(modHandle.Path())
if err != nil {
return jobIds, err
}

moduleCalls, err := idx.modStore.ModuleCalls(modHandle.Path())
var errs *multierror.Error

idx.logger.Printf("indexing installed module calls: %d", len(moduleCalls.Installed))
for _, mc := range moduleCalls.Installed {
fi, err := os.Stat(mc.Path)
if err != nil || !fi.IsDir() {
multierror.Append(errs, err)
continue
}
err = idx.modStore.Add(mc.Path)
if err != nil {
return jobIds, err
multierror.Append(errs, err)
continue
}

var errs *multierror.Error
mcHandle := document.DirHandleFromPath(mc.Path)
// copy path for queued jobs below
mcPath := mc.Path

for _, mc := range moduleCalls.Installed {
fi, err := os.Stat(mc.Path)
if err != nil || !fi.IsDir() {
multierror.Append(errs, err)
continue
}
err = idx.modStore.Add(mc.Path)
if err != nil {
multierror.Append(errs, err)
continue
}
refCollectionDeps := make(job.IDs, 0)

mcHandle := document.DirHandleFromPath(mc.Path)
// copy path for queued jobs below
mcPath := mc.Path
parseId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Func: func(ctx context.Context) error {
return module.ParseModuleConfiguration(idx.fs, idx.modStore, mcPath)
},
Type: op.OpTypeParseModuleConfiguration.String(),
})
if err != nil {
multierror.Append(errs, err)
} else {
jobIds = append(jobIds, parseId)
refCollectionDeps = append(refCollectionDeps, parseId)
}

id, err := idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
var metaId job.ID
if parseId != "" {
metaId, err = idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Type: op.OpTypeLoadModuleMetadata.String(),
Func: func(ctx context.Context) error {
return module.ParseModuleConfiguration(idx.fs, idx.modStore, mcPath)
},
Type: op.OpTypeParseModuleConfiguration.String(),
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
ids := make(job.IDs, 0)

id, err := idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Type: op.OpTypeLoadModuleMetadata.String(),
Func: func(ctx context.Context) error {
return module.LoadModuleMetadata(idx.modStore, mcPath)
},
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
return idx.collectReferences(ctx, mcHandle)
},
})
if err != nil {
return ids, err
} else {
ids = append(ids, id)
}

return ids, nil
return module.LoadModuleMetadata(idx.modStore, mcPath)
},
DependsOn: job.IDs{parseId},
})
if err != nil {
multierror.Append(errs, err)
continue
} else {
jobIds = append(jobIds, metaId)
refCollectionDeps = append(refCollectionDeps, metaId)
}
}

if parseId != "" {
ids, err := idx.collectReferences(mcHandle, refCollectionDeps)
if err != nil {
multierror.Append(errs, err)
} else {
jobIds = append(jobIds, ids...)
}
jobIds = append(jobIds, id)
}

varsParseId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Func: func(ctx context.Context) error {
return module.ParseVariables(idx.fs, idx.modStore, mcPath)
},
Type: op.OpTypeParseVariables.String(),
})
if err != nil {
multierror.Append(errs, err)
} else {
jobIds = append(jobIds, varsParseId)
}

id, err = idx.jobStore.EnqueueJob(job.Job{
if varsParseId != "" {
varsRefId, err := idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Func: func(ctx context.Context) error {
return module.ParseVariables(idx.fs, idx.modStore, mcPath)
},
Type: op.OpTypeParseVariables.String(),
Defer: func(ctx context.Context, jobErr error) (job.IDs, error) {
ids := make(job.IDs, 0)
id, err = idx.jobStore.EnqueueJob(job.Job{
Dir: mcHandle,
Func: func(ctx context.Context) error {
return module.DecodeVarsReferences(ctx, idx.modStore, idx.schemaStore, mcPath)
},
Type: op.OpTypeDecodeVarsReferences.String(),
})
if err != nil {
return ids, err
}
ids = append(ids, id)
return ids, err
return module.DecodeVarsReferences(ctx, idx.modStore, idx.schemaStore, mcPath)
},
Type: op.OpTypeDecodeVarsReferences.String(),
DependsOn: job.IDs{varsParseId},
})
if err != nil {
multierror.Append(errs, err)
continue
} else {
jobIds = append(jobIds, varsRefId)
}
jobIds = append(jobIds, id)
}

return jobIds, errs.ErrorOrNil()
}

return jobIds, errs.ErrorOrNil()
}

func (idx *Indexer) collectReferences(ctx context.Context, modHandle document.DirHandle) (job.IDs, error) {
func (idx *Indexer) collectReferences(modHandle document.DirHandle, dependsOn job.IDs) (job.IDs, error) {
ids := make(job.IDs, 0)

var errs *multierror.Error
Expand All @@ -118,7 +124,8 @@ func (idx *Indexer) collectReferences(ctx context.Context, modHandle document.Di
Func: func(ctx context.Context) error {
return module.DecodeReferenceTargets(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceTargets.String(),
Type: op.OpTypeDecodeReferenceTargets.String(),
DependsOn: dependsOn,
})
if err != nil {
errs = multierror.Append(errs, err)
Expand All @@ -131,7 +138,8 @@ func (idx *Indexer) collectReferences(ctx context.Context, modHandle document.Di
Func: func(ctx context.Context) error {
return module.DecodeReferenceOrigins(ctx, idx.modStore, idx.schemaStore, modHandle.Path())
},
Type: op.OpTypeDecodeReferenceOrigins.String(),
Type: op.OpTypeDecodeReferenceOrigins.String(),
DependsOn: dependsOn,
})
if err != nil {
errs = multierror.Append(errs, err)
Expand Down
Loading

0 comments on commit 48ab782

Please sign in to comment.