Small improvements (#1598)

Decrease the size of DISCARD file and WAL for Memtables.

db.go

@@ -925,19 +925,11 @@ func (db *DB) ensureRoomForWrite() error {
 	db.Lock()
 	defer db.Unlock()
 
-	var forceFlush bool
-	// We don't need to force flush the memtable in in-memory mode because the size of the WAL will
-	// always be zero.
-	if !forceFlush && !db.opt.InMemory {
-		// Force flush if memTable WAL is getting filled up.
-		forceFlush = int64(db.mt.wal.writeAt) > db.opt.ValueLogFileSize
-	}
-
-	if !forceFlush && db.mt.sl.MemSize() < db.opt.MemTableSize {
+	y.AssertTrue(db.mt != nil) // A nil mt indicates that DB is being closed.
+	if !db.mt.isFull() {
 		return nil
 	}
 
-	y.AssertTrue(db.mt != nil) // A nil mt indicates that DB is being closed.
 	select {
 	case db.flushChan <- flushTask{mt: db.mt}:
 		db.opt.Debugf("Flushing memtable, mt.size=%d size of flushChan: %d\n",
@@ -1637,7 +1629,10 @@ func (db *DB) dropAll() (func(), error) {
 // - Compact rest of the levels, Li->Li, picking tables which have Kp.
 // - Resume memtable flushes, compactions and writes.
 func (db *DB) DropPrefix(prefixes ...[]byte) error {
-	db.opt.Infof("DropPrefix Called %s", prefixes)
+	if len(prefixes) == 0 {
+		return nil
+	}
+	db.opt.Infof("DropPrefix called for %s", prefixes)
 	f, err := db.prepareToDrop()
 	if err != nil {
 		return err

discard.go

@@ -38,14 +38,12 @@ type discardStats struct {
 }
 
 const discardFname string = "DISCARD"
-const discardFsize int = 1 << 30
-const maxSlot int = 64 << 20
 
 func initDiscardStats(opt Options) (*discardStats, error) {
 	fname := path.Join(opt.ValueDir, discardFname)
 
 	// 1GB file can store 67M discard entries. Each entry is 16 bytes.
-	mf, err := z.OpenMmapFile(fname, os.O_CREATE|os.O_RDWR, discardFsize)
+	mf, err := z.OpenMmapFile(fname, os.O_CREATE|os.O_RDWR, 1<<20)
 	lf := &discardStats{
 		MmapFile: mf,
 		opt:      opt,
@@ -58,7 +56,7 @@ func initDiscardStats(opt Options) (*discardStats, error) {
 		return nil, y.Wrapf(err, "while opening file: %s\n", discardFname)
 	}
 
-	for slot := 0; slot < maxSlot; slot++ {
+	for slot := 0; slot < lf.maxSlot(); slot++ {
 		if lf.get(16*slot) == 0 {
 			lf.nextEmptySlot = slot
 			break
@@ -98,6 +96,10 @@ func (lf *discardStats) zeroOut() {
 	lf.set(lf.nextEmptySlot*16+8, 0)
 }
 
+func (lf *discardStats) maxSlot() int {
+	return len(lf.Data) / 16
+}
+
 // Update would update the discard stats for the given file id. If discard is
 // 0, it would return the current value of discard for the file. If discard is
 // < 0, it would set the current value of discard to zero for the file.
@@ -134,7 +136,9 @@ func (lf *discardStats) Update(fidu uint32, discard int64) int64 {
 
 	// Move to next slot.
 	lf.nextEmptySlot++
-	y.AssertTrue(lf.nextEmptySlot < maxSlot)
+	for lf.nextEmptySlot >= lf.maxSlot() {
+		y.Check(lf.Truncate(2 * int64(len(lf.Data))))
+	}
 	lf.zeroOut()
 
 	sort.Sort(lf)

memtable.go

@@ -125,8 +125,9 @@ func (db *DB) openMemTable(fid, flags int) (*memTable, error) {
 		path:     filepath,
 		registry: db.registry,
 		writeAt:  vlogHeaderSize,
+		opt:      db.opt,
 	}
-	lerr := mt.wal.open(filepath, flags, db.opt)
+	lerr := mt.wal.open(filepath, flags, 2*db.opt.MemTableSize)
 	if lerr != z.NewFile && lerr != nil {
 		return nil, y.Wrapf(lerr, "While opening memtable: %s", filepath)
 	}
@@ -170,6 +171,17 @@ func (mt *memTable) SyncWAL() error {
 	return mt.wal.Sync()
 }
 
+func (mt *memTable) isFull() bool {
+	if mt.sl.MemSize() >= mt.opt.MemTableSize {
+		return true
+	}
+	if mt.opt.InMemory {
+		// InMemory mode doesn't have any WAL.
+		return false
+	}
+	return int64(mt.wal.writeAt) >= mt.opt.MemTableSize
+}
+
 func (mt *memTable) Put(key []byte, value y.ValueStruct) error {
 	entry := &Entry{
 		Key:       key,
@@ -540,10 +552,8 @@ func (lf *logFile) zeroNextEntry() {
 	z.ZeroOut(lf.Data, int(lf.writeAt), int(lf.writeAt+maxHeaderSize))
 }
 
-func (lf *logFile) open(path string, flags int, opt Options) error {
-	lf.opt = opt
-
-	mf, ferr := z.OpenMmapFile(path, flags, 2*int(opt.ValueLogFileSize))
+func (lf *logFile) open(path string, flags int, fsize int64) error {
+	mf, ferr := z.OpenMmapFile(path, flags, int(fsize))
 	lf.MmapFile = mf
 
 	if ferr == z.NewFile {

value.go

@@ -508,8 +508,9 @@ func (vlog *valueLog) createVlogFile() (*logFile, error) {
 		path:     path,
 		registry: vlog.db.registry,
 		writeAt:  vlogHeaderSize,
+		opt:      vlog.opt,
 	}
-	err := lf.open(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, vlog.opt)
+	err := lf.open(path, os.O_RDWR|os.O_CREATE|os.O_EXCL, 2*vlog.opt.ValueLogFileSize)
 	if err != z.NewFile && err != nil {
 		return nil, err
 	}
@@ -574,7 +575,9 @@ func (vlog *valueLog) open(db *DB) error {
 		y.AssertTrue(ok)
 
 		// Just open in RDWR mode. This should not create a new log file.
-		if err := lf.open(vlog.fpath(fid), os.O_RDWR, vlog.opt); err != nil {
+		lf.opt = vlog.opt
+		if err := lf.open(vlog.fpath(fid), os.O_RDWR,
+			2*vlog.opt.ValueLogFileSize); err != nil {
 			return y.Wrapf(err, "Open existing file: %q", lf.path)
 		}
 		// We shouldn't delete the maxFid file.

value_test.go

@@ -990,7 +990,7 @@ func TestValueLogTruncate(t *testing.T) {
 	require.Equal(t, 4, int(db.mt.wal.fid))
 	fileStat, err := db.mt.wal.Fd.Stat()
 	require.NoError(t, err)
-	require.Equal(t, 2*db.opt.ValueLogFileSize, fileStat.Size())
+	require.Equal(t, 2*db.opt.MemTableSize, fileStat.Size())
 
 	fileCountAfterCorruption := len(db.Tables()) + len(db.imm) + 1 // +1 for db.mt
 	// We should have one memtable and one sst file.