gopls/internal/cache: async pull diagnostics and joined analysis

Make pull diagnostics async, to optimize handling of many simultaneous
requests for the diagnostics of open files. Accordingly, update the pull
diagnostic benchmark to request diagnostics concurrently.

Naively, these concurrent diagnostics would cause gopls to incur major
performance penalty due to duplicate analyses -- as much as 5x total CPU
in the pull diagnostic benchmark. To mitigate this cost, join ongoing
analysis operations using a shared global transient futureCache. This
reduces the benchmark back down close to its previous total CPU, with
reduced latency. Eyeballing the net delta of this CL in the benchmark,
it looks like +20% total CPU, -30% latency.

As a nice side effect, this more than eliminates the need to pre-seed
the file cache in the gopls marker tests. The shared futures provide
more consolidation of work than the pre-seeding, because of variance in
the cache keys of standard library packages, due to different gopls
options.

For golang/go#53275

Change-Id: Ie92bab4c140e3f86852531be8204b6574b254d8e
Reviewed-on: https://go-review.googlesource.com/c/tools/+/622036
LUCI-TryBot-Result: Go LUCI <[email protected]>
Auto-Submit: Robert Findley <[email protected]>
Reviewed-by: Alan Donovan <[email protected]>

Author: findleyr

gopls/internal/cache/analysis.go

@@ -259,7 +259,7 @@ func (s *Snapshot) Analyze(ctx context.Context, pkgs map[PackageID]*metadata.Pac
 			an = &analysisNode{
 				allNodes:     nodes,
 				fset:         fset,
-				fsource:      struct{ file.Source }{s}, // expose only ReadFile
+				fsource:      s, // expose only ReadFile
 				viewType:     s.View().Type(),
 				mp:           mp,
 				analyzers:    facty, // all nodes run at least the facty analyzers
@@ -305,6 +305,8 @@ func (s *Snapshot) Analyze(ctx context.Context, pkgs map[PackageID]*metadata.Pac
 			from.unfinishedSuccs.Add(+1) // incref
 			an.preds = append(an.preds, from)
 		}
+		// Increment unfinishedPreds even for root nodes (from==nil), so that their
+		// Action summaries are never cleared.
 		an.unfinishedPreds.Add(+1)
 		return an, nil
 	}
@@ -689,6 +691,19 @@ type actionSummary struct {
 	Err         string // "" => success
 }
 
+var (
+	// inFlightAnalyses records active analysis operations so that later requests
+	// can be satisfied by joining onto earlier requests that are still active.
+	//
+	// Note that persistent=false, so results are cleared once they are delivered
+	// to awaiting goroutines.
+	inFlightAnalyses = newFutureCache[file.Hash, *analyzeSummary](false)
+
+	// cacheLimit reduces parallelism of filecache updates.
+	// We allow more than typical GOMAXPROCS as it's a mix of CPU and I/O.
+	cacheLimit = make(chan unit, 32)
+)
+
 // runCached applies a list of analyzers (plus any others
 // transitively required by them) to a package.  It succeeds as long
 // as it could produce a types.Package, even if there were direct or
@@ -725,39 +740,41 @@ func (an *analysisNode) runCached(ctx context.Context) (*analyzeSummary, error)
 		return nil, bug.Errorf("internal error reading shared cache: %v", err)
 	} else {
 		// Cache miss: do the work.
-		var err error
-		summary, err = an.run(ctx)
+		cachedSummary, err := inFlightAnalyses.get(ctx, key, func(ctx context.Context) (*analyzeSummary, error) {
+			summary, err := an.run(ctx)
+			if err != nil {
+				return nil, err
+			}
+			if summary == nil { // debugging #66732 (can't happen)
+				bug.Reportf("analyzeNode.run returned nil *analyzeSummary")
+			}
+			go func() {
+				cacheLimit <- unit{}            // acquire token
+				defer func() { <-cacheLimit }() // release token
+
+				data := analyzeSummaryCodec.Encode(summary)
+				if false {
+					log.Printf("Set key=%d value=%d id=%s\n", len(key), len(data), an.mp.ID)
+				}
+				if err := filecache.Set(cacheKind, key, data); err != nil {
+					event.Error(ctx, "internal error updating analysis shared cache", err)
+				}
+			}()
+			return summary, nil
+		})
 		if err != nil {
 			return nil, err
 		}
-		if summary == nil { // debugging #66732 (can't happen)
-			bug.Reportf("analyzeNode.run returned nil *analyzeSummary")
-		}
-
-		an.unfinishedPreds.Add(+1) // incref
-		go func() {
-			defer an.decrefPreds() //decref
 
-			cacheLimit <- unit{}            // acquire token
-			defer func() { <-cacheLimit }() // release token
-
-			data := analyzeSummaryCodec.Encode(summary)
-			if false {
-				log.Printf("Set key=%d value=%d id=%s\n", len(key), len(data), an.mp.ID)
-			}
-			if err := filecache.Set(cacheKind, key, data); err != nil {
-				event.Error(ctx, "internal error updating analysis shared cache", err)
-			}
-		}()
+		// Copy the computed summary. In decrefPreds, we may zero out
+		// summary.actions, but can't mutate a shared result.
+		copy := *cachedSummary
+		summary = &copy
 	}
 
 	return summary, nil
 }
 
