gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mount: better combines connected intervals to write to volume servers

Browse Source
This commit is contained in:
Chris Lu
2020-01-22 23:00:04 -08:00
parent 6a5c037099
commit c2e589f202
4 changed files with 265 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
weed/filesys/dirty_page.go
View File

@@ -4,8 +4,8 @@ import (
"bytes"
"context"
"fmt"
"io"
"sync"
"sync/atomic"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
@@ -15,28 +15,19 @@ import (
)
type ContinuousDirtyPages struct {
hasData bool
Offset int64
Size int64
Data []byte
f *File
lock sync.Mutex
intervals *ContinuousIntervals
f *File
lock sync.Mutex
}
func newDirtyPages(file *File) *ContinuousDirtyPages {
return &ContinuousDirtyPages{
Data: nil,
f: file,
intervals: &ContinuousIntervals{},
f: file,
}
}
func (pages *ContinuousDirtyPages) releaseResource() {
if pages.Data != nil {
pages.f.wfs.bufPool.Put(pages.Data)
pages.Data = nil
atomic.AddInt32(&counter, -1)
glog.V(3).Infof("%s/%s releasing resource %d", pages.f.dir.Path, pages.f.Name, counter)
}
}
var counter = int32(0)
@@ -46,61 +37,29 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
pages.lock.Lock()
defer pages.lock.Unlock()
var chunk *filer_pb.FileChunk
glog.V(3).Infof("%s AddPage [%d,%d)", pages.f.fullpath(), offset, offset+int64(len(data)))
if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
// this is more than what buffer can hold.
return pages.flushAndSave(ctx, offset, data)
}
if pages.Data == nil {
pages.Data = pages.f.wfs.bufPool.Get().([]byte)
atomic.AddInt32(&counter, 1)
glog.V(3).Infof("%s/%s acquire resource %d", pages.f.dir.Path, pages.f.Name, counter)
}
if offset < pages.Offset || offset >= pages.Offset+int64(len(pages.Data)) ||
pages.Offset+int64(len(pages.Data)) < offset+int64(len(data)) {
// if the data is out of range,
// or buffer is full if adding new data,
// flush current buffer and add new data
glog.V(4).Infof("offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
if chunk, err = pages.saveExistingPagesToStorage(ctx); err == nil {
if chunk != nil {
glog.V(4).Infof("%s/%s add save [%d,%d)", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size))
chunks = append(chunks, chunk)
}
} else {
glog.V(0).Infof("%s/%s add save [%d,%d): %v", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), err)
return
}
pages.Offset = offset
glog.V(4).Infof("copy data0: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
copy(pages.Data, data)
pages.Size = int64(len(data))
hasOverlap := pages.intervals.AddInterval(data, offset)
if hasOverlap {
chunks, err = pages.saveExistingPagesToStorage(ctx)
pages.intervals.AddInterval(data, offset)
return
}
if offset != pages.Offset+pages.Size {
// when this happens, debug shows the data overlapping with existing data is empty
// the data is not just append
if offset == pages.Offset && int(pages.Size) < len(data) {
glog.V(4).Infof("copy data1: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
copy(pages.Data[pages.Size:], data[pages.Size:])
} else {
if pages.Size != 0 {
glog.V(1).Infof("%s/%s add page: pages [%d, %d) write [%d, %d)", pages.f.dir.Path, pages.f.Name, pages.Offset, pages.Offset+pages.Size, offset, offset+int64(len(data)))
}
return pages.flushAndSave(ctx, offset, data)
}
} else {
glog.V(4).Infof("copy data2: offset=%d, size=%d, existing pages offset=%d, pages size=%d, data=%d", offset, len(data), pages.Offset, pages.Size, len(pages.Data))
copy(pages.Data[offset-pages.Offset:], data)
}
var chunk *filer_pb.FileChunk
var hasSavedData bool
pages.Size = max(pages.Size, offset+int64(len(data))-pages.Offset)
if pages.intervals.TotalSize() > pages.f.wfs.option.ChunkSizeLimit {
chunk, hasSavedData, err = pages.saveExistingLargestPageToStorage(ctx)
if hasSavedData {
chunks = append(chunks, chunk)
}
}
return
}
@@ -108,22 +67,19 @@ func (pages *ContinuousDirtyPages) AddPage(ctx context.Context, offset int64, da
func (pages *ContinuousDirtyPages) flushAndSave(ctx context.Context, offset int64, data []byte) (chunks []*filer_pb.FileChunk, err error) {
var chunk *filer_pb.FileChunk
var newChunks []*filer_pb.FileChunk
// flush existing
if chunk, err = pages.saveExistingPagesToStorage(ctx); err == nil {
if chunk != nil {
glog.V(4).Infof("%s/%s flush existing [%d,%d) to %s", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), chunk.FileId)
chunks = append(chunks, chunk)
if newChunks, err = pages.saveExistingPagesToStorage(ctx); err == nil {
if newChunks != nil {
chunks = append(chunks, newChunks...)
}
} else {
glog.V(0).Infof("%s/%s failed to flush1 [%d,%d): %v", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), err)
return
}
pages.Size = 0
pages.Offset = 0
// flush the new page
if chunk, err = pages.saveToStorage(ctx, data, offset); err == nil {
if chunk, err = pages.saveToStorage(ctx, bytes.NewReader(data), offset, int64(len(data))); err == nil {
if chunk != nil {
glog.V(4).Infof("%s/%s flush big request [%d,%d) to %s", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size), chunk.FileId)
chunks = append(chunks, chunk)
@@ -136,37 +92,55 @@ func (pages *ContinuousDirtyPages) flushAndSave(ctx context.Context, offset int6
return
}
func (pages *ContinuousDirtyPages) FlushToStorage(ctx context.Context) (chunk *filer_pb.FileChunk, err error) {
func (pages *ContinuousDirtyPages) FlushToStorage(ctx context.Context) (chunks []*filer_pb.FileChunk, err error) {
pages.lock.Lock()
defer pages.lock.Unlock()
if pages.Size == 0 {
return nil, nil
}
return pages.saveExistingPagesToStorage(ctx)
}
if chunk, err = pages.saveExistingPagesToStorage(ctx); err == nil {
pages.Size = 0
pages.Offset = 0
if chunk != nil {
glog.V(4).Infof("%s/%s flush [%d,%d)", pages.f.dir.Path, pages.f.Name, chunk.Offset, chunk.Offset+int64(chunk.Size))
func (pages *ContinuousDirtyPages) saveExistingPagesToStorage(ctx context.Context) (chunks []*filer_pb.FileChunk, err error) {
var hasSavedData bool
var chunk *filer_pb.FileChunk
for {
chunk, hasSavedData, err = pages.saveExistingLargestPageToStorage(ctx)
if !hasSavedData {
return chunks, err
}
if err == nil {
chunks = append(chunks, chunk)
} else {
return
}
}
}
func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage(ctx context.Context) (chunk *filer_pb.FileChunk, hasSavedData bool, err error) {
maxList := pages.intervals.RemoveLargestIntervalLinkedList()
if maxList == nil {
return nil, false, nil
}
chunk, err = pages.saveToStorage(ctx, maxList.ToReader(), maxList.Offset(), maxList.Size())
if err == nil {
hasSavedData = true
glog.V(3).Infof("%s saveToStorage [%d,%d) %s", pages.f.fullpath(), maxList.Offset(), maxList.Offset()+maxList.Size(), chunk.FileId)
} else {
glog.V(0).Infof("%s saveToStorage [%d,%d): %v", pages.f.fullpath(), maxList.Offset(), maxList.Offset()+maxList.Size(), err)
return
}
return
}
func (pages *ContinuousDirtyPages) saveExistingPagesToStorage(ctx context.Context) (*filer_pb.FileChunk, error) {
if pages.Size == 0 {
return nil, nil
}
glog.V(0).Infof("%s/%s saveExistingPagesToStorage [%d,%d): Data len=%d", pages.f.dir.Path, pages.f.Name, pages.Offset, pages.Size, len(pages.Data))
return pages.saveToStorage(ctx, pages.Data[:pages.Size], pages.Offset)
}
func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte, offset int64) (*filer_pb.FileChunk, error) {
func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, reader io.Reader, offset int64, size int64) (*filer_pb.FileChunk, error) {
var fileId, host string
var auth security.EncodedJwt
@@ -195,8 +169,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte
}
fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
bufReader := bytes.NewReader(buf)
uploadResult, err := operation.Upload(fileUrl, pages.f.Name, bufReader, false, "", nil, auth)
uploadResult, err := operation.Upload(fileUrl, pages.f.Name, reader, false, "", nil, auth)
if err != nil {
glog.V(0).Infof("upload data %v to %s: %v", pages.f.Name, fileUrl, err)
return nil, fmt.Errorf("upload data: %v", err)
@@ -209,7 +182,7 @@ func (pages *ContinuousDirtyPages) saveToStorage(ctx context.Context, buf []byte
return &filer_pb.FileChunk{
FileId: fileId,
Offset: offset,
Size: uint64(len(buf)),
Size: uint64(size),
Mtime: time.Now().UnixNano(),
ETag: uploadResult.ETag,
}, nil
@@ -229,23 +202,11 @@ func min(x, y int64) int64 {
return y
}
func (pages *ContinuousDirtyPages) ReadDirtyData(ctx context.Context, data []byte, startOffset int64) (offset int64, size int, err error) {
bufSize := int64(len(data))
func (pages *ContinuousDirtyPages) ReadDirtyData(ctx context.Context, data []byte, startOffset int64) (offset int64, size int) {
pages.lock.Lock()
defer pages.lock.Unlock()
if startOffset+bufSize < pages.Offset {
return
}
if startOffset >= pages.Offset+pages.Size {
return
}
return pages.intervals.ReadData(data, startOffset)
offset = max(pages.Offset, startOffset)
stopOffset := min(pages.Offset+pages.Size, startOffset+bufSize)
size = int(stopOffset - offset)
copy(data[offset-startOffset:], pages.Data[offset-pages.Offset:stopOffset-pages.Offset])
return
}