ring.go raw

   1  /*
   2   * SPDX-FileCopyrightText: © Hypermode Inc. <hello@hypermode.com>
   3   * SPDX-License-Identifier: Apache-2.0
   4   */
   5  
   6  package ristretto
   7  
   8  import (
   9  	"sync"
  10  )
  11  
  12  // ringConsumer is the user-defined object responsible for receiving and
  13  // processing items in batches when buffers are drained.
  14  type ringConsumer interface {
  15  	Push([]uint64) bool
  16  }
  17  
  18  // ringStripe is a singular ring buffer that is not concurrent safe.
  19  type ringStripe struct {
  20  	cons ringConsumer
  21  	data []uint64
  22  	capa int
  23  }
  24  
  25  func newRingStripe(cons ringConsumer, capa int64) *ringStripe {
  26  	return &ringStripe{
  27  		cons: cons,
  28  		data: make([]uint64, 0, capa),
  29  		capa: int(capa),
  30  	}
  31  }
  32  
  33  // Push appends an item in the ring buffer and drains (copies items and
  34  // sends to Consumer) if full.
  35  func (s *ringStripe) Push(item uint64) {
  36  	s.data = append(s.data, item)
  37  	// Decide if the ring buffer should be drained.
  38  	if len(s.data) >= s.capa {
  39  		// Send elements to consumer and create a new ring stripe.
  40  		if s.cons.Push(s.data) {
  41  			s.data = make([]uint64, 0, s.capa)
  42  		} else {
  43  			s.data = s.data[:0]
  44  		}
  45  	}
  46  }
  47  
  48  // ringBuffer stores multiple buffers (stripes) and distributes Pushed items
  49  // between them to lower contention.
  50  //
  51  // This implements the "batching" process described in the BP-Wrapper paper
  52  // (section III part A).
  53  type ringBuffer struct {
  54  	pool *sync.Pool
  55  }
  56  
  57  // newRingBuffer returns a striped ring buffer. The Consumer in ringConfig will
  58  // be called when individual stripes are full and need to drain their elements.
  59  func newRingBuffer(cons ringConsumer, capa int64) *ringBuffer {
  60  	// LOSSY buffers use a very simple sync.Pool for concurrently reusing
  61  	// stripes. We do lose some stripes due to GC (unheld items in sync.Pool
  62  	// are cleared), but the performance gains generally outweigh the small
  63  	// percentage of elements lost. The performance primarily comes from
  64  	// low-level runtime functions used in the standard library that aren't
  65  	// available to us (such as runtime_procPin()).
  66  	return &ringBuffer{
  67  		pool: &sync.Pool{
  68  			New: func() interface{} { return newRingStripe(cons, capa) },
  69  		},
  70  	}
  71  }
  72  
  73  // Push adds an element to one of the internal stripes and possibly drains if
  74  // the stripe becomes full.
  75  func (b *ringBuffer) Push(item uint64) {
  76  	// Reuse or create a new stripe.
  77  	stripe := b.pool.Get().(*ringStripe)
  78  	stripe.Push(item)
  79  	b.pool.Put(stripe)
  80  }
  81