1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
13 "git.curoverse.com/arvados.git/sdk/go/arvados"
14 "git.curoverse.com/arvados.git/sdk/go/manifest"
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().
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
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")
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)
39 return nil, ErrNoManifest
41 m := manifest.Manifest{Text: mText}
42 return kc.ManifestFileReader(m, filename)
45 func (kc *KeepClient) ManifestFileReader(m manifest.Manifest, filename string) (arvados.File, error) {
46 return (&arvados.Collection{ManifestText: m.Text}).FileSystem(nil, kc).OpenFile(filename, os.O_RDONLY, 0)
51 segments []*manifest.FileSegment
52 size int64 // total file size
53 offset int64 // current read offset
55 // current/latest segment accessed -- might or might not match pos
56 seg *manifest.FileSegment
57 segStart int64 // position of segment relative to file
59 segNext []*manifest.FileSegment
63 // Close implements io.Closer.
64 func (f *file) Close() error {
71 // Read implements io.Reader.
72 func (f *file) Read(buf []byte) (int, error) {
73 if f.seg == nil || f.offset < f.segStart || f.offset >= f.segStart+int64(f.seg.Len) {
74 // f.seg does not cover the current read offset
75 // (f.pos). Iterate over f.segments to find the one
80 f.segNext = f.segments
81 for len(f.segNext) > 0 {
83 f.segNext = f.segNext[1:]
84 segEnd := f.segStart + int64(seg.Len)
85 if segEnd > f.offset {
91 f.readaheadDone = false
97 data, err := f.kc.cache().Get(f.kc, f.seg.Locator)
101 if len(data) < f.seg.Offset+f.seg.Len {
102 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)
104 f.segData = data[f.seg.Offset : f.seg.Offset+f.seg.Len]
106 // dataOff and dataLen denote a portion of f.segData
107 // corresponding to a portion of the file at f.offset.
108 dataOff := int(f.offset - f.segStart)
109 dataLen := f.seg.Len - dataOff
111 if !f.readaheadDone && len(f.segNext) > 0 && f.offset >= 1048576 && dataOff+dataLen > len(f.segData)/16 {
112 // If we have already read more than just the first
113 // few bytes of this file, and we have already
114 // consumed a noticeable portion of this segment, and
115 // there's more data for this file in the next segment
116 // ... then there's a good chance we are going to need
117 // the data for that next segment soon. Start getting
118 // it into the cache now.
119 go f.kc.cache().Get(f.kc, f.segNext[0].Locator)
120 f.readaheadDone = true
127 copy(buf[:n], f.segData[dataOff:dataOff+n])
132 // Seek implements io.Seeker.
133 func (f *file) Seek(offset int64, whence int) (int64, error) {
139 want = f.offset + offset
141 want = f.size + offset
143 return f.offset, fmt.Errorf("invalid whence %d", whence)
146 return f.offset, fmt.Errorf("attempted seek to %d", want)
155 // Size returns the file size in bytes.
156 func (f *file) Size() int64 {
160 func (f *file) load(m manifest.Manifest, path string) error {
163 for seg := range m.FileSegmentIterByName(path) {
164 f.segments = append(f.segments, seg)
165 f.size += int64(seg.Len)
167 if f.segments == nil {
168 return os.ErrNotExist