model.go 34 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292
  1. // Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
  2. //
  3. // This program is free software: you can redistribute it and/or modify it
  4. // under the terms of the GNU General Public License as published by the Free
  5. // Software Foundation, either version 3 of the License, or (at your option)
  6. // any later version.
  7. //
  8. // This program is distributed in the hope that it will be useful, but WITHOUT
  9. // ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10. // FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  11. // more details.
  12. //
  13. // You should have received a copy of the GNU General Public License along
  14. // with this program. If not, see <http://www.gnu.org/licenses/>.
  15. package model
  16. import (
  17. "bufio"
  18. "crypto/tls"
  19. "errors"
  20. "fmt"
  21. "io"
  22. "io/ioutil"
  23. "net"
  24. "os"
  25. "path/filepath"
  26. "strconv"
  27. "strings"
  28. "sync"
  29. "time"
  30. "github.com/syncthing/syncthing/internal/config"
  31. "github.com/syncthing/syncthing/internal/events"
  32. "github.com/syncthing/syncthing/internal/files"
  33. "github.com/syncthing/syncthing/internal/ignore"
  34. "github.com/syncthing/syncthing/internal/lamport"
  35. "github.com/syncthing/syncthing/internal/osutil"
  36. "github.com/syncthing/syncthing/internal/protocol"
  37. "github.com/syncthing/syncthing/internal/scanner"
  38. "github.com/syncthing/syncthing/internal/stats"
  39. "github.com/syncthing/syncthing/internal/versioner"
  40. "github.com/syndtr/goleveldb/leveldb"
  41. )
  42. type folderState int
  43. const (
  44. FolderIdle folderState = iota
  45. FolderScanning
  46. FolderSyncing
  47. FolderCleaning
  48. )
  49. func (s folderState) String() string {
  50. switch s {
  51. case FolderIdle:
  52. return "idle"
  53. case FolderScanning:
  54. return "scanning"
  55. case FolderCleaning:
  56. return "cleaning"
  57. case FolderSyncing:
  58. return "syncing"
  59. default:
  60. return "unknown"
  61. }
  62. }
  63. // How many files to send in each Index/IndexUpdate message.
  64. const (
  65. indexTargetSize = 250 * 1024 // Aim for making index messages no larger than 250 KiB (uncompressed)
  66. indexPerFileSize = 250 // Each FileInfo is approximately this big, in bytes, excluding BlockInfos
  67. IndexPerBlockSize = 40 // Each BlockInfo is approximately this big
  68. indexBatchSize = 1000 // Either way, don't include more files than this
  69. )
  70. type service interface {
  71. Serve()
  72. Stop()
  73. }
  74. type Model struct {
  75. cfg *config.ConfigWrapper
  76. db *leveldb.DB
  77. deviceName string
  78. clientName string
  79. clientVersion string
  80. folderCfgs map[string]config.FolderConfiguration // folder -> cfg
  81. folderFiles map[string]*files.Set // folder -> files
  82. folderDevices map[string][]protocol.DeviceID // folder -> deviceIDs
  83. deviceFolders map[protocol.DeviceID][]string // deviceID -> folders
  84. deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
  85. folderIgnores map[string]ignore.Patterns // folder -> list of ignore patterns
  86. folderRunners map[string]service // folder -> puller or scanner
  87. fmut sync.RWMutex // protects the above
  88. folderState map[string]folderState // folder -> state
  89. folderStateChanged map[string]time.Time // folder -> time when state changed
  90. smut sync.RWMutex
  91. protoConn map[protocol.DeviceID]protocol.Connection
  92. rawConn map[protocol.DeviceID]io.Closer
  93. deviceVer map[protocol.DeviceID]string
  94. pmut sync.RWMutex // protects protoConn and rawConn
  95. addedFolder bool
  96. started bool
  97. }
  98. var (
  99. ErrNoSuchFile = errors.New("no such file")
  100. ErrInvalid = errors.New("file is invalid")
  101. )
  102. // NewModel creates and starts a new model. The model starts in read-only mode,
  103. // where it sends index information to connected peers and responds to requests
  104. // for file data without altering the local folder in any way.
  105. func NewModel(cfg *config.ConfigWrapper, deviceName, clientName, clientVersion string, db *leveldb.DB) *Model {
  106. m := &Model{
  107. cfg: cfg,
  108. db: db,
  109. deviceName: deviceName,
  110. clientName: clientName,
  111. clientVersion: clientVersion,
  112. folderCfgs: make(map[string]config.FolderConfiguration),
  113. folderFiles: make(map[string]*files.Set),
  114. folderDevices: make(map[string][]protocol.DeviceID),
  115. deviceFolders: make(map[protocol.DeviceID][]string),
  116. deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
  117. folderIgnores: make(map[string]ignore.Patterns),
  118. folderRunners: make(map[string]service),
  119. folderState: make(map[string]folderState),
  120. folderStateChanged: make(map[string]time.Time),
  121. protoConn: make(map[protocol.DeviceID]protocol.Connection),
  122. rawConn: make(map[protocol.DeviceID]io.Closer),
  123. deviceVer: make(map[protocol.DeviceID]string),
  124. }
  125. var timeout = 20 * 60 // seconds
  126. if t := os.Getenv("STDEADLOCKTIMEOUT"); len(t) > 0 {
  127. it, err := strconv.Atoi(t)
  128. if err == nil {
  129. timeout = it
  130. }
  131. }
  132. deadlockDetect(&m.fmut, time.Duration(timeout)*time.Second)
  133. deadlockDetect(&m.smut, time.Duration(timeout)*time.Second)
  134. deadlockDetect(&m.pmut, time.Duration(timeout)*time.Second)
  135. return m
  136. }
  137. // StartRW starts read/write processing on the current model. When in
  138. // read/write mode the model will attempt to keep in sync with the cluster by
  139. // pulling needed files from peer devices.
  140. func (m *Model) StartFolderRW(folder string) {
  141. m.fmut.Lock()
  142. cfg, ok := m.folderCfgs[folder]
  143. if !ok {
  144. panic("cannot start nonexistent folder " + folder)
  145. }
  146. _, ok = m.folderRunners[folder]
  147. if ok {
  148. panic("cannot start already running folder " + folder)
  149. }
  150. p := &Puller{
  151. folder: folder,
  152. dir: cfg.Path,
  153. scanIntv: time.Duration(cfg.RescanIntervalS) * time.Second,
  154. model: m,
  155. ignorePerms: cfg.IgnorePerms,
  156. }
  157. m.folderRunners[folder] = p
  158. m.fmut.Unlock()
  159. if len(cfg.Versioning.Type) > 0 {
  160. factory, ok := versioner.Factories[cfg.Versioning.Type]
  161. if !ok {
  162. l.Fatalf("Requested versioning type %q that does not exist", cfg.Versioning.Type)
  163. }
  164. p.versioner = factory(folder, cfg.Path, cfg.Versioning.Params)
  165. }
  166. go p.Serve()
  167. }
  168. // StartRO starts read only processing on the current model. When in
  169. // read only mode the model will announce files to the cluster but not
  170. // pull in any external changes.
  171. func (m *Model) StartFolderRO(folder string) {
  172. m.fmut.Lock()
  173. cfg, ok := m.folderCfgs[folder]
  174. if !ok {
  175. panic("cannot start nonexistent folder " + folder)
  176. }
  177. _, ok = m.folderRunners[folder]
  178. if ok {
  179. panic("cannot start already running folder " + folder)
  180. }
  181. s := &Scanner{
  182. folder: folder,
  183. intv: time.Duration(cfg.RescanIntervalS) * time.Second,
  184. model: m,
  185. }
  186. m.folderRunners[folder] = s
  187. m.fmut.Unlock()
  188. go s.Serve()
  189. }
  190. type ConnectionInfo struct {
  191. protocol.Statistics
  192. Address string
  193. ClientVersion string
  194. }
  195. // ConnectionStats returns a map with connection statistics for each connected device.
  196. func (m *Model) ConnectionStats() map[string]ConnectionInfo {
  197. type remoteAddrer interface {
  198. RemoteAddr() net.Addr
  199. }
  200. m.pmut.RLock()
  201. m.fmut.RLock()
  202. var res = make(map[string]ConnectionInfo)
  203. for device, conn := range m.protoConn {
  204. ci := ConnectionInfo{
  205. Statistics: conn.Statistics(),
  206. ClientVersion: m.deviceVer[device],
  207. }
  208. if nc, ok := m.rawConn[device].(remoteAddrer); ok {
  209. ci.Address = nc.RemoteAddr().String()
  210. }
  211. res[device.String()] = ci
  212. }
  213. m.fmut.RUnlock()
  214. m.pmut.RUnlock()
  215. in, out := protocol.TotalInOut()
  216. res["total"] = ConnectionInfo{
  217. Statistics: protocol.Statistics{
  218. At: time.Now(),
  219. InBytesTotal: in,
  220. OutBytesTotal: out,
  221. },
  222. }
  223. return res
  224. }
  225. // Returns statistics about each device
  226. func (m *Model) DeviceStatistics() map[string]stats.DeviceStatistics {
  227. var res = make(map[string]stats.DeviceStatistics)
  228. for id := range m.cfg.Devices() {
  229. res[id.String()] = m.deviceStatRef(id).GetStatistics()
  230. }
  231. return res
  232. }
  233. // Returns the completion status, in percent, for the given device and folder.
  234. func (m *Model) Completion(device protocol.DeviceID, folder string) float64 {
  235. defer m.leveldbPanicWorkaround()
  236. var tot int64
  237. m.fmut.RLock()
  238. rf, ok := m.folderFiles[folder]
  239. m.fmut.RUnlock()
  240. if !ok {
  241. return 0 // Folder doesn't exist, so we hardly have any of it
  242. }
  243. rf.WithGlobalTruncated(func(f protocol.FileIntf) bool {
  244. if !f.IsDeleted() {
  245. tot += f.Size()
  246. }
  247. return true
  248. })
  249. if tot == 0 {
  250. return 100 // Folder is empty, so we have all of it
  251. }
  252. var need int64
  253. rf.WithNeedTruncated(device, func(f protocol.FileIntf) bool {
  254. if !f.IsDeleted() {
  255. need += f.Size()
  256. }
  257. return true
  258. })
  259. res := 100 * (1 - float64(need)/float64(tot))
  260. if debug {
  261. l.Debugf("%v Completion(%s, %q): %f (%d / %d)", m, device, folder, res, need, tot)
  262. }
  263. return res
  264. }
  265. func sizeOf(fs []protocol.FileInfo) (files, deleted int, bytes int64) {
  266. for _, f := range fs {
  267. fs, de, by := sizeOfFile(f)
  268. files += fs
  269. deleted += de
  270. bytes += by
  271. }
  272. return
  273. }
  274. func sizeOfFile(f protocol.FileIntf) (files, deleted int, bytes int64) {
  275. if !f.IsDeleted() {
  276. files++
  277. } else {
  278. deleted++
  279. }
  280. bytes += f.Size()
  281. return
  282. }
  283. // GlobalSize returns the number of files, deleted files and total bytes for all
  284. // files in the global model.
  285. func (m *Model) GlobalSize(folder string) (files, deleted int, bytes int64) {
  286. defer m.leveldbPanicWorkaround()
  287. m.fmut.RLock()
  288. defer m.fmut.RUnlock()
  289. if rf, ok := m.folderFiles[folder]; ok {
  290. rf.WithGlobalTruncated(func(f protocol.FileIntf) bool {
  291. fs, de, by := sizeOfFile(f)
  292. files += fs
  293. deleted += de
  294. bytes += by
  295. return true
  296. })
  297. }
  298. return
  299. }
  300. // LocalSize returns the number of files, deleted files and total bytes for all
  301. // files in the local folder.
  302. func (m *Model) LocalSize(folder string) (files, deleted int, bytes int64) {
  303. defer m.leveldbPanicWorkaround()
  304. m.fmut.RLock()
  305. defer m.fmut.RUnlock()
  306. if rf, ok := m.folderFiles[folder]; ok {
  307. rf.WithHaveTruncated(protocol.LocalDeviceID, func(f protocol.FileIntf) bool {
  308. if f.IsInvalid() {
  309. return true
  310. }
  311. fs, de, by := sizeOfFile(f)
  312. files += fs
  313. deleted += de
  314. bytes += by
  315. return true
  316. })
  317. }
  318. return
  319. }
  320. // NeedSize returns the number and total size of currently needed files.
  321. func (m *Model) NeedSize(folder string) (files int, bytes int64) {
  322. defer m.leveldbPanicWorkaround()
  323. m.fmut.RLock()
  324. defer m.fmut.RUnlock()
  325. if rf, ok := m.folderFiles[folder]; ok {
  326. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f protocol.FileIntf) bool {
  327. fs, de, by := sizeOfFile(f)
  328. files += fs + de
  329. bytes += by
  330. return true
  331. })
  332. }
  333. if debug {
  334. l.Debugf("%v NeedSize(%q): %d %d", m, folder, files, bytes)
  335. }
  336. return
  337. }
  338. // NeedFiles returns the list of currently needed files, stopping at maxFiles
  339. // files or maxBlocks blocks. Limits <= 0 are ignored.
  340. func (m *Model) NeedFolderFilesLimited(folder string, maxFiles, maxBlocks int) []protocol.FileInfo {
  341. defer m.leveldbPanicWorkaround()
  342. m.fmut.RLock()
  343. defer m.fmut.RUnlock()
  344. nblocks := 0
  345. if rf, ok := m.folderFiles[folder]; ok {
  346. fs := make([]protocol.FileInfo, 0, maxFiles)
  347. rf.WithNeed(protocol.LocalDeviceID, func(f protocol.FileIntf) bool {
  348. fi := f.(protocol.FileInfo)
  349. fs = append(fs, fi)
  350. nblocks += len(fi.Blocks)
  351. return (maxFiles <= 0 || len(fs) < maxFiles) && (maxBlocks <= 0 || nblocks < maxBlocks)
  352. })
  353. return fs
  354. }
  355. return nil
  356. }
  357. // Index is called when a new device is connected and we receive their full index.
  358. // Implements the protocol.Model interface.
  359. func (m *Model) Index(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo) {
  360. if debug {
  361. l.Debugf("IDX(in): %s %q: %d files", deviceID, folder, len(fs))
  362. }
  363. if !m.folderSharedWith(folder, deviceID) {
  364. events.Default.Log(events.FolderRejected, map[string]string{
  365. "folder": folder,
  366. "device": deviceID.String(),
  367. })
  368. l.Warnf("Unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  369. return
  370. }
  371. m.fmut.RLock()
  372. files, ok := m.folderFiles[folder]
  373. ignores, _ := m.folderIgnores[folder]
  374. m.fmut.RUnlock()
  375. if !ok {
  376. l.Fatalf("Index for nonexistant folder %q", folder)
  377. }
  378. for i := 0; i < len(fs); {
  379. lamport.Default.Tick(fs[i].Version)
  380. if ignores.Match(fs[i].Name) {
  381. fs[i] = fs[len(fs)-1]
  382. fs = fs[:len(fs)-1]
  383. } else {
  384. i++
  385. }
  386. }
  387. files.Replace(deviceID, fs)
  388. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  389. "device": deviceID.String(),
  390. "folder": folder,
  391. "items": len(fs),
  392. "version": files.LocalVersion(deviceID),
  393. })
  394. }
  395. // IndexUpdate is called for incremental updates to connected devices' indexes.
  396. // Implements the protocol.Model interface.
  397. func (m *Model) IndexUpdate(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo) {
  398. if debug {
  399. l.Debugf("%v IDXUP(in): %s / %q: %d files", m, deviceID, folder, len(fs))
  400. }
  401. if !m.folderSharedWith(folder, deviceID) {
  402. l.Infof("Update for unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  403. return
  404. }
  405. m.fmut.RLock()
  406. files, ok := m.folderFiles[folder]
  407. ignores, _ := m.folderIgnores[folder]
  408. m.fmut.RUnlock()
  409. if !ok {
  410. l.Fatalf("IndexUpdate for nonexistant folder %q", folder)
  411. }
  412. for i := 0; i < len(fs); {
  413. lamport.Default.Tick(fs[i].Version)
  414. if ignores.Match(fs[i].Name) {
  415. fs[i] = fs[len(fs)-1]
  416. fs = fs[:len(fs)-1]
  417. } else {
  418. i++
  419. }
  420. }
  421. files.Update(deviceID, fs)
  422. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  423. "device": deviceID.String(),
  424. "folder": folder,
  425. "items": len(fs),
  426. "version": files.LocalVersion(deviceID),
  427. })
  428. }
  429. func (m *Model) folderSharedWith(folder string, deviceID protocol.DeviceID) bool {
  430. m.fmut.RLock()
  431. defer m.fmut.RUnlock()
  432. for _, nfolder := range m.deviceFolders[deviceID] {
  433. if nfolder == folder {
  434. return true
  435. }
  436. }
  437. return false
  438. }
  439. func (m *Model) ClusterConfig(deviceID protocol.DeviceID, cm protocol.ClusterConfigMessage) {
  440. m.pmut.Lock()
  441. if cm.ClientName == "syncthing" {
  442. m.deviceVer[deviceID] = cm.ClientVersion
  443. } else {
  444. m.deviceVer[deviceID] = cm.ClientName + " " + cm.ClientVersion
  445. }
  446. m.pmut.Unlock()
  447. l.Infof(`Device %s client is "%s %s"`, deviceID, cm.ClientName, cm.ClientVersion)
  448. if name := cm.GetOption("name"); name != "" {
  449. l.Infof("Device %s name is %q", deviceID, name)
  450. device, ok := m.cfg.Devices()[deviceID]
  451. if ok && device.Name == "" {
  452. device.Name = name
  453. m.cfg.SetDevice(device)
  454. }
  455. }
  456. if m.cfg.Devices()[deviceID].Introducer {
  457. // This device is an introducer. Go through the announced lists of folders
  458. // and devices and add what we are missing.
  459. var changed bool
  460. for _, folder := range cm.Folders {
  461. // If we don't have this folder yet, skip it. Ideally, we'd
  462. // offer up something in the GUI to create the folder, but for the
  463. // moment we only handle folders that we already have.
  464. if _, ok := m.folderDevices[folder.ID]; !ok {
  465. continue
  466. }
  467. nextDevice:
  468. for _, device := range folder.Devices {
  469. var id protocol.DeviceID
  470. copy(id[:], device.ID)
  471. if _, ok := m.cfg.Devices()[id]; !ok {
  472. // The device is currently unknown. Add it to the config.
  473. l.Infof("Adding device %v to config (vouched for by introducer %v)", id, deviceID)
  474. newDeviceCfg := config.DeviceConfiguration{
  475. DeviceID: id,
  476. Compression: true,
  477. }
  478. // The introducers' introducers are also our introducers.
  479. if device.Flags&protocol.FlagIntroducer != 0 {
  480. l.Infof("Device %v is now also an introducer", id)
  481. newDeviceCfg.Introducer = true
  482. }
  483. m.cfg.SetDevice(newDeviceCfg)
  484. changed = true
  485. }
  486. for _, er := range m.deviceFolders[id] {
  487. if er == folder.ID {
  488. // We already share the folder with this device, so
  489. // nothing to do.
  490. continue nextDevice
  491. }
  492. }
  493. // We don't yet share this folder with this device. Add the device
  494. // to sharing list of the folder.
  495. l.Infof("Adding device %v to share %q (vouched for by introducer %v)", id, folder.ID, deviceID)
  496. m.deviceFolders[id] = append(m.deviceFolders[id], folder.ID)
  497. m.folderDevices[folder.ID] = append(m.folderDevices[folder.ID], id)
  498. folderCfg := m.cfg.Folders()[folder.ID]
  499. folderCfg.Devices = append(folderCfg.Devices, config.FolderDeviceConfiguration{
  500. DeviceID: id,
  501. })
  502. m.cfg.SetFolder(folderCfg)
  503. changed = true
  504. }
  505. }
  506. if changed {
  507. m.cfg.Save()
  508. }
  509. }
  510. }
  511. // Close removes the peer from the model and closes the underlying connection if possible.
  512. // Implements the protocol.Model interface.
  513. func (m *Model) Close(device protocol.DeviceID, err error) {
  514. l.Infof("Connection to %s closed: %v", device, err)
  515. events.Default.Log(events.DeviceDisconnected, map[string]string{
  516. "id": device.String(),
  517. "error": err.Error(),
  518. })
  519. m.pmut.Lock()
  520. m.fmut.RLock()
  521. for _, folder := range m.deviceFolders[device] {
  522. m.folderFiles[folder].Replace(device, nil)
  523. }
  524. m.fmut.RUnlock()
  525. conn, ok := m.rawConn[device]
  526. if ok {
  527. if conn, ok := conn.(*tls.Conn); ok {
  528. // If the underlying connection is a *tls.Conn, Close() does more
  529. // than it says on the tin. Specifically, it sends a TLS alert
  530. // message, which might block forever if the connection is dead
  531. // and we don't have a deadline site.
  532. conn.SetWriteDeadline(time.Now().Add(250 * time.Millisecond))
  533. }
  534. conn.Close()
  535. }
  536. delete(m.protoConn, device)
  537. delete(m.rawConn, device)
  538. delete(m.deviceVer, device)
  539. m.pmut.Unlock()
  540. }
  541. // Request returns the specified data segment by reading it from local disk.
  542. // Implements the protocol.Model interface.
  543. func (m *Model) Request(deviceID protocol.DeviceID, folder, name string, offset int64, size int) ([]byte, error) {
  544. // Verify that the requested file exists in the local model.
  545. m.fmut.RLock()
  546. r, ok := m.folderFiles[folder]
  547. m.fmut.RUnlock()
  548. if !ok {
  549. l.Warnf("Request from %s for file %s in nonexistent folder %q", deviceID, name, folder)
  550. return nil, ErrNoSuchFile
  551. }
  552. lf := r.Get(protocol.LocalDeviceID, name)
  553. if protocol.IsInvalid(lf.Flags) || protocol.IsDeleted(lf.Flags) {
  554. if debug {
  555. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d; invalid: %v", m, deviceID, folder, name, offset, size, lf)
  556. }
  557. return nil, ErrInvalid
  558. }
  559. if offset > lf.Size() {
  560. if debug {
  561. l.Debugf("%v REQ(in; nonexistent): %s: %q o=%d s=%d", m, deviceID, name, offset, size)
  562. }
  563. return nil, ErrNoSuchFile
  564. }
  565. if debug && deviceID != protocol.LocalDeviceID {
  566. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  567. }
  568. m.fmut.RLock()
  569. fn := filepath.Join(m.folderCfgs[folder].Path, name)
  570. m.fmut.RUnlock()
  571. fd, err := os.Open(fn) // XXX: Inefficient, should cache fd?
  572. if err != nil {
  573. return nil, err
  574. }
  575. defer fd.Close()
  576. buf := make([]byte, size)
  577. _, err = fd.ReadAt(buf, offset)
  578. if err != nil {
  579. return nil, err
  580. }
  581. return buf, nil
  582. }
  583. // ReplaceLocal replaces the local folder index with the given list of files.
  584. func (m *Model) ReplaceLocal(folder string, fs []protocol.FileInfo) {
  585. m.fmut.RLock()
  586. m.folderFiles[folder].ReplaceWithDelete(protocol.LocalDeviceID, fs)
  587. m.fmut.RUnlock()
  588. }
  589. func (m *Model) CurrentFolderFile(folder string, file string) protocol.FileInfo {
  590. m.fmut.RLock()
  591. f := m.folderFiles[folder].Get(protocol.LocalDeviceID, file)
  592. m.fmut.RUnlock()
  593. return f
  594. }
  595. func (m *Model) CurrentGlobalFile(folder string, file string) protocol.FileInfo {
  596. m.fmut.RLock()
  597. f := m.folderFiles[folder].GetGlobal(file)
  598. m.fmut.RUnlock()
  599. return f
  600. }
  601. type cFiler struct {
  602. m *Model
  603. r string
  604. }
  605. // Implements scanner.CurrentFiler
  606. func (cf cFiler) CurrentFile(file string) protocol.FileInfo {
  607. return cf.m.CurrentFolderFile(cf.r, file)
  608. }
  609. // ConnectedTo returns true if we are connected to the named device.
  610. func (m *Model) ConnectedTo(deviceID protocol.DeviceID) bool {
  611. m.pmut.RLock()
  612. _, ok := m.protoConn[deviceID]
  613. m.pmut.RUnlock()
  614. if ok {
  615. m.deviceWasSeen(deviceID)
  616. }
  617. return ok
  618. }
  619. func (m *Model) GetIgnores(folder string) ([]string, error) {
  620. var lines []string
  621. cfg, ok := m.folderCfgs[folder]
  622. if !ok {
  623. return lines, fmt.Errorf("Folder %s does not exist", folder)
  624. }
  625. m.fmut.Lock()
  626. defer m.fmut.Unlock()
  627. fd, err := os.Open(filepath.Join(cfg.Path, ".stignore"))
  628. if err != nil {
  629. if os.IsNotExist(err) {
  630. return lines, nil
  631. }
  632. l.Warnln("Loading .stignore:", err)
  633. return lines, err
  634. }
  635. defer fd.Close()
  636. scanner := bufio.NewScanner(fd)
  637. for scanner.Scan() {
  638. lines = append(lines, strings.TrimSpace(scanner.Text()))
  639. }
  640. return lines, nil
  641. }
  642. func (m *Model) SetIgnores(folder string, content []string) error {
  643. cfg, ok := m.folderCfgs[folder]
  644. if !ok {
  645. return fmt.Errorf("Folder %s does not exist", folder)
  646. }
  647. fd, err := ioutil.TempFile(cfg.Path, ".syncthing.stignore-"+folder)
  648. if err != nil {
  649. l.Warnln("Saving .stignore:", err)
  650. return err
  651. }
  652. defer os.Remove(fd.Name())
  653. for _, line := range content {
  654. _, err = fmt.Fprintln(fd, line)
  655. if err != nil {
  656. l.Warnln("Saving .stignore:", err)
  657. return err
  658. }
  659. }
  660. err = fd.Close()
  661. if err != nil {
  662. l.Warnln("Saving .stignore:", err)
  663. return err
  664. }
  665. file := filepath.Join(cfg.Path, ".stignore")
  666. err = osutil.Rename(fd.Name(), file)
  667. if err != nil {
  668. l.Warnln("Saving .stignore:", err)
  669. return err
  670. }
  671. return m.ScanFolder(folder)
  672. }
  673. // AddConnection adds a new peer connection to the model. An initial index will
  674. // be sent to the connected peer, thereafter index updates whenever the local
  675. // folder changes.
  676. func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection) {
  677. deviceID := protoConn.ID()
  678. m.pmut.Lock()
  679. if _, ok := m.protoConn[deviceID]; ok {
  680. panic("add existing device")
  681. }
  682. m.protoConn[deviceID] = protoConn
  683. if _, ok := m.rawConn[deviceID]; ok {
  684. panic("add existing device")
  685. }
  686. m.rawConn[deviceID] = rawConn
  687. cm := m.clusterConfig(deviceID)
  688. protoConn.ClusterConfig(cm)
  689. m.fmut.RLock()
  690. for _, folder := range m.deviceFolders[deviceID] {
  691. fs := m.folderFiles[folder]
  692. go sendIndexes(protoConn, folder, fs, m.folderIgnores[folder])
  693. }
  694. m.fmut.RUnlock()
  695. m.pmut.Unlock()
  696. m.deviceWasSeen(deviceID)
  697. }
  698. func (m *Model) deviceStatRef(deviceID protocol.DeviceID) *stats.DeviceStatisticsReference {
  699. m.fmut.Lock()
  700. defer m.fmut.Unlock()
  701. if sr, ok := m.deviceStatRefs[deviceID]; ok {
  702. return sr
  703. } else {
  704. sr = stats.NewDeviceStatisticsReference(m.db, deviceID)
  705. m.deviceStatRefs[deviceID] = sr
  706. return sr
  707. }
  708. }
  709. func (m *Model) deviceWasSeen(deviceID protocol.DeviceID) {
  710. m.deviceStatRef(deviceID).WasSeen()
  711. }
  712. func sendIndexes(conn protocol.Connection, folder string, fs *files.Set, ignores ignore.Patterns) {
  713. deviceID := conn.ID()
  714. name := conn.Name()
  715. var err error
  716. if debug {
  717. l.Debugf("sendIndexes for %s-%s/%q starting", deviceID, name, folder)
  718. }
  719. minLocalVer, err := sendIndexTo(true, 0, conn, folder, fs, ignores)
  720. for err == nil {
  721. time.Sleep(5 * time.Second)
  722. if fs.LocalVersion(protocol.LocalDeviceID) <= minLocalVer {
  723. continue
  724. }
  725. minLocalVer, err = sendIndexTo(false, minLocalVer, conn, folder, fs, ignores)
  726. }
  727. if debug {
  728. l.Debugf("sendIndexes for %s-%s/%q exiting: %v", deviceID, name, folder, err)
  729. }
  730. }
  731. func sendIndexTo(initial bool, minLocalVer uint64, conn protocol.Connection, folder string, fs *files.Set, ignores ignore.Patterns) (uint64, error) {
  732. deviceID := conn.ID()
  733. name := conn.Name()
  734. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  735. currentBatchSize := 0
  736. maxLocalVer := uint64(0)
  737. var err error
  738. fs.WithHave(protocol.LocalDeviceID, func(fi protocol.FileIntf) bool {
  739. f := fi.(protocol.FileInfo)
  740. if f.LocalVersion <= minLocalVer {
  741. return true
  742. }
  743. if f.LocalVersion > maxLocalVer {
  744. maxLocalVer = f.LocalVersion
  745. }
  746. if ignores.Match(f.Name) {
  747. return true
  748. }
  749. if len(batch) == indexBatchSize || currentBatchSize > indexTargetSize {
  750. if initial {
  751. if err = conn.Index(folder, batch); err != nil {
  752. return false
  753. }
  754. if debug {
  755. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (initial index)", deviceID, name, folder, len(batch), currentBatchSize)
  756. }
  757. initial = false
  758. } else {
  759. if err = conn.IndexUpdate(folder, batch); err != nil {
  760. return false
  761. }
  762. if debug {
  763. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (batched update)", deviceID, name, folder, len(batch), currentBatchSize)
  764. }
  765. }
  766. batch = make([]protocol.FileInfo, 0, indexBatchSize)
  767. currentBatchSize = 0
  768. }
  769. batch = append(batch, f)
  770. currentBatchSize += indexPerFileSize + len(f.Blocks)*IndexPerBlockSize
  771. return true
  772. })
  773. if initial && err == nil {
  774. err = conn.Index(folder, batch)
  775. if debug && err == nil {
  776. l.Debugf("sendIndexes for %s-%s/%q: %d files (small initial index)", deviceID, name, folder, len(batch))
  777. }
  778. } else if len(batch) > 0 && err == nil {
  779. err = conn.IndexUpdate(folder, batch)
  780. if debug && err == nil {
  781. l.Debugf("sendIndexes for %s-%s/%q: %d files (last batch)", deviceID, name, folder, len(batch))
  782. }
  783. }
  784. return maxLocalVer, err
  785. }
  786. func (m *Model) updateLocal(folder string, f protocol.FileInfo) {
  787. f.LocalVersion = 0
  788. m.fmut.RLock()
  789. m.folderFiles[folder].Update(protocol.LocalDeviceID, []protocol.FileInfo{f})
  790. m.fmut.RUnlock()
  791. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  792. "folder": folder,
  793. "name": f.Name,
  794. "modified": time.Unix(f.Modified, 0),
  795. "flags": fmt.Sprintf("0%o", f.Flags),
  796. "size": f.Size(),
  797. })
  798. }
  799. func (m *Model) requestGlobal(deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte) ([]byte, error) {
  800. m.pmut.RLock()
  801. nc, ok := m.protoConn[deviceID]
  802. m.pmut.RUnlock()
  803. if !ok {
  804. return nil, fmt.Errorf("requestGlobal: no such device: %s", deviceID)
  805. }
  806. if debug {
  807. l.Debugf("%v REQ(out): %s: %q / %q o=%d s=%d h=%x", m, deviceID, folder, name, offset, size, hash)
  808. }
  809. return nc.Request(folder, name, offset, size)
  810. }
  811. func (m *Model) AddFolder(cfg config.FolderConfiguration) {
  812. if m.started {
  813. panic("cannot add folder to started model")
  814. }
  815. if len(cfg.ID) == 0 {
  816. panic("cannot add empty folder id")
  817. }
  818. m.fmut.Lock()
  819. m.folderCfgs[cfg.ID] = cfg
  820. m.folderFiles[cfg.ID] = files.NewSet(cfg.ID, m.db)
  821. m.folderDevices[cfg.ID] = make([]protocol.DeviceID, len(cfg.Devices))
  822. for i, device := range cfg.Devices {
  823. m.folderDevices[cfg.ID][i] = device.DeviceID
  824. m.deviceFolders[device.DeviceID] = append(m.deviceFolders[device.DeviceID], cfg.ID)
  825. }
  826. m.addedFolder = true
  827. m.fmut.Unlock()
  828. }
  829. func (m *Model) ScanFolders() {
  830. m.fmut.RLock()
  831. var folders = make([]string, 0, len(m.folderCfgs))
  832. for folder := range m.folderCfgs {
  833. folders = append(folders, folder)
  834. }
  835. m.fmut.RUnlock()
  836. var wg sync.WaitGroup
  837. wg.Add(len(folders))
  838. for _, folder := range folders {
  839. folder := folder
  840. go func() {
  841. err := m.ScanFolder(folder)
  842. if err != nil {
  843. m.cfg.InvalidateFolder(folder, err.Error())
  844. }
  845. wg.Done()
  846. }()
  847. }
  848. wg.Wait()
  849. }
  850. func (m *Model) ScanFolder(folder string) error {
  851. return m.ScanFolderSub(folder, "")
  852. }
  853. func (m *Model) ScanFolderSub(folder, sub string) error {
  854. if p := filepath.Clean(filepath.Join(folder, sub)); !strings.HasPrefix(p, folder) {
  855. return errors.New("invalid subpath")
  856. }
  857. m.fmut.RLock()
  858. fs, ok := m.folderFiles[folder]
  859. dir := m.folderCfgs[folder].Path
  860. ignores, _ := ignore.Load(filepath.Join(dir, ".stignore"))
  861. m.folderIgnores[folder] = ignores
  862. w := &scanner.Walker{
  863. Dir: dir,
  864. Sub: sub,
  865. Ignores: ignores,
  866. BlockSize: protocol.BlockSize,
  867. TempNamer: defTempNamer,
  868. CurrentFiler: cFiler{m, folder},
  869. IgnorePerms: m.folderCfgs[folder].IgnorePerms,
  870. }
  871. m.fmut.RUnlock()
  872. if !ok {
  873. return errors.New("no such folder")
  874. }
  875. m.setState(folder, FolderScanning)
  876. fchan, err := w.Walk()
  877. if err != nil {
  878. return err
  879. }
  880. batchSize := 100
  881. batch := make([]protocol.FileInfo, 0, 00)
  882. for f := range fchan {
  883. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  884. "folder": folder,
  885. "name": f.Name,
  886. "modified": time.Unix(f.Modified, 0),
  887. "flags": fmt.Sprintf("0%o", f.Flags),
  888. "size": f.Size(),
  889. })
  890. if len(batch) == batchSize {
  891. fs.Update(protocol.LocalDeviceID, batch)
  892. batch = batch[:0]
  893. }
  894. batch = append(batch, f)
  895. }
  896. if len(batch) > 0 {
  897. fs.Update(protocol.LocalDeviceID, batch)
  898. }
  899. batch = batch[:0]
  900. // TODO: We should limit the Have scanning to start at sub
  901. seenPrefix := false
  902. fs.WithHaveTruncated(protocol.LocalDeviceID, func(fi protocol.FileIntf) bool {
  903. f := fi.(protocol.FileInfoTruncated)
  904. if !strings.HasPrefix(f.Name, sub) {
  905. // Return true so that we keep iterating, until we get to the part
  906. // of the tree we are interested in. Then return false so we stop
  907. // iterating when we've passed the end of the subtree.
  908. return !seenPrefix
  909. }
  910. seenPrefix = true
  911. if !protocol.IsDeleted(f.Flags) {
  912. if f.IsInvalid() {
  913. return true
  914. }
  915. if len(batch) == batchSize {
  916. fs.Update(protocol.LocalDeviceID, batch)
  917. batch = batch[:0]
  918. }
  919. if ignores.Match(f.Name) {
  920. // File has been ignored. Set invalid bit.
  921. nf := protocol.FileInfo{
  922. Name: f.Name,
  923. Flags: f.Flags | protocol.FlagInvalid,
  924. Modified: f.Modified,
  925. Version: f.Version, // The file is still the same, so don't bump version
  926. }
  927. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  928. "folder": folder,
  929. "name": f.Name,
  930. "modified": time.Unix(f.Modified, 0),
  931. "flags": fmt.Sprintf("0%o", f.Flags),
  932. "size": f.Size(),
  933. })
  934. batch = append(batch, nf)
  935. } else if _, err := os.Stat(filepath.Join(dir, f.Name)); err != nil && os.IsNotExist(err) {
  936. // File has been deleted
  937. nf := protocol.FileInfo{
  938. Name: f.Name,
  939. Flags: f.Flags | protocol.FlagDeleted,
  940. Modified: f.Modified,
  941. Version: lamport.Default.Tick(f.Version),
  942. }
  943. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  944. "folder": folder,
  945. "name": f.Name,
  946. "modified": time.Unix(f.Modified, 0),
  947. "flags": fmt.Sprintf("0%o", f.Flags),
  948. "size": f.Size(),
  949. })
  950. batch = append(batch, nf)
  951. }
  952. }
  953. return true
  954. })
  955. if len(batch) > 0 {
  956. fs.Update(protocol.LocalDeviceID, batch)
  957. }
  958. m.setState(folder, FolderIdle)
  959. return nil
  960. }
  961. // clusterConfig returns a ClusterConfigMessage that is correct for the given peer device
  962. func (m *Model) clusterConfig(device protocol.DeviceID) protocol.ClusterConfigMessage {
  963. cm := protocol.ClusterConfigMessage{
  964. ClientName: m.clientName,
  965. ClientVersion: m.clientVersion,
  966. Options: []protocol.Option{
  967. {
  968. Key: "name",
  969. Value: m.deviceName,
  970. },
  971. },
  972. }
  973. m.fmut.RLock()
  974. for _, folder := range m.deviceFolders[device] {
  975. cr := protocol.Folder{
  976. ID: folder,
  977. }
  978. for _, device := range m.folderDevices[folder] {
  979. // DeviceID is a value type, but with an underlying array. Copy it
  980. // so we don't grab aliases to the same array later on in device[:]
  981. device := device
  982. // TODO: Set read only bit when relevant
  983. cn := protocol.Device{
  984. ID: device[:],
  985. Flags: protocol.FlagShareTrusted,
  986. }
  987. if deviceCfg := m.cfg.Devices()[device]; deviceCfg.Introducer {
  988. cn.Flags |= protocol.FlagIntroducer
  989. }
  990. cr.Devices = append(cr.Devices, cn)
  991. }
  992. cm.Folders = append(cm.Folders, cr)
  993. }
  994. m.fmut.RUnlock()
  995. return cm
  996. }
  997. func (m *Model) setState(folder string, state folderState) {
  998. m.smut.Lock()
  999. oldState := m.folderState[folder]
  1000. changed, ok := m.folderStateChanged[folder]
  1001. if state != oldState {
  1002. m.folderState[folder] = state
  1003. m.folderStateChanged[folder] = time.Now()
  1004. eventData := map[string]interface{}{
  1005. "folder": folder,
  1006. "to": state.String(),
  1007. }
  1008. if ok {
  1009. eventData["duration"] = time.Since(changed).Seconds()
  1010. eventData["from"] = oldState.String()
  1011. }
  1012. events.Default.Log(events.StateChanged, eventData)
  1013. }
  1014. m.smut.Unlock()
  1015. }
  1016. func (m *Model) State(folder string) (string, time.Time) {
  1017. m.smut.RLock()
  1018. state := m.folderState[folder]
  1019. changed := m.folderStateChanged[folder]
  1020. m.smut.RUnlock()
  1021. return state.String(), changed
  1022. }
  1023. func (m *Model) Override(folder string) {
  1024. m.fmut.RLock()
  1025. fs := m.folderFiles[folder]
  1026. m.fmut.RUnlock()
  1027. m.setState(folder, FolderScanning)
  1028. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  1029. fs.WithNeed(protocol.LocalDeviceID, func(fi protocol.FileIntf) bool {
  1030. need := fi.(protocol.FileInfo)
  1031. if len(batch) == indexBatchSize {
  1032. fs.Update(protocol.LocalDeviceID, batch)
  1033. batch = batch[:0]
  1034. }
  1035. have := fs.Get(protocol.LocalDeviceID, need.Name)
  1036. if have.Name != need.Name {
  1037. // We are missing the file
  1038. need.Flags |= protocol.FlagDeleted
  1039. need.Blocks = nil
  1040. } else {
  1041. // We have the file, replace with our version
  1042. need = have
  1043. }
  1044. need.Version = lamport.Default.Tick(need.Version)
  1045. need.LocalVersion = 0
  1046. batch = append(batch, need)
  1047. return true
  1048. })
  1049. if len(batch) > 0 {
  1050. fs.Update(protocol.LocalDeviceID, batch)
  1051. }
  1052. m.setState(folder, FolderIdle)
  1053. }
  1054. // CurrentLocalVersion returns the change version for the given folder.
  1055. // This is guaranteed to increment if the contents of the local folder has
  1056. // changed.
  1057. func (m *Model) CurrentLocalVersion(folder string) uint64 {
  1058. m.fmut.Lock()
  1059. defer m.fmut.Unlock()
  1060. fs, ok := m.folderFiles[folder]
  1061. if !ok {
  1062. // The folder might not exist, since this can be called with a user
  1063. // specified folder name from the REST interface.
  1064. return 0
  1065. }
  1066. return fs.LocalVersion(protocol.LocalDeviceID)
  1067. }
  1068. // RemoteLocalVersion returns the change version for the given folder, as
  1069. // sent by remote peers. This is guaranteed to increment if the contents of
  1070. // the remote or global folder has changed.
  1071. func (m *Model) RemoteLocalVersion(folder string) uint64 {
  1072. m.fmut.Lock()
  1073. defer m.fmut.Unlock()
  1074. fs, ok := m.folderFiles[folder]
  1075. if !ok {
  1076. panic("bug: LocalVersion called for nonexistent folder " + folder)
  1077. }
  1078. var ver uint64
  1079. for _, n := range m.folderDevices[folder] {
  1080. ver += fs.LocalVersion(n)
  1081. }
  1082. return ver
  1083. }
  1084. func (m *Model) availability(folder string, file string) []protocol.DeviceID {
  1085. m.fmut.Lock()
  1086. defer m.fmut.Unlock()
  1087. fs, ok := m.folderFiles[folder]
  1088. if !ok {
  1089. return nil
  1090. }
  1091. return fs.Availability(file)
  1092. }
  1093. func (m *Model) String() string {
  1094. return fmt.Sprintf("model@%p", m)
  1095. }
  1096. func (m *Model) leveldbPanicWorkaround() {
  1097. // When an inconsistency is detected in leveldb we panic(). This is
  1098. // appropriate because it should never happen, but currently it does for
  1099. // some reason. However it only seems to trigger in the asynchronous full-
  1100. // database scans that happen due to REST and usage-reporting calls. In
  1101. // those places we defer to this workaround to catch the panic instead of
  1102. // taking down syncthing.
  1103. // This is just a band-aid and should be removed as soon as we have found
  1104. // a real root cause.
  1105. if pnc := recover(); pnc != nil {
  1106. if err, ok := pnc.(error); ok && strings.Contains(err.Error(), "leveldb") {
  1107. l.Warnln("recovered:", err)
  1108. } else {
  1109. // Any non-leveldb error is genuine and should continue panicing.
  1110. panic(err)
  1111. }
  1112. }
  1113. }