Reader.curChunk can strand a waiter goroutine: Cond wakeups are sent without rmux held, and Close never broadcasts

オープン
  • 何か問題がありますか？
#59 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
1〜2日
初心者へのやさしさ
56/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
go
領域
backend

調査の方向性

Start with the waiter loop and condition broadcasts in b2/reader.go, especially Reader.curChunk, setErr, and Close; the issue includes a regression test in b2/reader_wakeup_test.go. Read the retry block alongside related issue #57, then run the race detector command shown in the issue. Done means closed or failed readers leave no parked waiter goroutines and the existing test suites pass.

索引モデルが issue の本文から書いたものです。

説明

bug

Summary

(*Reader).curChunk starts a goroutine that waits on r.rcond (a sync.Cond) until the next chunk is ready, an error is set, or the context is done. Two things break that wait:

  1. Lost wakeups. The wait predicate reads r.err and r.ctx, but those are changed without holding rmux (the Cond's lock), and the fetch workers call r.rcond.Broadcast() without holding it either. A Broadcast that lands between the waiter's predicate check and its registration in Wait is lost.
  2. No wakeup on Close. Reader.Close cancels the context but never broadcasts, and idle fetch workers exit on <-r.ctx.Done() without broadcasting. So once a wakeup is lost, nothing else ever wakes the waiter.

The goroutine stays blocked in r.rcond.Wait() for the life of the process. The caller is not stuck — curChunk's outer select returns on <-r.ctx.Done() — which is why this goes unnoticed.

Affected: v0.8.0 (latest release) and current master — the relevant files are identical in both. Lower severity than #56-#58: it is timing-dependent.

Where it is (b2/reader.go)

  • The waiter loop: for r.chunks[r.chrid] == nil && r.getErr() == nil && r.ctx.Err() == nil { r.rcond.Wait() } (reader.go:192-196).
  • setErr sets r.err and calls r.cancel() under emux, not rmux (reader.go:84-90).
  • Every r.rcond.Broadcast() is issued without rmux held (reader.go:143, 148, 171, 179, 185).
  • Close (reader.go:78-82) cancels and removes the reader; no Broadcast.
  • Idle workers return on <-r.ctx.Done() (reader.go:117) without broadcasting.

Reproduction

This is a race, so the test is a stress loop: 20,000 readers whose downloads fail after a small varying delay, then a count of curChunk goroutines still parked. It reproduces reliably only under the race detector, which slows scheduling enough to hit the window.

Regression test (b2/reader_wakeup_test.go) 
package b2

import (
	"context"
	"errors"
	"runtime"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

// wakeRoot fails every download after a small, varying delay so the fetch
// worker's setErr+Broadcast races the curChunk waiter's predicate check.
type wakeRoot struct {
	*testRoot
	calls int32
}

func (r *wakeRoot) createBucket(ctx context.Context, name, btype string, info map[string]string, rules []LifecycleRule, sse *ServerSideEncryption) (b2BucketInterface, error) {
	b, err := r.testRoot.createBucket(ctx, name, btype, info, rules, sse)
	if err != nil {
		return nil, err
	}
	return &wakeBucket{testBucket: b.(*testBucket), r: r}, nil
}

type wakeBucket struct {
	*testBucket
	r *wakeRoot
}

func (b *wakeBucket) downloadFileByName(context.Context, string, int64, int64, bool) (b2FileReaderInterface, error) {
	n := atomic.AddInt32(&b.r.calls, 1)
	for spin := time.Now().Add(time.Duration(n%40) * time.Microsecond); time.Now().Before(spin); {
	}
	return nil, errors.New("injected download failure")
}

// Racy by nature: reliably red on v0.8.0 only under -race, which widens the
// window between the waiter's predicate check and its registration in Wait.
//
//	go test -race -run TestReaderCurChunkWaitersDoNotLeak ./b2/
func TestReaderCurChunkWaitersDoNotLeak(t *testing.T) {
	root := &wakeRoot{testRoot: &testRoot{bucketMap: make(map[string]map[string]string), errs: &errCont{}}}
	bucket, err := (&Client{backend: &beRoot{b2i: root}}).NewBucket(context.Background(), bucketName, &BucketAttrs{Type: Private})
	if err != nil {
		t.Fatal(err)
	}
	const readers = 20000
	for i := 0; i < readers; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		r := bucket.Object("obj").NewReader(ctx)
		_, _ = r.Read(make([]byte, 10))
		r.Close()
		cancel()
	}
	time.Sleep(500 * time.Millisecond)
	buf := make([]byte, 1<<24)
	if n := strings.Count(string(buf[:runtime.Stack(buf, true)]), "(*Reader).curChunk.func1"); n > 0 {
		t.Fatalf("goroutine leak: %d curChunk waiter goroutine(s) still parked after %d closed readers", n, readers)
	}
}

On v0.8.0 / master, go test -race -run TestReaderCurChunkWaitersDoNotLeak ./b2/:

--- FAIL: TestReaderCurChunkWaitersDoNotLeak (1.33s)
    reader_wakeup_test.go:61: goroutine leak: 2 curChunk waiter goroutine(s) still parked after 20000 closed readers

(2-3 per 20,000 readers in our runs. Without -race it did not reproduce in 100,000 iterations.)

Expected vs. actual

  • Expected: once a reader is closed or has failed, no goroutine of its is left waiting.
  • Actual: occasionally a curChunk waiter stays blocked in Wait forever.

Impact

A slow goroutine leak in programs that create many readers and see download errors or cancellations. No wrong data and no hung caller.

Proposed fix

Broadcast while holding rmux, so a wakeup cannot fall between the waiter's check and its Wait, and broadcast from Close as well. The diff below is written on top of the #57 fix (it also rewrites the Broadcast that #57 adds); on master alone it is the same change minus that one line.

Proposed diff (b2/reader.go, on top of #57) 
diff --git b2/reader.go b2/reader.go
index 74a0111..e689677 100644
--- b2/reader.go
+++ b2/reader.go
@@ -81,4 +81,7 @@ type rchunk struct {
 func (r *Reader) Close() error {
 	r.cancel()
+	if r.rcond != nil {
+		r.broadcast()
+	}
 	r.o.b.c.removeReader(r)
 	return nil
@@ -102,4 +105,12 @@ func (r *Reader) setErrNoCancel(err error) {
 }
 
+// broadcast wakes curChunk waiters. It takes rmux so the wakeup cannot land
+// between a waiter's predicate check and its registration in Wait.
+func (r *Reader) broadcast() {
+	r.rmux.Lock()
+	r.rcond.Broadcast()
+	r.rmux.Unlock()
+}
+
 func (r *Reader) getErr() error {
 	r.emux.RLock()
@@ -145,10 +156,10 @@ func (r *Reader) thread() {
 				r.chunks[chunkID] = buf
 				r.rmux.Unlock()
-				r.rcond.Broadcast()
+				r.broadcast()
 				return
 			}
 			if err != nil {
 				r.setErr(err)
-				r.rcond.Broadcast()
+				r.broadcast()
 				return
 			}
@@ -173,5 +184,5 @@ func (r *Reader) thread() {
 				if attempts >= maxShortReadAttempts {
 					r.setErr(fmt.Errorf("b2 reader %d: got %dB of %dB after %d attempts", chunkID, i, rsize, attempts))
-					r.rcond.Broadcast()
+					r.broadcast()
 					return
 				}
@@ -179,5 +190,5 @@ func (r *Reader) thread() {
 				if err := b.wait(r.ctx); err != nil {
 					r.setErr(err)
-					r.rcond.Broadcast()
+					r.broadcast()
 					return
 				}
@@ -187,5 +198,5 @@ func (r *Reader) thread() {
 			if err != nil {
 				r.setErr(err)
-				r.rcond.Broadcast()
+				r.broadcast()
 				return
 			}
@@ -193,5 +204,5 @@ func (r *Reader) thread() {
 			r.chunks[chunkID] = buf
 			r.rmux.Unlock()
-			r.rcond.Broadcast()
+			r.broadcast()
 		}
 	}()

rcond is created lazily in initFunc, so Close checks for nil (a reader closed before its first Read).

Validated on a copy of master with the #57 and #59 changes applied: the regression test passes under -race (three runs), go test ./... passes, and go test -race ./b2/ passes with no data-race reports.

Related

  • #57 edits the same retry block; apply this change after it.
主要言語
Go
スター
66
フォーク
18
平均マージ
1日 10時間
マージ済み PR（30日）
132

環境構築

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

Backblaze/blazer のほかの issue

Backblaze/blazer の issue をすべて見る

似ている issue

Go の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。