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 wal
       6             : 
       7             : import (
       8             :         "sync"
       9             : 
      10             :         "github.com/cockroachdb/errors"
      11             :         "github.com/cockroachdb/pebble/internal/base"
      12             : )
      13             : 
      14             : // TODO(sumeer): hide LogRecycler once rest of Pebble is using wal.Manager.
      15             : 
      16             : // LogRecycler recycles WAL log files. It holds a set of log file numbers that
      17             : // are available for reuse. Writing to a recycled log file is faster than to a
      18             : // new log file on some common filesystems (xfs, and ext3/4) due to avoiding
      19             : // metadata updates.
      20             : type LogRecycler struct {
      21             :         // The maximum number of log files to maintain for recycling.
      22             :         limit int
      23             : 
      24             :         // The minimum log number that is allowed to be recycled. Log numbers smaller
      25             :         // than this will be subject to immediate deletion. This is used to prevent
      26             :         // recycling a log written by a previous instance of the DB which may not
      27             :         // have had log recycling enabled. If that previous instance of the DB was
      28             :         // RocksDB, the old non-recyclable log record headers will be present.
      29             :         minRecycleLogNum base.DiskFileNum
      30             : 
      31             :         mu struct {
      32             :                 sync.Mutex
      33             :                 logs      []base.FileInfo
      34             :                 maxLogNum base.DiskFileNum
      35             :         }
      36             : }
      37             : 
      38             : // Init initialized the LogRecycler.
      39           2 : func (r *LogRecycler) Init(maxNumLogFiles int) {
      40           2 :         r.limit = maxNumLogFiles
      41           2 : }
      42             : 
      43             : // MinRecycleLogNum returns the current minimum log number that is allowed to
      44             : // be recycled.
      45           2 : func (r *LogRecycler) MinRecycleLogNum() NumWAL {
      46           2 :         return NumWAL(r.minRecycleLogNum)
      47           2 : }
      48             : 
      49             : // SetMinRecycleLogNum sets the minimum log number that is allowed to be
      50             : // recycled.
      51           2 : func (r *LogRecycler) SetMinRecycleLogNum(n NumWAL) {
      52           2 :         r.minRecycleLogNum = base.DiskFileNum(n)
      53           2 : }
      54             : 
      55             : // Add attempts to recycle the log file specified by logInfo. Returns true if
      56             : // the log file should not be deleted (i.e. the log is being recycled), and
      57             : // false otherwise.
      58           1 : func (r *LogRecycler) Add(logInfo base.FileInfo) bool {
      59           1 :         if logInfo.FileNum < r.minRecycleLogNum {
      60           1 :                 return false
      61           1 :         }
      62             : 
      63           1 :         r.mu.Lock()
      64           1 :         defer r.mu.Unlock()
      65           1 : 
      66           1 :         if logInfo.FileNum <= r.mu.maxLogNum {
      67           1 :                 // The log file number was already considered for recycling. Don't consider
      68           1 :                 // it again. This avoids a race between adding the same log file for
      69           1 :                 // recycling multiple times, and removing the log file for actual
      70           1 :                 // reuse. Note that we return true because the log was already considered
      71           1 :                 // for recycling and either it was deleted on the previous attempt (which
      72           1 :                 // means we shouldn't get here) or it was recycled and thus the file
      73           1 :                 // shouldn't be deleted.
      74           1 :                 return true
      75           1 :         }
      76           1 :         r.mu.maxLogNum = logInfo.FileNum
      77           1 :         if len(r.mu.logs) >= r.limit {
      78           1 :                 return false
      79           1 :         }
      80           1 :         r.mu.logs = append(r.mu.logs, logInfo)
      81           1 :         return true
      82             : }
      83             : 
      84             : // Peek returns the log at the head of the recycling queue, or the zero value
      85             : // fileInfo and false if the queue is empty.
      86           2 : func (r *LogRecycler) Peek() (base.FileInfo, bool) {
      87           2 :         r.mu.Lock()
      88           2 :         defer r.mu.Unlock()
      89           2 : 
      90           2 :         if len(r.mu.logs) == 0 {
      91           2 :                 return base.FileInfo{}, false
      92           2 :         }
      93           1 :         return r.mu.logs[0], true
      94             : }
      95             : 
      96             : // Stats return current stats.
      97           2 : func (r *LogRecycler) Stats() (count int, size uint64) {
      98           2 :         r.mu.Lock()
      99           2 :         defer r.mu.Unlock()
     100           2 :         count = len(r.mu.logs)
     101           2 :         for i := 0; i < count; i++ {
     102           1 :                 size += r.mu.logs[i].FileSize
     103           1 :         }
     104           2 :         return count, size
     105             : }
     106             : 
     107             : // Pop removes the log number at the head of the recycling queue, enforcing
     108             : // that it matches the specified logNum. An error is returned of the recycling
     109             : // queue is empty or the head log number does not match the specified one.
     110           1 : func (r *LogRecycler) Pop(logNum base.DiskFileNum) error {
     111           1 :         r.mu.Lock()
     112           1 :         defer r.mu.Unlock()
     113           1 : 
     114           1 :         if len(r.mu.logs) == 0 {
     115           1 :                 return errors.New("pebble: log recycler empty")
     116           1 :         }
     117           1 :         if r.mu.logs[0].FileNum != logNum {
     118           1 :                 return errors.Errorf("pebble: log recycler invalid %d vs %v", logNum, errors.Safe(fileInfoNums(r.mu.logs)))
     119           1 :         }
     120           1 :         r.mu.logs = r.mu.logs[1:]
     121           1 :         return nil
     122             : }
     123             : 
     124             : // LogNumsForTesting returns the current set of recyclable logs.
     125           1 : func (r *LogRecycler) LogNumsForTesting() []base.DiskFileNum {
     126           1 :         r.mu.Lock()
     127           1 :         defer r.mu.Unlock()
     128           1 :         return fileInfoNums(r.mu.logs)
     129           1 : }
     130             : 
     131           1 : func (r *LogRecycler) maxLogNumForTesting() base.DiskFileNum {
     132           1 :         r.mu.Lock()
     133           1 :         defer r.mu.Unlock()
     134           1 :         return r.mu.maxLogNum
     135           1 : }
     136             : 
     137           1 : func fileInfoNums(finfos []base.FileInfo) []base.DiskFileNum {
     138           1 :         if len(finfos) == 0 {
     139           1 :                 return nil
     140           1 :         }
     141           1 :         nums := make([]base.DiskFileNum, len(finfos))
     142           1 :         for i := range finfos {
     143           1 :                 nums[i] = finfos[i].FileNum
     144           1 :         }
     145           1 :         return nums
     146             : }