sdk/java: run positioned reads on one stream concurrently

pkg/chunk/cached_store.go

@@ -167,6 +167,41 @@ func (s *rSlice) ReadAt(ctx context.Context, page *Page, off int) (n int, err er
 	return len(p), nil
 }
 
+// ReadCachedAt reads only from the local cache; see CachedReader.
+func (s *rSlice) ReadCachedAt(p []byte, off int) (int, bool) {
+	if !s.store.conf.CacheEnabled() || off < 0 || off+len(p) > s.length {
+		return 0, false
+	}
+	got := 0
+	for got < len(p) {
+		indx := s.index(off)
+		boff := off % s.store.conf.BlockSize
+		l := min(len(p)-got, s.blockSize(indx)-boff)
+		key := s.key(indx)
+		start := time.Now()
+		r, err := s.store.bcache.load(key)
+		if err != nil {
+			return got, false
+		}
+		n, err := r.ReadAt(p[got:got+l], int64(boff))
+		if !s.store.conf.OSCache {
+			dropOSCache(r)
+		}
+		_ = r.Close()
+		if err != nil {
+			logger.Warnf("remove partial cached block %s: %d %s", key, n, err)
+			s.store.bcache.remove(key, false)
+			return got, false
+		}
+		s.store.cacheHits.Add(1)
+		s.store.cacheHitBytes.Add(float64(n))
+		s.store.cacheReadHist.Observe(time.Since(start).Seconds())
+		got += n
+		off += n
+	}
+	return got, true
+}
+
 func (s *rSlice) delete(indx int) error {
 	key := s.key(indx)
 	return s.store.delete(key)

pkg/chunk/cached_store_test.go

@@ -146,6 +146,87 @@ func TestStoreMemCache(t *testing.T) {
 		t.Fatalf("cache cnt %d used %d, expect both 0", cnt, used)
 	}
 }
