Line data    Source code

       1             : // Copyright 2023 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 objstorageprovider
       6             : 
       7             : import (
       8             :         "context"
       9             :         "fmt"
      10             :         "os"
      11             :         "sync"
      12             : 
      13             :         "github.com/cockroachdb/pebble/internal/invariants"
      14             :         "github.com/cockroachdb/pebble/objstorage"
      15             :         "github.com/cockroachdb/pebble/vfs"
      16             : )
      17             : 
      18             : const fileMaxReadaheadSize = 256 * 1024 /* 256KB */
      19             : 
      20             : // fileReadable implements objstorage.Readable on top of a vfs.File.
      21             : //
      22             : // The implementation might use Prealloc and might reopen the file with
      23             : // SequentialReadsOption.
      24             : type fileReadable struct {
      25             :         file vfs.File
      26             :         size int64
      27             : 
      28             :         // The following fields are used to possibly open the file again using the
      29             :         // sequential reads option (see vfsReadHandle).
      30             :         filename string
      31             :         fs       vfs.FS
      32             : }
      33             : 
      34             : var _ objstorage.Readable = (*fileReadable)(nil)
      35             : 
      36           1 : func newFileReadable(file vfs.File, fs vfs.FS, filename string) (*fileReadable, error) {
      37           1 :         info, err := file.Stat()
      38           1 :         if err != nil {
      39           1 :                 return nil, err
      40           1 :         }
      41           1 :         r := &fileReadable{
      42           1 :                 file:     file,
      43           1 :                 size:     info.Size(),
      44           1 :                 filename: filename,
      45           1 :                 fs:       fs,
      46           1 :         }
      47           1 :         invariants.SetFinalizer(r, func(obj interface{}) {
      48           1 :                 if obj.(*fileReadable).file != nil {
      49           0 :                         fmt.Fprintf(os.Stderr, "Readable was not closed")
      50           0 :                         os.Exit(1)
      51           0 :                 }
      52             :         })
      53           1 :         return r, nil
      54             : }
      55             : 
      56             : // ReadAt is part of the objstorage.Readable interface.
      57           1 : func (r *fileReadable) ReadAt(_ context.Context, p []byte, off int64) error {
      58           1 :         n, err := r.file.ReadAt(p, off)
      59           1 :         if invariants.Enabled && err == nil && n != len(p) {
      60           0 :                 panic("short read")
      61             :         }
      62           1 :         return err
      63             : }
      64             : 
      65             : // Close is part of the objstorage.Readable interface.
      66           1 : func (r *fileReadable) Close() error {
      67           1 :         defer func() { r.file = nil }()
      68           1 :         return r.file.Close()
      69             : }
      70             : 
      71             : // Size is part of the objstorage.Readable interface.
      72           1 : func (r *fileReadable) Size() int64 {
      73           1 :         return r.size
      74           1 : }
      75             : 
      76             : // NewReadHandle is part of the objstorage.Readable interface.
      77           1 : func (r *fileReadable) NewReadHandle(_ context.Context) objstorage.ReadHandle {
      78           1 :         rh := readHandlePool.Get().(*vfsReadHandle)
      79           1 :         rh.r = r
      80           1 :         rh.rs = makeReadaheadState(fileMaxReadaheadSize)
      81           1 :         return rh
      82           1 : }
      83             : 
      84             : type vfsReadHandle struct {
      85             :         r  *fileReadable
      86             :         rs readaheadState
      87             : 
      88             :         // sequentialFile holds a file descriptor to the same underlying File,
      89             :         // except with fadvise(FADV_SEQUENTIAL) called on it to take advantage of
      90             :         // OS-level readahead. Once this is non-nil, the other variables in
      91             :         // readaheadState don't matter much as we defer to OS-level readahead.
      92             :         sequentialFile vfs.File
      93             : }
      94             : 
      95             : var _ objstorage.ReadHandle = (*vfsReadHandle)(nil)
      96             : 
      97             : var readHandlePool = sync.Pool{
      98           1 :         New: func() interface{} {
      99           1 :                 i := &vfsReadHandle{}
     100           1 :                 // Note: this is a no-op if invariants are disabled or race is enabled.
     101           1 :                 invariants.SetFinalizer(i, func(obj interface{}) {
     102           1 :                         if obj.(*vfsReadHandle).r != nil {
     103           0 :                                 fmt.Fprintf(os.Stderr, "ReadHandle was not closed")
     104           0 :                                 os.Exit(1)
     105           0 :                         }
     106             :                 })
     107           1 :                 return i
     108             :         },
     109             : }
     110             : 
     111             : // Close is part of the objstorage.ReadHandle interface.
     112           1 : func (rh *vfsReadHandle) Close() error {
     113           1 :         var err error
     114           1 :         if rh.sequentialFile != nil {
     115           1 :                 err = rh.sequentialFile.Close()
     116           1 :         }
     117           1 :         *rh = vfsReadHandle{}
     118           1 :         readHandlePool.Put(rh)
     119           1 :         return err
     120             : }
     121             : 
     122             : // ReadAt is part of the objstorage.ReadHandle interface.
     123           1 : func (rh *vfsReadHandle) ReadAt(_ context.Context, p []byte, offset int64) error {
     124           1 :         var n int
     125           1 :         var err error
     126           1 :         if rh.sequentialFile != nil {
     127           1 :                 // Use OS-level read-ahead.
     128           1 :                 n, err = rh.sequentialFile.ReadAt(p, offset)
     129           1 :         } else {
     130           1 :                 if readaheadSize := rh.rs.maybeReadahead(offset, int64(len(p))); readaheadSize > 0 {
     131           1 :                         if readaheadSize >= fileMaxReadaheadSize {
     132           1 :                                 // We've reached the maximum readahead size. Beyond this point, rely on
     133           1 :                                 // OS-level readahead.
     134           1 :                                 rh.switchToOSReadahead()
     135           1 :                         } else {
     136           1 :                                 _ = rh.r.file.Prefetch(offset, readaheadSize)
     137           1 :                         }
     138             :                 }
     139           1 :                 n, err = rh.r.file.ReadAt(p, offset)
     140             :         }
     141           1 :         if invariants.Enabled && err == nil && n != len(p) {
     142           0 :                 panic("short read")
     143             :         }
     144           1 :         return err
     145             : }
     146             : 
     147             : // SetupForCompaction is part of the objstorage.ReadHandle interface.
     148           1 : func (rh *vfsReadHandle) SetupForCompaction() {
     149           1 :         rh.switchToOSReadahead()
     150           1 : }
     151             : 
     152           1 : func (rh *vfsReadHandle) switchToOSReadahead() {
     153           1 :         if rh.sequentialFile != nil {
     154           0 :                 return
     155           0 :         }
     156             : 
     157             :         // TODO(radu): we could share the reopened file descriptor across multiple
     158             :         // handles.
     159           1 :         f, err := rh.r.fs.Open(rh.r.filename, vfs.SequentialReadsOption)
     160           1 :         if err == nil {
     161           1 :                 rh.sequentialFile = f
     162           1 :         }
     163             : }
     164             : 
     165             : // RecordCacheHit is part of the objstorage.ReadHandle interface.
     166           1 : func (rh *vfsReadHandle) RecordCacheHit(_ context.Context, offset, size int64) {
     167           1 :         if rh.sequentialFile != nil {
     168           1 :                 // Using OS-level readahead, so do nothing.
     169           1 :                 return
     170           1 :         }
     171           1 :         rh.rs.recordCacheHit(offset, size)
     172             : }
     173             : 
     174             : // TestingCheckMaxReadahead returns true if the ReadHandle has switched to
     175             : // OS-level read-ahead.
     176           1 : func TestingCheckMaxReadahead(rh objstorage.ReadHandle) bool {
     177           1 :         switch rh := rh.(type) {
     178           1 :         case *vfsReadHandle:
     179           1 :                 return rh.sequentialFile != nil
     180           1 :         case *PreallocatedReadHandle:
     181           1 :                 return rh.sequentialFile != nil
     182           0 :         default:
     183           0 :                 panic("unknown ReadHandle type")
     184             :         }
     185             : }
     186             : 
     187             : // PreallocatedReadHandle is used to avoid an allocation in NewReadHandle; see
     188             : // UsePreallocatedReadHandle.
     189             : type PreallocatedReadHandle struct {
     190             :         vfsReadHandle
     191             : }
     192             : 
     193             : // Close is part of the objstorage.ReadHandle interface.
     194           1 : func (rh *PreallocatedReadHandle) Close() error {
     195           1 :         var err error
     196           1 :         if rh.sequentialFile != nil {
     197           1 :                 err = rh.sequentialFile.Close()
     198           1 :         }
     199           1 :         rh.vfsReadHandle = vfsReadHandle{}
     200           1 :         return err
     201             : }
     202             : 
     203             : // UsePreallocatedReadHandle is equivalent to calling readable.NewReadHandle()
     204             : // but uses the existing storage of a PreallocatedReadHandle when possible
     205             : // (currently this happens if we are reading from a local file).
     206             : // The returned handle still needs to be closed.
     207             : func UsePreallocatedReadHandle(
     208             :         ctx context.Context, readable objstorage.Readable, rh *PreallocatedReadHandle,
     209           1 : ) objstorage.ReadHandle {
     210           1 :         if r, ok := readable.(*fileReadable); ok {
     211           1 :                 // See fileReadable.NewReadHandle.
     212           1 :                 rh.vfsReadHandle = vfsReadHandle{r: r}
     213           1 :                 return rh
     214           1 :         }
     215           1 :         return readable.NewReadHandle(ctx)
     216             : }