Browse Source

shell meta load add concurrency (#4529)

* fix: increase speed cmd fs meta load

* fix: add wg
pull/5144/head
Konstantin Lebedev 1 year ago
committed by GitHub
parent
commit
183352c796
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 32
      weed/shell/command_fs_meta_load.go

32
weed/shell/command_fs_meta_load.go

@ -6,6 +6,7 @@ import (
"io" "io"
"os" "os"
"strings" "strings"
"sync"
"time" "time"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
@ -47,6 +48,7 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
metaLoadCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) metaLoadCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
c.dirPrefix = metaLoadCommand.String("dirPrefix", "", "load entries only with directories matching prefix") c.dirPrefix = metaLoadCommand.String("dirPrefix", "", "load entries only with directories matching prefix")
concurrency := metaLoadCommand.Int("concurrency", 1, "number of parallel meta load to filer")
verbose := metaLoadCommand.Bool("v", true, "verbose mode") verbose := metaLoadCommand.Bool("v", true, "verbose mode")
if err = metaLoadCommand.Parse(args[0 : len(args)-1]); err != nil { if err = metaLoadCommand.Parse(args[0 : len(args)-1]); err != nil {
return nil return nil
@ -64,6 +66,9 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
sizeBuf := make([]byte, 4) sizeBuf := make([]byte, 4)
waitChan := make(chan struct{}, *concurrency)
defer close(waitChan)
var wg sync.WaitGroup
for { for {
if n, err := dst.Read(sizeBuf); n != 4 { if n, err := dst.Read(sizeBuf); n != 4 {
@ -105,21 +110,34 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
} }
fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x") fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x")
if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
if fullEntry.Entry.IsDirectory {
wg.Wait()
if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
Directory: fullEntry.Dir, Directory: fullEntry.Dir,
Entry: fullEntry.Entry, Entry: fullEntry.Entry,
}); err != nil {
return err
}); errEntry != nil {
return errEntry
} }
if fullEntry.Entry.IsDirectory {
dirCount++ dirCount++
} else { } else {
wg.Add(1)
waitChan <- struct{}{}
go func(entry *filer_pb.FullEntry) {
if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
Directory: entry.Dir,
Entry: entry.Entry,
}); errEntry != nil {
err = errEntry
}
defer wg.Done()
<-waitChan
}(fullEntry)
if err != nil {
return err
}
fileCount++ fileCount++
} }
} }
}) })
if err == nil { if err == nil {

Loading…
Cancel
Save