Merge #150867 #150963

150867: storage/disk: deflake TestMonitorManager_monitorDisks r=RaduBerinde a=jbowens

**storage/disk: deflake TestMonitorManager_monitorDisks** 

Use `require.Eventually` to deflake this test.

Epic: none
Fixes: #148556.
Release note: none

**storage/disk: use context cancellation to stop monitoring** 

Address a pending TODO to use context cancellation, rather than an explicit
stop channel, to stop monitoring of a disk.

Epic: none
Release note: none

150963: roachtest: deflake activerecord r=fqazi a=fqazi

Fixes: #150649
Fixes: #150570

Release note: None

Co-authored-by: Jackson Owens <[email protected]>
Co-authored-by: Faizan Qazi <[email protected]>

Author: 3 people

pkg/cmd/roachtest/tests/activerecord_blocklist.go

@@ -24,13 +24,17 @@ var activeRecordBlocklist = blocklist{
 var activeRecordIgnoreList = blocklist{
 	`ActiveRecord::AdapterTestWithoutTransaction#test_create_with_query_cache`:                                                                                 "affected by autocommit_before_ddl",
 	`ActiveRecord::CockroachDBStructureDumpTest#test_structure_dump`:                                                                                           "flaky",
+	`ActiveRecord::ConnectionAdapters::ConnectionPoolThreadTest#test_checkout_fairness`:                                                                        "flaky",
 	`ActiveRecord::ConnectionAdapters::ConnectionPoolThreadTest#test_checkout_fairness_by_group`:                                                               "flaky",
 	`ActiveRecord::ConnectionAdapters::PostgreSQLAdapterTest#test_translate_no_connection_exception_to_not_established`:                                        "pg_terminate_backend not implemented",
 	`ActiveRecord::Encryption::EncryptableRecordTest#test_by_default,_it's_case_sensitive`:                                                                     "flaky",
 	`ActiveRecord::Encryption::EncryptableRecordTest#test_forced_encoding_for_deterministic_attributes_will_replace_invalid_characters`:                        "flaky",
 	`AssociationCallbacksTest#test_has_many_callbacks_for_destroy_on_parent`:                                                                                   "flaky",
 	`BasicsTest#test_default_values_are_deeply_dupped`:                                                                                                         "flaky",
 	`CascadedEagerLoadingTest#test_eager_association_loading_with_cascaded_three_levels_by_ping_pong`:                                                          "flaky",
+	`CockroachDB::AdapterForeignKeyTest#test_foreign_key_violations_are_translated_to_specific_exception_with_validate_false`:                                  "flaky",
+	`CockroachDB::AdapterForeignKeyTest#test_foreign_key_violations_on_delete_are_translated_to_specific_exception`:                                            "flaky",
+	`CockroachDB::AdapterForeignKeyTest#test_foreign_key_violations_on_insert_are_translated_to_specific_exception`:                                            "flaky",
 	`CockroachDB::FixturesTest#test_create_fixtures`:                                                                                                           "flaky",
 	`FixtureWithSetModelClassPrevailsOverNamingConventionTest#test_model_class_in_fixture_file_is_respected`:                                                   "flaky",
 	`InheritanceTest#test_eager_load_belongs_to_primary_key_quoting`:                                                                                           "flaky",

pkg/storage/disk/monitor.go

@@ -49,8 +49,8 @@ type MonitorManager struct {
 
 	mu struct {
 		syncutil.Mutex
-		stop  chan struct{}
-		disks []*monitoredDisk
+		cancel context.CancelFunc
+		disks  []*monitoredDisk
 	}
 }
 
@@ -90,13 +90,14 @@ func (m *MonitorManager) Monitor(path string) (*Monitor, error) {
 
 		// The design maintains the invariant that the disk stat polling loop
 		// is always running unless there are no disks being monitored.
-		if m.mu.stop == nil {
+		if m.mu.cancel == nil {
 			collector, err := newStatsCollector(m.fs)
 			if err != nil {
 				return nil, err
 			}
-			m.mu.stop = make(chan struct{})
-			go m.monitorDisks(collector, m.mu.stop)
+			ctx, cancel := context.WithCancel(context.Background())
+			m.mu.cancel = cancel
+			go m.monitorDisks(ctx, collector)
 		}
 	}
 	disk.refCount++
@@ -105,7 +106,7 @@ func (m *MonitorManager) Monitor(path string) (*Monitor, error) {
 }
 
 func (m *MonitorManager) unrefDisk(disk *monitoredDisk) {
-	var stop chan struct{}
+	var cancel context.CancelFunc
 	func() {
 		m.mu.Lock()
 		defer m.mu.Unlock()
@@ -123,14 +124,14 @@ func (m *MonitorManager) unrefDisk(disk *monitoredDisk) {
 			// If the MonitorManager has no disks left to monitor, the disk stat polling loop can
 			// be stopped.
 			if len(m.mu.disks) == 0 {
-				stop = m.mu.stop
-				m.mu.stop = nil
+				cancel = m.mu.cancel
+				m.mu.cancel = nil
 			}
 		}
 	}()
 
-	if stop != nil {
-		stop <- struct{}{}
+	if cancel != nil {
+		cancel()
 	}
 }
 
@@ -139,22 +140,15 @@ type statsCollector interface {
 }
 
 // monitorDisks runs a loop collecting disk stats for all monitored disks.
-//
-// NB: A stop channel must be passed down to ensure that the function terminates during the
-// race where the MonitorManager creates a new stop channel after unrefDisk sends a message
-// across the old stop channel.
-func (m *MonitorManager) monitorDisks(collector statsCollector, stop chan struct{}) {
-	// TODO(jackson): Should we propagate a context here to replace the stop
-	// channel?
-	ctx := context.TODO()
+// monitorDisks returns when the context is done.
+func (m *MonitorManager) monitorDisks(ctx context.Context, collector statsCollector) {
 	ticker := time.NewTicker(DefaultDiskStatsPollingInterval)
 	defer ticker.Stop()
 
 	every := log.Every(5 * time.Minute)
 	for {
 		select {
-		case <-stop:
-			close(stop)
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
 			m.mu.Lock()

pkg/storage/disk/monitor_test.go

@@ -6,6 +6,8 @@
 package disk
 
 import (
+	"context"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -17,13 +19,13 @@ import (
 )
 
 type spyCollector struct {
-	collectCallCount int
+	collectCallCount atomic.Int32
 }
 
 func (s *spyCollector) collect(
 	disks []*monitoredDisk, now time.Time,
 ) (countCollected int, err error) {
-	s.collectCallCount++
+	s.collectCallCount.Add(1)
 	return len(disks), nil
 }
 
@@ -42,12 +44,13 @@ func TestMonitorManager_monitorDisks(t *testing.T) {
 	manager.mu.disks = []*monitoredDisk{testDisk}
 
 	testCollector := &spyCollector{}
-	stop := make(chan struct{})
-	go manager.monitorDisks(testCollector, stop)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go manager.monitorDisks(ctx, testCollector)
 
-	time.Sleep(2 * DefaultDiskStatsPollingInterval)
-	stop <- struct{}{}
-	require.Greater(t, testCollector.collectCallCount, 0)
+	require.Eventually(t, func() bool {
+		return testCollector.collectCallCount.Load() > 0
+	}, 100*DefaultDiskStatsPollingInterval, DefaultDiskStatsPollingInterval)
 }
 
 func TestMonitor_StatsWindow(t *testing.T) {
@@ -134,8 +137,8 @@ func TestMonitor_Close(t *testing.T) {
 		},
 		refCount: 2,
 	}
-	stop := make(chan struct{})
-	manager.mu.stop = stop
+	ctx, cancel := context.WithCancel(context.Background())
+	manager.mu.cancel = cancel
 	manager.mu.disks = []*monitoredDisk{testDisk}
 	monitor1 := Monitor{monitoredDisk: testDisk}
 	monitor2 := Monitor{monitoredDisk: testDisk}
@@ -150,7 +153,7 @@ func TestMonitor_Close(t *testing.T) {
 	go monitor2.Close()
 	// If there are no monitors, stop the stat polling loop.
 	select {
-	case <-stop:
+	case <-ctx.Done():
 	case <-time.After(time.Second):
 		t.Fatal("Failed to receive stop signal")
 	}