@@ -2465,39 +2465,63 @@ func TestGetBlobMultipartPartHasDescriptorContentType(t *testing.T) {
24652465// - A reader-error mid-stream truncates the body (since the 206
24662466// headers have already been flushed) and is logged.
24672467
2468- // countingReader wraps a strings.Reader so a test can observe whether
2469- // the wrapper has been closed yet. It tracks open/max-open counters
2470- // shared with the test; the storage mock invokes its constructor on
2471- // every GetBlobPartial call, so any concurrent opens immediately
2472- // surface as a maxOpen > 1.
2473- type countingReader struct {
2474- * strings. Reader
2468+ // partialReaderOpenTracker records how many partial-blob readers are open at once and
2469+ // the peak concurrent count. The multipart test's GetBlobPartial mock calls NewReadCloser
2470+ // per range; overlapping opens show up as PeakOpens() > 1.
2471+ type partialReaderOpenTracker struct {
2472+ live atomic. Int32
2473+ peak atomic. Int32
2474+ }
24752475
2476- open * atomic.Int32
2477- maxOpen * atomic.Int32
2478- closed bool
2476+ // NewReadCloser returns a reader that registers in the tracker until Close.
2477+ func (t * partialReaderOpenTracker ) NewReadCloser (body string ) io.ReadCloser {
2478+ t .beginOpen ()
2479+
2480+ return & partialReaderReadCloser {
2481+ Reader : strings .NewReader (body ),
2482+ tracker : t ,
2483+ }
24792484}
24802485
2481- func newCountingReader (body string , open , maxOpen * atomic.Int32 ) * countingReader {
2482- cur := open .Add (1 )
2486+ func (t * partialReaderOpenTracker ) LiveOpens () int32 { return t .live .Load () }
2487+
2488+ func (t * partialReaderOpenTracker ) PeakOpens () int32 { return t .peak .Load () }
2489+
2490+ func (t * partialReaderOpenTracker ) endClose () { t .live .Add (- 1 ) }
2491+
2492+ // beginOpen increments the live-open count and sets peak := max(peak, newLiveCount).
2493+ //
2494+ // The for loop retries when CompareAndSwap fails: another goroutine can change peak
2495+ // after Load but before CompareAndSwap, so one attempt is not enough under contention.
2496+ func (t * partialReaderOpenTracker ) beginOpen () {
2497+ cur := t .live .Add (1 )
24832498
24842499 for {
2485- prev := maxOpen .Load ()
2486- if cur <= prev || maxOpen .CompareAndSwap (prev , cur ) {
2487- break
2500+ observedPeak := t .peak .Load ()
2501+ if cur <= observedPeak {
2502+ return
2503+ }
2504+ if t .peak .CompareAndSwap (observedPeak , cur ) {
2505+ return
24882506 }
24892507 }
2508+ }
24902509
2491- return & countingReader {Reader : strings .NewReader (body ), open : open , maxOpen : maxOpen }
2510+ // partialReaderReadCloser wraps a strings.Reader and only notifies the tracker on Close.
2511+ type partialReaderReadCloser struct {
2512+ * strings.Reader
2513+
2514+ tracker * partialReaderOpenTracker
2515+ closed bool
24922516}
24932517
2494- func (cr * countingReader ) Close () error {
2495- if cr .closed {
2518+ func (r * partialReaderReadCloser ) Close () error {
2519+ if r .closed {
24962520 return nil
24972521 }
24982522
2499- cr .closed = true
2500- cr . open . Add ( - 1 )
2523+ r .closed = true
2524+ r . tracker . endClose ( )
25012525
25022526 return nil
25032527}
@@ -2583,10 +2607,7 @@ func TestGetBlobMultipartContentLengthMatchesBody(t *testing.T) {
25832607func TestGetBlobMultipartOpensOneReaderAtATime (t * testing.T ) {
25842608 const blobBody = "0123456789abcdef0123456789abcdef" // 32 bytes
25852609
2586- var (
2587- open atomic.Int32
2588- maxOpen atomic.Int32
2589- )
2610+ var opens partialReaderOpenTracker
25902611
25912612 store := descriptorStore (t )
25922613 store .CheckBlobFn = func (repo string , digest godigest.Digest ) (bool , int64 , error ) {
@@ -2599,11 +2620,9 @@ func TestGetBlobMultipartOpensOneReaderAtATime(t *testing.T) {
25992620 from ,
26002621 to int64 ,
26012622 ) (io.ReadCloser , int64 , int64 , error ) {
2602- // Wrap a strings.Reader in a counter that increments on open
2603- // and decrements on close. The producer goroutine in
2604- // writeMultipartRanges should open and fully consume each
2605- // reader before opening the next.
2606- reader := newCountingReader (blobBody [from :to + 1 ], & open , & maxOpen )
2623+ // opens tracks live readers; Close decrements. writeMultipartRanges should fully
2624+ // consume each reader before opening the next.
2625+ reader := opens .NewReadCloser (blobBody [from : to + 1 ])
26072626
26082627 return reader , to - from + 1 , int64 (len (blobBody )), nil
26092628 }
@@ -2633,8 +2652,8 @@ func TestGetBlobMultipartOpensOneReaderAtATime(t *testing.T) {
26332652 // the open counter on every reader.
26342653 _ = drainResponseBody (t , resp )
26352654
2636- assert .Equal (t , int32 (0 ), open . Load (), "all readers must be closed by the time the body is drained" )
2637- assert .Equal (t , int32 (1 ), maxOpen . Load (),
2655+ assert .Equal (t , int32 (0 ), opens . LiveOpens (), "all readers must be closed by the time the body is drained" )
2656+ assert .Equal (t , int32 (1 ), opens . PeakOpens (),
26382657 "writeMultipartRanges must open at most one range reader at a time" )
26392658}
26402659
0 commit comments