package page_writer import ( "fmt" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util/mem" "sync" "sync/atomic" "time" ) type LogicChunkIndex int type UploadPipeline struct { filepath util.FullPath ChunkSize int64 writableChunks map[LogicChunkIndex]*MemChunk writableChunksLock sync.Mutex sealedChunks map[LogicChunkIndex]*SealedChunk sealedChunksLock sync.Mutex uploaders *util.LimitedConcurrentExecutor uploaderCount int32 uploaderCountCond *sync.Cond saveToStorageFn SaveToStorageFunc activeReadChunks map[LogicChunkIndex]int activeReadChunksLock sync.Mutex } type SealedChunk struct { chunk *MemChunk referenceCounter int // track uploading or reading processes } func (sc *SealedChunk) FreeReference(messageOnFree string) { sc.referenceCounter-- if sc.referenceCounter == 0 { glog.V(4).Infof("Free sealed chunk: %s", messageOnFree) mem.Free(sc.chunk.buf) } } func NewUploadPipeline(filepath util.FullPath, writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc) *UploadPipeline { return &UploadPipeline{ ChunkSize: chunkSize, writableChunks: make(map[LogicChunkIndex]*MemChunk), sealedChunks: make(map[LogicChunkIndex]*SealedChunk), uploaders: writers, uploaderCountCond: sync.NewCond(&sync.Mutex{}), saveToStorageFn: saveToStorageFn, filepath: filepath, activeReadChunks: make(map[LogicChunkIndex]int), } } func (cw *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) { cw.writableChunksLock.Lock() defer cw.writableChunksLock.Unlock() logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize) offsetRemainder := off % cw.ChunkSize memChunk, found := cw.writableChunks[logicChunkIndex] if !found { memChunk = &MemChunk{ buf: mem.Allocate(int(cw.ChunkSize)), usage: newChunkWrittenIntervalList(), } cw.writableChunks[logicChunkIndex] = memChunk } n = copy(memChunk.buf[offsetRemainder:], p) memChunk.usage.MarkWritten(offsetRemainder, offsetRemainder+int64(n)) cw.maybeMoveToSealed(memChunk, logicChunkIndex) return } func (cw *UploadPipeline) MaybeReadDataAt(p []byte, off int64) (maxStop int64) { logicChunkIndex := LogicChunkIndex(off / cw.ChunkSize) // read from sealed chunks first cw.sealedChunksLock.Lock() sealedChunk, found := cw.sealedChunks[logicChunkIndex] if found { sealedChunk.referenceCounter++ } cw.sealedChunksLock.Unlock() if found { maxStop = readMemChunk(sealedChunk.chunk, p, off, logicChunkIndex, cw.ChunkSize) glog.V(4).Infof("%s read sealed memchunk [%d,%d)", cw.filepath, off, maxStop) sealedChunk.FreeReference(fmt.Sprintf("%s finish reading chunk %d", cw.filepath, logicChunkIndex)) } // read from writable chunks last cw.writableChunksLock.Lock() defer cw.writableChunksLock.Unlock() writableChunk, found := cw.writableChunks[logicChunkIndex] if !found { return } writableMaxStop := readMemChunk(writableChunk, p, off, logicChunkIndex, cw.ChunkSize) glog.V(4).Infof("%s read writable memchunk [%d,%d)", cw.filepath, off, writableMaxStop) maxStop = max(maxStop, writableMaxStop) return } func (cw *UploadPipeline) FlushAll() { cw.writableChunksLock.Lock() defer cw.writableChunksLock.Unlock() for logicChunkIndex, memChunk := range cw.writableChunks { cw.moveToSealed(memChunk, logicChunkIndex) } cw.waitForCurrentWritersToComplete() } func (cw *UploadPipeline) LockForRead(startOffset, stopOffset int64) { startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize) stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize) if stopOffset%cw.ChunkSize > 0 { stopLogicChunkIndex += 1 } cw.activeReadChunksLock.Lock() defer cw.activeReadChunksLock.Unlock() for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ { if count, found := cw.activeReadChunks[i]; found { cw.activeReadChunks[i] = count + 1 } else { cw.activeReadChunks[i] = 1 } } } func (cw *UploadPipeline) UnlockForRead(startOffset, stopOffset int64) { startLogicChunkIndex := LogicChunkIndex(startOffset / cw.ChunkSize) stopLogicChunkIndex := LogicChunkIndex(stopOffset / cw.ChunkSize) if stopOffset%cw.ChunkSize > 0 { stopLogicChunkIndex += 1 } cw.activeReadChunksLock.Lock() defer cw.activeReadChunksLock.Unlock() for i := startLogicChunkIndex; i < stopLogicChunkIndex; i++ { if count, found := cw.activeReadChunks[i]; found { if count == 1 { delete(cw.activeReadChunks, i) } else { cw.activeReadChunks[i] = count - 1 } } } } func (cw *UploadPipeline) IsLocked(logicChunkIndex LogicChunkIndex) bool { cw.activeReadChunksLock.Lock() defer cw.activeReadChunksLock.Unlock() if count, found := cw.activeReadChunks[logicChunkIndex]; found { return count > 0 } return false } func (cw *UploadPipeline) waitForCurrentWritersToComplete() { cw.uploaderCountCond.L.Lock() t := int32(100) for { t = atomic.LoadInt32(&cw.uploaderCount) if t <= 0 { break } cw.uploaderCountCond.Wait() } cw.uploaderCountCond.L.Unlock() } func (cw *UploadPipeline) maybeMoveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) { if memChunk.usage.IsComplete(cw.ChunkSize) { cw.moveToSealed(memChunk, logicChunkIndex) } } func (cw *UploadPipeline) moveToSealed(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) { atomic.AddInt32(&cw.uploaderCount, 1) glog.V(4).Infof("%s uploaderCount %d ++> %d", cw.filepath, cw.uploaderCount-1, cw.uploaderCount) cw.sealedChunksLock.Lock() if oldMemChunk, found := cw.sealedChunks[logicChunkIndex]; found { oldMemChunk.FreeReference(fmt.Sprintf("%s replace chunk %d", cw.filepath, logicChunkIndex)) } sealedChunk := &SealedChunk{ chunk: memChunk, referenceCounter: 1, // default 1 is for uploading process } cw.sealedChunks[logicChunkIndex] = sealedChunk delete(cw.writableChunks, logicChunkIndex) cw.sealedChunksLock.Unlock() cw.uploaders.Execute(func() { // first add to the file chunks cw.saveOneChunk(sealedChunk.chunk, logicChunkIndex) // notify waiting process atomic.AddInt32(&cw.uploaderCount, -1) glog.V(4).Infof("%s uploaderCount %d --> %d", cw.filepath, cw.uploaderCount+1, cw.uploaderCount) // Lock and Unlock are not required, // but it may signal multiple times during one wakeup, // and the waiting goroutine may miss some of them! cw.uploaderCountCond.L.Lock() cw.uploaderCountCond.Broadcast() cw.uploaderCountCond.L.Unlock() // wait for readers for cw.IsLocked(logicChunkIndex) { time.Sleep(59 * time.Millisecond) } // then remove from sealed chunks cw.sealedChunksLock.Lock() defer cw.sealedChunksLock.Unlock() delete(cw.sealedChunks, logicChunkIndex) sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", cw.filepath, logicChunkIndex)) }) } func (cw *UploadPipeline) saveOneChunk(memChunk *MemChunk, logicChunkIndex LogicChunkIndex) { if cw.saveToStorageFn == nil { return } for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next { reader := util.NewBytesReader(memChunk.buf[t.StartOffset:t.stopOffset]) cw.saveToStorageFn(reader, int64(logicChunkIndex)*cw.ChunkSize+t.StartOffset, t.Size(), func() { }) } } func readMemChunk(memChunk *MemChunk, p []byte, off int64, logicChunkIndex LogicChunkIndex, chunkSize int64) (maxStop int64) { memChunkBaseOffset := int64(logicChunkIndex) * chunkSize for t := memChunk.usage.head.next; t != memChunk.usage.tail; t = t.next { logicStart := max(off, int64(logicChunkIndex)*chunkSize+t.StartOffset) logicStop := min(off+int64(len(p)), memChunkBaseOffset+t.stopOffset) if logicStart < logicStop { copy(p[logicStart-off:logicStop-off], memChunk.buf[logicStart-memChunkBaseOffset:logicStop-memChunkBaseOffset]) maxStop = max(maxStop, logicStop) } } return } func (p2 *UploadPipeline) Shutdown() { }