kqueue.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521
  1. // Copyright 2010 The Go Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. // +build freebsd openbsd netbsd dragonfly darwin
  5. package fsnotify
  6. import (
  7. "errors"
  8. "fmt"
  9. "io/ioutil"
  10. "os"
  11. "path/filepath"
  12. "sync"
  13. "time"
  14. "golang.org/x/sys/unix"
  15. )
  16. // Watcher watches a set of files, delivering events to a channel.
  17. type Watcher struct {
  18. Events chan Event
  19. Errors chan error
  20. done chan struct{} // Channel for sending a "quit message" to the reader goroutine
  21. kq int // File descriptor (as returned by the kqueue() syscall).
  22. mu sync.Mutex // Protects access to watcher data
  23. watches map[string]int // Map of watched file descriptors (key: path).
  24. externalWatches map[string]bool // Map of watches added by user of the library.
  25. dirFlags map[string]uint32 // Map of watched directories to fflags used in kqueue.
  26. paths map[int]pathInfo // Map file descriptors to path names for processing kqueue events.
  27. fileExists map[string]bool // Keep track of if we know this file exists (to stop duplicate create events).
  28. isClosed bool // Set to true when Close() is first called
  29. }
  30. type pathInfo struct {
  31. name string
  32. isDir bool
  33. }
  34. // NewWatcher establishes a new watcher with the underlying OS and begins waiting for events.
  35. func NewWatcher() (*Watcher, error) {
  36. kq, err := kqueue()
  37. if err != nil {
  38. return nil, err
  39. }
  40. w := &Watcher{
  41. kq: kq,
  42. watches: make(map[string]int),
  43. dirFlags: make(map[string]uint32),
  44. paths: make(map[int]pathInfo),
  45. fileExists: make(map[string]bool),
  46. externalWatches: make(map[string]bool),
  47. Events: make(chan Event),
  48. Errors: make(chan error),
  49. done: make(chan struct{}),
  50. }
  51. go w.readEvents()
  52. return w, nil
  53. }
  54. // Close removes all watches and closes the events channel.
  55. func (w *Watcher) Close() error {
  56. w.mu.Lock()
  57. if w.isClosed {
  58. w.mu.Unlock()
  59. return nil
  60. }
  61. w.isClosed = true
  62. // copy paths to remove while locked
  63. var pathsToRemove = make([]string, 0, len(w.watches))
  64. for name := range w.watches {
  65. pathsToRemove = append(pathsToRemove, name)
  66. }
  67. w.mu.Unlock()
  68. // unlock before calling Remove, which also locks
  69. for _, name := range pathsToRemove {
  70. w.Remove(name)
  71. }
  72. // send a "quit" message to the reader goroutine
  73. close(w.done)
  74. return nil
  75. }
  76. // Add starts watching the named file or directory (non-recursively).
  77. func (w *Watcher) Add(name string) error {
  78. w.mu.Lock()
  79. w.externalWatches[name] = true
  80. w.mu.Unlock()
  81. _, err := w.addWatch(name, noteAllEvents)
  82. return err
  83. }
  84. // Remove stops watching the the named file or directory (non-recursively).
  85. func (w *Watcher) Remove(name string) error {
  86. name = filepath.Clean(name)
  87. w.mu.Lock()
  88. watchfd, ok := w.watches[name]
  89. w.mu.Unlock()
  90. if !ok {
  91. return fmt.Errorf("can't remove non-existent kevent watch for: %s", name)
  92. }
  93. const registerRemove = unix.EV_DELETE
  94. if err := register(w.kq, []int{watchfd}, registerRemove, 0); err != nil {
  95. return err
  96. }
  97. unix.Close(watchfd)
  98. w.mu.Lock()
  99. isDir := w.paths[watchfd].isDir
  100. delete(w.watches, name)
  101. delete(w.paths, watchfd)
  102. delete(w.dirFlags, name)
  103. w.mu.Unlock()
  104. // Find all watched paths that are in this directory that are not external.
  105. if isDir {
  106. var pathsToRemove []string
  107. w.mu.Lock()
  108. for _, path := range w.paths {
  109. wdir, _ := filepath.Split(path.name)
  110. if filepath.Clean(wdir) == name {
  111. if !w.externalWatches[path.name] {
  112. pathsToRemove = append(pathsToRemove, path.name)
  113. }
  114. }
  115. }
  116. w.mu.Unlock()
  117. for _, name := range pathsToRemove {
  118. // Since these are internal, not much sense in propagating error
  119. // to the user, as that will just confuse them with an error about
  120. // a path they did not explicitly watch themselves.
  121. w.Remove(name)
  122. }
  123. }
  124. return nil
  125. }
  126. // Watch all events (except NOTE_EXTEND, NOTE_LINK, NOTE_REVOKE)
  127. const noteAllEvents = unix.NOTE_DELETE | unix.NOTE_WRITE | unix.NOTE_ATTRIB | unix.NOTE_RENAME
  128. // keventWaitTime to block on each read from kevent
  129. var keventWaitTime = durationToTimespec(100 * time.Millisecond)
  130. // addWatch adds name to the watched file set.
  131. // The flags are interpreted as described in kevent(2).
  132. // Returns the real path to the file which was added, if any, which may be different from the one passed in the case of symlinks.
  133. func (w *Watcher) addWatch(name string, flags uint32) (string, error) {
  134. var isDir bool
  135. // Make ./name and name equivalent
  136. name = filepath.Clean(name)
  137. w.mu.Lock()
  138. if w.isClosed {
  139. w.mu.Unlock()
  140. return "", errors.New("kevent instance already closed")
  141. }
  142. watchfd, alreadyWatching := w.watches[name]
  143. // We already have a watch, but we can still override flags.
  144. if alreadyWatching {
  145. isDir = w.paths[watchfd].isDir
  146. }
  147. w.mu.Unlock()
  148. if !alreadyWatching {
  149. fi, err := os.Lstat(name)
  150. if err != nil {
  151. return "", err
  152. }
  153. // Don't watch sockets.
  154. if fi.Mode()&os.ModeSocket == os.ModeSocket {
  155. return "", nil
  156. }
  157. // Don't watch named pipes.
  158. if fi.Mode()&os.ModeNamedPipe == os.ModeNamedPipe {
  159. return "", nil
  160. }
  161. // Follow Symlinks
  162. // Unfortunately, Linux can add bogus symlinks to watch list without
  163. // issue, and Windows can't do symlinks period (AFAIK). To maintain
  164. // consistency, we will act like everything is fine. There will simply
  165. // be no file events for broken symlinks.
  166. // Hence the returns of nil on errors.
  167. if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
  168. name, err = filepath.EvalSymlinks(name)
  169. if err != nil {
  170. return "", nil
  171. }
  172. w.mu.Lock()
  173. _, alreadyWatching = w.watches[name]
  174. w.mu.Unlock()
  175. if alreadyWatching {
  176. return name, nil
  177. }
  178. fi, err = os.Lstat(name)
  179. if err != nil {
  180. return "", nil
  181. }
  182. }
  183. watchfd, err = unix.Open(name, openMode, 0700)
  184. if watchfd == -1 {
  185. return "", err
  186. }
  187. isDir = fi.IsDir()
  188. }
  189. const registerAdd = unix.EV_ADD | unix.EV_CLEAR | unix.EV_ENABLE
  190. if err := register(w.kq, []int{watchfd}, registerAdd, flags); err != nil {
  191. unix.Close(watchfd)
  192. return "", err
  193. }
  194. if !alreadyWatching {
  195. w.mu.Lock()
  196. w.watches[name] = watchfd
  197. w.paths[watchfd] = pathInfo{name: name, isDir: isDir}
  198. w.mu.Unlock()
  199. }
  200. if isDir {
  201. // Watch the directory if it has not been watched before,
  202. // or if it was watched before, but perhaps only a NOTE_DELETE (watchDirectoryFiles)
  203. w.mu.Lock()
  204. watchDir := (flags&unix.NOTE_WRITE) == unix.NOTE_WRITE &&
  205. (!alreadyWatching || (w.dirFlags[name]&unix.NOTE_WRITE) != unix.NOTE_WRITE)
  206. // Store flags so this watch can be updated later
  207. w.dirFlags[name] = flags
  208. w.mu.Unlock()
  209. if watchDir {
  210. if err := w.watchDirectoryFiles(name); err != nil {
  211. return "", err
  212. }
  213. }
  214. }
  215. return name, nil
  216. }
  217. // readEvents reads from kqueue and converts the received kevents into
  218. // Event values that it sends down the Events channel.
  219. func (w *Watcher) readEvents() {
  220. eventBuffer := make([]unix.Kevent_t, 10)
  221. loop:
  222. for {
  223. // See if there is a message on the "done" channel
  224. select {
  225. case <-w.done:
  226. break loop
  227. default:
  228. }
  229. // Get new events
  230. kevents, err := read(w.kq, eventBuffer, &keventWaitTime)
  231. // EINTR is okay, the syscall was interrupted before timeout expired.
  232. if err != nil && err != unix.EINTR {
  233. select {
  234. case w.Errors <- err:
  235. case <-w.done:
  236. break loop
  237. }
  238. continue
  239. }
  240. // Flush the events we received to the Events channel
  241. for len(kevents) > 0 {
  242. kevent := &kevents[0]
  243. watchfd := int(kevent.Ident)
  244. mask := uint32(kevent.Fflags)
  245. w.mu.Lock()
  246. path := w.paths[watchfd]
  247. w.mu.Unlock()
  248. event := newEvent(path.name, mask)
  249. if path.isDir && !(event.Op&Remove == Remove) {
  250. // Double check to make sure the directory exists. This can happen when
  251. // we do a rm -fr on a recursively watched folders and we receive a
  252. // modification event first but the folder has been deleted and later
  253. // receive the delete event
  254. if _, err := os.Lstat(event.Name); os.IsNotExist(err) {
  255. // mark is as delete event
  256. event.Op |= Remove
  257. }
  258. }
  259. if event.Op&Rename == Rename || event.Op&Remove == Remove {
  260. w.Remove(event.Name)
  261. w.mu.Lock()
  262. delete(w.fileExists, event.Name)
  263. w.mu.Unlock()
  264. }
  265. if path.isDir && event.Op&Write == Write && !(event.Op&Remove == Remove) {
  266. w.sendDirectoryChangeEvents(event.Name)
  267. } else {
  268. // Send the event on the Events channel.
  269. select {
  270. case w.Events <- event:
  271. case <-w.done:
  272. break loop
  273. }
  274. }
  275. if event.Op&Remove == Remove {
  276. // Look for a file that may have overwritten this.
  277. // For example, mv f1 f2 will delete f2, then create f2.
  278. if path.isDir {
  279. fileDir := filepath.Clean(event.Name)
  280. w.mu.Lock()
  281. _, found := w.watches[fileDir]
  282. w.mu.Unlock()
  283. if found {
  284. // make sure the directory exists before we watch for changes. When we
  285. // do a recursive watch and perform rm -fr, the parent directory might
  286. // have gone missing, ignore the missing directory and let the
  287. // upcoming delete event remove the watch from the parent directory.
  288. if _, err := os.Lstat(fileDir); err == nil {
  289. w.sendDirectoryChangeEvents(fileDir)
  290. }
  291. }
  292. } else {
  293. filePath := filepath.Clean(event.Name)
  294. if fileInfo, err := os.Lstat(filePath); err == nil {
  295. w.sendFileCreatedEventIfNew(filePath, fileInfo)
  296. }
  297. }
  298. }
  299. // Move to next event
  300. kevents = kevents[1:]
  301. }
  302. }
  303. // cleanup
  304. err := unix.Close(w.kq)
  305. if err != nil {
  306. // only way the previous loop breaks is if w.done was closed so we need to async send to w.Errors.
  307. select {
  308. case w.Errors <- err:
  309. default:
  310. }
  311. }
  312. close(w.Events)
  313. close(w.Errors)
  314. }
  315. // newEvent returns an platform-independent Event based on kqueue Fflags.
  316. func newEvent(name string, mask uint32) Event {
  317. e := Event{Name: name}
  318. if mask&unix.NOTE_DELETE == unix.NOTE_DELETE {
  319. e.Op |= Remove
  320. }
  321. if mask&unix.NOTE_WRITE == unix.NOTE_WRITE {
  322. e.Op |= Write
  323. }
  324. if mask&unix.NOTE_RENAME == unix.NOTE_RENAME {
  325. e.Op |= Rename
  326. }
  327. if mask&unix.NOTE_ATTRIB == unix.NOTE_ATTRIB {
  328. e.Op |= Chmod
  329. }
  330. return e
  331. }
  332. func newCreateEvent(name string) Event {
  333. return Event{Name: name, Op: Create}
  334. }
  335. // watchDirectoryFiles to mimic inotify when adding a watch on a directory
  336. func (w *Watcher) watchDirectoryFiles(dirPath string) error {
  337. // Get all files
  338. files, err := ioutil.ReadDir(dirPath)
  339. if err != nil {
  340. return err
  341. }
  342. for _, fileInfo := range files {
  343. filePath := filepath.Join(dirPath, fileInfo.Name())
  344. filePath, err = w.internalWatch(filePath, fileInfo)
  345. if err != nil {
  346. return err
  347. }
  348. w.mu.Lock()
  349. w.fileExists[filePath] = true
  350. w.mu.Unlock()
  351. }
  352. return nil
  353. }
  354. // sendDirectoryEvents searches the directory for newly created files
  355. // and sends them over the event channel. This functionality is to have
  356. // the BSD version of fsnotify match Linux inotify which provides a
  357. // create event for files created in a watched directory.
  358. func (w *Watcher) sendDirectoryChangeEvents(dirPath string) {
  359. // Get all files
  360. files, err := ioutil.ReadDir(dirPath)
  361. if err != nil {
  362. select {
  363. case w.Errors <- err:
  364. case <-w.done:
  365. return
  366. }
  367. }
  368. // Search for new files
  369. for _, fileInfo := range files {
  370. filePath := filepath.Join(dirPath, fileInfo.Name())
  371. err := w.sendFileCreatedEventIfNew(filePath, fileInfo)
  372. if err != nil {
  373. return
  374. }
  375. }
  376. }
  377. // sendFileCreatedEvent sends a create event if the file isn't already being tracked.
  378. func (w *Watcher) sendFileCreatedEventIfNew(filePath string, fileInfo os.FileInfo) (err error) {
  379. w.mu.Lock()
  380. _, doesExist := w.fileExists[filePath]
  381. w.mu.Unlock()
  382. if !doesExist {
  383. // Send create event
  384. select {
  385. case w.Events <- newCreateEvent(filePath):
  386. case <-w.done:
  387. return
  388. }
  389. }
  390. // like watchDirectoryFiles (but without doing another ReadDir)
  391. filePath, err = w.internalWatch(filePath, fileInfo)
  392. if err != nil {
  393. return err
  394. }
  395. w.mu.Lock()
  396. w.fileExists[filePath] = true
  397. w.mu.Unlock()
  398. return nil
  399. }
  400. func (w *Watcher) internalWatch(name string, fileInfo os.FileInfo) (string, error) {
  401. if fileInfo.IsDir() {
  402. // mimic Linux providing delete events for subdirectories
  403. // but preserve the flags used if currently watching subdirectory
  404. w.mu.Lock()
  405. flags := w.dirFlags[name]
  406. w.mu.Unlock()
  407. flags |= unix.NOTE_DELETE | unix.NOTE_RENAME
  408. return w.addWatch(name, flags)
  409. }
  410. // watch file to mimic Linux inotify
  411. return w.addWatch(name, noteAllEvents)
  412. }
  413. // kqueue creates a new kernel event queue and returns a descriptor.
  414. func kqueue() (kq int, err error) {
  415. kq, err = unix.Kqueue()
  416. if kq == -1 {
  417. return kq, err
  418. }
  419. return kq, nil
  420. }
  421. // register events with the queue
  422. func register(kq int, fds []int, flags int, fflags uint32) error {
  423. changes := make([]unix.Kevent_t, len(fds))
  424. for i, fd := range fds {
  425. // SetKevent converts int to the platform-specific types:
  426. unix.SetKevent(&changes[i], fd, unix.EVFILT_VNODE, flags)
  427. changes[i].Fflags = fflags
  428. }
  429. // register the events
  430. success, err := unix.Kevent(kq, changes, nil, nil)
  431. if success == -1 {
  432. return err
  433. }
  434. return nil
  435. }
  436. // read retrieves pending events, or waits until an event occurs.
  437. // A timeout of nil blocks indefinitely, while 0 polls the queue.
  438. func read(kq int, events []unix.Kevent_t, timeout *unix.Timespec) ([]unix.Kevent_t, error) {
  439. n, err := unix.Kevent(kq, nil, events, timeout)
  440. if err != nil {
  441. return nil, err
  442. }
  443. return events[0:n], nil
  444. }
  445. // durationToTimespec prepares a timeout value
  446. func durationToTimespec(d time.Duration) unix.Timespec {
  447. return unix.NsecToTimespec(d.Nanoseconds())
  448. }