db.go
func (db *DB) handleFlushTask(ft flushTask) error {
	if !ft.mt.Empty() {
		// Store badger head even if vptr is zero, need it for readTs
		db.opt.Debugf("Storing value log head: %+v\n", ft.vptr)
		db.elog.Printf("Storing offset: %+v\n", ft.vptr)
		offset := make([]byte, vptrSize)
		ft.vptr.Encode(offset)

		// Pick the max commit ts, so in case of crash, our read ts would be higher than all the
		// commits.
		headTs := y.KeyWithTs(head, db.orc.nextTs())
		ft.mt.Put(headTs, y.ValueStruct{Value: offset})

		// Also store lfDiscardStats before flushing memtables
		discardStatsKey := y.KeyWithTs(lfDiscardStatsKey, 1)
		ft.mt.Put(discardStatsKey, y.ValueStruct{Value: db.vlog.encodedDiscardStats()})
	}

	fileID := db.lc.reserveFileID()
	fd, err := y.CreateSyncedFile(table.NewFilename(fileID, db.opt.Dir), true)
	if err != nil {
		return y.Wrap(err)
	}

	// Don't block just to sync the directory entry.
	dirSyncCh := make(chan error)
	go func() { dirSyncCh <- syncDir(db.opt.Dir) }()

	err = writeLevel0Table(ft, fd)
	dirSyncErr := <-dirSyncCh

	if err != nil {
		db.elog.Errorf("ERROR while writing to level 0: %v", err)
		return err
	}
	if dirSyncErr != nil {
		// Do dir sync as best effort. No need to return due to an error there.
		db.elog.Errorf("ERROR while syncing level directory: %v", dirSyncErr)
	}

	tbl, err := table.OpenTable(fd, db.opt.TableLoadingMode, nil)
	if err != nil {
		db.elog.Printf("ERROR while opening table: %v", err)
		return err
	}
	// We own a ref on tbl.
	err = db.lc.addLevel0Table(tbl) // This will incrRef (if we don't error, sure)
	tbl.DecrRef()                   // Releases our ref.
	return err
}
func (db *DB) handleFlushTask(ft flushTask) error {
	if !ft.mt.Empty() {
		// Store badger head even if vptr is zero, need it for readTs
		db.opt.Debugf("Storing value log head: %+v\n", ft.vptr)
		db.elog.Printf("Storing offset: %+v\n", ft.vptr)
		offset := make([]byte, vptrSize)
		ft.vptr.Encode(offset)

		// Pick the max commit ts, so in case of crash, our read ts would be higher than all the
		// commits.
		headTs := y.KeyWithTs(head, db.orc.nextTs())
		ft.mt.Put(headTs, y.ValueStruct{Value: offset})

		// Also store lfDiscardStats before flushing memtables
		discardStatsKey := y.KeyWithTs(lfDiscardStatsKey, 1)
		ft.mt.Put(discardStatsKey, y.ValueStruct{Value: db.vlog.encodedDiscardStats()})
	}

	fileID := db.lc.reserveFileID()
	fd, err := y.CreateSyncedFile(table.NewFilename(fileID, db.opt.Dir), true)
	if err != nil {
		return y.Wrap(err)
	}

	// Don't block just to sync the directory entry.
	dirSyncCh := make(chan error)
	go func() { dirSyncCh <- syncDir(db.opt.Dir) }()

	err = writeLevel0Table(ft, fd)
	dirSyncErr := <-dirSyncCh

	if err != nil {
		db.elog.Errorf("ERROR while writing to level 0: %v", err)
		return err
	}
	if dirSyncErr != nil {
		// Do dir sync as best effort. No need to return due to an error there.
		db.elog.Errorf("ERROR while syncing level directory: %v", dirSyncErr)
	}

	tbl, err := table.OpenTable(fd, db.opt.TableLoadingMode, nil)
	if err != nil {
		db.elog.Printf("ERROR while opening table: %v", err)
		return err
	}
	// We own a ref on tbl.
	err = db.lc.addLevel0Table(tbl) // This will incrRef (if we don't error, sure)
	tbl.DecrRef()                   // Releases our ref.
	return err
}
// handleFlushTask must be run serially.