vendor/github.com/rjeczalik/notify/watcher_readdcw.go

   1 // Copyright (c) 2014-2015 The Notify Authors. All rights reserved.
   2 // Use of this source code is governed by the MIT license that can be
   3 // found in the LICENSE file.
   4
   5 // +build windows
   6
   7 package notify
   8
   9 import (
  10         "errors"
  11         "runtime"
  12         "sync"
  13         "sync/atomic"
  14         "syscall"
  15         "unsafe"
  16 )
  17
  18 // readBufferSize defines the size of an array in which read statuses are stored.
  19 // The buffer have to be DWORD-aligned and, if notify is used in monitoring a
  20 // directory over the network, its size must not be greater than 64KB. Each of
  21 // watched directories uses its own buffer for storing events.
  22 const readBufferSize = 4096
  23
  24 // Since all operations which go through the Windows completion routine are done
  25 // asynchronously, filter may set one of the constants belor. They were defined
  26 // in order to distinguish whether current folder should be re-registered in
  27 // ReadDirectoryChangesW function or some control operations need to be executed.
  28 const (
  29         stateRewatch uint32 = 1 << (28 + iota)
  30         stateUnwatch
  31         stateCPClose
  32 )
  33
  34 // Filter used in current implementation was split into four segments:
  35 //  - bits  0-11 store ReadDirectoryChangesW filters,
  36 //  - bits 12-19 store File notify actions,
  37 //  - bits 20-27 store notify specific events and flags,
  38 //  - bits 28-31 store states which are used in loop's FSM.
  39 // Constants below are used as masks to retrieve only specific filter parts.
  40 const (
  41         onlyNotifyChanges uint32 = 0x00000FFF
  42         onlyNGlobalEvents uint32 = 0x0FF00000
  43         onlyMachineStates uint32 = 0xF0000000
  44 )
  45
  46 // grip represents a single watched directory. It stores the data required by
  47 // ReadDirectoryChangesW function. Only the filter, recursive, and handle members
  48 // may by modified by watcher implementation. Rest of the them have to remain
  49 // constant since they are used by Windows completion routine. This indicates that
  50 // grip can be removed only when all operations on the file handle are finished.
  51 type grip struct {
  52         handle    syscall.Handle
  53         filter    uint32
  54         recursive bool
  55         pathw     []uint16
  56         buffer    [readBufferSize]byte
  57         parent    *watched
  58         ovlapped  *overlappedEx
  59 }
  60
  61 // overlappedEx stores information used in asynchronous input and output.
  62 // Additionally, overlappedEx contains a pointer to 'grip' item which is used in
  63 // order to gather the structure in which the overlappedEx object was created.
  64 type overlappedEx struct {
  65         syscall.Overlapped
  66         parent *grip
  67 }
  68
  69 // newGrip creates a new file handle that can be used in overlapped operations.
  70 // Then, the handle is associated with I/O completion port 'cph' and its value
  71 // is stored in newly created 'grip' object.
  72 func newGrip(cph syscall.Handle, parent *watched, filter uint32) (*grip, error) {
  73         g := &grip{
  74                 handle:    syscall.InvalidHandle,
  75                 filter:    filter,
  76                 recursive: parent.recursive,
  77                 pathw:     parent.pathw,
  78                 parent:    parent,
  79                 ovlapped:  &overlappedEx{},
  80         }
  81         if err := g.register(cph); err != nil {
  82                 return nil, err
  83         }
  84         g.ovlapped.parent = g
  85         return g, nil
  86 }
  87
  88 // NOTE : Thread safe
  89 func (g *grip) register(cph syscall.Handle) (err error) {
  90         if g.handle, err = syscall.CreateFile(
  91                 &g.pathw[0],
  92                 syscall.FILE_LIST_DIRECTORY,
  93                 syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE|syscall.FILE_SHARE_DELETE,
  94                 nil,
  95                 syscall.OPEN_EXISTING,
  96                 syscall.FILE_FLAG_BACKUP_SEMANTICS|syscall.FILE_FLAG_OVERLAPPED,
  97                 0,
  98         ); err != nil {
  99                 return
 100         }
 101         if _, err = syscall.CreateIoCompletionPort(g.handle, cph, 0, 0); err != nil {
 102                 syscall.CloseHandle(g.handle)
 103                 return
 104         }
 105         return g.readDirChanges()
 106 }
 107
 108 // readDirChanges tells the system to store file change information in grip's
 109 // buffer. Directory changes that occur between calls to this function are added
 110 // to the buffer and then, returned with the next call.
 111 func (g *grip) readDirChanges() error {
 112         return syscall.ReadDirectoryChanges(
 113                 g.handle,
 114                 &g.buffer[0],
 115                 uint32(unsafe.Sizeof(g.buffer)),
 116                 g.recursive,
 117                 encode(g.filter),
 118                 nil,
 119                 (*syscall.Overlapped)(unsafe.Pointer(g.ovlapped)),
 120                 0,
 121         )
 122 }
 123
 124 // encode transforms a generic filter, which contains platform independent and
 125 // implementation specific bit fields, to value that can be used as NotifyFilter
 126 // parameter in ReadDirectoryChangesW function.
 127 func encode(filter uint32) uint32 {
 128         e := Event(filter & (onlyNGlobalEvents | onlyNotifyChanges))
 129         if e&dirmarker != 0 {
 130                 return uint32(FileNotifyChangeDirName)
 131         }
 132         if e&Create != 0 {
 133                 e = (e ^ Create) | FileNotifyChangeFileName
 134         }
 135         if e&Remove != 0 {
 136                 e = (e ^ Remove) | FileNotifyChangeFileName
 137         }
 138         if e&Write != 0 {
 139                 e = (e ^ Write) | FileNotifyChangeAttributes | FileNotifyChangeSize |
 140                         FileNotifyChangeCreation | FileNotifyChangeSecurity
 141         }
 142         if e&Rename != 0 {
 143                 e = (e ^ Rename) | FileNotifyChangeFileName
 144         }
 145         return uint32(e)
 146 }
 147
 148 // watched is made in order to check whether an action comes from a directory or
 149 // file. This approach requires two file handlers per single monitored folder. The
 150 // second grip handles actions which include creating or deleting a directory. If
 151 // these processes are not monitored, only the first grip is created.
 152 type watched struct {
 153         filter    uint32
 154         recursive bool
 155         count     uint8
 156         pathw     []uint16
 157         digrip    [2]*grip
 158 }
 159
 160 // newWatched creates a new watched instance. It splits the filter variable into
 161 // two parts. The first part is responsible for watching all events which can be
 162 // created for a file in watched directory structure and the second one watches
 163 // only directory Create/Remove actions. If all operations succeed, the Create
 164 // message is sent to I/O completion port queue for further processing.
 165 func newWatched(cph syscall.Handle, filter uint32, recursive bool,
 166         path string) (wd *watched, err error) {
 167         wd = &watched{
 168                 filter:    filter,
 169                 recursive: recursive,
 170         }
 171         if wd.pathw, err = syscall.UTF16FromString(path); err != nil {
 172                 return
 173         }
 174         if err = wd.recreate(cph); err != nil {
 175                 return
 176         }
 177         return wd, nil
 178 }
 179
 180 // TODO : doc
 181 func (wd *watched) recreate(cph syscall.Handle) (err error) {
 182         filefilter := wd.filter &^ uint32(FileNotifyChangeDirName)
 183         if err = wd.updateGrip(0, cph, filefilter == 0, filefilter); err != nil {
 184                 return
 185         }
 186         dirfilter := wd.filter & uint32(FileNotifyChangeDirName|Create|Remove)
 187         if err = wd.updateGrip(1, cph, dirfilter == 0, wd.filter|uint32(dirmarker)); err != nil {
 188                 return
 189         }
 190         wd.filter &^= onlyMachineStates
 191         return
 192 }
 193
 194 // TODO : doc
 195 func (wd *watched) updateGrip(idx int, cph syscall.Handle, reset bool,
 196         newflag uint32) (err error) {
 197         if reset {
 198                 wd.digrip[idx] = nil
 199         } else {
 200                 if wd.digrip[idx] == nil {
 201                         if wd.digrip[idx], err = newGrip(cph, wd, newflag); err != nil {
 202                                 wd.closeHandle()
 203                                 return
 204                         }
 205                 } else {
 206                         wd.digrip[idx].filter = newflag
 207                         wd.digrip[idx].recursive = wd.recursive
 208                         if err = wd.digrip[idx].register(cph); err != nil {
 209                                 wd.closeHandle()
 210                                 return
 211                         }
 212                 }
 213                 wd.count++
 214         }
 215         return
 216 }
 217
 218 // closeHandle closes handles that are stored in digrip array. Function always
 219 // tries to close all of the handlers before it exits, even when there are errors
 220 // returned from the operating system kernel.
 221 func (wd *watched) closeHandle() (err error) {
 222         for _, g := range wd.digrip {
 223                 if g != nil && g.handle != syscall.InvalidHandle {
 224                         switch suberr := syscall.CloseHandle(g.handle); {
 225                         case suberr == nil:
 226                                 g.handle = syscall.InvalidHandle
 227                         case err == nil:
 228                                 err = suberr
 229                         }
 230                 }
 231         }
 232         return
 233 }
 234
 235 // watcher implements Watcher interface. It stores a set of watched directories.
 236 // All operations which remove watched objects from map `m` must be performed in
 237 // loop goroutine since these structures are used internally by operating system.
 238 type readdcw struct {
 239         sync.Mutex
 240         m     map[string]*watched
 241         cph   syscall.Handle
 242         start bool
 243         wg    sync.WaitGroup
 244         c     chan<- EventInfo
 245 }
 246
 247 // NewWatcher creates new non-recursive watcher backed by ReadDirectoryChangesW.
 248 func newWatcher(c chan<- EventInfo) watcher {
 249         r := &readdcw{
 250                 m:   make(map[string]*watched),
 251                 cph: syscall.InvalidHandle,
 252                 c:   c,
 253         }
 254         runtime.SetFinalizer(r, func(r *readdcw) {
 255                 if r.cph != syscall.InvalidHandle {
 256                         syscall.CloseHandle(r.cph)
 257                 }
 258         })
 259         return r
 260 }
 261
 262 // Watch implements notify.Watcher interface.
 263 func (r *readdcw) Watch(path string, event Event) error {
 264         return r.watch(path, event, false)
 265 }
 266
 267 // RecursiveWatch implements notify.RecursiveWatcher interface.
 268 func (r *readdcw) RecursiveWatch(path string, event Event) error {
 269         return r.watch(path, event, true)
 270 }
 271
 272 // watch inserts a directory to the group of watched folders. If watched folder
 273 // already exists, function tries to rewatch it with new filters(NOT VALID). Moreover,
 274 // watch starts the main event loop goroutine when called for the first time.
 275 func (r *readdcw) watch(path string, event Event, recursive bool) (err error) {
 276         if event&^(All|fileNotifyChangeAll) != 0 {
 277                 return errors.New("notify: unknown event")
 278         }
 279         r.Lock()
 280         wd, ok := r.m[path]
 281         r.Unlock()
 282         if !ok {
 283                 if err = r.lazyinit(); err != nil {
 284                         return
 285                 }
 286                 r.Lock()
 287                 if wd, ok = r.m[path]; ok {
 288                         r.Unlock()
 289                         return
 290                 }
 291                 if wd, err = newWatched(r.cph, uint32(event), recursive, path); err != nil {
 292                         r.Unlock()
 293                         return
 294                 }
 295                 r.m[path] = wd
 296                 r.Unlock()
 297         }
 298         return nil
 299 }
 300
 301 // lazyinit creates an I/O completion port and starts the main event processing
 302 // loop. This method uses Double-Checked Locking optimization.
 303 func (r *readdcw) lazyinit() (err error) {
 304         invalid := uintptr(syscall.InvalidHandle)
 305         if atomic.LoadUintptr((*uintptr)(&r.cph)) == invalid {
 306                 r.Lock()
 307                 defer r.Unlock()
 308                 if atomic.LoadUintptr((*uintptr)(&r.cph)) == invalid {
 309                         cph := syscall.InvalidHandle
 310                         if cph, err = syscall.CreateIoCompletionPort(cph, 0, 0, 0); err != nil {
 311                                 return
 312                         }
 313                         r.cph, r.start = cph, true
 314                         go r.loop()
 315                 }
 316         }
 317         return
 318 }
 319
 320 // TODO(pknap) : doc
 321 func (r *readdcw) loop() {
 322         var n, key uint32
 323         var overlapped *syscall.Overlapped
 324         for {
 325                 err := syscall.GetQueuedCompletionStatus(r.cph, &n, &key, &overlapped, syscall.INFINITE)
 326                 if key == stateCPClose {
 327                         r.Lock()
 328                         handle := r.cph
 329                         r.cph = syscall.InvalidHandle
 330                         r.Unlock()
 331                         syscall.CloseHandle(handle)
 332                         r.wg.Done()
 333                         return
 334                 }
 335                 if overlapped == nil {
 336                         // TODO: check key == rewatch delete or 0(panic)
 337                         continue
 338                 }
 339                 overEx := (*overlappedEx)(unsafe.Pointer(overlapped))
 340                 if n == 0 {
 341                         r.loopstate(overEx)
 342                 } else {
 343                         r.loopevent(n, overEx)
 344                         if err = overEx.parent.readDirChanges(); err != nil {
 345                                 // TODO: error handling
 346                         }
 347                 }
 348         }
 349 }
 350
 351 // TODO(pknap) : doc
 352 func (r *readdcw) loopstate(overEx *overlappedEx) {
 353         filter := atomic.LoadUint32(&overEx.parent.parent.filter)
 354         if filter&onlyMachineStates == 0 {
 355                 return
 356         }
 357         if overEx.parent.parent.count--; overEx.parent.parent.count == 0 {
 358                 switch filter & onlyMachineStates {
 359                 case stateRewatch:
 360                         r.Lock()
 361                         overEx.parent.parent.recreate(r.cph)
 362                         r.Unlock()
 363                 case stateUnwatch:
 364                         r.Lock()
 365                         delete(r.m, syscall.UTF16ToString(overEx.parent.pathw))
 366                         r.Unlock()
 367                 case stateCPClose:
 368                 default:
 369                         panic(`notify: windows loopstate logic error`)
 370                 }
 371         }
 372 }
 373
 374 // TODO(pknap) : doc
 375 func (r *readdcw) loopevent(n uint32, overEx *overlappedEx) {
 376         events := []*event{}
 377         var currOffset uint32
 378         for {
 379                 raw := (*syscall.FileNotifyInformation)(unsafe.Pointer(&overEx.parent.buffer[currOffset]))
 380                 name := syscall.UTF16ToString((*[syscall.MAX_LONG_PATH]uint16)(unsafe.Pointer(&raw.FileName))[:raw.FileNameLength>>1])
 381                 events = append(events, &event{
 382                         pathw:  overEx.parent.pathw,
 383                         filter: overEx.parent.filter,
 384                         action: raw.Action,
 385                         name:   name,
 386                 })
 387                 if raw.NextEntryOffset == 0 {
 388                         break
 389                 }
 390                 if currOffset += raw.NextEntryOffset; currOffset >= n {
 391                         break
 392                 }
 393         }
 394         r.send(events)
 395 }
 396
 397 // TODO(pknap) : doc
 398 func (r *readdcw) send(es []*event) {
 399         for _, e := range es {
 400                 var syse Event
 401                 if e.e, syse = decode(e.filter, e.action); e.e == 0 && syse == 0 {
 402                         continue
 403                 }
 404                 switch {
 405                 case e.action == syscall.FILE_ACTION_MODIFIED:
 406                         e.ftype = fTypeUnknown
 407                 case e.filter&uint32(dirmarker) != 0:
 408                         e.ftype = fTypeDirectory
 409                 default:
 410                         e.ftype = fTypeFile
 411                 }
 412                 switch {
 413                 case e.e == 0:
 414                         e.e = syse
 415                 case syse != 0:
 416                         r.c <- &event{
 417                                 pathw:  e.pathw,
 418                                 name:   e.name,
 419                                 ftype:  e.ftype,
 420                                 action: e.action,
 421                                 filter: e.filter,
 422                                 e:      syse,
 423                         }
 424                 }
 425                 r.c <- e
 426         }
 427 }
 428
 429 // Rewatch implements notify.Rewatcher interface.
 430 func (r *readdcw) Rewatch(path string, oldevent, newevent Event) error {
 431         return r.rewatch(path, uint32(oldevent), uint32(newevent), false)
 432 }
 433
 434 // RecursiveRewatch implements notify.RecursiveRewatcher interface.
 435 func (r *readdcw) RecursiveRewatch(oldpath, newpath string, oldevent,
 436         newevent Event) error {
 437         if oldpath != newpath {
 438                 if err := r.unwatch(oldpath); err != nil {
 439                         return err
 440                 }
 441                 return r.watch(newpath, newevent, true)
 442         }
 443         return r.rewatch(newpath, uint32(oldevent), uint32(newevent), true)
 444 }
 445
 446 // TODO : (pknap) doc.
 447 func (r *readdcw) rewatch(path string, oldevent, newevent uint32, recursive bool) (err error) {
 448         if Event(newevent)&^(All|fileNotifyChangeAll) != 0 {
 449                 return errors.New("notify: unknown event")
 450         }
 451         var wd *watched
 452         r.Lock()
 453         if wd, err = r.nonStateWatched(path); err != nil {
 454                 r.Unlock()
 455                 return
 456         }
 457         if wd.filter&(onlyNotifyChanges|onlyNGlobalEvents) != oldevent {
 458                 panic(`notify: windows re-watcher logic error`)
 459         }
 460         wd.filter = stateRewatch | newevent
 461         wd.recursive, recursive = recursive, wd.recursive
 462         if err = wd.closeHandle(); err != nil {
 463                 wd.filter = oldevent
 464                 wd.recursive = recursive
 465                 r.Unlock()
 466                 return
 467         }
 468         r.Unlock()
 469         return
 470 }
 471
 472 // TODO : pknap
 473 func (r *readdcw) nonStateWatched(path string) (wd *watched, err error) {
 474         wd, ok := r.m[path]
 475         if !ok || wd == nil {
 476                 err = errors.New(`notify: ` + path + ` path is unwatched`)
 477                 return
 478         }
 479         if filter := atomic.LoadUint32(&wd.filter); filter&onlyMachineStates != 0 {
 480                 err = errors.New(`notify: another re/unwatching operation in progress`)
 481                 return
 482         }
 483         return
 484 }
 485
 486 // Unwatch implements notify.Watcher interface.
 487 func (r *readdcw) Unwatch(path string) error {
 488         return r.unwatch(path)
 489 }
 490
 491 // RecursiveUnwatch implements notify.RecursiveWatcher interface.
 492 func (r *readdcw) RecursiveUnwatch(path string) error {
 493         return r.unwatch(path)
 494 }
 495
 496 // TODO : pknap
 497 func (r *readdcw) unwatch(path string) (err error) {
 498         var wd *watched
 499         r.Lock()
 500         if wd, err = r.nonStateWatched(path); err != nil {
 501                 r.Unlock()
 502                 return
 503         }
 504         wd.filter |= stateUnwatch
 505         if err = wd.closeHandle(); err != nil {
 506                 wd.filter &^= stateUnwatch
 507                 r.Unlock()
 508                 return
 509         }
 510         r.Unlock()
 511         return
 512 }
 513
 514 // Close resets the whole watcher object, closes all existing file descriptors,
 515 // and sends stateCPClose state as completion key to the main watcher's loop.
 516 func (r *readdcw) Close() (err error) {
 517         r.Lock()
 518         if !r.start {
 519                 r.Unlock()
 520                 return nil
 521         }
 522         for _, wd := range r.m {
 523                 wd.filter &^= onlyMachineStates
 524                 wd.filter |= stateCPClose
 525                 if e := wd.closeHandle(); e != nil && err == nil {
 526                         err = e
 527                 }
 528         }
 529         r.start = false
 530         r.Unlock()
 531         r.wg.Add(1)
 532         if e := syscall.PostQueuedCompletionStatus(r.cph, 0, stateCPClose, nil); e != nil && err == nil {
 533                 return e
 534         }
 535         r.wg.Wait()
 536         return
 537 }
 538
 539 // decode creates a notify event from both non-raw filter and action which was
 540 // returned from completion routine. Function may return Event(0) in case when
 541 // filter was replaced by a new value which does not contain fields that are
 542 // valid with passed action.
 543 func decode(filter, action uint32) (Event, Event) {
 544         switch action {
 545         case syscall.FILE_ACTION_ADDED:
 546                 return gensys(filter, Create, FileActionAdded)
 547         case syscall.FILE_ACTION_REMOVED:
 548                 return gensys(filter, Remove, FileActionRemoved)
 549         case syscall.FILE_ACTION_MODIFIED:
 550                 return gensys(filter, Write, FileActionModified)
 551         case syscall.FILE_ACTION_RENAMED_OLD_NAME:
 552                 return gensys(filter, Rename, FileActionRenamedOldName)
 553         case syscall.FILE_ACTION_RENAMED_NEW_NAME:
 554                 return gensys(filter, Rename, FileActionRenamedNewName)
 555         }
 556         panic(`notify: cannot decode internal mask`)
 557 }
 558
 559 // gensys decides whether the Windows action, system-independent event or both
 560 // of them should be returned. Since the grip's filter may be atomically changed
 561 // during watcher lifetime, it is possible that neither Windows nor notify masks
 562 // are watched by the user when this function is called.
 563 func gensys(filter uint32, ge, se Event) (gene, syse Event) {
 564         isdir := filter&uint32(dirmarker) != 0
 565         if isdir && filter&uint32(FileNotifyChangeDirName) != 0 ||
 566                 !isdir && filter&uint32(FileNotifyChangeFileName) != 0 ||
 567                 filter&uint32(fileNotifyChangeModified) != 0 {
 568                 syse = se
 569         }
 570         if filter&uint32(ge) != 0 {
 571                 gene = ge
 572         }
 573         return
 574 }