-// cacheLimit reduces parallelism of cache updates.
-// We allow more than typical GOMAXPROCS as it's a mix of CPU and I/O.
-var cacheLimit = make(chan unit, 32)
-
 // analysisCacheKey returns a cache key that is a cryptographic digest
 // of the all the values that might affect type checking and analysis:
 // the analyzer names, package metadata, names and contents of

gopls/internal/server/diagnostics.go

@@ -29,6 +29,7 @@ import (
 	"golang.org/x/tools/gopls/internal/work"
 	"golang.org/x/tools/internal/event"
 	"golang.org/x/tools/internal/event/keys"
+	"golang.org/x/tools/internal/jsonrpc2"
 )
 
 // Diagnostic implements the textDocument/diagnostic LSP request, reporting
@@ -50,6 +51,9 @@ func (s *server) Diagnostic(ctx context.Context, params *protocol.DocumentDiagno
 		return nil, err
 	}
 	defer release()
+
+	jsonrpc2.Async(ctx) // allow asynchronous collection of diagnostics
+
 	uri := fh.URI()
 	kind := snapshot.FileKind(fh)
 	var diagnostics []*cache.Diagnostic

gopls/internal/test/integration/bench/diagnostic_test.go

@@ -5,6 +5,7 @@
 package bench
 
 import (
+	"sync"
 	"testing"
 
 	"golang.org/x/tools/gopls/internal/protocol"
@@ -61,17 +62,19 @@ func BenchmarkDiagnosePackageFiles(b *testing.B) {
 
 	for i := 0; i < b.N; i++ {
 		edit()
-		var diags []protocol.Diagnostic
+		var wg sync.WaitGroup
 		for _, file := range files {
-			fileDiags := env.Diagnostics(file)
-			for _, d := range fileDiags {
-				if d.Severity == protocol.SeverityError {
-					diags = append(diags, d)
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				fileDiags := env.Diagnostics(file)
+				for _, d := range fileDiags {
+					if d.Severity == protocol.SeverityError {
+						b.Errorf("unexpected error diagnostic: %s", d.Message)
+					}
 				}
-			}
-		}
-		if len(diags) != 0 {
-			b.Fatalf("got %d error diagnostics, want 0\ndiagnostics:\n%v", len(diags), diags)
+			}()
 		}
+		wg.Wait()
 	}
 }

gopls/internal/test/marker/marker_test.go

@@ -27,7 +27,6 @@ import (
 	"sort"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
@@ -108,10 +107,6 @@ func Test(t *testing.T) {
 	// Opt: use a shared cache.
 	cache := cache.New(nil)
 
-	// Opt: seed the cache and file cache by type-checking and analyzing common
-	// standard library packages.
-	seedCache(t, cache)
-
 	for _, test := range tests {
 		test := test
 		t.Run(test.name, func(t *testing.T) {
@@ -290,58 +285,6 @@ func Test(t *testing.T) {
 	}
 }
 
-// seedCache populates the file cache by type checking and analyzing standard
-// library packages that are reachable from tests.
-//
-// Most tests are themselves small codebases, and yet may reference large
-// amounts of standard library code. Since tests are heavily parallelized, they
-// naively end up type checking and analyzing many of the same standard library
-// packages. By seeding the cache, we ensure cache hits for these standard
-// library packages, significantly reducing the amount of work done by each
-// test.
-//
-// The following command was used to determine the set of packages to import
-// below:
-//
-//	rm -rf ~/.cache/gopls && \
-//	 go test -count=1 ./internal/test/marker -cpuprofile=prof -v
-//
-// Look through the individual test timings to see which tests are slow, then
-// look through the imports of slow tests to see which standard library
-// packages are imported. Choose high level packages such as go/types that
-// import others such as fmt or go/ast. After doing so, re-run the command and
-// verify that the total samples in the collected profile decreased.
-func seedCache(t *testing.T, cache *cache.Cache) {
-	start := time.Now()
-
-	// The the doc string for details on how this seed was produced.
-	seed := `package p
-import (
-	_ "net/http"
-	_ "sort"
-	_ "go/types"
-	_ "testing"
-)
-`
-
-	// Create a test environment for the seed file.
-	env := newEnv(t, cache, map[string][]byte{"p.go": []byte(seed)}, nil, nil, fake.EditorConfig{})
-	// See other TODO: this cleanup logic is too messy.
-	defer env.Editor.Shutdown(context.Background()) // ignore error
-	defer env.Sandbox.Close()                       // ignore error
-	env.Awaiter.Await(context.Background(), integration.InitialWorkspaceLoad)
-
-	// Opening the file is necessary to trigger analysis.
-	env.OpenFile("p.go")
-
-	// As a checksum, verify that the file has no errors after analysis.
-	// This isn't strictly necessary, but helps avoid incorrect seeding due to
-	// typos.
-	env.AfterChange(integration.NoDiagnostics())
-
-	t.Logf("warming the cache took %s", time.Since(start))
-}
-
 // A marker holds state for the execution of a single @marker
 // annotation in the source.
 type marker struct {