1 files changed, 65 insertions, 0 deletions

diff --git a/weed/util/chunk_cache/chunk_cache.go b/weed/util/chunk_cache/chunk_cache.go
index 3615aee0e..40d24b322 100644
--- a/weed/util/chunk_cache/chunk_cache.go
+++ b/weed/util/chunk_cache/chunk_cache.go
@@ -1,14 +1,18 @@
 package chunk_cache
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 )
 
+var ErrorOutOfBounds = errors.New("attempt to read out of bounds")
+
 type ChunkCache interface {
 	GetChunk(fileId string, minSize uint64) (data []byte)
+	GetChunkSlice(fileId string, offset, length uint64) []byte
 	SetChunk(fileId string, data []byte)
 }
 
@@ -22,6 +26,8 @@ type TieredChunkCache struct {
 	onDiskCacheSizeLimit2 uint64
 }
 
+var _ ChunkCache = &TieredChunkCache{}
+
 func NewTieredChunkCache(maxEntries int64, dir string, diskSizeInUnit int64, unitSize int64) *TieredChunkCache {
 
 	c := &TieredChunkCache{
@@ -87,6 +93,58 @@ func (c *TieredChunkCache) doGetChunk(fileId string, minSize uint64) (data []byt
 
 }
 
+func (c *TieredChunkCache) GetChunkSlice(fileId string, offset, length uint64) []byte {
+	if c == nil {
+		return nil
+	}
+
+	c.RLock()
+	defer c.RUnlock()
+
+	return c.doGetChunkSlice(fileId, offset, length)
+}
+
+func (c *TieredChunkCache) doGetChunkSlice(fileId string, offset, length uint64) (data []byte) {
+
+	minSize := offset + length
+	if minSize <= c.onDiskCacheSizeLimit0 {
+		data, err := c.memCache.getChunkSlice(fileId, offset, length)
+		if err != nil {
+			glog.Errorf("failed to read from memcache: %s", err)
+		}
+		if len(data) >= int(minSize) {
+			return data
+		}
+	}
+
+	fid, err := needle.ParseFileIdFromString(fileId)
+	if err != nil {
+		glog.Errorf("failed to parse file id %s", fileId)
+		return nil
+	}
+
+	if minSize <= c.onDiskCacheSizeLimit0 {
+		data = c.diskCaches[0].getChunkSlice(fid.Key, offset, length)
+		if len(data) >= int(minSize) {
+			return data
+		}
+	}
+	if minSize <= c.onDiskCacheSizeLimit1 {
+		data = c.diskCaches[1].getChunkSlice(fid.Key, offset, length)
+		if len(data) >= int(minSize) {
+			return data
+		}
+	}
+	{
+		data = c.diskCaches[2].getChunkSlice(fid.Key, offset, length)
+		if len(data) >= int(minSize) {
+			return data
+		}
+	}
+
+	return nil
+}
+
 func (c *TieredChunkCache) SetChunk(fileId string, data []byte) {
 	if c == nil {
 		return
@@ -131,3 +189,10 @@ func (c *TieredChunkCache) Shutdown() {
 		diskCache.shutdown()
 	}
 }
+
+func min(x, y int) int {
+	if x < y {
+		return x
+	}
+	return y
+}