diff options
| author | chrislu <chris.lu@gmail.com> | 2022-12-20 12:52:30 -0800 |
|---|---|---|
| committer | chrislu <chris.lu@gmail.com> | 2022-12-20 12:52:30 -0800 |
| commit | f7beba8515ddfc138f7d068207baa0765c03ec0c (patch) | |
| tree | a9e5a77be9ec110d17f2965612f0047b2a8303b5 | |
| parent | 28d479e5c059bdb041aeaff1075a738990740bf2 (diff) | |
| download | seaweedfs-f7beba8515ddfc138f7d068207baa0765c03ec0c.tar.xz seaweedfs-f7beba8515ddfc138f7d068207baa0765c03ec0c.zip | |
unlock before submitting the uploading jobs
| -rw-r--r-- | weed/mount/page_writer/upload_pipeline.go | 6 |
1 files changed, 5 insertions, 1 deletions
diff --git a/weed/mount/page_writer/upload_pipeline.go b/weed/mount/page_writer/upload_pipeline.go index f54a87a02..bce062a6e 100644 --- a/weed/mount/page_writer/upload_pipeline.go +++ b/weed/mount/page_writer/upload_pipeline.go @@ -73,8 +73,9 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool) (n fullness = chunkFullness } } - up.moveToSealed(up.writableChunks[fullestChunkIndex], fullestChunkIndex) + fullWritableChunk := up.writableChunks[fullestChunkIndex] delete(up.writableChunks, fullestChunkIndex) + up.moveToSealed(fullWritableChunk, fullestChunkIndex) // fmt.Printf("flush chunk %d with %d bytes written\n", logicChunkIndex, fullness) } if isSequential && @@ -155,6 +156,8 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic up.sealedChunks[logicChunkIndex] = sealedChunk delete(up.writableChunks, logicChunkIndex) + // unlock before submitting the uploading jobs + up.chunksLock.Unlock() up.uploaders.Execute(func() { // first add to the file chunks 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)) }) + up.chunksLock.Lock() } func (up *UploadPipeline) Shutdown() { |