Browse Source

filer.sync: limit concurrency when fetching file chunks

fix https://github.com/seaweedfs/seaweedfs/issues/3787
pull/3789/head
chrislu 2 years ago
parent
commit
0452ae6a6c
  1. 4
      weed/replication/sink/filersink/fetch_write.go
  2. 2
      weed/replication/sink/filersink/filer_sink.go

4
weed/replication/sink/filersink/fetch_write.go

@ -24,7 +24,8 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
var wg sync.WaitGroup var wg sync.WaitGroup
for chunkIndex, sourceChunk := range sourceChunks { for chunkIndex, sourceChunk := range sourceChunks {
wg.Add(1) wg.Add(1)
go func(chunk *filer_pb.FileChunk, index int) {
fs.executor.Execute(func() {
func(chunk *filer_pb.FileChunk, index int) {
defer wg.Done() defer wg.Done()
replicatedChunk, e := fs.replicateOneChunk(chunk, path) replicatedChunk, e := fs.replicateOneChunk(chunk, path)
if e != nil { if e != nil {
@ -33,6 +34,7 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, path st
} }
replicatedChunks[index] = replicatedChunk replicatedChunks[index] = replicatedChunk
}(sourceChunk, chunkIndex) }(sourceChunk, chunkIndex)
})
} }
wg.Wait() wg.Wait()

2
weed/replication/sink/filersink/filer_sink.go

@ -32,6 +32,7 @@ type FilerSink struct {
address string address string
writeChunkByFiler bool writeChunkByFiler bool
isIncremental bool isIncremental bool
executor *util.LimitedConcurrentExecutor
} }
func init() { func init() {
@ -53,6 +54,7 @@ func (fs *FilerSink) IsIncremental() bool {
func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error { func (fs *FilerSink) Initialize(configuration util.Configuration, prefix string) error {
fs.isIncremental = configuration.GetBool(prefix + "is_incremental") fs.isIncremental = configuration.GetBool(prefix + "is_incremental")
fs.dataCenter = configuration.GetString(prefix + "dataCenter") fs.dataCenter = configuration.GetString(prefix + "dataCenter")
fs.executor = util.NewLimitedConcurrentExecutor(32)
return fs.DoInitialize( return fs.DoInitialize(
"", "",
configuration.GetString(prefix+"grpcAddress"), configuration.GetString(prefix+"grpcAddress"),

Loading…
Cancel
Save