Add ReaderAt support to content store

Signed-off-by: Derek McGowan <derek@mcgstyle.net>

content/content.go

@@ -30,6 +30,7 @@ var (
 
 type Provider interface {
 	Reader(ctx context.Context, dgst digest.Digest) (io.ReadCloser, error)
+	ReaderAt(ctx context.Context, dgst digest.Digest) (io.ReaderAt, error)
 }
 
 type Ingester interface {

content/readerat.go

@@ -0,0 +1,26 @@
+package content
+
+import (
+	"io"
+	"os"
+)
+
+// readerat implements io.ReaderAt in a completely stateless manner by opening
+// the referenced file for each call to ReadAt.
+type readerAt struct {
+	f string
+}
+
+func (ra readerAt) ReadAt(p []byte, offset int64) (int, error) {
+	fp, err := os.Open(ra.f)
+	if err != nil {
+		return 0, err
+	}
+	defer fp.Close()
+
+	if _, err := fp.Seek(offset, io.SeekStart); err != nil {
+		return 0, err
+	}
+
+	return fp.Read(p)
+}

content/store.go

@@ -58,11 +58,7 @@ func (s *store) info(dgst digest.Digest, fi os.FileInfo) Info {
 	}
 }
 
-// Open returns an io.ReadCloser for the blob.
-//
-// TODO(stevvooe): This would work much better as an io.ReaderAt in practice.
-// Right now, we are doing type assertion to tease that out, but it won't scale
-// well.
+// Reader returns an io.ReadCloser for the blob.
 func (s *store) Reader(ctx context.Context, dgst digest.Digest) (io.ReadCloser, error) {
 	fp, err := os.Open(s.blobPath(dgst))
 	if err != nil {
@@ -75,6 +71,11 @@ func (s *store) Reader(ctx context.Context, dgst digest.Digest) (io.ReadCloser,
 	return fp, nil
 }
 
+// ReaderAt returns an io.ReaderAt for the blob.
+func (s *store) ReaderAt(ctx context.Context, dgst digest.Digest) (io.ReaderAt, error) {
+	return readerAt{f: s.blobPath(dgst)}, nil
+}
+
 // Delete removes a blob by its digest.
 //
 // While this is safe to do concurrently, safe exist-removal logic must hold

services/content/reader.go

@@ -1,7 +1,10 @@
 package content
 
 import (
+	"context"
+
 	contentapi "github.com/containerd/containerd/api/services/content"
+	digest "github.com/opencontainers/go-digest"
 )
 
 type remoteReader struct {
@@ -42,8 +45,38 @@ func (rr *remoteReader) Read(p []byte) (n int, err error) {
 	return
 }
 
-// TODO(stevvooe): Implemente io.ReaderAt.
-
 func (rr *remoteReader) Close() error {
 	return rr.client.CloseSend()
 }
+
+type remoteReaderAt struct {
+	ctx    context.Context
+	digest digest.Digest
+	client contentapi.ContentClient
+}
+
+func (ra *remoteReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
+	rr := &contentapi.ReadRequest{
+		Digest: ra.digest,
+		Offset: off,
+		Size_:  int64(len(p)),
+	}
+	rc, err := ra.client.Read(ra.ctx, rr)
+	if err != nil {
+		return 0, err
+	}
+
+	for len(p) > 0 {
+		var resp *contentapi.ReadResponse
+		// fill our buffer up until we can fill p.
+		resp, err = rc.Recv()
+		if err != nil {
+			return n, err
+		}
+
+		copied := copy(p, resp.Data)
+		n += copied
+		p = p[copied:]
+	}
+	return n, nil
+}

services/content/store.go

@@ -85,6 +85,14 @@ func (rs *remoteStore) Reader(ctx context.Context, dgst digest.Digest) (io.ReadC
 	}, nil
 }
 
+func (rs *remoteStore) ReaderAt(ctx context.Context, dgst digest.Digest) (io.ReaderAt, error) {
+	return &remoteReaderAt{
+		ctx:    ctx,
+		digest: dgst,
+		client: rs.client,
+	}, nil
+}
+
 func (rs *remoteStore) Status(ctx context.Context, re string) ([]content.Status, error) {
 	resp, err := rs.client.Status(ctx, &contentapi.StatusRequest{
 		Regexp: re,