// startWatchAsync tries to start the filesystem watching and retries every minute on failure.
// It is a convenience function that should not be used except in startWatch.
188847
func (f *folder) startWatchAsync(ctx context.Context) {
	timer := time.NewTimer(0)
	for {
		select {
		case <-timer.C:
			eventChan, err := f.Filesystem().Watch(".", f.ignores, ctx, f.IgnorePerms)
			f.watchMut.Lock()
			prevErr := f.watchErr
			f.watchErr = err
			f.watchMut.Unlock()
			if err != prevErr {
				data := map[string]interface{}{
					"folder": f.ID,
				}
				if prevErr != nil {
					data["from"] = prevErr.Error()
				}
				if err != nil {
					data["to"] = err.Error()
				}
				events.Default.Log(events.FolderWatchStateChanged, data)
			}
			if err != nil {
				if prevErr == errWatchNotStarted {
					l.Infof("Error while trying to start filesystem watcher for folder %s, trying again in 1min: %v", f.Description(), err)
				} else {
					l.Debugf("Repeat error while trying to start filesystem watcher for folder %s, trying again in 1min: %v", f.Description(), err)
				}
				timer.Reset(time.Minute)
				continue
			}
			f.watchMut.Lock()
			defer f.watchMut.Unlock()
			watchaggregator.Aggregate(eventChan, f.watchChan, f.FolderConfiguration, f.model.cfg, ctx)
			l.Debugln("Started filesystem watcher for folder", f.Description())
			return
		case <-ctx.Done():
			return
		}
	}
}
438569
func (p *Project) Watch(wg *sync.WaitGroup) {
	var err error
	// change channel
	p.stop = make(chan bool)
	// init a new watcher
	p.watcher, err = NewFileWatcher(p.parent.Settings.Legacy)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		close(p.stop)
		p.watcher.Close()
	}()
	// before start checks
	p.Before()
	// start watcher
	go p.Reload("", p.stop)
