main.go

Overview

This file implements an integration test for BadgerDB's value log garbage collection (GC). It simulates write and read operations, triggers value log GC periodically, and verifies data consistency after GC runs.

Detailed Documentation

maxValue

var maxValue int64 = 10000000

• Purpose: Defines the maximum value for generating random keys.

suffix

var suffix = make([]byte, 128)

• Purpose: Defines a byte slice used as a suffix to increase the size of values written to the database.

testSuite

type testSuite struct {
	sync.Mutex
	vals map[uint64]uint64

	count atomic.Uint64 // Not under mutex lock.
}

• Purpose: Represents a test suite for managing concurrent write and read operations.
  • sync.Mutex: A mutex to protect the vals map.
  • vals: A map to store key-value pairs for validation during read operations.
  • count: An atomic counter to generate unique keys.

encoded(i uint64) []byte

func encoded(i uint64) []byte {
	out := make([]byte, 8)
	binary.BigEndian.PutUint64(out, i)
	return out
}

• Purpose: Encodes a uint64 integer into a byte slice using big-endian byte order.
  • Parameters:
    • i: The uint64 integer to encode.
  • Returns: A byte slice representing the encoded integer.

(*testSuite) write(db *badger.DB) error

func (s *testSuite) write(db *badger.DB) error {
	return db.Update(func(txn *badger.Txn) error {
		for i := 0; i < 10; i++ {
			// These keys would be overwritten.
			keyi := uint64(rand.Int63n(maxValue))
			key := encoded(keyi)
			vali := s.count.Add(1)
			val := encoded(vali)
			val = append(val, suffix...)
			if err := txn.SetEntry(badger.NewEntry(key, val)); err != nil {
				return err
			}
		}
		for i := 0; i < 20; i++ {
			// These keys would be new and never overwritten.
			keyi := s.count.Add(1)
			if keyi%1000000 == 0 {
				log.Printf("Count: %d\n", keyi)
			}
			key := encoded(keyi)
			val := append(key, suffix...)
			if err := txn.SetEntry(badger.NewEntry(key, val)); err != nil {
				return err
			}
		}
		return nil
	})
}

• Purpose: Writes data to the BadgerDB database. It writes both overwritten and new keys.
  • Parameters:
    • db: A pointer to the BadgerDB database.
  • Returns: An error, if any occurred during the write operation.

(*testSuite) read(db *badger.DB) error

func (s *testSuite) read(db *badger.DB) error {
	max := int64(s.count.Load())
	keyi := uint64(rand.Int63n(max))
	key := encoded(keyi)

	err := db.View(func(txn *badger.Txn) error {
		item, err := txn.Get(key)
		if err != nil {
			return err
		}
		val, err := item.ValueCopy(nil)
		if err != nil {
			return err
		}
		y.AssertTruef(len(val) == len(suffix)+8, "Found val of len: %d\n", len(val))
		vali := binary.BigEndian.Uint64(val[0:8])
		s.Lock()
		expected := s.vals[keyi]
		if vali < expected {
			log.Fatalf("Expected: %d. Found: %d. Key: %d\n", expected, vali, keyi)
		} else if vali == expected {
			// pass
		} else {
			s.vals[keyi] = vali
		}
		s.Unlock()
		return nil
	})
	if err == badger.ErrKeyNotFound {
		return nil
	}
	return err
}

• Purpose: Reads data from the BadgerDB database and validates the retrieved values.
  • Parameters:
    • db: A pointer to the BadgerDB database.
  • Returns: An error, if any occurred during the read operation, except for badger.ErrKeyNotFound.

main()

func main() {
	fmt.Println("Badger Integration test for value log GC.")

	dir := "/mnt/drive/badgertest"
	os.RemoveAll(dir)

	db, err := badger.Open(badger.DefaultOptions(dir).
		WithSyncWrites(false))
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	go func() {
		_ = http.ListenAndServe("localhost:8080", nil)
	}()

	closer := z.NewCloser(11)
	go func() {
		// Run value log GC.
		defer closer.Done()
		var count int
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()
		for range ticker.C {
		again:
			select {
			case <-closer.HasBeenClosed():
				log.Printf("Num times value log GC was successful: %d\n", count)
				return
			default:
			}
			log.Printf("Starting a value log GC")
			err := db.RunValueLogGC(0.1)
			log.Printf("Result of value log GC: %v\n", err)
			if err == nil {
				count++
				goto again
			}
		}
	}()

	s := testSuite{vals: make(map[uint64]uint64)}
	s.count.Store(uint64(maxValue))
	var numLoops atomic.Uint64
	ticker := time.NewTicker(5 * time.Second)
	for i := 0; i < 10; i++ {
		go func() {
			defer closer.Done()
			for {
				if err := s.write(db); err != nil {
					log.Fatal(err)
				}
				for j := 0; j < 10; j++ {
					if err := s.read(db); err != nil {
						log.Fatal(err)
					}
				}
				nl := numLoops.Add(1)
				select {
				case <-closer.HasBeenClosed():
					return
				case <-ticker.C:
					log.Printf("Num loops: %d\n", nl)
				default:
				}
			}
		}()
	}
	time.Sleep(5 * time.Minute)
	log.Println("Signaling...")
	closer.SignalAndWait()
	log.Println("Wait done. Now iterating over everything.")

	err = db.View(func(txn *badger.Txn) error {
		iopts := badger.DefaultIteratorOptions
		itr := txn.NewIterator(iopts)
		defer itr.Close()

		var total, tested int
		for itr.Rewind(); itr.Valid(); itr.Next() {
			item := itr.Item()
			key := item.Key()
			keyi := binary.BigEndian.Uint64(key)
			total++

			val, err := item.ValueCopy(nil)
			if err != nil {
				return err
			}
			if len(val) < 8 {
				log.Printf("Unexpected value: %x\n", val)
				continue
			}
			vali := binary.BigEndian.Uint64(val[0:8])

			expected, ok := s.vals[keyi] // Not all keys must be in vals map.
			if ok {
				tested++
				if vali < expected {
					// vali must be equal or greater than what's in the map.
					log.Fatalf("Expected: %d. Got: %d. Key: %d\n", expected, vali, keyi)
				}
			}
		}
		log.Printf("Total iterated: %d. Tested values: %d\n", total, tested)
		return nil
	})
	if err != nil {
		log.Fatalf("Error while iterating: %v", err)
	}
	log.Println("Iteration done. Test successful.")
	time.Sleep(time.Minute) // Time to do some poking around.
}

• Purpose: The main function that sets up and runs the integration test. It performs the following steps:
  • Initializes BadgerDB with specific options, disabling SyncWrites.
  • Starts a goroutine for profiling via HTTP.
  • Starts a goroutine to periodically run value log GC.
  • Starts multiple goroutines to perform concurrent write and read operations using the testSuite.
  • Waits for a specified duration.
  • Signals the closer to stop the GC and write/read goroutines.
  • Iterates over all keys in the database to validate data consistency.
  • Logs the test results.
  • Sleeps for an additional minute to allow for manual inspection.

Code Examples

None.

Getting Started Relevance