pools.go

     1  // Copyright 2020 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  //go:build goexperiment.jsonv2
     6  
     7  package jsontext
     8  
     9  import (
    10  	"bytes"
    11  	"io"
    12  	"math/bits"
    13  	"sync"
    14  )
    15  
    16  // TODO(https://go.dev/issue/47657): Use sync.PoolOf.
    17  
    18  var (
    19  	// This owns the internal buffer since there is no io.Writer to output to.
    20  	// Since the buffer can get arbitrarily large in normal usage,
    21  	// there is statistical tracking logic to determine whether to recycle
    22  	// the internal buffer or not based on a history of utilization.
    23  	bufferedEncoderPool = &sync.Pool{New: func() any { return new(Encoder) }}
    24  
    25  	// This owns the internal buffer, but it is only used to temporarily store
    26  	// buffered JSON before flushing it to the underlying io.Writer.
    27  	// In a sufficiently efficient streaming mode, we do not expect the buffer
    28  	// to grow arbitrarily large. Thus, we avoid recycling large buffers.
    29  	streamingEncoderPool = &sync.Pool{New: func() any { return new(Encoder) }}
    30  
    31  	// This does not own the internal buffer since
    32  	// it is taken directly from the provided bytes.Buffer.
    33  	bytesBufferEncoderPool = &sync.Pool{New: func() any { return new(Encoder) }}
    34  )
    35  
    36  // bufferStatistics is statistics to track buffer utilization.
    37  // It is used to determine whether to recycle a buffer or not
    38  // to avoid https://go.dev/issue/23199.
    39  type bufferStatistics struct {
    40  	strikes int // number of times the buffer was under-utilized
    41  	prevLen int // length of previous buffer
    42  }
    43  
    44  func getBufferedEncoder(opts ...Options) *Encoder {
    45  	e := bufferedEncoderPool.Get().(*Encoder)
    46  	if e.s.Buf == nil {
    47  		// Round up to nearest 2ⁿ to make best use of malloc size classes.
    48  		// See runtime/sizeclasses.go on Go1.15.
    49  		// Logical OR with 63 to ensure 64 as the minimum buffer size.
    50  		n := 1 << bits.Len(uint(e.s.bufStats.prevLen|63))
    51  		e.s.Buf = make([]byte, 0, n)
    52  	}
    53  	e.s.reset(e.s.Buf[:0], nil, opts...)
    54  	return e
    55  }
    56  func putBufferedEncoder(e *Encoder) {
    57  	// Recycle large buffers only if sufficiently utilized.
    58  	// If a buffer is under-utilized enough times sequentially,
    59  	// then it is discarded, ensuring that a single large buffer
    60  	// won't be kept alive by a continuous stream of small usages.
    61  	//
    62  	// The worst case utilization is computed as:
    63  	//	MIN_UTILIZATION_THRESHOLD / (1 + MAX_NUM_STRIKES)
    64  	//
    65  	// For the constants chosen below, this is (25%)/(1+4) ⇒ 5%.
    66  	// This may seem low, but it ensures a lower bound on
    67  	// the absolute worst-case utilization. Without this check,
    68  	// this would be theoretically 0%, which is infinitely worse.
    69  	//
    70  	// See https://go.dev/issue/27735.
    71  	switch {
    72  	case cap(e.s.Buf) <= 4<<10: // always recycle buffers smaller than 4KiB
    73  		e.s.bufStats.strikes = 0
    74  	case cap(e.s.Buf)/4 <= len(e.s.Buf): // at least 25% utilization
    75  		e.s.bufStats.strikes = 0
    76  	case e.s.bufStats.strikes < 4: // at most 4 strikes
    77  		e.s.bufStats.strikes++
    78  	default: // discard the buffer; too large and too often under-utilized
    79  		e.s.bufStats.strikes = 0
    80  		e.s.bufStats.prevLen = len(e.s.Buf) // heuristic for size to allocate next time
    81  		e.s.Buf = nil
    82  	}
    83  	bufferedEncoderPool.Put(e)
    84  }
    85  
    86  func getStreamingEncoder(w io.Writer, opts ...Options) *Encoder {
    87  	if _, ok := w.(*bytes.Buffer); ok {
    88  		e := bytesBufferEncoderPool.Get().(*Encoder)
    89  		e.s.reset(nil, w, opts...) // buffer taken from bytes.Buffer
    90  		return e
    91  	} else {
    92  		e := streamingEncoderPool.Get().(*Encoder)
    93  		e.s.reset(e.s.Buf[:0], w, opts...) // preserve existing buffer
    94  		return e
    95  	}
    96  }
    97  func putStreamingEncoder(e *Encoder) {
    98  	if _, ok := e.s.wr.(*bytes.Buffer); ok {
    99  		bytesBufferEncoderPool.Put(e)
   100  	} else {
   101  		if cap(e.s.Buf) > 64<<10 {
   102  			e.s.Buf = nil // avoid pinning arbitrarily large amounts of memory
   103  		}
   104  		streamingEncoderPool.Put(e)
   105  	}
   106  }
   107  
   108  var (
   109  	// This does not own the internal buffer since it is externally provided.
   110  	bufferedDecoderPool = &sync.Pool{New: func() any { return new(Decoder) }}
   111  
   112  	// This owns the internal buffer, but it is only used to temporarily store
   113  	// buffered JSON fetched from the underlying io.Reader.
   114  	// In a sufficiently efficient streaming mode, we do not expect the buffer
   115  	// to grow arbitrarily large. Thus, we avoid recycling large buffers.
   116  	streamingDecoderPool = &sync.Pool{New: func() any { return new(Decoder) }}
   117  
   118  	// This does not own the internal buffer since
   119  	// it is taken directly from the provided bytes.Buffer.
   120  	bytesBufferDecoderPool = bufferedDecoderPool
   121  )
   122  
   123  func getBufferedDecoder(b []byte, opts ...Options) *Decoder {
   124  	d := bufferedDecoderPool.Get().(*Decoder)
   125  	d.s.reset(b, nil, opts...)
   126  	return d
   127  }
   128  func putBufferedDecoder(d *Decoder) {
   129  	bufferedDecoderPool.Put(d)
   130  }
   131  
   132  func getStreamingDecoder(r io.Reader, opts ...Options) *Decoder {
   133  	if _, ok := r.(*bytes.Buffer); ok {
   134  		d := bytesBufferDecoderPool.Get().(*Decoder)
   135  		d.s.reset(nil, r, opts...) // buffer taken from bytes.Buffer
   136  		return d
   137  	} else {
   138  		d := streamingDecoderPool.Get().(*Decoder)
   139  		d.s.reset(d.s.buf[:0], r, opts...) // preserve existing buffer
   140  		return d
   141  	}
   142  }
   143  func putStreamingDecoder(d *Decoder) {
   144  	if _, ok := d.s.rd.(*bytes.Buffer); ok {
   145  		bytesBufferDecoderPool.Put(d)
   146  	} else {
   147  		if cap(d.s.buf) > 64<<10 {
   148  			d.s.buf = nil // avoid pinning arbitrarily large amounts of memory
   149  		}
   150  		streamingDecoderPool.Put(d)
   151  	}
   152  }
   153
View as plain text