L:
	for {
		select {
		case event := <-p.watcher.Events():
			if p.parent.Settings.Recovery.Events {
				log.Println("File:", event.Name, "LastFile:", p.last.file, "Time:", time.Now(), "LastTime:", p.last.time)
			}
			if time.Now().Truncate(time.Second).After(p.last.time) {
				// switch event type
				switch event.Op {
				case fsnotify.Chmod:
				case fsnotify.Remove:
					p.watcher.Remove(event.Name)
					if p.Validate(event.Name, false) && ext(event.Name) != "" {
						// stop and restart
						close(p.stop)
						p.stop = make(chan bool)
						p.Change(event)
						go p.Reload("", p.stop)
					}
				default:
					if p.Validate(event.Name, true) {
						fi, err := os.Stat(event.Name)
						if err != nil {
							continue
						}
						if fi.IsDir() {
							filepath.Walk(event.Name, p.walk)
						} else {
							// stop and restart
							close(p.stop)
							p.stop = make(chan bool)
							p.Change(event)
							go p.Reload(event.Name, p.stop)
							p.last.time = time.Now().Truncate(time.Second)
							p.last.file = event.Name
						}
					}
				}
			}
		case err := <-p.watcher.Errors():
			p.Err(err)
		case <-p.exit:
			p.After()
			break L
		}
	}
	wg.Done()
}
196187
func (watcher *RecursiveWatcher) Start() {
	go func() {
		for {
			select {
			case event := <-watcher.Event:
				watcher.handleEvent(event)
			case err := <-watcher.Error:
				watcher.ErrorEvents <- err
				log.Println("error:", err)
			}
		}
	}()
}
981285
func (eventBroadcaster *eventBroadcasterImpl) StartEventWatcher(eventHandler func(*v1.Event)) watch.Interface {
	watcher := eventBroadcaster.Watch()
	go func() {
		defer utilruntime.HandleCrash()
		for watchEvent := range watcher.ResultChan() {
			event, ok := watchEvent.Object.(*v1.Event)
			if !ok {
				// This is all local, so there's no reason this should
				// ever happen.
				continue
			}
			eventHandler(event)
		}
	}()
	return watcher
}
75413
func (p *pin) BeginWatch(edge Edge, callback IRQEvent) error {
	p.SetMode(ModeInput)
	if err := write([]byte(edge), p.edgePath); err != nil {
		return err
	}

	var event syscall.EpollEvent
	event.Events = syscall.EPOLLIN | (syscall.EPOLLET & 0xffffffff) | syscall.EPOLLPRI

	fd := int(p.valueFile.Fd())

	p.callback = callback
	watchEventCallbacks[fd] = p

	if err := syscall.SetNonblock(fd, true); err != nil {
		return err
	}

	event.Fd = int32(fd)

	if err := syscall.EpollCtl(epollFD, syscall.EPOLL_CTL_ADD, fd, &event); err != nil {
		return err
	}

	return nil

}
767155
func (s *Server) StartExitMonitor() {
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		logrus.Fatalf("Failed to create new watch: %v", err)
	}
	defer watcher.Close()

	done := make(chan struct{})
	go func() {
		for {
			select {
			case event := <-watcher.Events:
				logrus.Debugf("event: %v", event)
				if event.Op&fsnotify.Create == fsnotify.Create {
					containerID := filepath.Base(event.Name)
					logrus.Debugf("container or sandbox exited: %v", containerID)
					c := s.GetContainer(containerID)
					if c != nil {
						logrus.Debugf("container exited and found: %v", containerID)
						err := s.Runtime().UpdateContainerStatus(c)
						if err != nil {
							logrus.Warnf("Failed to update container status %s: %v", containerID, err)
						} else {
							s.ContainerStateToDisk(c)
						}
					} else {
						sb := s.GetSandbox(containerID)
						if sb != nil {
							c := sb.InfraContainer()
							if c == nil {
								logrus.Warnf("no infra container set for sandbox: %v", containerID)
								continue
							}
							logrus.Debugf("sandbox exited and found: %v", containerID)
							err := s.Runtime().UpdateContainerStatus(c)
							if err != nil {
								logrus.Warnf("Failed to update sandbox infra container status %s: %v", c.ID(), err)
							} else {
								s.ContainerStateToDisk(c)
							}
						}
					}
				}
			case err := <-watcher.Errors:
				logrus.Debugf("watch error: %v", err)
				return
			case <-s.monitorsChan:
				logrus.Debug("closing exit monitor...")
				close(done)
				return
			}
		}
	}()
	if err := watcher.Add(s.config.ContainerExitsDir); err != nil {
		logrus.Errorf("watcher.Add(%q) failed: %s", s.config.ContainerExitsDir, err)
		close(done)
	}
	<-done
}
1246956
func (c *Client) WatchTask(id string) *WatchTasksResult {
	// during watch we don't want to have a timeout
	// Store the old timeout so we can restore when we are through
	oldTimeout := c.http.Timeout
	c.http.Timeout = time.Duration(0)

	r := &WatchTasksResult{
		EventChan: make(chan *rbody.StreamedTaskEvent),
		DoneChan:  make(chan struct{}),
	}

	url := fmt.Sprintf("%s/tasks/%v/watch", c.prefix, id)
	req, err := http.NewRequest("GET", url, nil)
	addAuth(req, c.Username, c.Password)
	if err != nil {
		r.Err = err
		r.Close()
		return r
	}
	resp, err := c.http.Do(req)
	if err != nil {
		if strings.Contains(err.Error(), "tls: oversized record") || strings.Contains(err.Error(), "malformed HTTP response") {
			r.Err = fmt.Errorf("error connecting to API URI: %s. Do you have an http/https mismatch?", c.URL)
		} else {
			r.Err = err
		}
		r.Close()
		return r
	}

	if resp.StatusCode != 200 {
		ar, err := httpRespToAPIResp(resp)
		if err != nil {
			r.Err = err
		} else {
			r.Err = errors.New(ar.Meta.Message)
		}
		r.Close()
		return r
	}

	// Start watching
	go func() {
		reader := bufio.NewReader(resp.Body)
		defer func() { c.http.Timeout = oldTimeout }()
		for {
			select {
			case <-r.DoneChan:
				resp.Body.Close()
				return
			default:
				line, _ := reader.ReadBytes('\n')
				sline := string(line)
				if sline == "" || sline == "\n" {
					continue
				}
				if strings.HasPrefix(sline, "data:") {
					sline = strings.TrimPrefix(sline, "data:")
					line = []byte(sline)
				}
				ste := &rbody.StreamedTaskEvent{}
				err := json.Unmarshal(line, ste)
				if err != nil {
					r.Err = err
					r.Close()
					return
				}
				switch ste.EventType {
				case rbody.TaskWatchTaskDisabled:
					r.EventChan <- ste
					r.Close()
				case rbody.TaskWatchTaskStopped, rbody.TaskWatchTaskEnded, rbody.TaskWatchTaskStarted, rbody.TaskWatchMetricEvent:
					r.EventChan <- ste
				}
			}
		}
	}()
	return r
}
1038514
func (o *RsyncOptions) WatchAndSync() error {

	// these variables must be accessed while holding the changeLock
	// mutex as they are shared between goroutines to communicate
	// sync state/events.
	var (
		changeLock sync.Mutex
		dirty      bool
		lastChange time.Time
		watchError error
	)

	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		return fmt.Errorf("error setting up filesystem watcher: %v", err)
	}
	defer watcher.Close()

	go func() {
		for {
			select {
			case event := <-watcher.Events:
				changeLock.Lock()
				klog.V(5).Infof("filesystem watch event: %s", event)
				lastChange = time.Now()
				dirty = true
				if event.Op&fsnotify.Remove == fsnotify.Remove {
					if e := watcher.Remove(event.Name); e != nil {
						klog.V(5).Infof("error removing watch for %s: %v", event.Name, e)
					}
				} else {
					if e := fsnotification.AddRecursiveWatch(watcher, event.Name); e != nil && watchError == nil {
						watchError = e
					}
				}
				changeLock.Unlock()
			case err := <-watcher.Errors:
				changeLock.Lock()
				watchError = fmt.Errorf("error watching filesystem for changes: %v", err)
				changeLock.Unlock()
			}
		}
	}()

	err = fsnotification.AddRecursiveWatch(watcher, o.Source.Path)
	if err != nil {
		return fmt.Errorf("error watching source path %s: %v", o.Source.Path, err)
	}

	delay := 2 * time.Second
	ticker := time.NewTicker(delay)
	defer ticker.Stop()
	for {
		changeLock.Lock()
		if watchError != nil {
			return watchError
		}
		// if a change happened more than 'delay' seconds ago, sync it now.
		// if a change happened less than 'delay' seconds ago, sleep for 'delay' seconds
		// and see if more changes happen, we don't want to sync when
		// the filesystem is in the middle of changing due to a massive
		// set of changes (such as a local build in progress).
		if dirty && time.Now().After(lastChange.Add(delay)) {
			klog.V(1).Info("Synchronizing filesystem changes...")
			err = o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut)
			if err != nil {
				return err
			}
			klog.V(1).Info("Done.")
			dirty = false
		}
		changeLock.Unlock()
		<-ticker.C
	}
}
54560
func (c *Int) Watch(watch IntWatch) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.watches = append(c.watches, watch)
}
1248606
func (f *folder) restartWatch() {
	f.stopWatch()
	f.startWatch()
	f.scanSubdirs(nil)
}
800859
func (ns *NamedStopwatch) Start(name string) {
	if ns == nil {
		return // if we're not using stopwatches we just do nothing
	}
	ns.Lock()
	defer ns.Unlock()

	ns.start(name)
}
1322516
func (c *FakeApplications) Watch(opts v1.ListOptions) (watch.Interface, error) {
	return c.Fake.
		InvokesWatch(testing.NewWatchAction(applicationsResource, c.ns, opts))

}
885631
func (st *State) StartSync() {
	if advanceable, ok := st.clock().(hasAdvance); ok {
		// The amount of time we advance here just needs to be more
		// than 10ms as that is the minimum time the txnwatcher
		// is waiting on, however one second is more human noticeable.
		// The state testing StateSuite type changes the polling interval
		// of the pool's txnwatcher to be one second.
		advanceable.Advance(time.Second)
	}
	if syncable, ok := st.workers.txnLogWatcher().(hasStartSync); ok {
		syncable.StartSync()
	}
	st.workers.pingBatcherWorker().Sync()
	st.workers.presenceWatcher().Sync()
}
97251
func (f *asyncFlusher) start(numParallel int) {
	f.chunks = make(chan chunk)
	for i := 0; i < numParallel; i++ {
		f.wg.Add(1)
		go func() {
			defer f.wg.Done()
			f.uploaderLoop()
		}()
	}
}
39953
func (adapter *Registry) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error) {

	adapter.access.Lock()
	defer adapter.access.Unlock()

	if _, found := adapter.subscriptions[resyncName]; found {
		return nil, errors.New("Already watching " + resyncName)
	}

	adapter.subscriptions[resyncName] = &Subscription{
		ResyncName:  resyncName,
		ChangeChan:  changeChan,
		ResyncChan:  resyncChan,
		CloseChan:   make(chan string),
		KeyPrefixes: keyPrefixes,
	}

	return &WatchDataReg{resyncName, adapter}, nil
}
1286604
func (shardSwap *shardSchemaSwap) startHealthWatchers(ctx context.Context) error {
	shardSwap.allTablets = make(map[string]*discovery.TabletStats)

	shardSwap.tabletHealthCheck = discovery.NewHealthCheck(*vtctl.HealthcheckRetryDelay, *vtctl.HealthCheckTimeout)
	shardSwap.tabletHealthCheck.SetListener(shardSwap, true /* sendDownEvents */)

	topoServer := shardSwap.parent.topoServer
	cellList, err := topoServer.GetKnownCells(shardSwap.parent.ctx)
	if err != nil {
		return err
	}
	for _, cell := range cellList {
		watcher := discovery.NewShardReplicationWatcher(
			ctx,
			topoServer,
			shardSwap.tabletHealthCheck,
			cell,
			shardSwap.parent.keyspace,
			shardSwap.shardName,
			*vtctl.HealthCheckTimeout,
			discovery.DefaultTopoReadConcurrency)
		shardSwap.tabletWatchers = append(shardSwap.tabletWatchers, watcher)
	}
	for _, watcher := range shardSwap.tabletWatchers {
		if err := watcher.WaitForInitialTopology(); err != nil {
			return err
		}
	}
	shardSwap.tabletHealthCheck.WaitForInitialStatsUpdates()

	// Wait for all health connections to be either established or get broken.
waitHealthChecks:
	for {
		tabletList := shardSwap.getTabletList()
		for _, tabletStats := range tabletList {
			if tabletStats.Stats == nil && tabletStats.LastError == nil {
				select {
				case <-shardSwap.parent.ctx.Done():
					return shardSwap.parent.ctx.Err()
				case <-time.After(100 * time.Millisecond):
				}
				continue waitHealthChecks
			}
		}
		break waitHealthChecks
	}
	return nil
}
717888
func (w *Watcher) Start(d time.Duration) error {
	// Return an error if d is less than 1 nanosecond.
	if d < time.Nanosecond {
		return ErrDurationTooShort
	}

	// Make sure the Watcher is not already running.
	w.mu.Lock()
	if w.running {
		w.mu.Unlock()
		return ErrWatcherRunning
	}
	w.running = true
	w.mu.Unlock()

	// Unblock w.Wait().
	w.wg.Done()

	for {
		// done lets the inner polling cycle loop know when the
		// current cycle's method has finished executing.
		done := make(chan struct{})

		// Any events that are found are first piped to evt before
		// being sent to the main Event channel.
		evt := make(chan Event)

		// Retrieve the file list for all watched file's and dirs.
		fileList := w.retrieveFileList()

		// cancel can be used to cancel the current event polling function.
		cancel := make(chan struct{})

		// Look for events.
		go func() {
			w.pollEvents(fileList, evt, cancel)
			done <- struct{}{}
		}()

		// numEvents holds the number of events for the current cycle.
		numEvents := 0

	inner:
		for {
			select {
			case <-w.close:
				close(cancel)
				close(w.Closed)
				return nil
			case event := <-evt:
				if len(w.ops) > 0 { // Filter Ops.
					_, found := w.ops[event.Op]
					if !found {
						continue
					}
				}
				numEvents++
				if w.maxEvents > 0 && numEvents > w.maxEvents {
					close(cancel)
					break inner
				}
				w.Event <- event
			case <-done: // Current cycle is finished.
				break inner
			}
		}

		// Update the file's list.
		w.mu.Lock()
		w.files = fileList
		w.mu.Unlock()

		// Sleep and then continue to the next loop iteration.
		time.Sleep(d)
	}
}