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
}
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 ( " \n " , ft . vptr ) db . elog . Printf ( " \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 ( " " , err ) return err } if dirSyncErr != nil { // Do dir sync as best effort. No need to return due to an error there. db . elog . Errorf ( " " , dirSyncErr ) } tbl , err := table . OpenTable ( fd , db . opt . TableLoadingMode , nil ) if err != nil { db . elog . Printf ( " " , 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.
handleFlushTask must be run serially .