1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
package filesys
import (
"bytes"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"io"
"sync"
"time"
)
type ContinuousDirtyPages struct {
intervals *ContinuousIntervals
f *File
writeWaitGroup sync.WaitGroup
chunkSaveErrChan chan error
chunkSaveErrChanClosed bool
lock sync.Mutex
collection string
replication string
}
func newDirtyPages(file *File) *ContinuousDirtyPages {
return &ContinuousDirtyPages{
intervals: &ContinuousIntervals{},
f: file,
chunkSaveErrChan: make(chan error, 8),
}
}
func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
glog.V(4).Infof("%s AddPage [%d,%d) of %d bytes", pages.f.fullpath(), offset, offset+int64(len(data)), pages.f.entry.Attributes.FileSize)
if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
// this is more than what buffer can hold.
pages.flushAndSave(offset, data)
}
pages.intervals.AddInterval(data, offset)
if pages.intervals.TotalSize() > pages.f.wfs.option.ChunkSizeLimit {
pages.saveExistingLargestPageToStorage()
}
return
}
func (pages *ContinuousDirtyPages) flushAndSave(offset int64, data []byte) {
// flush existing
pages.saveExistingPagesToStorage()
// flush the new page
pages.saveToStorage(bytes.NewReader(data), offset, int64(len(data)))
return
}
func (pages *ContinuousDirtyPages) saveExistingPagesToStorage() {
for pages.saveExistingLargestPageToStorage() {
}
}
func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedData bool) {
maxList := pages.intervals.RemoveLargestIntervalLinkedList()
if maxList == nil {
return false
}
fileSize := int64(pages.f.entry.Attributes.FileSize)
chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
if chunkSize == 0 {
return false
}
pages.saveToStorage(maxList.ToReader(), maxList.Offset(), chunkSize)
return true
}
func (pages *ContinuousDirtyPages) saveToStorage(reader io.Reader, offset int64, size int64) {
if pages.chunkSaveErrChanClosed {
pages.chunkSaveErrChan = make(chan error, 8)
pages.chunkSaveErrChanClosed = false
}
mtime := time.Now().UnixNano()
pages.writeWaitGroup.Add(1)
go func() {
defer pages.writeWaitGroup.Done()
dir, _ := pages.f.fullpath().DirAndName()
reader = io.LimitReader(reader, size)
chunk, collection, replication, err := pages.f.wfs.saveDataAsChunk(dir)(reader, pages.f.Name, offset)
if err != nil {
glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), offset, offset+size, err)
pages.chunkSaveErrChan <- err
return
}
chunk.Mtime = mtime
pages.collection, pages.replication = collection, replication
pages.f.addChunks([]*filer_pb.FileChunk{chunk})
pages.chunkSaveErrChan <- nil
}()
}
func max(x, y int64) int64 {
if x > y {
return x
}
return y
}
func min(x, y int64) int64 {
if x < y {
return x
}
return y
}
func (pages *ContinuousDirtyPages) ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64) {
return pages.intervals.ReadDataAt(data, startOffset)
}
|
