rwfolder.go 49 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729
  1. // Copyright (C) 2014 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at http://mozilla.org/MPL/2.0/.
  6. package model
  7. import (
  8. "errors"
  9. "fmt"
  10. "math/rand"
  11. "os"
  12. "path/filepath"
  13. "runtime"
  14. "sort"
  15. "strings"
  16. "time"
  17. "github.com/syncthing/syncthing/lib/config"
  18. "github.com/syncthing/syncthing/lib/db"
  19. "github.com/syncthing/syncthing/lib/events"
  20. "github.com/syncthing/syncthing/lib/fs"
  21. "github.com/syncthing/syncthing/lib/ignore"
  22. "github.com/syncthing/syncthing/lib/osutil"
  23. "github.com/syncthing/syncthing/lib/protocol"
  24. "github.com/syncthing/syncthing/lib/scanner"
  25. "github.com/syncthing/syncthing/lib/symlinks"
  26. "github.com/syncthing/syncthing/lib/sync"
  27. "github.com/syncthing/syncthing/lib/versioner"
  28. )
  29. func init() {
  30. folderFactories[config.FolderTypeReadWrite] = newRWFolder
  31. }
  32. // A pullBlockState is passed to the puller routine for each block that needs
  33. // to be fetched.
  34. type pullBlockState struct {
  35. *sharedPullerState
  36. block protocol.BlockInfo
  37. }
  38. // A copyBlocksState is passed to copy routine if the file has blocks to be
  39. // copied.
  40. type copyBlocksState struct {
  41. *sharedPullerState
  42. blocks []protocol.BlockInfo
  43. }
  44. // Which filemode bits to preserve
  45. const retainBits = os.ModeSetgid | os.ModeSetuid | os.ModeSticky
  46. var (
  47. activity = newDeviceActivity()
  48. errNoDevice = errors.New("peers who had this file went away, or the file has changed while syncing. will retry later")
  49. )
  50. const (
  51. dbUpdateHandleDir = iota
  52. dbUpdateDeleteDir
  53. dbUpdateHandleFile
  54. dbUpdateDeleteFile
  55. dbUpdateShortcutFile
  56. dbUpdateHandleSymlink
  57. )
  58. const (
  59. defaultCopiers = 1
  60. defaultPullers = 16
  61. defaultPullerSleep = 10 * time.Second
  62. defaultPullerPause = 60 * time.Second
  63. )
  64. type dbUpdateJob struct {
  65. file protocol.FileInfo
  66. jobType int
  67. }
  68. type rwFolder struct {
  69. folder
  70. mtimeFS *fs.MtimeFS
  71. dir string
  72. versioner versioner.Versioner
  73. ignorePerms bool
  74. order config.PullOrder
  75. maxConflicts int
  76. sleep time.Duration
  77. pause time.Duration
  78. allowSparse bool
  79. checkFreeSpace bool
  80. ignoreDelete bool
  81. fsync bool
  82. copiers int
  83. pullers int
  84. queue *jobQueue
  85. dbUpdates chan dbUpdateJob
  86. pullTimer *time.Timer
  87. remoteIndex chan struct{} // An index update was received, we should re-evaluate needs
  88. errors map[string]string // path -> error string
  89. errorsMut sync.Mutex
  90. initialScanCompleted chan (struct{}) // exposed for testing
  91. }
  92. func newRWFolder(model *Model, cfg config.FolderConfiguration, ver versioner.Versioner, mtimeFS *fs.MtimeFS) service {
  93. f := &rwFolder{
  94. folder: folder{
  95. stateTracker: newStateTracker(cfg.ID),
  96. scan: newFolderScanner(cfg),
  97. stop: make(chan struct{}),
  98. model: model,
  99. },
  100. mtimeFS: mtimeFS,
  101. dir: cfg.Path(),
  102. versioner: ver,
  103. ignorePerms: cfg.IgnorePerms,
  104. copiers: cfg.Copiers,
  105. pullers: cfg.Pullers,
  106. order: cfg.Order,
  107. maxConflicts: cfg.MaxConflicts,
  108. allowSparse: !cfg.DisableSparseFiles,
  109. checkFreeSpace: cfg.MinDiskFreePct != 0,
  110. ignoreDelete: cfg.IgnoreDelete,
  111. fsync: cfg.Fsync,
  112. queue: newJobQueue(),
  113. pullTimer: time.NewTimer(time.Second),
  114. remoteIndex: make(chan struct{}, 1), // This needs to be 1-buffered so that we queue a notification if we're busy doing a pull when it comes.
  115. errorsMut: sync.NewMutex(),
  116. initialScanCompleted: make(chan struct{}),
  117. }
  118. f.configureCopiersAndPullers(cfg)
  119. return f
  120. }
  121. func (f *rwFolder) configureCopiersAndPullers(cfg config.FolderConfiguration) {
  122. if f.copiers == 0 {
  123. f.copiers = defaultCopiers
  124. }
  125. if f.pullers == 0 {
  126. f.pullers = defaultPullers
  127. }
  128. if cfg.PullerPauseS == 0 {
  129. f.pause = defaultPullerPause
  130. } else {
  131. f.pause = time.Duration(cfg.PullerPauseS) * time.Second
  132. }
  133. if cfg.PullerSleepS == 0 {
  134. f.sleep = defaultPullerSleep
  135. } else {
  136. f.sleep = time.Duration(cfg.PullerSleepS) * time.Second
  137. }
  138. }
  139. // Helper function to check whether either the ignorePerm flag has been
  140. // set on the local host or the FlagNoPermBits has been set on the file/dir
  141. // which is being pulled.
  142. func (f *rwFolder) ignorePermissions(file protocol.FileInfo) bool {
  143. return f.ignorePerms || file.NoPermissions
  144. }
  145. // Serve will run scans and pulls. It will return when Stop()ed or on a
  146. // critical error.
  147. func (f *rwFolder) Serve() {
  148. l.Debugln(f, "starting")
  149. defer l.Debugln(f, "exiting")
  150. defer func() {
  151. f.pullTimer.Stop()
  152. f.scan.timer.Stop()
  153. // TODO: Should there be an actual FolderStopped state?
  154. f.setState(FolderIdle)
  155. }()
  156. var prevSec int64
  157. var prevIgnoreHash string
  158. for {
  159. select {
  160. case <-f.stop:
  161. return
  162. case <-f.remoteIndex:
  163. prevSec = 0
  164. f.pullTimer.Reset(0)
  165. l.Debugln(f, "remote index updated, rescheduling pull")
  166. case <-f.pullTimer.C:
  167. select {
  168. case <-f.initialScanCompleted:
  169. default:
  170. // We don't start pulling files until a scan has been completed.
  171. l.Debugln(f, "skip (initial)")
  172. f.pullTimer.Reset(f.sleep)
  173. continue
  174. }
  175. f.model.fmut.RLock()
  176. curIgnores := f.model.folderIgnores[f.folderID]
  177. f.model.fmut.RUnlock()
  178. if newHash := curIgnores.Hash(); newHash != prevIgnoreHash {
  179. // The ignore patterns have changed. We need to re-evaluate if
  180. // there are files we need now that were ignored before.
  181. l.Debugln(f, "ignore patterns have changed, resetting prevVer")
  182. prevSec = 0
  183. prevIgnoreHash = newHash
  184. }
  185. // RemoteSequence() is a fast call, doesn't touch the database.
  186. curSeq, ok := f.model.RemoteSequence(f.folderID)
  187. if !ok || curSeq == prevSec {
  188. l.Debugln(f, "skip (curSeq == prevSeq)", prevSec, ok)
  189. f.pullTimer.Reset(f.sleep)
  190. continue
  191. }
  192. if err := f.model.CheckFolderHealth(f.folderID); err != nil {
  193. l.Infoln("Skipping folder", f.folderID, "pull due to folder error:", err)
  194. f.pullTimer.Reset(f.sleep)
  195. continue
  196. }
  197. l.Debugln(f, "pulling", prevSec, curSeq)
  198. f.setState(FolderSyncing)
  199. f.clearErrors()
  200. tries := 0
  201. for {
  202. tries++
  203. changed := f.pullerIteration(curIgnores)
  204. l.Debugln(f, "changed", changed)
  205. if changed == 0 {
  206. // No files were changed by the puller, so we are in
  207. // sync. Remember the local version number and
  208. // schedule a resync a little bit into the future.
  209. if lv, ok := f.model.RemoteSequence(f.folderID); ok && lv < curSeq {
  210. // There's a corner case where the device we needed
  211. // files from disconnected during the puller
  212. // iteration. The files will have been removed from
  213. // the index, so we've concluded that we don't need
  214. // them, but at the same time we have the local
  215. // version that includes those files in curVer. So we
  216. // catch the case that sequence might have
  217. // decreased here.
  218. l.Debugln(f, "adjusting curVer", lv)
  219. curSeq = lv
  220. }
  221. prevSec = curSeq
  222. l.Debugln(f, "next pull in", f.sleep)
  223. f.pullTimer.Reset(f.sleep)
  224. break
  225. }
  226. if tries > 10 {
  227. // We've tried a bunch of times to get in sync, but
  228. // we're not making it. Probably there are write
  229. // errors preventing us. Flag this with a warning and
  230. // wait a bit longer before retrying.
  231. l.Infof("Folder %q isn't making progress. Pausing puller for %v.", f.folderID, f.pause)
  232. l.Debugln(f, "next pull in", f.pause)
  233. if folderErrors := f.currentErrors(); len(folderErrors) > 0 {
  234. events.Default.Log(events.FolderErrors, map[string]interface{}{
  235. "folder": f.folderID,
  236. "errors": folderErrors,
  237. })
  238. }
  239. f.pullTimer.Reset(f.pause)
  240. break
  241. }
  242. }
  243. f.setState(FolderIdle)
  244. // The reason for running the scanner from within the puller is that
  245. // this is the easiest way to make sure we are not doing both at the
  246. // same time.
  247. case <-f.scan.timer.C:
  248. err := f.scanSubdirsIfHealthy(nil)
  249. f.scan.Reschedule()
  250. if err != nil {
  251. continue
  252. }
  253. select {
  254. case <-f.initialScanCompleted:
  255. default:
  256. l.Infoln("Completed initial scan (rw) of folder", f.folderID)
  257. close(f.initialScanCompleted)
  258. }
  259. case req := <-f.scan.now:
  260. req.err <- f.scanSubdirsIfHealthy(req.subdirs)
  261. case next := <-f.scan.delay:
  262. f.scan.timer.Reset(next)
  263. }
  264. }
  265. }
  266. func (f *rwFolder) IndexUpdated() {
  267. select {
  268. case f.remoteIndex <- struct{}{}:
  269. default:
  270. // We might be busy doing a pull and thus not reading from this
  271. // channel. The channel is 1-buffered, so one notification will be
  272. // queued to ensure we recheck after the pull, but beyond that we must
  273. // make sure to not block index receiving.
  274. }
  275. }
  276. func (f *rwFolder) String() string {
  277. return fmt.Sprintf("rwFolder/%s@%p", f.folderID, f)
  278. }
  279. // pullerIteration runs a single puller iteration for the given folder and
  280. // returns the number items that should have been synced (even those that
  281. // might have failed). One puller iteration handles all files currently
  282. // flagged as needed in the folder.
  283. func (f *rwFolder) pullerIteration(ignores *ignore.Matcher) int {
  284. pullChan := make(chan pullBlockState)
  285. copyChan := make(chan copyBlocksState)
  286. finisherChan := make(chan *sharedPullerState)
  287. updateWg := sync.NewWaitGroup()
  288. copyWg := sync.NewWaitGroup()
  289. pullWg := sync.NewWaitGroup()
  290. doneWg := sync.NewWaitGroup()
  291. l.Debugln(f, "c", f.copiers, "p", f.pullers)
  292. f.dbUpdates = make(chan dbUpdateJob)
  293. updateWg.Add(1)
  294. go func() {
  295. // dbUpdaterRoutine finishes when f.dbUpdates is closed
  296. f.dbUpdaterRoutine()
  297. updateWg.Done()
  298. }()
  299. for i := 0; i < f.copiers; i++ {
  300. copyWg.Add(1)
  301. go func() {
  302. // copierRoutine finishes when copyChan is closed
  303. f.copierRoutine(copyChan, pullChan, finisherChan)
  304. copyWg.Done()
  305. }()
  306. }
  307. for i := 0; i < f.pullers; i++ {
  308. pullWg.Add(1)
  309. go func() {
  310. // pullerRoutine finishes when pullChan is closed
  311. f.pullerRoutine(pullChan, finisherChan)
  312. pullWg.Done()
  313. }()
  314. }
  315. doneWg.Add(1)
  316. // finisherRoutine finishes when finisherChan is closed
  317. go func() {
  318. f.finisherRoutine(finisherChan)
  319. doneWg.Done()
  320. }()
  321. f.model.fmut.RLock()
  322. folderFiles := f.model.folderFiles[f.folderID]
  323. f.model.fmut.RUnlock()
  324. // !!!
  325. // WithNeed takes a database snapshot (by necessity). By the time we've
  326. // handled a bunch of files it might have become out of date and we might
  327. // be attempting to sync with an old version of a file...
  328. // !!!
  329. changed := 0
  330. fileDeletions := map[string]protocol.FileInfo{}
  331. dirDeletions := []protocol.FileInfo{}
  332. buckets := map[string][]protocol.FileInfo{}
  333. handleItem := func(fi protocol.FileInfo) bool {
  334. switch {
  335. case fi.IsDeleted():
  336. // A deleted file, directory or symlink
  337. if fi.IsDirectory() {
  338. dirDeletions = append(dirDeletions, fi)
  339. } else {
  340. fileDeletions[fi.Name] = fi
  341. df, ok := f.model.CurrentFolderFile(f.folderID, fi.Name)
  342. // Local file can be already deleted, but with a lower version
  343. // number, hence the deletion coming in again as part of
  344. // WithNeed, furthermore, the file can simply be of the wrong
  345. // type if we haven't yet managed to pull it.
  346. if ok && !df.IsDeleted() && !df.IsSymlink() && !df.IsDirectory() {
  347. // Put files into buckets per first hash
  348. key := string(df.Blocks[0].Hash)
  349. buckets[key] = append(buckets[key], df)
  350. }
  351. }
  352. changed++
  353. case fi.IsDirectory() && !fi.IsSymlink():
  354. l.Debugln("Handling directory", fi.Name)
  355. f.handleDir(fi)
  356. changed++
  357. case fi.IsSymlink():
  358. l.Debugln("Handling symlink", fi.Name)
  359. f.handleSymlink(fi)
  360. changed++
  361. default:
  362. return false
  363. }
  364. return true
  365. }
  366. folderFiles.WithNeed(protocol.LocalDeviceID, func(intf db.FileIntf) bool {
  367. // Needed items are delivered sorted lexicographically. We'll handle
  368. // directories as they come along, so parents before children. Files
  369. // are queued and the order may be changed later.
  370. if shouldIgnore(intf, ignores, f.ignoreDelete) {
  371. return true
  372. }
  373. if err := fileValid(intf); err != nil {
  374. // The file isn't valid so we can't process it. Pretend that we
  375. // tried and set the error for the file.
  376. f.newError(intf.FileName(), err)
  377. changed++
  378. return true
  379. }
  380. file := intf.(protocol.FileInfo)
  381. l.Debugln(f, "handling", file.Name)
  382. if !handleItem(file) {
  383. // A new or changed file or symlink. This is the only case where
  384. // we do stuff concurrently in the background. We only queue
  385. // files where we are connected to at least one device that has
  386. // the file.
  387. devices := folderFiles.Availability(file.Name)
  388. for _, dev := range devices {
  389. if f.model.ConnectedTo(dev) {
  390. f.queue.Push(file.Name, file.Size, file.ModTime())
  391. changed++
  392. break
  393. }
  394. }
  395. }
  396. return true
  397. })
  398. // Reorder the file queue according to configuration
  399. switch f.order {
  400. case config.OrderRandom:
  401. f.queue.Shuffle()
  402. case config.OrderAlphabetic:
  403. // The queue is already in alphabetic order.
  404. case config.OrderSmallestFirst:
  405. f.queue.SortSmallestFirst()
  406. case config.OrderLargestFirst:
  407. f.queue.SortLargestFirst()
  408. case config.OrderOldestFirst:
  409. f.queue.SortOldestFirst()
  410. case config.OrderNewestFirst:
  411. f.queue.SortNewestFirst()
  412. }
  413. // Process the file queue
  414. nextFile:
  415. for {
  416. select {
  417. case <-f.stop:
  418. // Stop processing files if the puller has been told to stop.
  419. break
  420. default:
  421. }
  422. fileName, ok := f.queue.Pop()
  423. if !ok {
  424. break
  425. }
  426. fi, ok := f.model.CurrentGlobalFile(f.folderID, fileName)
  427. if !ok {
  428. // File is no longer in the index. Mark it as done and drop it.
  429. f.queue.Done(fileName)
  430. continue
  431. }
  432. // Handles races where an index update arrives changing what the file
  433. // is between queueing and retrieving it from the queue, effectively
  434. // changing how the file should be handled.
  435. if handleItem(fi) {
  436. continue
  437. }
  438. // Check our list of files to be removed for a match, in which case
  439. // we can just do a rename instead.
  440. key := string(fi.Blocks[0].Hash)
  441. for i, candidate := range buckets[key] {
  442. if scanner.BlocksEqual(candidate.Blocks, fi.Blocks) {
  443. // Remove the candidate from the bucket
  444. lidx := len(buckets[key]) - 1
  445. buckets[key][i] = buckets[key][lidx]
  446. buckets[key] = buckets[key][:lidx]
  447. // candidate is our current state of the file, where as the
  448. // desired state with the delete bit set is in the deletion
  449. // map.
  450. desired := fileDeletions[candidate.Name]
  451. // Remove the pending deletion (as we perform it by renaming)
  452. delete(fileDeletions, candidate.Name)
  453. f.renameFile(desired, fi)
  454. f.queue.Done(fileName)
  455. continue nextFile
  456. }
  457. }
  458. // Handle the file normally, by coping and pulling, etc.
  459. f.handleFile(fi, copyChan, finisherChan)
  460. }
  461. // Signal copy and puller routines that we are done with the in data for
  462. // this iteration. Wait for them to finish.
  463. close(copyChan)
  464. copyWg.Wait()
  465. close(pullChan)
  466. pullWg.Wait()
  467. // Signal the finisher chan that there will be no more input.
  468. close(finisherChan)
  469. // Wait for the finisherChan to finish.
  470. doneWg.Wait()
  471. for _, file := range fileDeletions {
  472. l.Debugln("Deleting file", file.Name)
  473. f.deleteFile(file)
  474. }
  475. for i := range dirDeletions {
  476. dir := dirDeletions[len(dirDeletions)-i-1]
  477. l.Debugln("Deleting dir", dir.Name)
  478. f.deleteDir(dir, ignores)
  479. }
  480. // Wait for db updates to complete
  481. close(f.dbUpdates)
  482. updateWg.Wait()
  483. return changed
  484. }
  485. // handleDir creates or updates the given directory
  486. func (f *rwFolder) handleDir(file protocol.FileInfo) {
  487. // Used in the defer closure below, updated by the function body. Take
  488. // care not declare another err.
  489. var err error
  490. events.Default.Log(events.ItemStarted, map[string]string{
  491. "folder": f.folderID,
  492. "item": file.Name,
  493. "type": "dir",
  494. "action": "update",
  495. })
  496. defer func() {
  497. events.Default.Log(events.ItemFinished, map[string]interface{}{
  498. "folder": f.folderID,
  499. "item": file.Name,
  500. "error": events.Error(err),
  501. "type": "dir",
  502. "action": "update",
  503. })
  504. }()
  505. realName, err := rootedJoinedPath(f.dir, file.Name)
  506. if err != nil {
  507. f.newError(file.Name, err)
  508. return
  509. }
  510. mode := os.FileMode(file.Permissions & 0777)
  511. if f.ignorePermissions(file) {
  512. mode = 0777
  513. }
  514. if shouldDebug() {
  515. curFile, _ := f.model.CurrentFolderFile(f.folderID, file.Name)
  516. l.Debugf("need dir\n\t%v\n\t%v", file, curFile)
  517. }
  518. info, err := f.mtimeFS.Lstat(realName)
  519. switch {
  520. // There is already something under that name, but it's a file/link.
  521. // Most likely a file/link is getting replaced with a directory.
  522. // Remove the file/link and fall through to directory creation.
  523. case err == nil && (!info.IsDir() || info.Mode()&os.ModeSymlink != 0):
  524. err = osutil.InWritableDir(os.Remove, realName)
  525. if err != nil {
  526. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  527. f.newError(file.Name, err)
  528. return
  529. }
  530. fallthrough
  531. // The directory doesn't exist, so we create it with the right
  532. // mode bits from the start.
  533. case err != nil && os.IsNotExist(err):
  534. // We declare a function that acts on only the path name, so
  535. // we can pass it to InWritableDir. We use a regular Mkdir and
  536. // not MkdirAll because the parent should already exist.
  537. mkdir := func(path string) error {
  538. err = os.Mkdir(path, mode)
  539. if err != nil || f.ignorePermissions(file) {
  540. return err
  541. }
  542. // Stat the directory so we can check its permissions.
  543. info, err := f.mtimeFS.Lstat(path)
  544. if err != nil {
  545. return err
  546. }
  547. // Mask for the bits we want to preserve and add them in to the
  548. // directories permissions.
  549. return os.Chmod(path, mode|(info.Mode()&retainBits))
  550. }
  551. if err = osutil.InWritableDir(mkdir, realName); err == nil {
  552. f.dbUpdates <- dbUpdateJob{file, dbUpdateHandleDir}
  553. } else {
  554. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  555. f.newError(file.Name, err)
  556. }
  557. return
  558. // Weird error when stat()'ing the dir. Probably won't work to do
  559. // anything else with it if we can't even stat() it.
  560. case err != nil:
  561. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  562. f.newError(file.Name, err)
  563. return
  564. }
  565. // The directory already exists, so we just correct the mode bits. (We
  566. // don't handle modification times on directories, because that sucks...)
  567. // It's OK to change mode bits on stuff within non-writable directories.
  568. if f.ignorePermissions(file) {
  569. f.dbUpdates <- dbUpdateJob{file, dbUpdateHandleDir}
  570. } else if err := os.Chmod(realName, mode|(info.Mode()&retainBits)); err == nil {
  571. f.dbUpdates <- dbUpdateJob{file, dbUpdateHandleDir}
  572. } else {
  573. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  574. f.newError(file.Name, err)
  575. }
  576. }
  577. // handleSymlink creates or updates the given symlink
  578. func (f *rwFolder) handleSymlink(file protocol.FileInfo) {
  579. // Used in the defer closure below, updated by the function body. Take
  580. // care not declare another err.
  581. var err error
  582. events.Default.Log(events.ItemStarted, map[string]string{
  583. "folder": f.folderID,
  584. "item": file.Name,
  585. "type": "symlink",
  586. "action": "update",
  587. })
  588. defer func() {
  589. events.Default.Log(events.ItemFinished, map[string]interface{}{
  590. "folder": f.folderID,
  591. "item": file.Name,
  592. "error": events.Error(err),
  593. "type": "symlink",
  594. "action": "update",
  595. })
  596. }()
  597. realName, err := rootedJoinedPath(f.dir, file.Name)
  598. if err != nil {
  599. f.newError(file.Name, err)
  600. return
  601. }
  602. if shouldDebug() {
  603. curFile, _ := f.model.CurrentFolderFile(f.folderID, file.Name)
  604. l.Debugf("need symlink\n\t%v\n\t%v", file, curFile)
  605. }
  606. if len(file.SymlinkTarget) == 0 {
  607. // Index entry from a Syncthing predating the support for including
  608. // the link target in the index entry. We log this as an error.
  609. err = errors.New("incompatible symlink entry; rescan with newer Syncthing on source")
  610. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  611. f.newError(file.Name, err)
  612. return
  613. }
  614. if _, err = f.mtimeFS.Lstat(realName); err == nil {
  615. // There is already something under that name. Remove it to replace
  616. // with the symlink. This also handles the "change symlink type"
  617. // path.
  618. err = osutil.InWritableDir(os.Remove, realName)
  619. if err != nil {
  620. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  621. f.newError(file.Name, err)
  622. return
  623. }
  624. }
  625. tt := symlinks.TargetFile
  626. if file.IsDirectory() {
  627. tt = symlinks.TargetDirectory
  628. }
  629. // We declare a function that acts on only the path name, so
  630. // we can pass it to InWritableDir.
  631. createLink := func(path string) error {
  632. return symlinks.Create(path, file.SymlinkTarget, tt)
  633. }
  634. if err = osutil.InWritableDir(createLink, realName); err == nil {
  635. f.dbUpdates <- dbUpdateJob{file, dbUpdateHandleSymlink}
  636. } else {
  637. l.Infof("Puller (folder %q, dir %q): %v", f.folderID, file.Name, err)
  638. f.newError(file.Name, err)
  639. }
  640. }
  641. // deleteDir attempts to delete the given directory
  642. func (f *rwFolder) deleteDir(file protocol.FileInfo, matcher *ignore.Matcher) {
  643. // Used in the defer closure below, updated by the function body. Take
  644. // care not declare another err.
  645. var err error
  646. events.Default.Log(events.ItemStarted, map[string]string{
  647. "folder": f.folderID,
  648. "item": file.Name,
  649. "type": "dir",
  650. "action": "delete",
  651. })
  652. defer func() {
  653. events.Default.Log(events.ItemFinished, map[string]interface{}{
  654. "folder": f.folderID,
  655. "item": file.Name,
  656. "error": events.Error(err),
  657. "type": "dir",
  658. "action": "delete",
  659. })
  660. }()
  661. realName, err := rootedJoinedPath(f.dir, file.Name)
  662. if err != nil {
  663. f.newError(file.Name, err)
  664. return
  665. }
  666. // Delete any temporary files lying around in the directory
  667. dir, _ := os.Open(realName)
  668. if dir != nil {
  669. files, _ := dir.Readdirnames(-1)
  670. for _, dirFile := range files {
  671. fullDirFile := filepath.Join(file.Name, dirFile)
  672. if defTempNamer.IsTemporary(dirFile) || (matcher != nil && matcher.Match(fullDirFile).IsDeletable()) {
  673. os.RemoveAll(filepath.Join(f.dir, fullDirFile))
  674. }
  675. }
  676. dir.Close()
  677. }
  678. err = osutil.InWritableDir(os.Remove, realName)
  679. if err == nil || os.IsNotExist(err) {
  680. // It was removed or it doesn't exist to start with
  681. f.dbUpdates <- dbUpdateJob{file, dbUpdateDeleteDir}
  682. } else if _, serr := f.mtimeFS.Lstat(realName); serr != nil && !os.IsPermission(serr) {
  683. // We get an error just looking at the directory, and it's not a
  684. // permission problem. Lets assume the error is in fact some variant
  685. // of "file does not exist" (possibly expressed as some parent being a
  686. // file and not a directory etc) and that the delete is handled.
  687. f.dbUpdates <- dbUpdateJob{file, dbUpdateDeleteDir}
  688. } else {
  689. l.Infof("Puller (folder %q, dir %q): delete: %v", f.folderID, file.Name, err)
  690. f.newError(file.Name, err)
  691. }
  692. }
  693. // deleteFile attempts to delete the given file
  694. func (f *rwFolder) deleteFile(file protocol.FileInfo) {
  695. // Used in the defer closure below, updated by the function body. Take
  696. // care not declare another err.
  697. var err error
  698. events.Default.Log(events.ItemStarted, map[string]string{
  699. "folder": f.folderID,
  700. "item": file.Name,
  701. "type": "file",
  702. "action": "delete",
  703. })
  704. defer func() {
  705. events.Default.Log(events.ItemFinished, map[string]interface{}{
  706. "folder": f.folderID,
  707. "item": file.Name,
  708. "error": events.Error(err),
  709. "type": "file",
  710. "action": "delete",
  711. })
  712. }()
  713. realName, err := rootedJoinedPath(f.dir, file.Name)
  714. if err != nil {
  715. f.newError(file.Name, err)
  716. return
  717. }
  718. cur, ok := f.model.CurrentFolderFile(f.folderID, file.Name)
  719. if ok && f.inConflict(cur.Version, file.Version) {
  720. // There is a conflict here. Move the file to a conflict copy instead
  721. // of deleting. Also merge with the version vector we had, to indicate
  722. // we have resolved the conflict.
  723. file.Version = file.Version.Merge(cur.Version)
  724. err = osutil.InWritableDir(f.moveForConflict, realName)
  725. } else if f.versioner != nil {
  726. err = osutil.InWritableDir(f.versioner.Archive, realName)
  727. } else {
  728. err = osutil.InWritableDir(os.Remove, realName)
  729. }
  730. if err == nil || os.IsNotExist(err) {
  731. // It was removed or it doesn't exist to start with
  732. f.dbUpdates <- dbUpdateJob{file, dbUpdateDeleteFile}
  733. } else if _, serr := f.mtimeFS.Lstat(realName); serr != nil && !os.IsPermission(serr) {
  734. // We get an error just looking at the file, and it's not a permission
  735. // problem. Lets assume the error is in fact some variant of "file
  736. // does not exist" (possibly expressed as some parent being a file and
  737. // not a directory etc) and that the delete is handled.
  738. f.dbUpdates <- dbUpdateJob{file, dbUpdateDeleteFile}
  739. } else {
  740. l.Infof("Puller (folder %q, file %q): delete: %v", f.folderID, file.Name, err)
  741. f.newError(file.Name, err)
  742. }
  743. }
  744. // renameFile attempts to rename an existing file to a destination
  745. // and set the right attributes on it.
  746. func (f *rwFolder) renameFile(source, target protocol.FileInfo) {
  747. // Used in the defer closure below, updated by the function body. Take
  748. // care not declare another err.
  749. var err error
  750. events.Default.Log(events.ItemStarted, map[string]string{
  751. "folder": f.folderID,
  752. "item": source.Name,
  753. "type": "file",
  754. "action": "delete",
  755. })
  756. events.Default.Log(events.ItemStarted, map[string]string{
  757. "folder": f.folderID,
  758. "item": target.Name,
  759. "type": "file",
  760. "action": "update",
  761. })
  762. defer func() {
  763. events.Default.Log(events.ItemFinished, map[string]interface{}{
  764. "folder": f.folderID,
  765. "item": source.Name,
  766. "error": events.Error(err),
  767. "type": "file",
  768. "action": "delete",
  769. })
  770. events.Default.Log(events.ItemFinished, map[string]interface{}{
  771. "folder": f.folderID,
  772. "item": target.Name,
  773. "error": events.Error(err),
  774. "type": "file",
  775. "action": "update",
  776. })
  777. }()
  778. l.Debugln(f, "taking rename shortcut", source.Name, "->", target.Name)
  779. from, err := rootedJoinedPath(f.dir, source.Name)
  780. if err != nil {
  781. f.newError(source.Name, err)
  782. return
  783. }
  784. to, err := rootedJoinedPath(f.dir, target.Name)
  785. if err != nil {
  786. f.newError(target.Name, err)
  787. return
  788. }
  789. if f.versioner != nil {
  790. err = osutil.Copy(from, to)
  791. if err == nil {
  792. err = osutil.InWritableDir(f.versioner.Archive, from)
  793. }
  794. } else {
  795. err = osutil.TryRename(from, to)
  796. }
  797. if err == nil {
  798. // The file was renamed, so we have handled both the necessary delete
  799. // of the source and the creation of the target. Fix-up the metadata,
  800. // and update the local index of the target file.
  801. f.dbUpdates <- dbUpdateJob{source, dbUpdateDeleteFile}
  802. err = f.shortcutFile(target)
  803. if err != nil {
  804. l.Infof("Puller (folder %q, file %q): rename from %q metadata: %v", f.folderID, target.Name, source.Name, err)
  805. f.newError(target.Name, err)
  806. return
  807. }
  808. f.dbUpdates <- dbUpdateJob{target, dbUpdateHandleFile}
  809. } else {
  810. // We failed the rename so we have a source file that we still need to
  811. // get rid of. Attempt to delete it instead so that we make *some*
  812. // progress. The target is unhandled.
  813. err = osutil.InWritableDir(os.Remove, from)
  814. if err != nil {
  815. l.Infof("Puller (folder %q, file %q): delete %q after failed rename: %v", f.folderID, target.Name, source.Name, err)
  816. f.newError(target.Name, err)
  817. return
  818. }
  819. f.dbUpdates <- dbUpdateJob{source, dbUpdateDeleteFile}
  820. }
  821. }
  822. // This is the flow of data and events here, I think...
  823. //
  824. // +-----------------------+
  825. // | | - - - - > ItemStarted
  826. // | handleFile | - - - - > ItemFinished (on shortcuts)
  827. // | |
  828. // +-----------------------+
  829. // |
  830. // | copyChan (copyBlocksState; unless shortcut taken)
  831. // |
  832. // | +-----------------------+
  833. // | | +-----------------------+
  834. // +--->| | |
  835. // | | copierRoutine |
  836. // +-| |
  837. // +-----------------------+
  838. // |
  839. // | pullChan (sharedPullerState)
  840. // |
  841. // | +-----------------------+
  842. // | | +-----------------------+
  843. // +-->| | |
  844. // | | pullerRoutine |
  845. // +-| |
  846. // +-----------------------+
  847. // |
  848. // | finisherChan (sharedPullerState)
  849. // |
  850. // | +-----------------------+
  851. // | | |
  852. // +-->| finisherRoutine | - - - - > ItemFinished
  853. // | |
  854. // +-----------------------+
  855. // handleFile queues the copies and pulls as necessary for a single new or
  856. // changed file.
  857. func (f *rwFolder) handleFile(file protocol.FileInfo, copyChan chan<- copyBlocksState, finisherChan chan<- *sharedPullerState) {
  858. curFile, hasCurFile := f.model.CurrentFolderFile(f.folderID, file.Name)
  859. if hasCurFile && len(curFile.Blocks) == len(file.Blocks) && scanner.BlocksEqual(curFile.Blocks, file.Blocks) {
  860. // We are supposed to copy the entire file, and then fetch nothing. We
  861. // are only updating metadata, so we don't actually *need* to make the
  862. // copy.
  863. l.Debugln(f, "taking shortcut on", file.Name)
  864. events.Default.Log(events.ItemStarted, map[string]string{
  865. "folder": f.folderID,
  866. "item": file.Name,
  867. "type": "file",
  868. "action": "metadata",
  869. })
  870. f.queue.Done(file.Name)
  871. err := f.shortcutFile(file)
  872. events.Default.Log(events.ItemFinished, map[string]interface{}{
  873. "folder": f.folderID,
  874. "item": file.Name,
  875. "error": events.Error(err),
  876. "type": "file",
  877. "action": "metadata",
  878. })
  879. if err != nil {
  880. l.Infoln("Puller: shortcut:", err)
  881. f.newError(file.Name, err)
  882. } else {
  883. f.dbUpdates <- dbUpdateJob{file, dbUpdateShortcutFile}
  884. }
  885. return
  886. }
  887. // Figure out the absolute filenames we need once and for all
  888. tempName, err := rootedJoinedPath(f.dir, defTempNamer.TempName(file.Name))
  889. if err != nil {
  890. f.newError(file.Name, err)
  891. return
  892. }
  893. realName, err := rootedJoinedPath(f.dir, file.Name)
  894. if err != nil {
  895. f.newError(file.Name, err)
  896. return
  897. }
  898. if hasCurFile && !curFile.IsDirectory() && !curFile.IsSymlink() {
  899. // Check that the file on disk is what we expect it to be according to
  900. // the database. If there's a mismatch here, there might be local
  901. // changes that we don't know about yet and we should scan before
  902. // touching the file. If we can't stat the file we'll just pull it.
  903. if info, err := f.mtimeFS.Lstat(realName); err == nil {
  904. if !info.ModTime().Equal(curFile.ModTime()) || info.Size() != curFile.Size {
  905. l.Debugln("file modified but not rescanned; not pulling:", realName)
  906. // Scan() is synchronous (i.e. blocks until the scan is
  907. // completed and returns an error), but a scan can't happen
  908. // while we're in the puller routine. Request the scan in the
  909. // background and it'll be handled when the current pulling
  910. // sweep is complete. As we do retries, we'll queue the scan
  911. // for this file up to ten times, but the last nine of those
  912. // scans will be cheap...
  913. go f.scan.Scan([]string{file.Name})
  914. return
  915. }
  916. }
  917. }
  918. scanner.PopulateOffsets(file.Blocks)
  919. var blocks []protocol.BlockInfo
  920. var blocksSize int64
  921. var reused []int32
  922. // Check for an old temporary file which might have some blocks we could
  923. // reuse.
  924. tempBlocks, err := scanner.HashFile(tempName, protocol.BlockSize, nil)
  925. if err == nil {
  926. // Check for any reusable blocks in the temp file
  927. tempCopyBlocks, _ := scanner.BlockDiff(tempBlocks, file.Blocks)
  928. // block.String() returns a string unique to the block
  929. existingBlocks := make(map[string]struct{}, len(tempCopyBlocks))
  930. for _, block := range tempCopyBlocks {
  931. existingBlocks[block.String()] = struct{}{}
  932. }
  933. // Since the blocks are already there, we don't need to get them.
  934. for i, block := range file.Blocks {
  935. _, ok := existingBlocks[block.String()]
  936. if !ok {
  937. blocks = append(blocks, block)
  938. blocksSize += int64(block.Size)
  939. } else {
  940. reused = append(reused, int32(i))
  941. }
  942. }
  943. // The sharedpullerstate will know which flags to use when opening the
  944. // temp file depending if we are reusing any blocks or not.
  945. if len(reused) == 0 {
  946. // Otherwise, discard the file ourselves in order for the
  947. // sharedpuller not to panic when it fails to exclusively create a
  948. // file which already exists
  949. osutil.InWritableDir(os.Remove, tempName)
  950. }
  951. } else {
  952. // Copy the blocks, as we don't want to shuffle them on the FileInfo
  953. blocks = append(blocks, file.Blocks...)
  954. blocksSize = file.Size
  955. }
  956. if f.checkFreeSpace {
  957. if free, err := osutil.DiskFreeBytes(f.dir); err == nil && free < blocksSize {
  958. l.Warnf(`Folder "%s": insufficient disk space in %s for %s: have %.2f MiB, need %.2f MiB`, f.folderID, f.dir, file.Name, float64(free)/1024/1024, float64(blocksSize)/1024/1024)
  959. f.newError(file.Name, errors.New("insufficient space"))
  960. return
  961. }
  962. }
  963. // Shuffle the blocks
  964. for i := range blocks {
  965. j := rand.Intn(i + 1)
  966. blocks[i], blocks[j] = blocks[j], blocks[i]
  967. }
  968. events.Default.Log(events.ItemStarted, map[string]string{
  969. "folder": f.folderID,
  970. "item": file.Name,
  971. "type": "file",
  972. "action": "update",
  973. })
  974. s := sharedPullerState{
  975. file: file,
  976. folder: f.folderID,
  977. tempName: tempName,
  978. realName: realName,
  979. copyTotal: len(blocks),
  980. copyNeeded: len(blocks),
  981. reused: len(reused),
  982. updated: time.Now(),
  983. available: reused,
  984. availableUpdated: time.Now(),
  985. ignorePerms: f.ignorePermissions(file),
  986. version: curFile.Version,
  987. mut: sync.NewRWMutex(),
  988. sparse: f.allowSparse,
  989. created: time.Now(),
  990. }
  991. l.Debugf("%v need file %s; copy %d, reused %v", f, file.Name, len(blocks), reused)
  992. cs := copyBlocksState{
  993. sharedPullerState: &s,
  994. blocks: blocks,
  995. }
  996. copyChan <- cs
  997. }
  998. // shortcutFile sets file mode and modification time, when that's the only
  999. // thing that has changed.
  1000. func (f *rwFolder) shortcutFile(file protocol.FileInfo) error {
  1001. realName, err := rootedJoinedPath(f.dir, file.Name)
  1002. if err != nil {
  1003. f.newError(file.Name, err)
  1004. return err
  1005. }
  1006. if !f.ignorePermissions(file) {
  1007. if err := os.Chmod(realName, os.FileMode(file.Permissions&0777)); err != nil {
  1008. l.Infof("Puller (folder %q, file %q): shortcut: chmod: %v", f.folderID, file.Name, err)
  1009. f.newError(file.Name, err)
  1010. return err
  1011. }
  1012. }
  1013. f.mtimeFS.Chtimes(realName, file.ModTime(), file.ModTime()) // never fails
  1014. // This may have been a conflict. We should merge the version vectors so
  1015. // that our clock doesn't move backwards.
  1016. if cur, ok := f.model.CurrentFolderFile(f.folderID, file.Name); ok {
  1017. file.Version = file.Version.Merge(cur.Version)
  1018. }
  1019. return nil
  1020. }
  1021. // copierRoutine reads copierStates until the in channel closes and performs
  1022. // the relevant copies when possible, or passes it to the puller routine.
  1023. func (f *rwFolder) copierRoutine(in <-chan copyBlocksState, pullChan chan<- pullBlockState, out chan<- *sharedPullerState) {
  1024. buf := make([]byte, protocol.BlockSize)
  1025. for state := range in {
  1026. dstFd, err := state.tempFile()
  1027. if err != nil {
  1028. // Nothing more to do for this failed file, since we couldn't create a temporary for it.
  1029. out <- state.sharedPullerState
  1030. continue
  1031. }
  1032. if f.model.progressEmitter != nil {
  1033. f.model.progressEmitter.Register(state.sharedPullerState)
  1034. }
  1035. folderRoots := make(map[string]string)
  1036. var folders []string
  1037. f.model.fmut.RLock()
  1038. for folder, cfg := range f.model.folderCfgs {
  1039. folderRoots[folder] = cfg.Path()
  1040. folders = append(folders, folder)
  1041. }
  1042. f.model.fmut.RUnlock()
  1043. for _, block := range state.blocks {
  1044. if f.allowSparse && state.reused == 0 && block.IsEmpty() {
  1045. // The block is a block of all zeroes, and we are not reusing
  1046. // a temp file, so there is no need to do anything with it.
  1047. // If we were reusing a temp file and had this block to copy,
  1048. // it would be because the block in the temp file was *not* a
  1049. // block of all zeroes, so then we should not skip it.
  1050. // Pretend we copied it.
  1051. state.copiedFromOrigin()
  1052. continue
  1053. }
  1054. buf = buf[:int(block.Size)]
  1055. found := f.model.finder.Iterate(folders, block.Hash, func(folder, file string, index int32) bool {
  1056. inFile, err := rootedJoinedPath(folderRoots[folder], file)
  1057. if err != nil {
  1058. return false
  1059. }
  1060. fd, err := os.Open(inFile)
  1061. if err != nil {
  1062. return false
  1063. }
  1064. _, err = fd.ReadAt(buf, protocol.BlockSize*int64(index))
  1065. fd.Close()
  1066. if err != nil {
  1067. return false
  1068. }
  1069. hash, err := scanner.VerifyBuffer(buf, block)
  1070. if err != nil {
  1071. if hash != nil {
  1072. l.Debugf("Finder block mismatch in %s:%s:%d expected %q got %q", folder, file, index, block.Hash, hash)
  1073. err = f.model.finder.Fix(folder, file, index, block.Hash, hash)
  1074. if err != nil {
  1075. l.Warnln("finder fix:", err)
  1076. }
  1077. } else {
  1078. l.Debugln("Finder failed to verify buffer", err)
  1079. }
  1080. return false
  1081. }
  1082. _, err = dstFd.WriteAt(buf, block.Offset)
  1083. if err != nil {
  1084. state.fail("dst write", err)
  1085. }
  1086. if file == state.file.Name {
  1087. state.copiedFromOrigin()
  1088. }
  1089. return true
  1090. })
  1091. if state.failed() != nil {
  1092. break
  1093. }
  1094. if !found {
  1095. state.pullStarted()
  1096. ps := pullBlockState{
  1097. sharedPullerState: state.sharedPullerState,
  1098. block: block,
  1099. }
  1100. pullChan <- ps
  1101. } else {
  1102. state.copyDone(block)
  1103. }
  1104. }
  1105. out <- state.sharedPullerState
  1106. }
  1107. }
  1108. func (f *rwFolder) pullerRoutine(in <-chan pullBlockState, out chan<- *sharedPullerState) {
  1109. for state := range in {
  1110. if state.failed() != nil {
  1111. out <- state.sharedPullerState
  1112. continue
  1113. }
  1114. // Get an fd to the temporary file. Technically we don't need it until
  1115. // after fetching the block, but if we run into an error here there is
  1116. // no point in issuing the request to the network.
  1117. fd, err := state.tempFile()
  1118. if err != nil {
  1119. out <- state.sharedPullerState
  1120. continue
  1121. }
  1122. if f.allowSparse && state.reused == 0 && state.block.IsEmpty() {
  1123. // There is no need to request a block of all zeroes. Pretend we
  1124. // requested it and handled it correctly.
  1125. state.pullDone(state.block)
  1126. out <- state.sharedPullerState
  1127. continue
  1128. }
  1129. var lastError error
  1130. candidates := f.model.Availability(f.folderID, state.file.Name, state.file.Version, state.block)
  1131. for {
  1132. // Select the least busy device to pull the block from. If we found no
  1133. // feasible device at all, fail the block (and in the long run, the
  1134. // file).
  1135. selected, found := activity.leastBusy(candidates)
  1136. if !found {
  1137. if lastError != nil {
  1138. state.fail("pull", lastError)
  1139. } else {
  1140. state.fail("pull", errNoDevice)
  1141. }
  1142. break
  1143. }
  1144. candidates = removeAvailability(candidates, selected)
  1145. // Fetch the block, while marking the selected device as in use so that
  1146. // leastBusy can select another device when someone else asks.
  1147. activity.using(selected)
  1148. buf, lastError := f.model.requestGlobal(selected.ID, f.folderID, state.file.Name, state.block.Offset, int(state.block.Size), state.block.Hash, selected.FromTemporary)
  1149. activity.done(selected)
  1150. if lastError != nil {
  1151. l.Debugln("request:", f.folderID, state.file.Name, state.block.Offset, state.block.Size, "returned error:", lastError)
  1152. continue
  1153. }
  1154. // Verify that the received block matches the desired hash, if not
  1155. // try pulling it from another device.
  1156. _, lastError = scanner.VerifyBuffer(buf, state.block)
  1157. if lastError != nil {
  1158. l.Debugln("request:", f.folderID, state.file.Name, state.block.Offset, state.block.Size, "hash mismatch")
  1159. continue
  1160. }
  1161. // Save the block data we got from the cluster
  1162. _, err = fd.WriteAt(buf, state.block.Offset)
  1163. if err != nil {
  1164. state.fail("save", err)
  1165. } else {
  1166. state.pullDone(state.block)
  1167. }
  1168. break
  1169. }
  1170. out <- state.sharedPullerState
  1171. }
  1172. }
  1173. func (f *rwFolder) performFinish(state *sharedPullerState) error {
  1174. // Set the correct permission bits on the new file
  1175. if !f.ignorePermissions(state.file) {
  1176. if err := os.Chmod(state.tempName, os.FileMode(state.file.Permissions&0777)); err != nil {
  1177. return err
  1178. }
  1179. }
  1180. if stat, err := f.mtimeFS.Lstat(state.realName); err == nil {
  1181. // There is an old file or directory already in place. We need to
  1182. // handle that.
  1183. switch {
  1184. case stat.IsDir() || stat.Mode()&os.ModeSymlink != 0:
  1185. // It's a directory or a symlink. These are not versioned or
  1186. // archived for conflicts, only removed (which of course fails for
  1187. // non-empty directories).
  1188. // TODO: This is the place where we want to remove temporary files
  1189. // and future hard ignores before attempting a directory delete.
  1190. // Should share code with f.deletDir().
  1191. if err = osutil.InWritableDir(os.Remove, state.realName); err != nil {
  1192. return err
  1193. }
  1194. case f.inConflict(state.version, state.file.Version):
  1195. // The new file has been changed in conflict with the existing one. We
  1196. // should file it away as a conflict instead of just removing or
  1197. // archiving. Also merge with the version vector we had, to indicate
  1198. // we have resolved the conflict.
  1199. state.file.Version = state.file.Version.Merge(state.version)
  1200. if err = osutil.InWritableDir(f.moveForConflict, state.realName); err != nil {
  1201. return err
  1202. }
  1203. case f.versioner != nil:
  1204. // If we should use versioning, let the versioner archive the old
  1205. // file before we replace it. Archiving a non-existent file is not
  1206. // an error.
  1207. if err = f.versioner.Archive(state.realName); err != nil {
  1208. return err
  1209. }
  1210. }
  1211. }
  1212. // Replace the original content with the new one. If it didn't work,
  1213. // leave the temp file in place for reuse.
  1214. if err := osutil.TryRename(state.tempName, state.realName); err != nil {
  1215. return err
  1216. }
  1217. // Set the correct timestamp on the new file
  1218. f.mtimeFS.Chtimes(state.realName, state.file.ModTime(), state.file.ModTime()) // never fails
  1219. // Record the updated file in the index
  1220. f.dbUpdates <- dbUpdateJob{state.file, dbUpdateHandleFile}
  1221. return nil
  1222. }
  1223. func (f *rwFolder) finisherRoutine(in <-chan *sharedPullerState) {
  1224. for state := range in {
  1225. if closed, err := state.finalClose(); closed {
  1226. l.Debugln(f, "closing", state.file.Name)
  1227. f.queue.Done(state.file.Name)
  1228. if err == nil {
  1229. err = f.performFinish(state)
  1230. }
  1231. if err != nil {
  1232. l.Infoln("Puller: final:", err)
  1233. f.newError(state.file.Name, err)
  1234. }
  1235. events.Default.Log(events.ItemFinished, map[string]interface{}{
  1236. "folder": f.folderID,
  1237. "item": state.file.Name,
  1238. "error": events.Error(err),
  1239. "type": "file",
  1240. "action": "update",
  1241. })
  1242. if f.model.progressEmitter != nil {
  1243. f.model.progressEmitter.Deregister(state)
  1244. }
  1245. }
  1246. }
  1247. }
  1248. // Moves the given filename to the front of the job queue
  1249. func (f *rwFolder) BringToFront(filename string) {
  1250. f.queue.BringToFront(filename)
  1251. }
  1252. func (f *rwFolder) Jobs() ([]string, []string) {
  1253. return f.queue.Jobs()
  1254. }
  1255. // dbUpdaterRoutine aggregates db updates and commits them in batches no
  1256. // larger than 1000 items, and no more delayed than 2 seconds.
  1257. func (f *rwFolder) dbUpdaterRoutine() {
  1258. const (
  1259. maxBatchSize = 1000
  1260. maxBatchTime = 2 * time.Second
  1261. )
  1262. batch := make([]dbUpdateJob, 0, maxBatchSize)
  1263. files := make([]protocol.FileInfo, 0, maxBatchSize)
  1264. tick := time.NewTicker(maxBatchTime)
  1265. defer tick.Stop()
  1266. var changedFiles []string
  1267. var changedDirs []string
  1268. if f.fsync {
  1269. changedFiles = make([]string, 0, maxBatchSize)
  1270. changedDirs = make([]string, 0, maxBatchSize)
  1271. }
  1272. syncFilesOnce := func(files []string, syncFn func(string) error) {
  1273. sort.Strings(files)
  1274. var lastFile string
  1275. for _, file := range files {
  1276. if lastFile == file {
  1277. continue
  1278. }
  1279. lastFile = file
  1280. if err := syncFn(file); err != nil {
  1281. l.Infof("fsync %q failed: %v", file, err)
  1282. }
  1283. }
  1284. }
  1285. handleBatch := func() {
  1286. found := false
  1287. var lastFile protocol.FileInfo
  1288. for _, job := range batch {
  1289. files = append(files, job.file)
  1290. if f.fsync {
  1291. // collect changed files and dirs
  1292. switch job.jobType {
  1293. case dbUpdateHandleFile, dbUpdateShortcutFile:
  1294. changedFiles = append(changedFiles, filepath.Join(f.dir, job.file.Name))
  1295. case dbUpdateHandleDir:
  1296. changedDirs = append(changedDirs, filepath.Join(f.dir, job.file.Name))
  1297. case dbUpdateHandleSymlink:
  1298. // fsyncing symlinks is only supported by MacOS, ignore
  1299. }
  1300. if job.jobType != dbUpdateShortcutFile {
  1301. changedDirs = append(changedDirs, filepath.Dir(filepath.Join(f.dir, job.file.Name)))
  1302. }
  1303. }
  1304. if job.file.IsInvalid() || (job.file.IsDirectory() && !job.file.IsSymlink()) {
  1305. continue
  1306. }
  1307. if job.jobType&(dbUpdateHandleFile|dbUpdateDeleteFile) == 0 {
  1308. continue
  1309. }
  1310. found = true
  1311. lastFile = job.file
  1312. }
  1313. if f.fsync {
  1314. // sync files and dirs to disk
  1315. syncFilesOnce(changedFiles, osutil.SyncFile)
  1316. changedFiles = changedFiles[:0]
  1317. syncFilesOnce(changedDirs, osutil.SyncDir)
  1318. changedDirs = changedDirs[:0]
  1319. }
  1320. // All updates to file/folder objects that originated remotely
  1321. // (across the network) use this call to updateLocals
  1322. f.model.updateLocalsFromPulling(f.folderID, files)
  1323. if found {
  1324. f.model.receivedFile(f.folderID, lastFile)
  1325. }
  1326. batch = batch[:0]
  1327. files = files[:0]
  1328. }
  1329. loop:
  1330. for {
  1331. select {
  1332. case job, ok := <-f.dbUpdates:
  1333. if !ok {
  1334. break loop
  1335. }
  1336. job.file.Sequence = 0
  1337. batch = append(batch, job)
  1338. if len(batch) == maxBatchSize {
  1339. handleBatch()
  1340. }
  1341. case <-tick.C:
  1342. if len(batch) > 0 {
  1343. handleBatch()
  1344. }
  1345. }
  1346. }
  1347. if len(batch) > 0 {
  1348. handleBatch()
  1349. }
  1350. }
  1351. func (f *rwFolder) inConflict(current, replacement protocol.Vector) bool {
  1352. if current.Concurrent(replacement) {
  1353. // Obvious case
  1354. return true
  1355. }
  1356. if replacement.Counter(f.model.shortID) > current.Counter(f.model.shortID) {
  1357. // The replacement file contains a higher version for ourselves than
  1358. // what we have. This isn't supposed to be possible, since it's only
  1359. // we who can increment that counter. We take it as a sign that
  1360. // something is wrong (our index may have been corrupted or removed)
  1361. // and flag it as a conflict.
  1362. return true
  1363. }
  1364. return false
  1365. }
  1366. func removeAvailability(availabilities []Availability, availability Availability) []Availability {
  1367. for i := range availabilities {
  1368. if availabilities[i] == availability {
  1369. availabilities[i] = availabilities[len(availabilities)-1]
  1370. return availabilities[:len(availabilities)-1]
  1371. }
  1372. }
  1373. return availabilities
  1374. }
  1375. func (f *rwFolder) moveForConflict(name string) error {
  1376. if strings.Contains(filepath.Base(name), ".sync-conflict-") {
  1377. l.Infoln("Conflict for", name, "which is already a conflict copy; not copying again.")
  1378. if err := os.Remove(name); err != nil && !os.IsNotExist(err) {
  1379. return err
  1380. }
  1381. return nil
  1382. }
  1383. if f.maxConflicts == 0 {
  1384. if err := os.Remove(name); err != nil && !os.IsNotExist(err) {
  1385. return err
  1386. }
  1387. return nil
  1388. }
  1389. ext := filepath.Ext(name)
  1390. withoutExt := name[:len(name)-len(ext)]
  1391. newName := withoutExt + time.Now().Format(".sync-conflict-20060102-150405") + ext
  1392. err := os.Rename(name, newName)
  1393. if os.IsNotExist(err) {
  1394. // We were supposed to move a file away but it does not exist. Either
  1395. // the user has already moved it away, or the conflict was between a
  1396. // remote modification and a local delete. In either way it does not
  1397. // matter, go ahead as if the move succeeded.
  1398. err = nil
  1399. }
  1400. if f.maxConflicts > -1 {
  1401. matches, gerr := osutil.Glob(withoutExt + ".sync-conflict-????????-??????" + ext)
  1402. if gerr == nil && len(matches) > f.maxConflicts {
  1403. sort.Sort(sort.Reverse(sort.StringSlice(matches)))
  1404. for _, match := range matches[f.maxConflicts:] {
  1405. gerr = os.Remove(match)
  1406. if gerr != nil {
  1407. l.Debugln(f, "removing extra conflict", gerr)
  1408. }
  1409. }
  1410. } else if gerr != nil {
  1411. l.Debugln(f, "globbing for conflicts", gerr)
  1412. }
  1413. }
  1414. return err
  1415. }
  1416. func (f *rwFolder) newError(path string, err error) {
  1417. f.errorsMut.Lock()
  1418. defer f.errorsMut.Unlock()
  1419. // We might get more than one error report for a file (i.e. error on
  1420. // Write() followed by Close()); we keep the first error as that is
  1421. // probably closer to the root cause.
  1422. if _, ok := f.errors[path]; ok {
  1423. return
  1424. }
  1425. f.errors[path] = err.Error()
  1426. }
  1427. func (f *rwFolder) clearErrors() {
  1428. f.errorsMut.Lock()
  1429. f.errors = make(map[string]string)
  1430. f.errorsMut.Unlock()
  1431. }
  1432. func (f *rwFolder) currentErrors() []fileError {
  1433. f.errorsMut.Lock()
  1434. errors := make([]fileError, 0, len(f.errors))
  1435. for path, err := range f.errors {
  1436. errors = append(errors, fileError{path, err})
  1437. }
  1438. sort.Sort(fileErrorList(errors))
  1439. f.errorsMut.Unlock()
  1440. return errors
  1441. }
  1442. // A []fileError is sent as part of an event and will be JSON serialized.
  1443. type fileError struct {
  1444. Path string `json:"path"`
  1445. Err string `json:"error"`
  1446. }
  1447. type fileErrorList []fileError
  1448. func (l fileErrorList) Len() int {
  1449. return len(l)
  1450. }
  1451. func (l fileErrorList) Less(a, b int) bool {
  1452. return l[a].Path < l[b].Path
  1453. }
  1454. func (l fileErrorList) Swap(a, b int) {
  1455. l[a], l[b] = l[b], l[a]
  1456. }
  1457. // fileValid returns nil when the file is valid for processing, or an error if it's not
  1458. func fileValid(file db.FileIntf) error {
  1459. switch {
  1460. case file.IsDeleted():
  1461. // We don't care about file validity if we're not supposed to have it
  1462. return nil
  1463. case !symlinks.Supported && file.IsSymlink():
  1464. return errUnsupportedSymlink
  1465. case runtime.GOOS == "windows" && windowsInvalidFilename(file.FileName()):
  1466. return errInvalidFilename
  1467. }
  1468. return nil
  1469. }
  1470. var windowsDisallowedCharacters = string([]rune{
  1471. '<', '>', ':', '"', '|', '?', '*',
  1472. 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10,
  1473. 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,
  1474. 21, 22, 23, 24, 25, 26, 27, 28, 29, 30,
  1475. 31,
  1476. })
  1477. func windowsInvalidFilename(name string) bool {
  1478. // None of the path components should end in space
  1479. for _, part := range strings.Split(name, `\`) {
  1480. if len(part) == 0 {
  1481. continue
  1482. }
  1483. if part[len(part)-1] == ' ' {
  1484. // Names ending in space are not valid.
  1485. return true
  1486. }
  1487. }
  1488. // The path must not contain any disallowed characters
  1489. return strings.ContainsAny(name, windowsDisallowedCharacters)
  1490. }