Cap root module loading via worker pool (#256)

* deps: Add github.com/gammazero/workerpool

* Cap root module loading via worker pool

go.mod

@@ -6,6 +6,7 @@ require (
  github.com/apparentlymart/go-textseg v1.0.0
  github.com/creachadair/jrpc2 v0.10.0
  github.com/fsnotify/fsnotify v1.4.9
+ github.com/gammazero/workerpool v1.0.0
  github.com/google/go-cmp v0.4.1
  github.com/hashicorp/go-multierror v1.1.0
  github.com/hashicorp/go-version v1.2.0

go.sum

@@ -26,6 +26,10 @@ github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/gammazero/deque v0.0.0-20200227231300-1e9af0e52b46 h1:iX4+rD9Fjdx8SkmSO/O5WAIX/j79ll3kuqv5VdYt9J8=
+github.com/gammazero/deque v0.0.0-20200227231300-1e9af0e52b46/go.mod h1:D90+MBHVc9Sk1lJAbEVgws0eYEurY4mv2TDso3Nxh3w=
+github.com/gammazero/workerpool v1.0.0 h1:MfkJc6KL0tAmjrRDS203AZz3F+84Uod9YbL8KjpcQ00=
+github.com/gammazero/workerpool v1.0.0/go.mod h1:/XWO2YAUUpPi3smDlFBl0vpX0JHwUomDM/oRMwRmnSs=
 github.com/go-test/deep v1.0.3 h1:ZrJSEWsXzPOxaZnFteGEfooLba+ju3FYIbOrS+rQd68=
 github.com/go-test/deep v1.0.3/go.mod h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
 github.com/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=

internal/terraform/rootmodule/root_module_manager.go

@@ -6,9 +6,11 @@ import (
  "log"
  "os"
  "path/filepath"
+ "runtime"
  "strings"
  "time"
 
+ "github.com/gammazero/workerpool"
  "github.com/hashicorp/terraform-config-inspect/tfconfig"
  "github.com/hashicorp/terraform-ls/internal/terraform/discovery"
  "github.com/hashicorp/terraform-ls/internal/terraform/exec"
@@ -22,6 +24,7 @@ type rootModuleManager struct {
  filesystem tfconfig.FS
 
  syncLoading bool
+ workerPool *workerpool.WorkerPool
  logger *log.Logger
 
  // terraform discovery
@@ -40,9 +43,14 @@ func NewRootModuleManager(fs tfconfig.FS) RootModuleManager {
 
 func newRootModuleManager(fs tfconfig.FS) *rootModuleManager {
  d := &discovery.Discovery{}
+
+ defaultSize := 3 * runtime.NumCPU()
+ wp := workerpool.New(defaultSize)
+
  rmm := &rootModuleManager{
  rms: make([]*rootModule, 0),
  filesystem: fs,
+ workerPool: wp,
  logger: defaultLogger,
  tfDiscoFunc: d.LookPath,
  tfNewExecutor: exec.NewExecutor,
@@ -51,6 +59,14 @@ func newRootModuleManager(fs tfconfig.FS) *rootModuleManager {
  return rmm
 }
 
+func (rmm *rootModuleManager) WorkerPoolSize() int {
+ return rmm.workerPool.Size()
+}
+
+func (rmm *rootModuleManager) WorkerQueueSize() int {
+ return rmm.workerPool.WaitingQueueSize()
+}
+
 func (rmm *rootModuleManager) defaultRootModuleFactory(ctx context.Context, dir string) (*rootModule, error) {
  rm := newRootModule(rmm.filesystem, dir)
 
@@ -106,10 +122,11 @@ func (rmm *rootModuleManager) AddAndStartLoadingRootModule(ctx context.Context,
  }
 
  rmm.logger.Printf("asynchronously loading root module %s", dir)
- err = rm.StartLoading()
- if err != nil {
- return rm, err
- }
+ rmm.workerPool.Submit(func() {
+ rm := rm
+ err := rm.load(context.Background())
+ rm.setLoadErr(err)
+ })
 
  return rm, nil
 }
@@ -253,6 +270,7 @@ func (rmm *rootModuleManager) CancelLoading() {
  rm.CancelLoading()
  rmm.logger.Printf("loading cancelled for %s", rm.Path())
  }
+ rmm.workerPool.Stop()
 }
 
 // rootModuleDirFromPath strips known lock file paths and filenames

internal/terraform/rootmodule/types.go

@@ -43,6 +43,8 @@ type RootModuleManager interface {
  SetTerraformExecTimeout(timeout time.Duration)
 
  AddAndStartLoadingRootModule(ctx context.Context, dir string) (RootModule, error)
+ WorkerPoolSize() int
+ WorkerQueueSize() int
  ListRootModules() RootModules
  PathsToWatch() []string
  RootModuleByPath(path string) (RootModule, error)

langserver/handlers/service.go

@@ -6,6 +6,7 @@ import (
  "fmt"
  "io/ioutil"
  "log"
+ "time"
 
  "github.com/creachadair/jrpc2"
  "github.com/creachadair/jrpc2/code"
@@ -73,6 +74,14 @@ func (svc *service) Assigner() (jrpc2.Assigner, error) {
  svc.modMgr = svc.newRootModuleManager(fs)
  svc.modMgr.SetLogger(svc.logger)
 
+ svc.logger.Printf("Worker pool size set to %d", svc.modMgr.WorkerPoolSize())
+
+ tr := newTickReporter(5 * time.Second)
+ tr.AddReporter(func() {
+ svc.logger.Printf("Root modules waiting to be loaded: %d", svc.modMgr.WorkerQueueSize())
+ })
+ tr.StartReporting(svc.sessCtx)
+
  svc.walker = svc.newWalker()
 
  // The following is set via CLI flags, hence available in the server context

langserver/handlers/tick_reporter.go

@@ -0,0 +1,40 @@
+package handlers
+
+import (
+ "context"
+ "time"
+)
+
+func newTickReporter(d time.Duration) *tickReporter {
+ return &tickReporter{
+ t: time.NewTicker(d),
+ rfs: make([]reportFunc, 0),
+ }
+}
+
+type reportFunc func()
+
+type tickReporter struct {
+ t *time.Ticker
+ rfs []reportFunc
+}
+
+func (tr *tickReporter) AddReporter(f reportFunc) {
+ tr.rfs = append(tr.rfs, f)
+}
+
+func (tr *tickReporter) StartReporting(ctx context.Context) {
+ go func(ctx context.Context, tr *tickReporter) {
+ for {
+ select {
+ case <-ctx.Done():
+ tr.t.Stop()
+ return
+ case <-tr.t.C:
+ for _, rf := range tr.rfs {
+ rf()
+ }
+ }
+ }
+ }(ctx, tr)
+}