database/sql: prioritize closingMutex.Lock over RLock when no rlocks

We use a custom RWMutex (closingMutex) which permits recursive RLocks
to avoid deadlocks (see #78304). When a close operation is made
concurrently with a single read operation in a loop--for example,
when closing a Rows which is actively being scanned over--this
can cause the read to starve the close.

Prioritize Lock over RLock when there are no read-locks held:

- If a closingMutex is read-locked, a recursive RLock will succeed
  and can starve out concurrent closes.

- If the last read-lock on a closingMutex is dropped,
  a blocked Lock will take precedence over a new RLock.

This avoids the recursive rlock deadlock from #78304, while
allowing a close to consistently interrupt a read loop.

Change-Id: Ibdb7739b64fa76f90c133981ce72e1986a6a6964
Reviewed-on: https://go-review.googlesource.com/c/go/+/771580
Reviewed-by: Michael Pratt <mpratt@google.com>
LUCI-TryBot-Result: golang-scoped@luci-project-accounts.iam.gserviceaccount.com <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>
Auto-Submit: Damien Neil <dneil@google.com>

src/database/sql/closemu.go

@@ -92,7 +92,12 @@ func (m *closingMutex) Unlock() {
 func (m *closingMutex) TryRLock() bool {
 	for {
 		x := m.state.Load()
-		if x < 0 {
+		// Fail if the mutex is write locked (x < 0)
+		// or unlocked with a writer waiting (x == 1).
+		//
+		// If the mutex is read-locked (x > 1), try to add a reader
+		// even if this starves out a waiting writer.
+		if x < 0 || x == 1 {
 			return false
 		}
 		if m.state.CompareAndSwap(x, x+2) {

src/database/sql/closemu_test.go

@@ -87,7 +87,45 @@ func TestClosingMutex(t *testing.T) {
 			t.Fatalf("m.Lock(): still blocking after RUnlock")
 		}
 		m.Unlock()
+	})
+}
 
+func TestClosingMutexLockStarvation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		// Run this test for a few iterations, to avoid racy successes.
+		for range 100 {
+			var m closingMutex
+
+			// With the mutex RLocked, try to Lock it. Lock blocks.
+			m.RLock()
+			locked := false
+			go func() {
+				m.Lock()
+				locked = true
+				m.Unlock()
+			}()
+			synctest.Wait()
+			if locked {
+				t.Errorf("lock acquired while mutex is rlocked")
+			}
+
+			// Add and drop another RLock. Lock is still blocking.
+			m.RLock()
+			m.RUnlock()
+			if locked {
+				t.Errorf("lock acquired while mutex is double-rlocked")
+			}
+
+			// Drop and reacquire the RLock.
+			// The blocking Lock should always acquire the mutex
+			// before the RLock succeeds.
+			m.RUnlock()
+			m.RLock()
+			if !locked {
+				t.Errorf("lock not acquired when rlock dropped")
+			}
+			m.RUnlock()
+		}
 	})
 }