seaweedfs/weed/filesys/page_writer/upload_pipeline.go

177 lines
5.2 KiB
Go
Raw Normal View History

2022-01-17 09:53:56 +00:00
package page_writer
import (
2022-01-17 22:15:10 +00:00
"fmt"
2022-01-17 09:53:56 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/util"
"sync"
"sync/atomic"
"time"
2022-01-17 09:53:56 +00:00
)
type LogicChunkIndex int
2022-01-17 09:53:56 +00:00
type UploadPipeline struct {
filepath util.FullPath
ChunkSize int64
2022-01-22 12:18:54 +00:00
writableChunks map[LogicChunkIndex]PageChunk
writableChunksLock sync.Mutex
sealedChunks map[LogicChunkIndex]*SealedChunk
sealedChunksLock sync.Mutex
2022-01-22 09:46:10 +00:00
uploaders *util.LimitedConcurrentExecutor
uploaderCount int32
uploaderCountCond *sync.Cond
2022-01-22 09:43:14 +00:00
saveToStorageFn SaveToStorageFunc
activeReadChunks map[LogicChunkIndex]int
activeReadChunksLock sync.Mutex
swapFile *SwapFile
2022-01-17 09:53:56 +00:00
}
type SealedChunk struct {
2022-01-22 12:18:54 +00:00
chunk PageChunk
2022-01-17 09:53:56 +00:00
referenceCounter int // track uploading or reading processes
}
2022-01-17 22:15:10 +00:00
func (sc *SealedChunk) FreeReference(messageOnFree string) {
2022-01-17 09:53:56 +00:00
sc.referenceCounter--
if sc.referenceCounter == 0 {
2022-01-17 22:15:10 +00:00
glog.V(4).Infof("Free sealed chunk: %s", messageOnFree)
2022-01-22 12:18:54 +00:00
sc.chunk.FreeResource()
2022-01-17 09:53:56 +00:00
}
}
func NewUploadPipeline(filepath util.FullPath, writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc, swapFileDir string) *UploadPipeline {
2022-01-17 09:53:56 +00:00
return &UploadPipeline{
2022-01-22 09:46:10 +00:00
ChunkSize: chunkSize,
2022-01-22 12:18:54 +00:00
writableChunks: make(map[LogicChunkIndex]PageChunk),
2022-01-22 09:46:10 +00:00
sealedChunks: make(map[LogicChunkIndex]*SealedChunk),
uploaders: writers,
uploaderCountCond: sync.NewCond(&sync.Mutex{}),
saveToStorageFn: saveToStorageFn,
filepath: filepath,
activeReadChunks: make(map[LogicChunkIndex]int),
swapFile: NewSwapFile(swapFileDir, chunkSize),
2022-01-17 09:53:56 +00:00
}
}
func (up *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
up.writableChunksLock.Lock()
defer up.writableChunksLock.Unlock()
2022-01-17 09:53:56 +00:00
logicChunkIndex := LogicChunkIndex(off / up.ChunkSize)
2022-01-17 09:53:56 +00:00
memChunk, found := up.writableChunks[logicChunkIndex]
2022-01-17 09:53:56 +00:00
if !found {
2022-01-22 16:11:01 +00:00
if len(up.writableChunks) < 16 {
memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
} else {
memChunk = up.swapFile.NewTempFileChunk(logicChunkIndex)
if memChunk == nil {
memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
}
}
up.writableChunks[logicChunkIndex] = memChunk
2022-01-17 09:53:56 +00:00
}
2022-01-22 14:34:29 +00:00
n = memChunk.WriteDataAt(p, off)
up.maybeMoveToSealed(memChunk, logicChunkIndex)
2022-01-17 09:53:56 +00:00
return
}
func (up *UploadPipeline) MaybeReadDataAt(p []byte, off int64) (maxStop int64) {
logicChunkIndex := LogicChunkIndex(off / up.ChunkSize)
2022-01-17 09:53:56 +00:00
// read from sealed chunks first
up.sealedChunksLock.Lock()
sealedChunk, found := up.sealedChunks[logicChunkIndex]
2022-01-17 09:53:56 +00:00
if found {
sealedChunk.referenceCounter++
}
up.sealedChunksLock.Unlock()
2022-01-17 09:53:56 +00:00
if found {
2022-01-22 13:40:10 +00:00
maxStop = sealedChunk.chunk.ReadDataAt(p, off)
glog.V(4).Infof("%s read sealed memchunk [%d,%d)", up.filepath, off, maxStop)
sealedChunk.FreeReference(fmt.Sprintf("%s finish reading chunk %d", up.filepath, logicChunkIndex))
2022-01-17 09:53:56 +00:00
}
// read from writable chunks last
up.writableChunksLock.Lock()
defer up.writableChunksLock.Unlock()
writableChunk, found := up.writableChunks[logicChunkIndex]
2022-01-17 09:53:56 +00:00
if !found {
return
}
2022-01-22 13:40:10 +00:00
writableMaxStop := writableChunk.ReadDataAt(p, off)
glog.V(4).Infof("%s read writable memchunk [%d,%d)", up.filepath, off, writableMaxStop)
2022-01-18 04:41:00 +00:00
maxStop = max(maxStop, writableMaxStop)
2022-01-17 09:53:56 +00:00
return
}
func (up *UploadPipeline) FlushAll() {
up.writableChunksLock.Lock()
defer up.writableChunksLock.Unlock()
2022-01-17 09:53:56 +00:00
for logicChunkIndex, memChunk := range up.writableChunks {
up.moveToSealed(memChunk, logicChunkIndex)
2022-01-17 09:53:56 +00:00
}
up.waitForCurrentWritersToComplete()
2022-01-17 09:53:56 +00:00
}
func (up *UploadPipeline) maybeMoveToSealed(memChunk PageChunk, logicChunkIndex LogicChunkIndex) {
2022-01-22 13:40:10 +00:00
if memChunk.IsComplete() {
up.moveToSealed(memChunk, logicChunkIndex)
2022-01-17 09:53:56 +00:00
}
}
func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex LogicChunkIndex) {
atomic.AddInt32(&up.uploaderCount, 1)
glog.V(4).Infof("%s uploaderCount %d ++> %d", up.filepath, up.uploaderCount-1, up.uploaderCount)
2022-01-17 09:53:56 +00:00
up.sealedChunksLock.Lock()
2022-01-17 09:53:56 +00:00
if oldMemChunk, found := up.sealedChunks[logicChunkIndex]; found {
oldMemChunk.FreeReference(fmt.Sprintf("%s replace chunk %d", up.filepath, logicChunkIndex))
2022-01-17 09:53:56 +00:00
}
sealedChunk := &SealedChunk{
chunk: memChunk,
referenceCounter: 1, // default 1 is for uploading process
}
up.sealedChunks[logicChunkIndex] = sealedChunk
delete(up.writableChunks, logicChunkIndex)
2022-01-17 09:53:56 +00:00
up.sealedChunksLock.Unlock()
2022-01-17 09:53:56 +00:00
up.uploaders.Execute(func() {
2022-01-17 21:40:41 +00:00
// first add to the file chunks
sealedChunk.chunk.SaveContent(up.saveToStorageFn)
2022-01-17 09:53:56 +00:00
// notify waiting process
atomic.AddInt32(&up.uploaderCount, -1)
glog.V(4).Infof("%s uploaderCount %d --> %d", up.filepath, up.uploaderCount+1, up.uploaderCount)
2022-01-17 09:53:56 +00:00
// Lock and Unlock are not required,
// but it may signal multiple times during one wakeup,
// and the waiting goroutine may miss some of them!
up.uploaderCountCond.L.Lock()
up.uploaderCountCond.Broadcast()
up.uploaderCountCond.L.Unlock()
// wait for readers
for up.IsLocked(logicChunkIndex) {
time.Sleep(59 * time.Millisecond)
}
// then remove from sealed chunks
up.sealedChunksLock.Lock()
defer up.sealedChunksLock.Unlock()
delete(up.sealedChunks, logicChunkIndex)
sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", up.filepath, logicChunkIndex))
2022-01-17 09:53:56 +00:00
})
}
func (up *UploadPipeline) Shutdown() {
up.swapFile.FreeResource()
2022-01-17 09:53:56 +00:00
}