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

Abierto
  • ¿Algo no está bien?
#59 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
1-2 días
Aptitud para principiantes
56/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
go
Área
backend

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.
Lenguaje dominante
Go
Estrellas
66
Forks
18
Merge medio
1 d 10 h
PR fusionados (30 d)
132

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de Backblaze/blazer

Todos los issues de Backblaze/blazer

Issues similares

Más issues de Go

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.