|
|
@@ -5,6 +5,7 @@ import (
|
|
|
"io"
|
|
|
|
|
|
"github.com/chrislusf/seaweedfs/weed/util"
|
|
|
+ "github.com/valyala/bytebufferpool"
|
|
|
)
|
|
|
|
|
|
type IntervalNode struct {
|
|
|
@@ -12,6 +13,15 @@ type IntervalNode struct {
|
|
|
Offset int64
|
|
|
Size int64
|
|
|
Next *IntervalNode
|
|
|
+ Buffer *bytebufferpool.ByteBuffer
|
|
|
+}
|
|
|
+
|
|
|
+func (l *IntervalNode) Bytes() []byte {
|
|
|
+ data := l.Data
|
|
|
+ if data == nil {
|
|
|
+ data = l.Buffer.Bytes()
|
|
|
+ }
|
|
|
+ return data
|
|
|
}
|
|
|
|
|
|
type IntervalLinkedList struct {
|
|
|
@@ -23,16 +33,39 @@ type ContinuousIntervals struct {
|
|
|
lists []*IntervalLinkedList
|
|
|
}
|
|
|
|
|
|
+func NewIntervalLinkedList(head, tail *IntervalNode) *IntervalLinkedList {
|
|
|
+ list := &IntervalLinkedList{
|
|
|
+ Head: head,
|
|
|
+ Tail: tail,
|
|
|
+ }
|
|
|
+ return list
|
|
|
+}
|
|
|
+
|
|
|
+func (list *IntervalLinkedList) Destroy() {
|
|
|
+ for t := list.Head; t != nil; t = t.Next {
|
|
|
+ if t.Buffer != nil {
|
|
|
+ bytebufferpool.Put(t.Buffer)
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
func (list *IntervalLinkedList) Offset() int64 {
|
|
|
return list.Head.Offset
|
|
|
}
|
|
|
func (list *IntervalLinkedList) Size() int64 {
|
|
|
return list.Tail.Offset + list.Tail.Size - list.Head.Offset
|
|
|
}
|
|
|
+
|
|
|
func (list *IntervalLinkedList) addNodeToTail(node *IntervalNode) {
|
|
|
// glog.V(4).Infof("add to tail [%d,%d) + [%d,%d) => [%d,%d)", list.Head.Offset, list.Tail.Offset+list.Tail.Size, node.Offset, node.Offset+node.Size, list.Head.Offset, node.Offset+node.Size)
|
|
|
- list.Tail.Next = node
|
|
|
- list.Tail = node
|
|
|
+ if list.Tail.Buffer == nil {
|
|
|
+ list.Tail.Buffer = bytebufferpool.Get()
|
|
|
+ list.Tail.Buffer.Write(list.Tail.Data)
|
|
|
+ list.Tail.Data = nil
|
|
|
+ }
|
|
|
+ list.Tail.Buffer.Write(node.Data)
|
|
|
+ list.Tail.Size += int64(len(node.Data))
|
|
|
+ return
|
|
|
}
|
|
|
func (list *IntervalLinkedList) addNodeToHead(node *IntervalNode) {
|
|
|
// glog.V(4).Infof("add to head [%d,%d) + [%d,%d) => [%d,%d)", node.Offset, node.Offset+node.Size, list.Head.Offset, list.Tail.Offset+list.Tail.Size, node.Offset, list.Tail.Offset+list.Tail.Size)
|
|
|
@@ -47,7 +80,7 @@ func (list *IntervalLinkedList) ReadData(buf []byte, start, stop int64) {
|
|
|
nodeStart, nodeStop := max(start, t.Offset), min(stop, t.Offset+t.Size)
|
|
|
if nodeStart < nodeStop {
|
|
|
// glog.V(0).Infof("copying start=%d stop=%d t=[%d,%d) t.data=%d => bufSize=%d nodeStart=%d, nodeStop=%d", start, stop, t.Offset, t.Offset+t.Size, len(t.Data), len(buf), nodeStart, nodeStop)
|
|
|
- copy(buf[nodeStart-start:], t.Data[nodeStart-t.Offset:nodeStop-t.Offset])
|
|
|
+ copy(buf[nodeStart-start:], t.Bytes()[nodeStart-t.Offset:nodeStop-t.Offset])
|
|
|
}
|
|
|
|
|
|
if t.Next == nil {
|
|
|
@@ -72,8 +105,15 @@ func subList(list *IntervalLinkedList, start, stop int64) *IntervalLinkedList {
|
|
|
// skip non overlapping IntervalNode
|
|
|
continue
|
|
|
}
|
|
|
+ data := t.Bytes()[nodeStart-t.Offset : nodeStop-t.Offset]
|
|
|
+ if t.Data == nil {
|
|
|
+ // need to clone if the bytes is from byte buffer
|
|
|
+ t := make([]byte, len(data))
|
|
|
+ copy(t, data)
|
|
|
+ data = t
|
|
|
+ }
|
|
|
nodes = append(nodes, &IntervalNode{
|
|
|
- Data: t.Data[nodeStart-t.Offset : nodeStop-t.Offset],
|
|
|
+ Data: data,
|
|
|
Offset: nodeStart,
|
|
|
Size: nodeStop - nodeStart,
|
|
|
Next: nil,
|
|
|
@@ -82,10 +122,7 @@ func subList(list *IntervalLinkedList, start, stop int64) *IntervalLinkedList {
|
|
|
for i := 1; i < len(nodes); i++ {
|
|
|
nodes[i-1].Next = nodes[i]
|
|
|
}
|
|
|
- return &IntervalLinkedList{
|
|
|
- Head: nodes[0],
|
|
|
- Tail: nodes[len(nodes)-1],
|
|
|
- }
|
|
|
+ return NewIntervalLinkedList(nodes[0], nodes[len(nodes)-1])
|
|
|
}
|
|
|
|
|
|
func (c *ContinuousIntervals) AddInterval(data []byte, offset int64) {
|
|
|
@@ -144,10 +181,7 @@ func (c *ContinuousIntervals) AddInterval(data []byte, offset int64) {
|
|
|
nextList.addNodeToHead(interval)
|
|
|
}
|
|
|
if prevList == nil && nextList == nil {
|
|
|
- c.lists = append(c.lists, &IntervalLinkedList{
|
|
|
- Head: interval,
|
|
|
- Tail: interval,
|
|
|
- })
|
|
|
+ c.lists = append(c.lists, NewIntervalLinkedList(interval, interval))
|
|
|
}
|
|
|
|
|
|
return
|
|
|
@@ -155,11 +189,13 @@ func (c *ContinuousIntervals) AddInterval(data []byte, offset int64) {
|
|
|
|
|
|
func (c *ContinuousIntervals) RemoveLargestIntervalLinkedList() *IntervalLinkedList {
|
|
|
var maxSize int64
|
|
|
- maxIndex := -1
|
|
|
+ maxIndex, maxOffset := -1, int64(-1)
|
|
|
+ println("in memory list:", len(c.lists))
|
|
|
for k, list := range c.lists {
|
|
|
- if maxSize <= list.Size() {
|
|
|
- maxSize = list.Size()
|
|
|
- maxIndex = k
|
|
|
+ listSize := list.Size()
|
|
|
+ if maxSize < listSize || (maxSize == listSize && list.Offset() < maxOffset ) {
|
|
|
+ maxSize = listSize
|
|
|
+ maxIndex, maxOffset = k, list.Offset()
|
|
|
}
|
|
|
}
|
|
|
if maxSize <= 0 {
|
|
|
@@ -202,10 +238,10 @@ func (c *ContinuousIntervals) ReadDataAt(data []byte, startOffset int64) (maxSto
|
|
|
func (l *IntervalLinkedList) ToReader() io.Reader {
|
|
|
var readers []io.Reader
|
|
|
t := l.Head
|
|
|
- readers = append(readers, util.NewBytesReader(t.Data))
|
|
|
+ readers = append(readers, util.NewBytesReader(t.Bytes()))
|
|
|
for t.Next != nil {
|
|
|
t = t.Next
|
|
|
- readers = append(readers, bytes.NewReader(t.Data))
|
|
|
+ readers = append(readers, bytes.NewReader(t.Bytes()))
|
|
|
}
|
|
|
if len(readers) == 1 {
|
|
|
return readers[0]
|
Chris Lu