sdk/go/keepclient/collectionreader.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package keepclient
   6
   7 import (
   8         "errors"
   9         "fmt"
  10         "io"
  11         "os"
  12
  13         "git.curoverse.com/arvados.git/sdk/go/arvados"
  14         "git.curoverse.com/arvados.git/sdk/go/manifest"
  15 )
  16
  17 const (
  18         // After reading a data block from Keep, cfReader slices it up
  19         // and sends the slices to a buffered channel to be consumed
  20         // by the caller via Read().
  21         //
  22         // dataSliceSize is the maximum size of the slices, and
  23         // therefore the maximum number of bytes that will be returned
  24         // by a single call to Read().
  25         dataSliceSize = 1 << 20
  26 )
  27
  28 // ErrNoManifest indicates the given collection has no manifest
  29 // information (e.g., manifest_text was excluded by a "select"
  30 // parameter when retrieving the collection record).
  31 var ErrNoManifest = errors.New("Collection has no manifest")
  32
  33 // CollectionFileReader returns a Reader that reads content from a single file
  34 // in the collection. The filename must be relative to the root of the
  35 // collection.  A leading prefix of "/" or "./" in the filename is ignored.
  36 func (kc *KeepClient) CollectionFileReader(collection map[string]interface{}, filename string) (arvados.File, error) {
  37         mText, ok := collection["manifest_text"].(string)
  38         if !ok {
  39                 return nil, ErrNoManifest
  40         }
  41         m := manifest.Manifest{Text: mText}
  42         return kc.ManifestFileReader(m, filename)
  43 }
  44
  45 func (kc *KeepClient) ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error) {
  46         fs, err := (&arvados.Collection{ManifestText: m.Text}).FileSystem(nil, kc)
  47         if err != nil {
  48                 return nil, err
  49         }
  50         return fs.OpenFile(filename, os.O_RDONLY, 0)
  51 }
  52
  53 type file struct {
  54         kc       *KeepClient
  55         segments []*manifest.FileSegment
  56         size     int64 // total file size
  57         offset   int64 // current read offset
  58
  59         // current/latest segment accessed -- might or might not match pos
  60         seg           *manifest.FileSegment
  61         segStart      int64 // position of segment relative to file
  62         segData       []byte
  63         segNext       []*manifest.FileSegment
  64         readaheadDone bool
  65 }
  66
  67 // Close implements io.Closer.
  68 func (f *file) Close() error {
  69         f.kc = nil
  70         f.segments = nil
  71         f.segData = nil
  72         return nil
  73 }
  74
  75 // Read implements io.Reader.
  76 func (f *file) Read(buf []byte) (int, error) {
  77         if f.seg == nil || f.offset < f.segStart || f.offset >= f.segStart+int64(f.seg.Len) {
  78                 // f.seg does not cover the current read offset
  79                 // (f.pos).  Iterate over f.segments to find the one
  80                 // that does.
  81                 f.seg = nil
  82                 f.segStart = 0
  83                 f.segData = nil
  84                 f.segNext = f.segments
  85                 for len(f.segNext) > 0 {
  86                         seg := f.segNext[0]
  87                         f.segNext = f.segNext[1:]
  88                         segEnd := f.segStart + int64(seg.Len)
  89                         if segEnd > f.offset {
  90                                 f.seg = seg
  91                                 break
  92                         }
  93                         f.segStart = segEnd
  94                 }
  95                 f.readaheadDone = false
  96         }
  97         if f.seg == nil {
  98                 return 0, io.EOF
  99         }
 100         if f.segData == nil {
 101                 data, err := f.kc.cache().Get(f.kc, f.seg.Locator)
 102                 if err != nil {
 103                         return 0, err
 104                 }
 105                 if len(data) < f.seg.Offset+f.seg.Len {
 106                         return 0, fmt.Errorf("invalid segment (offset %d len %d) in %d-byte block %s", f.seg.Offset, f.seg.Len, len(data), f.seg.Locator)
 107                 }
 108                 f.segData = data[f.seg.Offset : f.seg.Offset+f.seg.Len]
 109         }
 110         // dataOff and dataLen denote a portion of f.segData
 111         // corresponding to a portion of the file at f.offset.
 112         dataOff := int(f.offset - f.segStart)
 113         dataLen := f.seg.Len - dataOff
 114
 115         if !f.readaheadDone && len(f.segNext) > 0 && f.offset >= 1048576 && dataOff+dataLen > len(f.segData)/16 {
 116                 // If we have already read more than just the first
 117                 // few bytes of this file, and we have already
 118                 // consumed a noticeable portion of this segment, and
 119                 // there's more data for this file in the next segment
 120                 // ... then there's a good chance we are going to need
 121                 // the data for that next segment soon. Start getting
 122                 // it into the cache now.
 123                 go f.kc.cache().Get(f.kc, f.segNext[0].Locator)
 124                 f.readaheadDone = true
 125         }
 126
 127         n := len(buf)
 128         if n > dataLen {
 129                 n = dataLen
 130         }
 131         copy(buf[:n], f.segData[dataOff:dataOff+n])
 132         f.offset += int64(n)
 133         return n, nil
 134 }
 135
 136 // Seek implements io.Seeker.
 137 func (f *file) Seek(offset int64, whence int) (int64, error) {
 138         var want int64
 139         switch whence {
 140         case io.SeekStart:
 141                 want = offset
 142         case io.SeekCurrent:
 143                 want = f.offset + offset
 144         case io.SeekEnd:
 145                 want = f.size + offset
 146         default:
 147                 return f.offset, fmt.Errorf("invalid whence %d", whence)
 148         }
 149         if want < 0 {
 150                 return f.offset, fmt.Errorf("attempted seek to %d", want)
 151         }
 152         if want > f.size {
 153                 want = f.size
 154         }
 155         f.offset = want
 156         return f.offset, nil
 157 }
 158
 159 // Size returns the file size in bytes.
 160 func (f *file) Size() int64 {
 161         return f.size
 162 }
 163
 164 func (f *file) load(m manifest.Manifest, path string) error {
 165         f.segments = nil
 166         f.size = 0
 167         for seg := range m.FileSegmentIterByName(path) {
 168                 f.segments = append(f.segments, seg)
 169                 f.size += int64(seg.Len)
 170         }
 171         if f.segments == nil {
 172                 return os.ErrNotExist
 173         }
 174         return nil
 175 }