lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-09-18 18:47:55 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

unlock before submitting the uploading jobs

Browse Source
This commit is contained in:
chrislu
2022-12-20 12:52:30 -08:00
parent 28d479e5c0
commit f7beba8515
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
weed/mount/page_writer/upload_pipeline.go
View File

@@ -73,8 +73,9 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool) (n
fullness = chunkFullness fullness = chunkFullness
} }
} }
up.moveToSealed(up.writableChunks[fullestChunkIndex], fullestChunkIndex) fullWritableChunk := up.writableChunks[fullestChunkIndex]
delete(up.writableChunks, fullestChunkIndex) delete(up.writableChunks, fullestChunkIndex)
up.moveToSealed(fullWritableChunk, fullestChunkIndex)
// fmt.Printf("flush chunk %d with %d bytes written\n", logicChunkIndex, fullness) // fmt.Printf("flush chunk %d with %d bytes written\n", logicChunkIndex, fullness)
} }
if isSequential && if isSequential &&
@@ -155,6 +156,8 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic
up.sealedChunks[logicChunkIndex] = sealedChunk up.sealedChunks[logicChunkIndex] = sealedChunk
delete(up.writableChunks, logicChunkIndex) delete(up.writableChunks, logicChunkIndex)
// unlock before submitting the uploading jobs
up.chunksLock.Unlock()
up.uploaders.Execute(func() { up.uploaders.Execute(func() {
// first add to the file chunks // first add to the file chunks
sealedChunk.chunk.SaveContent(up.saveToStorageFn) sealedChunk.chunk.SaveContent(up.saveToStorageFn)
@@ -181,6 +184,7 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic
sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", up.filepath, logicChunkIndex)) sealedChunk.FreeReference(fmt.Sprintf("%s finished uploading chunk %d", up.filepath, logicChunkIndex))
}) })
up.chunksLock.Lock()
} }
func (up *UploadPipeline) Shutdown() { func (up *UploadPipeline) Shutdown() {