Move Closer from y to z (#1498)

The `Closer` type was moved by badger/y package to ristretto/z via dgraph-io/ristretto#191

badger/cmd/read_bench.go

@@ -31,6 +31,7 @@ import (
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 )
 
 var readBenchCmd = &cobra.Command{
@@ -84,7 +85,7 @@ func fullScanDB(db *badger.DB) {
 
 	startTime = time.Now()
 	// Print the stats
-	c := y.NewCloser(0)
+	c := z.NewCloser(0)
 	c.AddRunning(1)
 	go printStats(c)
 
@@ -142,7 +143,7 @@ func readBench(cmd *cobra.Command, args []string) error {
 		fmt.Println("DB is empty, hence returning")
 		return nil
 	}
-	c := y.NewCloser(0)
+	c := z.NewCloser(0)
 	startTime = time.Now()
 	for i := 0; i < numGoroutines; i++ {
 		c.AddRunning(1)
@@ -159,7 +160,7 @@ func readBench(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-func printStats(c *y.Closer) {
+func printStats(c *z.Closer) {
 	defer c.Done()
 
 	t := time.NewTicker(time.Second)
@@ -181,7 +182,7 @@ func printStats(c *y.Closer) {
 	}
 }
 
-func readKeys(db *badger.DB, c *y.Closer, keys [][]byte) {
+func readKeys(db *badger.DB, c *z.Closer, keys [][]byte) {
 	defer c.Done()
 	r := rand.New(rand.NewSource(time.Now().Unix()))
 	for {

badger/cmd/write_bench.go

@@ -35,6 +35,7 @@ import (
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 )
 
 var writeBenchCmd = &cobra.Command{
@@ -272,7 +273,7 @@ func writeBench(cmd *cobra.Command, args []string) error {
 
 	startTime = time.Now()
 	num := uint64(numKeys * mil)
-	c := y.NewCloser(4)
+	c := z.NewCloser(4)
 	go reportStats(c, db)
 	go dropAll(c, db)
 	go dropPrefix(c, db)
@@ -324,7 +325,7 @@ func showKeysStats(db *badger.DB) {
 		moveKeyCount, internalKeyCount)
 }
 
-func reportStats(c *y.Closer, db *badger.DB) {
+func reportStats(c *z.Closer, db *badger.DB) {
 	defer c.Done()
 
 	t := time.NewTicker(time.Second)
@@ -374,7 +375,7 @@ func reportStats(c *y.Closer, db *badger.DB) {
 	}
 }
 
-func runGC(c *y.Closer, db *badger.DB) {
+func runGC(c *z.Closer, db *badger.DB) {
 	defer c.Done()
 	period, err := time.ParseDuration(gcPeriod)
 	y.Check(err)
@@ -398,7 +399,7 @@ func runGC(c *y.Closer, db *badger.DB) {
 	}
 }
 
-func dropAll(c *y.Closer, db *badger.DB) {
+func dropAll(c *z.Closer, db *badger.DB) {
 	defer c.Done()
 	dropPeriod, err := time.ParseDuration(dropAllPeriod)
 	y.Check(err)
@@ -429,7 +430,7 @@ func dropAll(c *y.Closer, db *badger.DB) {
 	}
 }
 
-func dropPrefix(c *y.Closer, db *badger.DB) {
+func dropPrefix(c *z.Closer, db *badger.DB) {
 	defer c.Done()
 	dropPeriod, err := time.ParseDuration(dropPrefixPeriod)
 	y.Check(err)

db.go

@@ -37,6 +37,7 @@ import (
 	"github.com/dgraph-io/badger/v2/table"
 	"github.com/dgraph-io/badger/v2/y"
 	"github.com/dgraph-io/ristretto"
+	"github.com/dgraph-io/ristretto/z"
 	humanize "github.com/dustin/go-humanize"
 	"github.com/pkg/errors"
 )
@@ -50,12 +51,12 @@ var (
 )
 
 type closers struct {
-	updateSize *y.Closer
-	compactors *y.Closer
-	memtable   *y.Closer
-	writes     *y.Closer
-	valueGC    *y.Closer
-	pub        *y.Closer
+	updateSize *z.Closer
+	compactors *z.Closer
+	memtable   *z.Closer
+	writes     *z.Closer
+	valueGC    *z.Closer
+	pub        *z.Closer
 }
 
 // DB provides the various functions required to interact with Badger.
@@ -358,7 +359,7 @@ func Open(opt Options) (db *DB, err error) {
 		return db, err
 	}
 	db.calculateSize()
-	db.closers.updateSize = y.NewCloser(1)
+	db.closers.updateSize = z.NewCloser(1)
 	go db.updateSize(db.closers.updateSize)
 	db.mt = skl.NewSkiplist(arenaSize(opt))
 
@@ -371,10 +372,10 @@ func Open(opt Options) (db *DB, err error) {
 	db.vlog.init(db)
 
 	if !opt.ReadOnly {
-		db.closers.compactors = y.NewCloser(1)
+		db.closers.compactors = z.NewCloser(1)
 		db.lc.startCompact(db.closers.compactors)
 
-		db.closers.memtable = y.NewCloser(1)
+		db.closers.memtable = z.NewCloser(1)
 		go func() {
 			_ = db.flushMemtable(db.closers.memtable) // Need levels controller to be up.
 		}()
@@ -392,7 +393,7 @@ func Open(opt Options) (db *DB, err error) {
 		vptr.Decode(vs.Value)
 	}
 
-	replayCloser := y.NewCloser(1)
+	replayCloser := z.NewCloser(1)
 	go db.doWrites(replayCloser)
 
 	if err = db.vlog.open(db, vptr, db.replayFunction()); err != nil {
@@ -409,15 +410,15 @@ func Open(opt Options) (db *DB, err error) {
 	db.orc.readMark.Done(db.orc.nextTxnTs)
 	db.orc.incrementNextTs()
 
-	db.closers.writes = y.NewCloser(1)
+	db.closers.writes = z.NewCloser(1)
 	go db.doWrites(db.closers.writes)
 
 	if !db.opt.InMemory {
-		db.closers.valueGC = y.NewCloser(1)
+		db.closers.valueGC = z.NewCloser(1)
 		go db.vlog.waitOnGC(db.closers.valueGC)
 	}
 
-	db.closers.pub = y.NewCloser(1)
+	db.closers.pub = z.NewCloser(1)
 	go db.pub.listenForUpdates(db.closers.pub)
 
 	valueDirLockGuard = nil
@@ -843,7 +844,7 @@ func (db *DB) sendToWriteCh(entries []*Entry) (*request, error) {
 	return req, nil
 }
 
-func (db *DB) doWrites(lc *y.Closer) {
+func (db *DB) doWrites(lc *z.Closer) {
 	defer lc.Done()
 	pendingCh := make(chan struct{}, 1)
 
@@ -1104,7 +1105,7 @@ func (db *DB) handleFlushTask(ft flushTask) error {
 
 // flushMemtable must keep running until we send it an empty flushTask. If there
 // are errors during handling the flush task, we'll retry indefinitely.
-func (db *DB) flushMemtable(lc *y.Closer) error {
+func (db *DB) flushMemtable(lc *z.Closer) error {
 	defer lc.Done()
 
 	for ft := range db.flushChan {
@@ -1190,7 +1191,7 @@ func (db *DB) calculateSize() {
 	y.VlogSize.Set(db.opt.ValueDir, newInt(vlogSize))
 }
 
-func (db *DB) updateSize(lc *y.Closer) {
+func (db *DB) updateSize(lc *z.Closer) {
 	defer lc.Done()
 	if db.opt.InMemory {
 		return
@@ -1449,7 +1450,7 @@ func (db *DB) stopCompactions() {
 func (db *DB) startCompactions() {
 	// Resume compactions.
 	if db.closers.compactors != nil {
-		db.closers.compactors = y.NewCloser(1)
+		db.closers.compactors = z.NewCloser(1)
 		db.lc.startCompact(db.closers.compactors)
 	}
 }
@@ -1458,7 +1459,7 @@ func (db *DB) startMemoryFlush() {
 	// Start memory fluhser.
 	if db.closers.memtable != nil {
 		db.flushChan = make(chan flushTask, db.opt.NumMemtables)
-		db.closers.memtable = y.NewCloser(1)
+		db.closers.memtable = z.NewCloser(1)
 		go func() {
 			_ = db.flushMemtable(db.closers.memtable)
 		}()
@@ -1549,7 +1550,7 @@ func (db *DB) blockWrite() error {
 }
 
 func (db *DB) unblockWrite() {
-	db.closers.writes = y.NewCloser(1)
+	db.closers.writes = z.NewCloser(1)
 	go db.doWrites(db.closers.writes)
 
 	// Resume writes.
@@ -1717,7 +1718,7 @@ func (db *DB) Subscribe(ctx context.Context, cb func(kv *KVList) error, prefixes
 		return ErrNilCallback
 	}
 
-	c := y.NewCloser(1)
+	c := z.NewCloser(1)
 	recvCh, id := db.pub.newSubscriber(c, prefixes...)
 	slurp := func(batch *pb.KVList) error {
 		for {

go.mod

@@ -5,7 +5,7 @@ go 1.12
 require (
 	github.com/DataDog/zstd v1.4.1
 	github.com/cespare/xxhash v1.1.0
-	github.com/dgraph-io/ristretto v0.0.4-0.20200820164438-623d8ef1614b
+	github.com/dgraph-io/ristretto v0.0.4-0.20200906165740-41ebdbffecfd
 	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2
 	github.com/dustin/go-humanize v1.0.0
 	github.com/golang/protobuf v1.3.1

go.sum

@@ -13,8 +13,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dgraph-io/ristretto v0.0.4-0.20200820164438-623d8ef1614b h1:/g8jOqvD1UzHTOwENtkqcLmMLzTcN18P3ut8aSUZ45g=
-github.com/dgraph-io/ristretto v0.0.4-0.20200820164438-623d8ef1614b/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70dGTu2u+Ahh6E=
+github.com/dgraph-io/ristretto v0.0.4-0.20200906165740-41ebdbffecfd h1:KoJOtZf+6wpQaDTuOWGuo61GxcPBIfhwRxRTaTWGCTc=
+github.com/dgraph-io/ristretto v0.0.4-0.20200906165740-41ebdbffecfd/go.mod h1:YylP9MpCYGVZQrly/j/diqcdUetCRRePeBB0c2VGXsA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=

integration/testgc/main.go

@@ -15,6 +15,7 @@ import (
 	"github.com/dgraph-io/badger/v2"
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 )
 
 var maxValue int64 = 10000000
@@ -115,7 +116,7 @@ func main() {
 		_ = http.ListenAndServe("localhost:8080", nil)
 	}()
 
-	closer := y.NewCloser(11)
+	closer := z.NewCloser(11)
 	go func() {
 		// Run value log GC.
 		defer closer.Done()

levels.go

@@ -32,6 +32,7 @@ import (
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/table"
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 	"github.com/pkg/errors"
 )
 
@@ -362,7 +363,7 @@ func (s *levelsController) dropPrefixes(prefixes [][]byte) error {
 	return nil
 }
 
-func (s *levelsController) startCompact(lc *y.Closer) {
+func (s *levelsController) startCompact(lc *z.Closer) {
 	n := s.kv.opt.NumCompactors
 	lc.AddRunning(n - 1)
 	for i := 0; i < n; i++ {
@@ -372,7 +373,7 @@ func (s *levelsController) startCompact(lc *y.Closer) {
 	}
 }
 
-func (s *levelsController) runCompactor(id int, lc *y.Closer) {
+func (s *levelsController) runCompactor(id int, lc *z.Closer) {
 	defer lc.Done()
 
 	randomDelay := time.NewTimer(time.Duration(rand.Int31n(1000)) * time.Millisecond)

managed_db_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 	"github.com/stretchr/testify/require"
 )
 
@@ -320,7 +321,7 @@ func TestDropAllRace(t *testing.T) {
 
 	N := 10000
 	// Start a goroutine to keep trying to write to DB while DropAll happens.
-	closer := y.NewCloser(1)
+	closer := z.NewCloser(1)
 	go func() {
 		defer closer.Done()
 		ticker := time.NewTicker(time.Millisecond)
@@ -540,7 +541,7 @@ func TestDropPrefixRace(t *testing.T) {
 
 	N := 10000
 	// Start a goroutine to keep trying to write to DB while DropPrefix happens.
-	closer := y.NewCloser(1)
+	closer := z.NewCloser(1)
 	go func() {
 		defer closer.Done()
 		ticker := time.NewTicker(time.Millisecond)

merge.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 	"github.com/pkg/errors"
 )
 
@@ -30,7 +31,7 @@ type MergeOperator struct {
 	f      MergeFunc
 	db     *DB
 	key    []byte
-	closer *y.Closer
+	closer *z.Closer
 }
 
 // MergeFunc accepts two byte slices, one representing an existing value, and
@@ -49,7 +50,7 @@ func (db *DB) GetMergeOperator(key []byte,
 		f:      f,
 		db:     db,
 		key:    key,
-		closer: y.NewCloser(1),
+		closer: z.NewCloser(1),
 	}
 
 	go op.runCompactions(dur)

publisher.go

@@ -22,12 +22,13 @@ import (
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/trie"
 	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/ristretto/z"
 )
 
 type subscriber struct {
 	prefixes  [][]byte
 	sendCh    chan<- *pb.KVList
-	subCloser *y.Closer
+	subCloser *z.Closer
 }
 
 type publisher struct {
@@ -47,7 +48,7 @@ func newPublisher() *publisher {
 	}
 }
 
-func (p *publisher) listenForUpdates(c *y.Closer) {
+func (p *publisher) listenForUpdates(c *z.Closer) {
 	defer func() {
 		p.cleanSubscribers()
 		c.Done()
@@ -108,7 +109,7 @@ func (p *publisher) publishUpdates(reqs requests) {
 	}
 }
 
-func (p *publisher) newSubscriber(c *y.Closer, prefixes ...[]byte) (<-chan *pb.KVList, uint64) {
+func (p *publisher) newSubscriber(c *z.Closer, prefixes ...[]byte) (<-chan *pb.KVList, uint64) {
 	p.Lock()
 	defer p.Unlock()
 	ch := make(chan *pb.KVList, 1000)