gibheer
/
bleve
0
0
Fork 0
Code

scorch tracks zap files that can't be removed yet 

A race & solution found by Marty Schoch... consider a case when the
merger might grab a nextSegmentID, like 4, but takes awhile to
complete.  Meanwhile, the persister grabs the nextSegmentID of 5, but
finishes its persistence work fast, and then loops to cleanup any old
files.  The simple approach of checking a "highest segment ID" of 5 is
wrong now, because the deleter now thinks that segment 4's zap file is
(incorrectly) ok to delete.

The solution in this commit is to track an ephemeral map of filenames
which are ineligibleForRemoval, because they're still being written
(by the merger) and haven't been fully incorporated into the rootBolt
yet.

The merger adds to that ineligibleForRemoval map as it starts a merged
zap file, the persister cleans up entries from that map when it
persists zap filenames into the rootBolt, and the deleter (part of the
persister's loop) consults the map before performing any actual zap
file deletions.
This commit is contained in:
Steve Yen 2017-12-14 10:49:33 -08:00
parent bd742caf65
commit 2be5eb4427
3 changed files with 60 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
index/scorch/merge.go
View File

@ -129,14 +129,17 @@ func (s *Scorch) planMergeAtSnapshot(ourSnapshot *IndexSnapshot) error {
}
}
filename := fmt.Sprintf("%08x.zap", newSegmentID)
filename := zapFileName(newSegmentID)
s.markIneligibleForRemoval(filename)
path := s.path + string(os.PathSeparator) + filename
newDocNums, err := zap.Merge(segmentsToMerge, docsToDrop, path, 1024)
if err != nil {
s.unmarkIneligibleForRemoval(filename)
return fmt.Errorf("merging failed: %v", err)
}
segment, err := zap.Open(path)
if err != nil {
s.unmarkIneligibleForRemoval(filename)
return err
}
sm := &segmentMerge{

39
index/scorch/persister.go
View File

@ -34,8 +34,6 @@ import (
type notificationChan chan struct{}
func (s *Scorch) persisterLoop() {
s.removeOldData(true)
var notify notificationChan
var lastPersistedEpoch uint64
OUTER:
@ -52,7 +50,6 @@ OUTER:
ourSnapshot.AddRef()
s.rootLock.RUnlock()
//for ourSnapshot.epoch != lastPersistedEpoch {
if ourSnapshot.epoch != lastPersistedEpoch {
// lets get started
err := s.persistSnapshot(ourSnapshot)
@ -110,7 +107,7 @@ OUTER:
// woken up, next loop should pick up work
}
}
s.removeOldData(false)
s.removeOldData()
}
s.asyncTasks.Done()
}
@ -159,6 +156,7 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
}
}
var filenames []string
newSegmentPaths := make(map[uint64]string)
// first ensure that each segment in this snapshot has been persisted
@ -171,7 +169,7 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
switch seg := segmentSnapshot.segment.(type) {
case *mem.Segment:
// need to persist this to disk
filename := fmt.Sprintf("%08x.zap", segmentSnapshot.id)
filename := zapFileName(segmentSnapshot.id)
path := s.path + string(os.PathSeparator) + filename
err2 := zap.PersistSegment(seg, path, 1024)
if err2 != nil {
@ -182,6 +180,7 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
if err != nil {
return err
}
filenames = append(filenames, filename)
case *zap.Segment:
path := seg.Path()
filename := strings.TrimPrefix(path, s.path+string(os.PathSeparator))
@ -189,6 +188,7 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
if err != nil {
return err
}
filenames = append(filenames, filename)
default:
return fmt.Errorf("unknown segment type: %T", seg)
}
@ -255,6 +255,9 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
for k, v := range s.root.internal {
newIndexSnapshot.internal[k] = v
}
for _, filename := range filenames {
delete(s.ineligibleForRemoval, filename)
}
rootPrev := s.root
s.root = newIndexSnapshot
s.rootLock.Unlock()
@ -272,6 +275,10 @@ func (s *Scorch) persistSnapshot(snapshot *IndexSnapshot) error {
return nil
}
func zapFileName(epoch uint64) string {
return fmt.Sprintf("%012x.zap", epoch)
}
// bolt snapshot code
var boltSnapshotsBucket = []byte{'s'}
@ -409,16 +416,16 @@ func (p uint64Descending) Len() int { return len(p) }
func (p uint64Descending) Less(i, j int) bool { return p[i] > p[j] }
func (p uint64Descending) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (s *Scorch) removeOldData(force bool) {
func (s *Scorch) removeOldData() {
removed, err := s.removeOldBoltSnapshots()
if err != nil {
log.Printf("got err removing old bolt snapshots: %v", err)
}
if force || removed > 0 {
if removed > 0 {
err = s.removeOldZapFiles()
if err != nil {
log.Printf("go err removing old zap files: %v", err)
log.Printf("got err removing old zap files: %v", err)
}
}
}
@ -478,7 +485,7 @@ func (s *Scorch) removeOldBoltSnapshots() (numRemoved int, err error) {
// Removes any *.zap files which aren't listed in the rootBolt.
func (s *Scorch) removeOldZapFiles() error {
liveFileNames, highestName, err := s.loadZapFileNames()
liveFileNames, err := s.loadZapFileNames()
if err != nil {
return err
}
@ -488,10 +495,12 @@ func (s *Scorch) removeOldZapFiles() error {
return err
}
s.rootLock.RLock()
for _, finfo := range currFileInfos {
fname := finfo.Name()
if filepath.Ext(fname) == ".zap" {
if _, exists := liveFileNames[fname]; !exists && fname < highestName {
if _, exists := liveFileNames[fname]; !exists && !s.ineligibleForRemoval[fname] {
err := os.Remove(s.path + string(os.PathSeparator) + fname)
if err != nil {
log.Printf("got err removing file: %s, err: %v", fname, err)
@ -500,13 +509,14 @@ func (s *Scorch) removeOldZapFiles() error {
}
}
s.rootLock.RUnlock()
return nil
}
// Returns the *.zap file names that are listed in the rootBolt.
func (s *Scorch) loadZapFileNames() (map[string]struct{}, string, error) {
func (s *Scorch) loadZapFileNames() (map[string]struct{}, error) {
rv := map[string]struct{}{}
var highest string
err := s.rootBolt.View(func(tx *bolt.Tx) error {
snapshots := tx.Bucket(boltSnapshotsBucket)
if snapshots == nil {
@ -532,14 +542,11 @@ func (s *Scorch) loadZapFileNames() (map[string]struct{}, string, error) {
continue
}
pathString := string(pathBytes)
if pathString > highest {
highest = pathString
}
rv[string(pathString)] = struct{}{}
}
}
return nil
})
return rv, highest, err
return rv, err
}

41
index/scorch/scorch.go
View File

@ -60,16 +60,19 @@ type Scorch struct {
rootBolt *bolt.DB
asyncTasks sync.WaitGroup
eligibleForRemoval []uint64 // Index snapshot epoch's that are safe to GC.
eligibleForRemoval []uint64 // Index snapshot epochs that are safe to GC.
ineligibleForRemoval map[string]bool // Filenames that should not be GC'ed yet.
}
func NewScorch(storeName string, config map[string]interface{}, analysisQueue *index.AnalysisQueue) (index.Index, error) {
rv := &Scorch{
version: Version,
config: config,
analysisQueue: analysisQueue,
stats: &Stats{},
nextSnapshotEpoch: 1,
version: Version,
config: config,
analysisQueue: analysisQueue,
stats: &Stats{},
nextSnapshotEpoch: 1,
closeCh: make(chan struct{}),
ineligibleForRemoval: map[string]bool{},
}
rv.root = &IndexSnapshot{parent: rv, refs: 1}
ro, ok := config["read_only"].(bool)
@ -113,16 +116,24 @@ func (s *Scorch) Open() error {
// now see if there is any existing state to load
err = s.loadFromBolt()
if err != nil {
_ = s.Close()
return err
}
}
s.closeCh = make(chan struct{})
s.introductions = make(chan *segmentIntroduction)
s.merges = make(chan *segmentMerge)
s.introducerNotifier = make(chan notificationChan)
s.persisterNotifier = make(chan notificationChan)
if !s.readOnly && s.path != "" {
err := s.removeOldZapFiles() // Before persister or merger create any new files.
if err != nil {
_ = s.Close()
return err
}
}
s.asyncTasks.Add(1)
go s.mainLoop()
@ -145,7 +156,9 @@ func (s *Scorch) Close() (err error) {
if s.rootBolt != nil {
err = s.rootBolt.Close()
s.rootLock.Lock()
_ = s.root.DecRef()
if s.root != nil {
_ = s.root.DecRef()
}
s.root = nil
s.rootLock.Unlock()
}
@ -334,6 +347,18 @@ func (s *Scorch) AddEligibleForRemoval(epoch uint64) {
s.rootLock.Unlock()
}
func (s *Scorch) markIneligibleForRemoval(filename string) {
s.rootLock.Lock()
s.ineligibleForRemoval[filename] = true
s.rootLock.Unlock()
}
func (s *Scorch) unmarkIneligibleForRemoval(filename string) {
s.rootLock.Lock()
delete(s.ineligibleForRemoval, filename)
s.rootLock.Unlock()
}
func init() {
registry.RegisterIndexType(Name, NewScorch)
}