db.go
func Open(opt Options) (db *DB, err error) {
	opt.maxBatchSize = (15 * opt.MaxTableSize) / 100
	opt.maxBatchCount = opt.maxBatchSize / int64(skl.MaxNodeSize)

	if opt.ValueThreshold > math.MaxUint16-16 {
		return nil, ErrValueThreshold
	}

	if opt.ReadOnly {
		// Can't truncate if the DB is read only.
		opt.Truncate = false
		// Do not perform compaction in read only mode.
		opt.CompactL0OnClose = false
	}

	for _, path := range []string{opt.Dir, opt.ValueDir} {
		dirExists, err := exists(path)
		if err != nil {
			return nil, y.Wrapf(err, "Invalid Dir: %q", path)
		}
		if !dirExists {
			if opt.ReadOnly {
				return nil, y.Wrapf(err, "Cannot find Dir for read-only open: %q", path)
			}
			// Try to create the directory
			err = os.Mkdir(path, 0700)
			if err != nil {
				return nil, y.Wrapf(err, "Error Creating Dir: %q", path)
			}
		}
	}
	absDir, err := filepath.Abs(opt.Dir)
	if err != nil {
		return nil, err
	}
	absValueDir, err := filepath.Abs(opt.ValueDir)
	if err != nil {
		return nil, err
	}
	var dirLockGuard, valueDirLockGuard *directoryLockGuard
	dirLockGuard, err = acquireDirectoryLock(opt.Dir, lockFile, opt.ReadOnly)
	if err != nil {
		return nil, err
	}
	defer func() {
		if dirLockGuard != nil {
			_ = dirLockGuard.release()
		}
	}()
	if absValueDir != absDir {
		valueDirLockGuard, err = acquireDirectoryLock(opt.ValueDir, lockFile, opt.ReadOnly)
		if err != nil {
			return nil, err
		}
		defer func() {
			if valueDirLockGuard != nil {
				_ = valueDirLockGuard.release()
			}
		}()
	}
	if !(opt.ValueLogFileSize <= 2<<30 && opt.ValueLogFileSize >= 1<<20) {
		return nil, ErrValueLogSize
	}
	if !(opt.ValueLogLoadingMode == options.FileIO ||
		opt.ValueLogLoadingMode == options.MemoryMap) {
		return nil, ErrInvalidLoadingMode
	}
	manifestFile, manifest, err := openOrCreateManifestFile(opt.Dir, opt.ReadOnly)
	if err != nil {
		return nil, err
	}
	defer func() {
		if manifestFile != nil {
			_ = manifestFile.close()
		}
	}()

	db = &DB{
		imm:           make([]*skl.Skiplist, 0, opt.NumMemtables),
		flushChan:     make(chan flushTask, opt.NumMemtables),
		writeCh:       make(chan *request, kvWriteChCapacity),
		opt:           opt,
		manifest:      manifestFile,
		elog:          trace.NewEventLog("Badger", "DB"),
		dirLockGuard:  dirLockGuard,
		valueDirGuard: valueDirLockGuard,
		orc:           newOracle(opt),
	}

	// Calculate initial size.
	db.calculateSize()
	db.closers.updateSize = y.NewCloser(1)
	go db.updateSize(db.closers.updateSize)
	db.mt = skl.NewSkiplist(arenaSize(opt))

	// newLevelsController potentially loads files in directory.
	if db.lc, err = newLevelsController(db, &manifest); err != nil {
		return nil, err
	}

	if !opt.ReadOnly {
		db.closers.compactors = y.NewCloser(1)
		db.lc.startCompact(db.closers.compactors)

		db.closers.memtable = y.NewCloser(1)
		go db.flushMemtable(db.closers.memtable) // Need levels controller to be up.
	}

	headKey := y.KeyWithTs(head, math.MaxUint64)
	// Need to pass with timestamp, lsm get removes the last 8 bytes and compares key
	vs, err := db.get(headKey)
	if err != nil {
		return nil, errors.Wrap(err, "Retrieving head")
	}
	db.orc.nextTxnTs = vs.Version
	var vptr valuePointer
	if len(vs.Value) > 0 {
		vptr.Decode(vs.Value)
	}

	replayCloser := y.NewCloser(1)
	go db.doWrites(replayCloser)

	if err = db.vlog.open(db, vptr, db.replayFunction()); err != nil {
		return db, err
	}
	replayCloser.SignalAndWait() // Wait for replay to be applied first.

	// Let's advance nextTxnTs to one more than whatever we observed via
	// replaying the logs.
	db.orc.txnMark.Done(db.orc.nextTxnTs)
	// In normal mode, we must update readMark so older versions of keys can be removed during
	// compaction when run in offline mode via the flatten tool.
	db.orc.readMark.Done(db.orc.nextTxnTs)
	db.orc.nextTxnTs++

	db.writeCh = make(chan *request, kvWriteChCapacity)
	db.closers.writes = y.NewCloser(1)
	go db.doWrites(db.closers.writes)

	db.closers.valueGC = y.NewCloser(1)
	go db.vlog.waitOnGC(db.closers.valueGC)

	valueDirLockGuard = nil
	dirLockGuard = nil
	manifestFile = nil
	return db, nil
}
func Open(opt Options) (db *DB, err error) {
	opt.maxBatchSize = (15 * opt.MaxTableSize) / 100
	opt.maxBatchCount = opt.maxBatchSize / int64(skl.MaxNodeSize)

	if opt.ValueThreshold > math.MaxUint16-16 {
		return nil, ErrValueThreshold
	}

	if opt.ReadOnly {
		// Can't truncate if the DB is read only.
		opt.Truncate = false
		// Do not perform compaction in read only mode.
		opt.CompactL0OnClose = false
	}

	for _, path := range []string{opt.Dir, opt.ValueDir} {
		dirExists, err := exists(path)
		if err != nil {
			return nil, y.Wrapf(err, "Invalid Dir: %q", path)
		}
		if !dirExists {
			if opt.ReadOnly {
				return nil, y.Wrapf(err, "Cannot find Dir for read-only open: %q", path)
			}
			// Try to create the directory
			err = os.Mkdir(path, 0700)
			if err != nil {
				return nil, y.Wrapf(err, "Error Creating Dir: %q", path)
			}
		}
	}
	absDir, err := filepath.Abs(opt.Dir)
	if err != nil {
		return nil, err
	}
	absValueDir, err := filepath.Abs(opt.ValueDir)
	if err != nil {
		return nil, err
	}
	var dirLockGuard, valueDirLockGuard *directoryLockGuard
	dirLockGuard, err = acquireDirectoryLock(opt.Dir, lockFile, opt.ReadOnly)
	if err != nil {
		return nil, err
	}
	defer func() {
		if dirLockGuard != nil {
			_ = dirLockGuard.release()
		}
	}()
	if absValueDir != absDir {
		valueDirLockGuard, err = acquireDirectoryLock(opt.ValueDir, lockFile, opt.ReadOnly)
		if err != nil {
			return nil, err
		}
		defer func() {
			if valueDirLockGuard != nil {
				_ = valueDirLockGuard.release()
			}
		}()
	}
	if !(opt.ValueLogFileSize <= 2<<30 && opt.ValueLogFileSize >= 1<<20) {
		return nil, ErrValueLogSize
	}
	if !(opt.ValueLogLoadingMode == options.FileIO ||
		opt.ValueLogLoadingMode == options.MemoryMap) {
		return nil, ErrInvalidLoadingMode
	}
	manifestFile, manifest, err := openOrCreateManifestFile(opt.Dir, opt.ReadOnly)
	if err != nil {
		return nil, err
	}
	defer func() {
		if manifestFile != nil {
			_ = manifestFile.close()
		}
	}()

	db = &DB{
		imm:           make([]*skl.Skiplist, 0, opt.NumMemtables),
		flushChan:     make(chan flushTask, opt.NumMemtables),
		writeCh:       make(chan *request, kvWriteChCapacity),
		opt:           opt,
		manifest:      manifestFile,
		elog:          trace.NewEventLog("Badger", "DB"),
		dirLockGuard:  dirLockGuard,
		valueDirGuard: valueDirLockGuard,
		orc:           newOracle(opt),
	}

	// Calculate initial size.
	db.calculateSize()
	db.closers.updateSize = y.NewCloser(1)
	go db.updateSize(db.closers.updateSize)
	db.mt = skl.NewSkiplist(arenaSize(opt))

	// newLevelsController potentially loads files in directory.
	if db.lc, err = newLevelsController(db, &manifest); err != nil {
		return nil, err
	}

	if !opt.ReadOnly {
		db.closers.compactors = y.NewCloser(1)
		db.lc.startCompact(db.closers.compactors)

		db.closers.memtable = y.NewCloser(1)
		go db.flushMemtable(db.closers.memtable) // Need levels controller to be up.
	}

	headKey := y.KeyWithTs(head, math.MaxUint64)
	// Need to pass with timestamp, lsm get removes the last 8 bytes and compares key
	vs, err := db.get(headKey)
	if err != nil {
		return nil, errors.Wrap(err, "Retrieving head")
	}
	db.orc.nextTxnTs = vs.Version
	var vptr valuePointer
	if len(vs.Value) > 0 {
		vptr.Decode(vs.Value)
	}

	replayCloser := y.NewCloser(1)
	go db.doWrites(replayCloser)

	if err = db.vlog.open(db, vptr, db.replayFunction()); err != nil {
		return db, err
	}
	replayCloser.SignalAndWait() // Wait for replay to be applied first.

	// Let's advance nextTxnTs to one more than whatever we observed via
	// replaying the logs.
	db.orc.txnMark.Done(db.orc.nextTxnTs)
	// In normal mode, we must update readMark so older versions of keys can be removed during
	// compaction when run in offline mode via the flatten tool.
	db.orc.readMark.Done(db.orc.nextTxnTs)
	db.orc.nextTxnTs++

	db.writeCh = make(chan *request, kvWriteChCapacity)
	db.closers.writes = y.NewCloser(1)
	go db.doWrites(db.closers.writes)

	db.closers.valueGC = y.NewCloser(1)
	go db.vlog.waitOnGC(db.closers.valueGC)

	valueDirLockGuard = nil
	dirLockGuard = nil
	manifestFile = nil
	return db, nil
}
func Open ( opt Options ) ( db * DB , err error ) { opt . maxBatchSize = ( 15 * opt . MaxTableSize ) / 100 opt . maxBatchCount = opt . maxBatchSize / int64 ( skl . MaxNodeSize ) if opt . ValueThreshold > math . MaxUint16 - 16 { return nil , ErrValueThreshold } if opt . ReadOnly { // Can't truncate if the DB is read only. opt . Truncate = false // Do not perform compaction in read only mode. opt . CompactL0OnClose = false } for _ , path := range [ ] string { opt . Dir , opt . ValueDir } { dirExists , err := exists ( path ) if err != nil { return nil , y . Wrapf ( err , " " , path ) } if ! dirExists { if opt . ReadOnly { return nil , y . Wrapf ( err , " " , path ) } // Try to create the directory err = os . Mkdir ( path , 0700 ) if err != nil { return nil , y . Wrapf ( err , " " , path ) } } } absDir , err := filepath . Abs ( opt . Dir ) if err != nil { return nil , err } absValueDir , err := filepath . Abs ( opt . ValueDir ) if err != nil { return nil , err } var dirLockGuard , valueDirLockGuard * directoryLockGuard dirLockGuard , err = acquireDirectoryLock ( opt . Dir , lockFile , opt . ReadOnly ) if err != nil { return nil , err } defer func ( ) { if dirLockGuard != nil { _ = dirLockGuard . release ( ) } } ( ) if absValueDir != absDir { valueDirLockGuard , err = acquireDirectoryLock ( opt . ValueDir , lockFile , opt . ReadOnly ) if err != nil { return nil , err } defer func ( ) { if valueDirLockGuard != nil { _ = valueDirLockGuard . release ( ) } } ( ) } if ! ( opt . ValueLogFileSize <= 2 << 30 && opt . ValueLogFileSize >= 1 << 20 ) { return nil , ErrValueLogSize } if ! ( opt . ValueLogLoadingMode == options . FileIO || opt . ValueLogLoadingMode == options . MemoryMap ) { return nil , ErrInvalidLoadingMode } manifestFile , manifest , err := openOrCreateManifestFile ( opt . Dir , opt . ReadOnly ) if err != nil { return nil , err } defer func ( ) { if manifestFile != nil { _ = manifestFile . close ( ) } } ( ) db = & DB { imm : make ( [ ] * skl . Skiplist , 0 , opt . NumMemtables ) , flushChan : make ( chan flushTask , opt . NumMemtables ) , writeCh : make ( chan * request , kvWriteChCapacity ) , opt : opt , manifest : manifestFile , elog : trace . NewEventLog ( " " , " " ) , dirLockGuard : dirLockGuard , valueDirGuard : valueDirLockGuard , orc : newOracle ( opt ) , } // Calculate initial size. db . calculateSize ( ) db . closers . updateSize = y . NewCloser ( 1 ) go db . updateSize ( db . closers . updateSize ) db . mt = skl . NewSkiplist ( arenaSize ( opt ) ) // newLevelsController potentially loads files in directory. if db . lc , err = newLevelsController ( db , & manifest ) ; err != nil { return nil , err } if ! opt . ReadOnly { db . closers . compactors = y . NewCloser ( 1 ) db . lc . startCompact ( db . closers . compactors ) db . closers . memtable = y . NewCloser ( 1 ) go db . flushMemtable ( db . closers . memtable ) // Need levels controller to be up. } headKey := y . KeyWithTs ( head , math . MaxUint64 ) // Need to pass with timestamp, lsm get removes the last 8 bytes and compares key vs , err := db . get ( headKey ) if err != nil { return nil , errors . Wrap ( err , " " ) } db . orc . nextTxnTs = vs . Version var vptr valuePointer if len ( vs . Value ) > 0 { vptr . Decode ( vs . Value ) } replayCloser := y . NewCloser ( 1 ) go db . doWrites ( replayCloser ) if err = db . vlog . open ( db , vptr , db . replayFunction ( ) ) ; err != nil { return db , err } replayCloser . SignalAndWait ( ) // Wait for replay to be applied first. // Let's advance nextTxnTs to one more than whatever we observed via // replaying the logs. db . orc . txnMark . Done ( db . orc . nextTxnTs ) // In normal mode, we must update readMark so older versions of keys can be removed during // compaction when run in offline mode via the flatten tool. db . orc . readMark . Done ( db . orc . nextTxnTs ) db . orc . nextTxnTs ++ db . writeCh = make ( chan * request , kvWriteChCapacity ) db . closers . writes = y . NewCloser ( 1 ) go db . doWrites ( db . closers . writes ) db . closers . valueGC = y . NewCloser ( 1 ) go db . vlog . waitOnGC ( db . closers . valueGC ) valueDirLockGuard = nil dirLockGuard = nil manifestFile = nil return db , nil }
// Open returns a new DB object.
Open returns a new DB object .