gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

use a sliding window of in-memory writable chunks

Browse source
This commit is contained in:
chrislu 2022-01-23 23:02:05 -08:00
parent 520591e6ea
commit 3bba2124ef
5 changed files with 26 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
weed/filesys/page_writer/chunk_interval_list.go
View file

@ -51,6 +51,12 @@ func (list *ChunkWrittenIntervalList) MarkWritten(startOffset, stopOffset int64)
func (list *ChunkWrittenIntervalList) IsComplete(chunkSize int64) bool { func (list *ChunkWrittenIntervalList) IsComplete(chunkSize int64) bool {
return list.size() == 1 && list.head.next.isComplete(chunkSize) return list.size() == 1 && list.head.next.isComplete(chunkSize)
} }
func (list *ChunkWrittenIntervalList) WrittenSize() (writtenByteCount int64) {
for t := list.head; t != nil; t = t.next {
writtenByteCount += t.Size()
}
return
}
func (list *ChunkWrittenIntervalList) addInterval(interval *ChunkWrittenInterval) { func (list *ChunkWrittenIntervalList) addInterval(interval *ChunkWrittenInterval) {

1
weed/filesys/page_writer/page_chunk.go
View file

@ -11,5 +11,6 @@ type PageChunk interface {
WriteDataAt(src []byte, offset int64) (n int) WriteDataAt(src []byte, offset int64) (n int)
ReadDataAt(p []byte, off int64) (maxStop int64) ReadDataAt(p []byte, off int64) (maxStop int64)
IsComplete() bool IsComplete() bool
WrittenSize() int64
SaveContent(saveFn SaveToStorageFunc) SaveContent(saveFn SaveToStorageFunc)
} }

4
weed/filesys/page_writer/page_chunk_mem.go
View file

@ -53,6 +53,10 @@ func (mc *MemChunk) IsComplete() bool {
return mc.usage.IsComplete(mc.chunkSize) return mc.usage.IsComplete(mc.chunkSize)
} }
func (mc *MemChunk) WrittenSize() int64 {
return mc.usage.WrittenSize()
}
func (mc *MemChunk) SaveContent(saveFn SaveToStorageFunc) { func (mc *MemChunk) SaveContent(saveFn SaveToStorageFunc) {
if saveFn == nil { if saveFn == nil {
return return

4
weed/filesys/page_writer/page_chunk_swapfile.go
View file

@ -101,6 +101,10 @@ func (sc *SwapFileChunk) IsComplete() bool {
return sc.usage.IsComplete(sc.swapfile.chunkSize) return sc.usage.IsComplete(sc.swapfile.chunkSize)
} }
func (sc *SwapFileChunk) WrittenSize() int64 {
return sc.usage.WrittenSize()
}
func (sc *SwapFileChunk) SaveContent(saveFn SaveToStorageFunc) { func (sc *SwapFileChunk) SaveContent(saveFn SaveToStorageFunc) {
if saveFn == nil { if saveFn == nil {
return return

14
weed/filesys/page_writer/upload_pipeline.go
View file

@ -65,10 +65,18 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
if len(up.writableChunks) < 16 { if len(up.writableChunks) < 16 {
memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize) memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
} else { } else {
memChunk = up.swapFile.NewTempFileChunk(logicChunkIndex) fullestChunkIndex, fullness := LogicChunkIndex(-1), int64(0)
if memChunk == nil { for lci, mc := range up.writableChunks {
memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize) chunkFullness := mc.WrittenSize()
if fullness < chunkFullness {
fullestChunkIndex = lci
fullness = chunkFullness
}
} }
up.moveToSealed(up.writableChunks[fullestChunkIndex], fullestChunkIndex)
delete(up.writableChunks, fullestChunkIndex)
fmt.Printf("flush chunk %d with %d bytes written", logicChunkIndex, fullness)
memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
} }
up.writableChunks[logicChunkIndex] = memChunk up.writableChunks[logicChunkIndex] = memChunk
} }