Browse Source

fix: #1226

pull/1228/head
HongyanShen 5 years ago
parent
commit
81610ed006
  1. 11
      weed/replication/sink/filersink/fetch_write.go
  2. 10
      weed/replication/sink/s3sink/s3_sink.go

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

@ -19,17 +19,20 @@ func (fs *FilerSink) replicateChunks(sourceChunks []*filer_pb.FileChunk, dir str
if len(sourceChunks) == 0 { if len(sourceChunks) == 0 {
return return
} }
replicatedChunks = make([]*filer_pb.FileChunk, len(sourceChunks))
var wg sync.WaitGroup var wg sync.WaitGroup
for _, sourceChunk := range sourceChunks {
for chunkIndex, sourceChunk := range sourceChunks {
wg.Add(1) wg.Add(1)
go func(chunk *filer_pb.FileChunk) {
go func(chunk *filer_pb.FileChunk, index int) {
defer wg.Done() defer wg.Done()
replicatedChunk, e := fs.replicateOneChunk(chunk, dir) replicatedChunk, e := fs.replicateOneChunk(chunk, dir)
if e != nil { if e != nil {
err = e err = e
} }
replicatedChunks = append(replicatedChunks, replicatedChunk)
}(sourceChunk)
replicatedChunks[index] = replicatedChunk
}(sourceChunk, chunkIndex)
} }
wg.Wait() wg.Wait()

10
weed/replication/sink/s3sink/s3_sink.go

@ -91,7 +91,6 @@ func (s3sink *S3Sink) DeleteEntry(key string, isDirectory, deleteIncludeChunks b
} }
func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry) error { func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
key = cleanKey(key) key = cleanKey(key)
if entry.IsDirectory { if entry.IsDirectory {
@ -106,19 +105,20 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry) error {
totalSize := filer2.TotalSize(entry.Chunks) totalSize := filer2.TotalSize(entry.Chunks)
chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize)) chunkViews := filer2.ViewFromChunks(entry.Chunks, 0, int(totalSize))
var parts []*s3.CompletedPart
parts := make([]*s3.CompletedPart, len(chunkViews))
var wg sync.WaitGroup var wg sync.WaitGroup
for chunkIndex, chunk := range chunkViews { for chunkIndex, chunk := range chunkViews {
partId := chunkIndex + 1 partId := chunkIndex + 1
wg.Add(1) wg.Add(1)
go func(chunk *filer2.ChunkView) {
go func(chunk *filer2.ChunkView, index int) {
defer wg.Done() defer wg.Done()
if part, uploadErr := s3sink.uploadPart(key, uploadId, partId, chunk); uploadErr != nil { if part, uploadErr := s3sink.uploadPart(key, uploadId, partId, chunk); uploadErr != nil {
err = uploadErr err = uploadErr
} else { } else {
parts = append(parts, part)
parts[index] = part
} }
}(chunk)
}(chunk, chunkIndex)
} }
wg.Wait() wg.Wait()

Loading…
Cancel
Save