diff --git a/lib/model/folder_sendrecv.go b/lib/model/folder_sendrecv.go index a4a2cd0e3..e86011982 100644 --- a/lib/model/folder_sendrecv.go +++ b/lib/model/folder_sendrecv.go @@ -143,6 +143,9 @@ func newSendReceiveFolder(model *model, ignores *ignore.Matcher, cfg config.Fold f.puller = f if f.Copiers == 0 { + // "Copiers" is effectively the concurrency level for a number of + // different processes in the folder runner, not only the specific + // copy step. TODO: Rename this config option at some point. f.Copiers = defaultCopiers } @@ -243,10 +246,10 @@ func (f *sendReceiveFolder) pullerIteration(ctx context.Context, scanChan chan<- f.tempPullErrors = make(map[string]string) f.errorsMut.Unlock() - pullChan := make(chan pullBlockState) - copyChan := make(chan copyBlocksState) - finisherChan := make(chan *sharedPullerState) - dbUpdateChan := make(chan dbUpdateJob) + pullChan := make(chan pullBlockState, f.Copiers) + copyChan := make(chan copyBlocksState, f.Copiers) + finisherChan := make(chan *sharedPullerState, f.Copiers) + dbUpdateChan := make(chan dbUpdateJob, f.Copiers) var pullWg sync.WaitGroup var copyWg sync.WaitGroup @@ -274,9 +277,11 @@ func (f *sendReceiveFolder) pullerIteration(ctx context.Context, scanChan chan<- }) // finisherRoutine finishes when finisherChan is closed - doneWg.Go(func() { - f.finisherRoutine(ctx, finisherChan, dbUpdateChan, scanChan) - }) + for range f.Copiers { + doneWg.Go(func() { + f.finisherRoutine(ctx, finisherChan, dbUpdateChan, scanChan) + }) + } fileDeletions, dirDeletions, err := f.processNeeded(ctx, dbUpdateChan, copyChan, scanChan) @@ -1772,21 +1777,10 @@ func (f *sendReceiveFolder) dbUpdaterRoutine(dbUpdateChan <-chan dbUpdateJob) in tick := time.NewTicker(maxBatchTime) defer tick.Stop() batch := NewFileInfoBatch(func(files []protocol.FileInfo) error { - // sync directories - for dir := range changedDirs { - delete(changedDirs, dir) - if !f.DisableFsync { - fd, err := f.mtimefs.Open(dir) - if err != nil { - f.sl.Debug("Fsync failed", slogutil.FilePath(dir), slogutil.Error(err)) - continue - } - if err := fd.Sync(); err != nil { - f.sl.Debug("Fsync failed", slogutil.FilePath(dir), slogutil.Error(err)) - } - fd.Close() - } + if !f.DisableFsync { + f.fsyncDirs(changedDirs) } + clear(changedDirs) // All updates to file/folder objects that originated remotely // (across the network) use this call to updateLocals @@ -1840,6 +1834,27 @@ loop: return changed } +func (f *sendReceiveFolder) fsyncDirs(changedDirs map[string]struct{}) { + var wg sync.WaitGroup + sem := make(chan struct{}, f.Copiers) + for dir := range changedDirs { + sem <- struct{}{} + wg.Go(func() { + defer func() { <-sem }() + fd, err := f.mtimefs.Open(dir) + if err != nil { + f.sl.Debug("Fsync failed", slogutil.FilePath(dir), slogutil.Error(err)) + return + } + if err := fd.Sync(); err != nil { + f.sl.Debug("Fsync failed", slogutil.FilePath(dir), slogutil.Error(err)) + } + fd.Close() + }) + } + wg.Wait() +} + // pullScannerRoutine aggregates paths to be scanned after pulling. The scan is // scheduled once when scanChan is closed (scanning can not happen during pulling). func (f *sendReceiveFolder) pullScannerRoutine(ctx context.Context, scanChan <-chan string) { diff --git a/test/h1/config.xml b/test/h1/config.xml index 0853c8630..6b0ad2b4f 100644 --- a/test/h1/config.xml +++ b/test/h1/config.xml @@ -1,5 +1,5 @@ - - + + basic @@ -37,6 +37,7 @@ false false false + true 0 0 @@ -72,6 +73,8 @@ false abc123 default + 604800 + / @@ -165,6 +168,7 @@ false false false + true 1024 4096 diff --git a/test/h2/config.xml b/test/h2/config.xml index b2acd6efa..3711b3216 100644 --- a/test/h2/config.xml +++ b/test/h2/config.xml @@ -1,5 +1,5 @@ - + basic @@ -31,7 +31,7 @@ false standard standard - false + true true false false