+func TestReadCachedAt(t *testing.T) {
+	mem, _ := object.CreateStorage("mem", "", "", "", "")
+	conf := defaultConf
+	conf.CacheDir = "memory"
+	conf.CacheFullBlock = true
+	store := NewCachedStore(mem, conf, nil)
+	size := conf.BlockSize + 100 // two blocks
+	data := make([]byte, size)
+	for i := range data {
+		data[i] = byte(i * 3)
+	}
+	w := store.NewWriter(1, 0)
+	if _, err := w.WriteAt(data, 0); err != nil {
+		t.Fatalf("write: %s", err)
+	}
+	if err := w.Finish(size); err != nil {
+		t.Fatalf("finish: %s", err)
+	}
+	r := store.NewReader(1, size)
+	cr := r.(CachedReader)
+
+	// nothing cached yet
+	buf := make([]byte, 100)
+	if _, ok := cr.ReadCachedAt(buf, 0); ok {
+		t.Fatalf("read of uncached block should fail")
+	}
+	// populate the cache through the regular path
+	p := NewOffPage(size)
+	if n, err := r.ReadAt(ctx, p, 0); err != nil || n != size {
+		t.Fatalf("read: (%d,%s)", n, err)
+	}
+	p.Release()
+
+	// across the block boundary
+	off := conf.BlockSize - 50
+	if n, ok := cr.ReadCachedAt(buf, off); !ok || n != len(buf) {
+		t.Fatalf("cached read at %d: (%d,%v)", off, n, ok)
+	}
+	if !bytes.Equal(buf, data[off:off+len(buf)]) {
+		t.Fatalf("cached read at %d: data mismatch", off)
+	}
+	// beyond the slice
+	if _, ok := cr.ReadCachedAt(buf, size-50); ok {
+		t.Fatalf("read beyond the slice should fail")
+	}
+	// evicting one block fails only the reads that need it
+	if err := store.EvictCache(1, uint32(size), []Range{{0, 1}}); err != nil {
+		t.Fatalf("evict: %s", err)
+	}
+	if _, ok := cr.ReadCachedAt(buf, off); ok {
+		t.Fatalf("read of evicted block should fail")
+	}
+	if n, ok := cr.ReadCachedAt(buf, conf.BlockSize); !ok || n != len(buf) {
+		t.Fatalf("cached read at %d: (%d,%v)", conf.BlockSize, n, ok)
+	}
+	if !bytes.Equal(buf, data[conf.BlockSize:conf.BlockSize+len(buf)]) {
+		t.Fatalf("cached read at %d: data mismatch", conf.BlockSize)
+	}
+
+	// a cached block shorter than the data it claims is dropped
+	key := sliceForRead(1, size, store.(*cachedStore)).key(1)
+	bcache := store.(*cachedStore).bcache
+	bcache.remove(key, false)
+	short := NewOffPage(10)
+	bcache.cache(key, short, true, false)
+	short.Release()
+	if _, ok := cr.ReadCachedAt(buf, conf.BlockSize); ok {
+		t.Fatalf("read of a partial cached block should fail")
+	}
+	if _, ok := bcache.exist(key); ok {
+		t.Fatalf("partial cached block should be removed")
+	}
+
+	// cache disabled
+	conf.CacheSize = 0
+	store = NewCachedStore(mem, conf, nil)
+	if _, ok := store.NewReader(1, size).(CachedReader).ReadCachedAt(buf, conf.BlockSize); ok {
+		t.Fatalf("read with cache disabled should fail")
+	}
+}
+
 func TestStoreCompressed(t *testing.T) {
 	mem, _ := object.CreateStorage("mem", "", "", "", "")
 	conf := defaultConf

pkg/chunk/chunk.go

@@ -27,6 +27,14 @@ type Reader interface {
 	ReadAt(ctx context.Context, p *Page, off int) (int, error)
 }
 
+// CachedReader is implemented by readers that can serve a range from the
+// local cache alone, without contacting the object storage.
+type CachedReader interface {
+	// ReadCachedAt fills p from the local cache. It returns false as soon as
+	// any part of the range is not cached; the caller then falls back to ReadAt.
+	ReadCachedAt(p []byte, off int) (int, bool)
+}
+
 type Writer interface {
 	io.WriterAt
 	ID() uint64

pkg/fs/fs.go

@@ -174,6 +174,8 @@ type File struct {
 	dircache []os.FileInfo
 	entries  []*meta.Entry
 	data     []byte
+	preads   sync.WaitGroup // positioned reads running outside the lock
+	closed   bool
 }
 
 func NewFileSystem(conf *vfs.Config, m meta.Meta, d chunk.ChunkStore, registry *prometheus.Registry) (*FileSystem, error) {
@@ -1336,37 +1338,62 @@ func (f *File) Pread(ctx meta.Context, b []byte, offset int64) (n int, err error
 	defer task.End()
 	l := vfs.NewLogContext(ctx)
 	defer func() { f.fs.log(l, "Pread (%s,%d,%d): (%d,%s)", f.path, len(b), offset, n, errstr(err)) }()
+	// Hold the lock only while touching the File's own state; the read itself
+	// may wait on the object storage, and positioned reads on one file must be
+	// able to overlap there (vfs.FileReader is safe for concurrent Read/Close).
 	f.Lock()
-	defer f.Unlock()
-	n, err = f.pread(ctx, b, offset)
-	return
+	rdata, b, n, err := f.preparePread(ctx, b, offset)
+	if rdata != nil {
+		f.preads.Add(1)
+		defer f.preads.Done()
+	}
+	f.Unlock()
+	if rdata == nil {
+		return
+	}
+	return f.readData(ctx, rdata, b, offset)
 }
 
 func (f *File) pread(ctx meta.Context, b []byte, offset int64) (n int, err error) {
+	rdata, b, n, err := f.preparePread(ctx, b, offset)
+	if rdata == nil {
+		return
+	}
+	return f.readData(ctx, rdata, b, offset)
+}
+
+// preparePread must be called with f locked. It returns a nil reader when the
+// request is already complete (EOF, in-memory data, or a flush error).
+func (f *File) preparePread(ctx meta.Context, b []byte, offset int64) (rdata vfs.FileReader, buf []byte, n int, err error) {
+	if f.closed {
+		return nil, nil, 0, syscall.EBADF
+	}
 	if offset >= f.info.Size() {
-		return 0, io.EOF
+		return nil, nil, 0, io.EOF
 	}
 	if int64(len(b))+offset > f.info.Size() {
 		b = b[:f.info.Size()-offset]
 	}
 	if f.data != nil {
 		n := copy(b, f.data[offset:])
-		return n, nil
+		return nil, nil, n, nil
 	}
 	if f.wdata != nil {
 		eno := f.wdata.Flush(ctx)
 		if eno != 0 {
-			err = eno
-			return
+			return nil, nil, 0, eno
 		}
 	}
 	if f.rdata == nil {
 		f.rdata = f.fs.reader.Open(f.inode, uint64(f.info.Size()))
 	}
+	return f.rdata, b, 0, nil
+}
 
-	got, eno := f.rdata.Read(ctx, uint64(offset), b)
+func (f *File) readData(ctx meta.Context, rdata vfs.FileReader, b []byte, offset int64) (n int, err error) {
+	got, eno := rdata.Read(ctx, uint64(offset), b)
 	for eno == syscall.EAGAIN {
-		got, eno = f.rdata.Read(ctx, uint64(offset), b)
+		got, eno = rdata.Read(ctx, uint64(offset), b)
 	}
 	if eno != 0 {
 		err = eno
@@ -1474,6 +1501,10 @@ func (f *File) Close(ctx meta.Context) (err syscall.Errno) {
 	f.Lock()
 	defer f.Unlock()
 	if f.flags != 0 && !f.info.IsDir() {
+		// Later reads fail with EBADF instead of opening a new reader on a
+		// closed file; then drain the positioned reads already in flight.
+		f.closed = true
+		f.preads.Wait()
 		f.offset = 0
 		if f.rdata != nil {
 			rdata := f.rdata