Line data    Source code

       1             : // Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
       2             : // of this source code is governed by a BSD-style license that can be found in
       3             : // the LICENSE file.
       4             : 
       5             : package pebble
       6             : 
       7             : import (
       8             :         "sync"
       9             : 
      10             :         "github.com/cockroachdb/errors"
      11             : )
      12             : 
      13             : type logRecycler struct {
      14             :         // The maximum number of log files to maintain for recycling.
      15             :         limit int
      16             : 
      17             :         // The minimum log number that is allowed to be recycled. Log numbers smaller
      18             :         // than this will be subject to immediate deletion. This is used to prevent
      19             :         // recycling a log written by a previous instance of the DB which may not
      20             :         // have had log recycling enabled. If that previous instance of the DB was
      21             :         // RocksDB, the old non-recyclable log record headers will be present.
      22             :         minRecycleLogNum FileNum
      23             : 
      24             :         mu struct {
      25             :                 sync.Mutex
      26             :                 logs      []fileInfo
      27             :                 maxLogNum FileNum
      28             :         }
      29             : }
      30             : 
      31             : // add attempts to recycle the log file specified by logInfo. Returns true if
      32             : // the log file should not be deleted (i.e. the log is being recycled), and
      33             : // false otherwise.
      34           1 : func (r *logRecycler) add(logInfo fileInfo) bool {
      35           1 :         if logInfo.fileNum.FileNum() < r.minRecycleLogNum {
      36           1 :                 return false
      37           1 :         }
      38             : 
      39           1 :         r.mu.Lock()
      40           1 :         defer r.mu.Unlock()
      41           1 : 
      42           1 :         if logInfo.fileNum.FileNum() <= r.mu.maxLogNum {
      43           1 :                 // The log file number was already considered for recycling. Don't consider
      44           1 :                 // it again. This avoids a race between adding the same log file for
      45           1 :                 // recycling multiple times, and removing the log file for actual
      46           1 :                 // reuse. Note that we return true because the log was already considered
      47           1 :                 // for recycling and either it was deleted on the previous attempt (which
      48           1 :                 // means we shouldn't get here) or it was recycled and thus the file
      49           1 :                 // shouldn't be deleted.
      50           1 :                 return true
      51           1 :         }
      52           1 :         r.mu.maxLogNum = logInfo.fileNum.FileNum()
      53           1 :         if len(r.mu.logs) >= r.limit {
      54           1 :                 return false
      55           1 :         }
      56           1 :         r.mu.logs = append(r.mu.logs, logInfo)
      57           1 :         return true
      58             : }
      59             : 
      60             : // peek returns the log at the head of the recycling queue, or the zero value
      61             : // fileInfo and false if the queue is empty.
      62           2 : func (r *logRecycler) peek() (fileInfo, bool) {
      63           2 :         r.mu.Lock()
      64           2 :         defer r.mu.Unlock()
      65           2 : 
      66           2 :         if len(r.mu.logs) == 0 {
      67           2 :                 return fileInfo{}, false
      68           2 :         }
      69           1 :         return r.mu.logs[0], true
      70             : }
      71             : 
      72           2 : func (r *logRecycler) stats() (count int, size uint64) {
      73           2 :         r.mu.Lock()
      74           2 :         defer r.mu.Unlock()
      75           2 :         count = len(r.mu.logs)
      76           2 :         for i := 0; i < count; i++ {
      77           1 :                 size += r.mu.logs[i].fileSize
      78           1 :         }
      79           2 :         return count, size
      80             : }
      81             : 
      82             : // pop removes the log number at the head of the recycling queue, enforcing
      83             : // that it matches the specified logNum. An error is returned of the recycling
      84             : // queue is empty or the head log number does not match the specified one.
      85           1 : func (r *logRecycler) pop(logNum FileNum) error {
      86           1 :         r.mu.Lock()
      87           1 :         defer r.mu.Unlock()
      88           1 : 
      89           1 :         if len(r.mu.logs) == 0 {
      90           1 :                 return errors.New("pebble: log recycler empty")
      91           1 :         }
      92           1 :         if r.mu.logs[0].fileNum.FileNum() != logNum {
      93           1 :                 return errors.Errorf("pebble: log recycler invalid %d vs %d", errors.Safe(logNum), errors.Safe(fileInfoNums(r.mu.logs)))
      94           1 :         }
      95           1 :         r.mu.logs = r.mu.logs[1:]
      96           1 :         return nil
      97             : }
      98             : 
      99           1 : func fileInfoNums(finfos []fileInfo) []FileNum {
     100           1 :         if len(finfos) == 0 {
     101           1 :                 return nil
     102           1 :         }
     103           1 :         nums := make([]FileNum, len(finfos))
     104           1 :         for i := range finfos {
     105           1 :                 nums[i] = finfos[i].fileNum.FileNum()
     106           1 :         }
     107           1 :         return nums
     108             : }