lint: make a smarter mutex lock linter

This commit udpates the lock linter to work by keeping track of locks and if we find any if conditions, loops, or function calls before finding a matching unlock it will report. It will also report if the matching unlock is >5 lines away from the lock. It will ignore cases where a `nolint:deferunlock` is present. Release note: None
cockroachdb · Sep 5, 2023 · d80a9f5 · d80a9f5
1 parent 8368312
commit d80a9f5
Show file tree

Hide file tree

Showing 285 changed files with 573 additions and 999 deletions.
diff --git a/build/bazelutil/nogo_config.json b/build/bazelutil/nogo_config.json
diff --git a/pkg/acceptance/localcluster/cluster.go b/pkg/acceptance/localcluster/cluster.go
@@ -475,7 +475,6 @@ func (n *Node) Alive() bool {
 func (n *Node) StatusClient(ctx context.Context) serverpb.StatusClient {
 	n.Lock()
 	existingClient := n.statusClient
-	// nolint:deferunlock
 	n.Unlock()
 
 	if existingClient != nil {
@@ -589,7 +588,6 @@ func (n *Node) startAsyncInnerLocked(ctx context.Context, joins ...string) error
 		_ = errors.As(waitErr, &execErr)
 		n.Lock()
 		n.setNotRunningLocked(execErr)
-		// nolint:deferunlock
 		n.Unlock()
 	}(n.cmd)
 
@@ -697,7 +695,6 @@ func (n *Node) waitUntilLive(dur time.Duration) error {
 		if n.cmd != nil {
 			pid = n.cmd.Process.Pid
 		}
-		// nolint:deferunlock
 		n.Unlock()
 		if pid == 0 {
 			log.Info(ctx, "process already quit")
@@ -720,14 +717,12 @@ func (n *Node) waitUntilLive(dur time.Duration) error {
 		if n.Cfg.RPCPort == 0 {
 			n.Lock()
 			n.rpcPort = pgURL.Port()
-			// nolint:deferunlock
 			n.Unlock()
 		}
 
 		pgURL.Path = n.Cfg.DB
 		n.Lock()
 		n.pgURL = pgURL.String()
-		// nolint:deferunlock
 		n.Unlock()
 
 		var uiURL *url.URL
@@ -744,7 +739,6 @@ func (n *Node) waitUntilLive(dur time.Duration) error {
 		// http port is required but isn't initialized yet.
 		n.Lock()
 		n.db = makeDB(n.pgURL, n.Cfg.NumWorkers, n.Cfg.DB)
-		// nolint:deferunlock
 		n.Unlock()
 
 		{
@@ -775,7 +769,6 @@ func (n *Node) Kill() {
 	for ok := false; !ok; {
 		n.Lock()
 		ok = n.cmd == nil
-		// nolint:deferunlock
 		n.Unlock()
 	}
 }
@@ -811,7 +804,6 @@ func (n *Node) Signal(s os.Signal) {
 func (n *Node) Wait() *exec.ExitError {
 	n.Lock()
 	ch := n.notRunning
-	// nolint:deferunlock
 	n.Unlock()
 	if ch == nil {
 		log.Warning(context.Background(), "(*Node).Wait called when node was not running")

diff --git a/pkg/ccl/auditloggingccl/audit_log_config.go b/pkg/ccl/auditloggingccl/audit_log_config.go
@@ -85,7 +85,6 @@ func UpdateAuditConfigOnChange(
 	}
 	acl.Lock()
 	acl.Config = config
-	// nolint:deferunlock
 	acl.Unlock()
 }
 

diff --git a/pkg/ccl/backupccl/backup_test.go b/pkg/ccl/backupccl/backup_test.go
@@ -2593,11 +2593,9 @@ func TestBackupRestoreDuringUserDefinedTypeChange(t *testing.T) {
 										if numTypeChangesStarted == len(tc.queries) {
 											close(typeChangesStarted)
 										}
-										// nolint:deferunlock
 										mu.Unlock()
 										<-waitForBackup
 									} else {
-										// nolint:deferunlock
 										mu.Unlock()
 									}
 									return nil
@@ -2626,7 +2624,6 @@ CREATE TYPE d.greeting AS ENUM ('hello', 'howdy', 'hi');
 						if numTypeChangesFinished == totalQueries {
 							close(typeChangesFinished)
 						}
-						// nolint:deferunlock
 						mu.Unlock()
 					}(query, len(tc.queries))
 				}
@@ -6208,7 +6205,6 @@ func TestRestoreErrorPropagates(t *testing.T) {
 
 	jobsTableKey.Lock()
 	jobsTableKey.key = tc.ApplicationLayer(0).Codec().TablePrefix(uint32(systemschema.JobsTable.GetID()))
-	// nolint:deferunlock
 	jobsTableKey.Unlock()
 
 	runner.Exec(t, `SET CLUSTER SETTING jobs.metrics.interval.poll = '30s'`)
@@ -6391,7 +6387,6 @@ INSERT INTO foo.bar VALUES (110), (210), (310), (410), (510)`)
 	startingSpan := mkSpan(id1, "/Tenant/10/Table/:id/1", "/Tenant/10/Table/:id/2")
 	mu.Lock()
 	require.Equal(t, []string{startingSpan.String()}, mu.exportRequestSpans)
-	// nolint:deferunlock
 	mu.Unlock()
 	resetStateVars()
 
@@ -6402,7 +6397,6 @@ INSERT INTO foo.bar VALUES (110), (210), (310), (410), (510)`)
 	resumeSpan := mkSpan(id1, "/Tenant/10/Table/:id/1/510/0", "/Tenant/10/Table/:id/2")
 	mu.Lock()
 	require.Equal(t, []string{startingSpan.String(), resumeSpan.String()}, mu.exportRequestSpans)
-	// nolint:deferunlock
 	mu.Unlock()
 	resetStateVars()
 
@@ -6421,7 +6415,6 @@ INSERT INTO foo.bar VALUES (110), (210), (310), (410), (510)`)
 	}
 	mu.Lock()
 	require.Equal(t, expected, mu.exportRequestSpans)
-	// nolint:deferunlock
 	mu.Unlock()
 	resetStateVars()
 
@@ -6455,7 +6448,6 @@ INSERT INTO baz.bar VALUES (110, 'a'), (210, 'b'), (310, 'c'), (410, 'd'), (510,
 	}
 	mu.Lock()
 	require.Equal(t, expected, mu.exportRequestSpans)
-	// nolint:deferunlock
 	mu.Unlock()
 	resetStateVars()
 

diff --git a/pkg/ccl/backupccl/backupinfo/manifest_handling.go b/pkg/ccl/backupccl/backupinfo/manifest_handling.go
@@ -1541,7 +1541,6 @@ func GetBackupManifests(
 				manifests[i] = desc
 			}
 			subMem.Shrink(ctx, size)
-			// nolint:deferunlock
 			memMu.Unlock()
 
 			return err

diff --git a/pkg/ccl/backupccl/restore_job.go b/pkg/ccl/backupccl/restore_job.go
@@ -318,7 +318,6 @@ func restore(
 		introducedSpanFrontier,
 		targetRestoreSpanSize.Get(&execCtx.ExecCfg().Settings.SV),
 		progressTracker.useFrontier)
-	// nolint:deferunlock
 	progressTracker.mu.Unlock()
 	if err != nil {
 		return roachpb.RowCount{}, err
@@ -2154,7 +2153,6 @@ func insertStats(
 					mu.completedBatches++
 					remainingBatches := totalNumBatches - mu.completedBatches
 					completedBatches := mu.completedBatches
-					// nolint:deferunlock
 					mu.Unlock()
 					if insertStatsProgress.ShouldLog() {
 						logStatsProgress(remainingBatches, completedBatches)

diff --git a/pkg/ccl/backupccl/tenant_backup_nemesis_test.go b/pkg/ccl/backupccl/tenant_backup_nemesis_test.go
@@ -428,15 +428,13 @@ func (r *randomBackupNemesis) Stop() {
 func (r *randomBackupNemesis) TablesToCheck() []string {
 	r.mu.Lock()
 	ret := append([]string(nil), r.mu.tablesToCheck...)
-	// nolint:deferunlock
 	r.mu.Unlock()
 	return ret
 }
 
 func (r *randomBackupNemesis) addTable(name string) {
 	r.mu.Lock()
 	r.mu.tablesToCheck = append(r.mu.tablesToCheck, name)
-	// nolint:deferunlock
 	r.mu.Unlock()
 }
 

diff --git a/pkg/ccl/changefeedccl/cdctest/schema_registry.go b/pkg/ccl/changefeedccl/cdctest/schema_registry.go
@@ -207,7 +207,6 @@ func (r *SchemaRegistry) EncodedAvroToNative(b []byte) (interface{}, error) {
 
 	r.mu.Lock()
 	jsonSchema := r.mu.schemas[id]
-	// nolint:deferunlock
 	r.mu.Unlock()
 	codec, err := goavro.NewCodec(jsonSchema)
 	if err != nil {

diff --git a/pkg/ccl/changefeedccl/changefeed_processors.go b/pkg/ccl/changefeedccl/changefeed_processors.go
@@ -1187,7 +1187,6 @@ func (cf *changeFrontier) Start(ctx context.Context) {
 	cf.metricsID = cf.metrics.mu.id
 	cf.metrics.mu.id++
 	sli.RunningCount.Inc(1)
-	// nolint:deferunlock
 	cf.metrics.mu.Unlock()
 
 	cf.sliMetricsID = cf.sliMetrics.claimId()
@@ -1230,7 +1229,6 @@ func (cf *changeFrontier) closeMetrics() {
 	}
 	delete(cf.metrics.mu.resolved, cf.metricsID)
 	cf.metricsID = -1
-	// nolint:deferunlock
 	cf.metrics.mu.Unlock()
 
 	cf.sliMetrics.closeId(cf.sliMetricsID)
@@ -1383,7 +1381,6 @@ func (cf *changeFrontier) forwardFrontier(resolved jobspb.ResolvedSpan) error {
 		if cf.metricsID != -1 {
 			cf.metrics.mu.resolved[cf.metricsID] = newResolved
 		}
-		// nolint:deferunlock
 		cf.metrics.mu.Unlock()
 
 		return cf.maybeEmitResolved(newResolved)

diff --git a/pkg/ccl/changefeedccl/event_processing.go b/pkg/ccl/changefeedccl/event_processing.go
@@ -646,15 +646,13 @@ func (c *parallelEventConsumer) incInFlight() {
 	c.mu.Lock()
 	c.mu.inFlight++
 	c.metrics.ParallelConsumerInFlightEvents.Update(int64(c.mu.inFlight))
-	// nolint:deferunlock
 	c.mu.Unlock()
 }
 
 func (c *parallelEventConsumer) decInFlight() {
 	c.mu.Lock()
 	c.mu.inFlight--
 	notifyFlush := c.mu.waiting && c.mu.inFlight == 0
-	// nolint:deferunlock
 	c.mu.Unlock()
 
 	// If someone is waiting on a flush, signal to them
@@ -708,7 +706,6 @@ func (c *parallelEventConsumer) Flush(ctx context.Context) error {
 		c.mu.Lock()
 		c.mu.waiting = false
 		c.mu.flushFrontier = c.spanFrontier.Frontier()
-		// nolint:deferunlock
 		c.mu.Unlock()
 		return nil
 	}

diff --git a/pkg/ccl/changefeedccl/kvevent/blocking_buffer.go b/pkg/ccl/changefeedccl/kvevent/blocking_buffer.go
@@ -150,7 +150,6 @@ func (b *blockingBuffer) pop() (e Event, ok bool, err error) {
 func (b *blockingBuffer) notifyOutOfQuota(canFlush bool) {
 	b.mu.Lock()
 	b.mu.canFlush = canFlush
-	// nolint:deferunlock
 	b.mu.Unlock()
 
 	if canFlush {
@@ -166,7 +165,6 @@ func (b *blockingBuffer) notifyOutOfQuota(canFlush bool) {
 func (b *blockingBuffer) producerBlocked() {
 	b.mu.Lock()
 	b.mu.numBlocked++
-	// nolint:deferunlock
 	b.mu.Unlock()
 }
 
@@ -185,7 +183,6 @@ func (b *blockingBuffer) quotaAcquiredAfterWait() {
 		// Clear out canFlush since we know that producers no longer blocked.
 		b.mu.canFlush = false
 	}
-	// nolint:deferunlock
 	b.mu.Unlock()
 }
 

diff --git a/pkg/ccl/changefeedccl/metrics.go b/pkg/ccl/changefeedccl/metrics.go
@@ -848,7 +848,6 @@ func MakeMetrics(histogramWindow time.Duration) metric.Struct {
 				maxBehind = behind
 			}
 		}
-		// nolint:deferunlock
 		m.mu.Unlock()
 		return maxBehind.Nanoseconds()
 	})

diff --git a/pkg/ccl/changefeedccl/schema_registry.go b/pkg/ccl/changefeedccl/schema_registry.go
@@ -148,7 +148,6 @@ func newConfluentSchemaRegistry(
 		}
 		schemaRegistrySingletons.cachePerEndpoint[baseURL] = src
 	}
-	// nolint:deferunlock
 	schemaRegistrySingletons.mu.Unlock()
 
 	s, err := getAndDeleteParams(u)

diff --git a/pkg/ccl/changefeedccl/schemafeed/schema_feed.go b/pkg/ccl/changefeedccl/schemafeed/schema_feed.go
@@ -280,7 +280,6 @@ func (tf *schemaFeed) Run(ctx context.Context) error {
 func (tf *schemaFeed) primeInitialTableDescs(ctx context.Context) error {
 	tf.mu.Lock()
 	initialTableDescTs := tf.mu.highWater
-	// nolint:deferunlock
 	tf.mu.Unlock()
 	var initialDescs []catalog.Descriptor
 
@@ -313,7 +312,6 @@ func (tf *schemaFeed) primeInitialTableDescs(ctx context.Context) error {
 		tbl := desc.(catalog.TableDescriptor)
 		tf.mu.typeDeps.ingestTable(tbl)
 	}
-	// nolint:deferunlock
 	tf.mu.Unlock()
 
 	return tf.ingestDescriptors(ctx, hlc.Timestamp{}, initialTableDescTs, initialDescs, tf.validateDescriptor)
@@ -512,7 +510,6 @@ func (tf *schemaFeed) pauseOrResumePolling(
 func (tf *schemaFeed) highWater() hlc.Timestamp {
 	tf.mu.Lock()
 	highWater := tf.mu.highWater
-	// nolint:deferunlock
 	tf.mu.Unlock()
 	return highWater
 }

diff --git a/pkg/ccl/changefeedccl/sink_kafka.go b/pkg/ccl/changefeedccl/sink_kafka.go
@@ -52,7 +52,6 @@ func (l *maybeLocker) Lock() {
 }
 func (l *maybeLocker) Unlock() {
 	if l.locked {
-		// nolint:deferunlock
 		l.wrapped.Unlock()
 		l.locked = false
 	}
@@ -422,7 +421,6 @@ func (s *kafkaSink) Flush(ctx context.Context) error {
 	if !immediateFlush {
 		s.mu.flushCh = flushCh
 	}
-	// nolint:deferunlock
 	s.mu.Unlock()
 
 	if immediateFlush {
@@ -439,7 +437,6 @@ func (s *kafkaSink) Flush(ctx context.Context) error {
 		s.mu.Lock()
 		flushErr := s.mu.flushErr
 		s.mu.flushErr = nil
-		// nolint:deferunlock
 		s.mu.Unlock()
 		return flushErr
 	}

diff --git a/pkg/ccl/changefeedccl/sink_pubsub_v2.go b/pkg/ccl/changefeedccl/sink_pubsub_v2.go
@@ -142,11 +142,9 @@ func (sc *pubsubSinkClient) maybeCreateTopic(topic string) error {
 	sc.mu.RLock()
 	_, ok := sc.mu.topicCache[topic]
 	if ok {
-		// nolint:deferunlock
 		sc.mu.RUnlock()
 		return nil
 	}
-	// nolint:deferunlock
 	sc.mu.RUnlock()
 	sc.mu.Lock()
 	defer sc.mu.Unlock()

diff --git a/pkg/ccl/changefeedccl/sink_test.go b/pkg/ccl/changefeedccl/sink_test.go
@@ -177,7 +177,6 @@ func (p *asyncProducerMock) consume() (cleanup func()) {
 			case m := <-p.inputCh:
 				p.mu.Lock()
 				p.mu.outstanding = append(p.mu.outstanding, m)
-				// nolint:deferunlock
 				p.mu.Unlock()
 			}
 		}
@@ -196,7 +195,6 @@ func (p *asyncProducerMock) acknowledge(n int, ch chan *sarama.ProducerMessage)
 		p.mu.Lock()
 		outstanding = append(outstanding, p.mu.outstanding...)
 		p.mu.outstanding = p.mu.outstanding[:0]
-		// nolint:deferunlock
 		p.mu.Unlock()
 
 		for _, m := range outstanding {

diff --git a/pkg/ccl/multiregionccl/datadriven_test.go b/pkg/ccl/multiregionccl/datadriven_test.go
@@ -249,12 +249,10 @@ SET CLUSTER SETTING kv.allocator.min_lease_transfer_interval = 5m
 					// Setup tracing for the input.
 					mu.Lock()
 					traceStmt = d.Input
-					// nolint:deferunlock
 					mu.Unlock()
 					defer func() {
 						mu.Lock()
 						traceStmt = ""
-						// nolint:deferunlock
 						mu.Unlock()
 					}()
 

diff --git a/pkg/ccl/multiregionccl/region_test.go b/pkg/ccl/multiregionccl/region_test.go
@@ -199,13 +199,11 @@ func TestConcurrentAddDropRegions(t *testing.T) {
 						if firstOp {
 							firstOp = false
 							close(firstOpStarted)
-							// nolint:deferunlock
 							mu.Unlock()
 							// Don't promote any members before the second operation reaches
 							// the schema changer as well.
 							<-secondOpFinished
 						} else {
-							// nolint:deferunlock
 							mu.Unlock()
 						}
 						return nil

diff --git a/pkg/ccl/multitenantccl/tenantcostclient/tenant_side.go b/pkg/ccl/multitenantccl/tenantcostclient/tenant_side.go
@@ -441,7 +441,6 @@ func (c *tenantSideCostController) onTick(ctx context.Context, newTime time.Time
 	c.mu.consumption.PGWireEgressBytes += deltaPGWireEgressBytes
 	c.mu.consumption.RU += float64(ru)
 	newConsumption := c.mu.consumption
-	// nolint:deferunlock
 	c.mu.Unlock()
 
 	// Update the average RUs consumed per second, based on the latest stats.
@@ -880,7 +879,6 @@ func (c *tenantSideCostController) onExternalIO(
 	if c.shouldAccountForExternalIORUs() {
 		c.mu.consumption.RU += float64(totalRU)
 	}
-	// nolint:deferunlock
 	c.mu.Unlock()
 
 	return nil