<commit_before>// Copyright 2015 Google Inc. All Rights Reserved.
// This file is available under the Apache license.

package watcher

import (
	"context"
	"expvar"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	"github.com/golang/glog"
	"github.com/pkg/errors"
)

var (
	errorCount = expvar.NewInt("log_watcher_errors_total")
)

type watch struct {
	ps []Processor
	fi os.FileInfo
}

// hasChanged indicates that a FileInfo has changed.
// http://apenwarr.ca/log/20181113 suggests that comparing mtime is
// insufficient for sub-second resolution on many platforms, and we can do
// better by comparing a few fields in the FileInfo.  This set of tests is less
// than the ones suggested in the blog post, but seem sufficient for making
// tests (notably, sub-millisecond accuracy) pass quickly.  mtime-only diff has
// caused race conditions in test and likely caused strange behaviour in
// production environments.
func hasChanged(a, b os.FileInfo) bool {
	if a == nil || b == nil {
		glog.V(2).Info("One or both FileInfos are nil")
		return true
	}
	if a.ModTime() != b.ModTime() {
		glog.V(2).Info("modtimes differ")
		return true
	}
	if a.Size() != b.Size() {
		glog.V(2).Info("sizes differ")
		return true
	}
	if a.Mode() != b.Mode() {
		glog.V(2).Info("modes differ")
		return true
	}
	return false
}

// LogWatcher implements a Watcher for watching real filesystems.
type LogWatcher struct {
	watcher    *fsnotify.Watcher
	pollTicker *time.Ticker

	watchedMu sync.RWMutex // protects `watched'
	watched   map[string]*watch

	stopTicks chan struct{} // Channel to notify ticker to stop.

	ticksDone  chan struct{} // Channel to notify when the ticks handler is done.
	eventsDone chan struct{} // Channel to notify when the events handler is done.

	pollMu sync.Mutex // protects `Poll()`

	closeOnce sync.Once
}

// NewLogWatcher returns a new LogWatcher, or returns an error.
func NewLogWatcher(pollInterval time.Duration, enableFsnotify bool) (*LogWatcher, error) {
	var f *fsnotify.Watcher
	if enableFsnotify {
		var err error
		f, err = fsnotify.NewWatcher()
		if err != nil {
			glog.Warning(err)
		}
	}
	w := &LogWatcher{
		watcher: f,
		watched: make(map[string]*watch),
	}
	if pollInterval > 0 {
		w.pollTicker = time.NewTicker(pollInterval)
		w.stopTicks = make(chan struct{})
		w.ticksDone = make(chan struct{})
		go w.runTicks()
		glog.V(2).Infof("started ticker with %s interval", pollInterval)
	}
	if f != nil {
		w.eventsDone = make(chan struct{})
		go w.runEvents()
	}
	return w, nil
}

func (w *LogWatcher) sendEvent(e Event) {
	w.watchedMu.RLock()
	watch, ok := w.watched[e.Pathname]
	w.watchedMu.RUnlock()
	if !ok {
		d := filepath.Dir(e.Pathname)
		w.watchedMu.RLock()
		watch, ok = w.watched[d]
		w.watchedMu.RUnlock()
		if !ok {
			glog.V(2).Infof("No watch for path %q", e.Pathname)
			return
		}
	}
	w.sendWatchedEvent(watch, e)
}

// Send an event to a watch; all locks assumed to be held.
func (w *LogWatcher) sendWatchedEvent(watch *watch, e Event) {
	for _, p := range watch.ps {
		p.ProcessFileEvent(context.TODO(), e)
	}
}

func (w *LogWatcher) runTicks() {
	defer close(w.ticksDone)

	if w.pollTicker == nil {
		return
	}

	for {
		select {
		case <-w.pollTicker.C:
			w.Poll()
		case <-w.stopTicks:
			w.pollTicker.Stop()
			return
		}
	}
}

// Poll all watched objects for updates, dispatching events if required.
func (w *LogWatcher) Poll() {
	w.pollMu.Lock()
	defer w.pollMu.Unlock()
	glog.V(2).Info("Polling watched files.")
	w.watchedMu.RLock()
	for n, watch := range w.watched {
		w.watchedMu.RUnlock()
		w.pollWatchedPath(n, watch)
		w.watchedMu.RLock()
	}
	w.watchedMu.RUnlock()
}

// pollWatchedPathLocked polls an already-watched path for updates.
func (w *LogWatcher) pollWatchedPath(pathname string, watched *watch) {
	glog.V(2).Infof("Stat %q", pathname)
	fi, err := os.Stat(pathname)
	if err != nil {
		glog.V(1).Info(err)<commit_msg>Send delete events when the file is not statted after a poll<commit_after>// Copyright 2015 Google Inc. All Rights Reserved.
// This file is available under the Apache license.

package watcher

import (
	"context"
	"expvar"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	"github.com/golang/glog"
	"github.com/pkg/errors"
)

var (
	errorCount = expvar.NewInt("log_watcher_errors_total")
)

type watch struct {
	ps []Processor
	fi os.FileInfo
}

// hasChanged indicates that a FileInfo has changed.
// http://apenwarr.ca/log/20181113 suggests that comparing mtime is
// insufficient for sub-second resolution on many platforms, and we can do
// better by comparing a few fields in the FileInfo.  This set of tests is less
// than the ones suggested in the blog post, but seem sufficient for making
// tests (notably, sub-millisecond accuracy) pass quickly.  mtime-only diff has
// caused race conditions in test and likely caused strange behaviour in
// production environments.
func hasChanged(a, b os.FileInfo) bool {
	if a == nil || b == nil {
		glog.V(2).Info("One or both FileInfos are nil")
		return true
	}
	if a.ModTime() != b.ModTime() {
		glog.V(2).Info("modtimes differ")
		return true
	}
	if a.Size() != b.Size() {
		glog.V(2).Info("sizes differ")
		return true
	}
	if a.Mode() != b.Mode() {
		glog.V(2).Info("modes differ")
		return true
	}
	return false
}

// LogWatcher implements a Watcher for watching real filesystems.
type LogWatcher struct {
	watcher    *fsnotify.Watcher
	pollTicker *time.Ticker

	watchedMu sync.RWMutex // protects `watched'
	watched   map[string]*watch

	stopTicks chan struct{} // Channel to notify ticker to stop.

	ticksDone  chan struct{} // Channel to notify when the ticks handler is done.
	eventsDone chan struct{} // Channel to notify when the events handler is done.

	pollMu sync.Mutex // protects `Poll()`

	closeOnce sync.Once
}

// NewLogWatcher returns a new LogWatcher, or returns an error.
func NewLogWatcher(pollInterval time.Duration, enableFsnotify bool) (*LogWatcher, error) {
	var f *fsnotify.Watcher
	if enableFsnotify {
		var err error
		f, err = fsnotify.NewWatcher()
		if err != nil {
			glog.Warning(err)
		}
	}
	w := &LogWatcher{
		watcher: f,
		watched: make(map[string]*watch),
	}
	if pollInterval > 0 {
		w.pollTicker = time.NewTicker(pollInterval)
		w.stopTicks = make(chan struct{})
		w.ticksDone = make(chan struct{})
		go w.runTicks()
		glog.V(2).Infof("started ticker with %s interval", pollInterval)
	}
	if f != nil {
		w.eventsDone = make(chan struct{})
		go w.runEvents()
	}
	return w, nil
}

func (w *LogWatcher) sendEvent(e Event) {
	w.watchedMu.RLock()
	watch, ok := w.watched[e.Pathname]
	w.watchedMu.RUnlock()
	if !ok {
		d := filepath.Dir(e.Pathname)
		w.watchedMu.RLock()
		watch, ok = w.watched[d]
		w.watchedMu.RUnlock()
		if !ok {
			glog.V(2).Infof("No watch for path %q", e.Pathname)
			return
		}
	}
	w.sendWatchedEvent(watch, e)
}

// Send an event to a watch; all locks assumed to be held.
func (w *LogWatcher) sendWatchedEvent(watch *watch, e Event) {
	for _, p := range watch.ps {
		p.ProcessFileEvent(context.TODO(), e)
	}
}

func (w *LogWatcher) runTicks() {
	defer close(w.ticksDone)

	if w.pollTicker == nil {
		return
	}

	for {
		select {
		case <-w.pollTicker.C:
			w.Poll()
		case <-w.stopTicks:
			w.pollTicker.Stop()
			return
		}
	}
}

// Poll all watched objects for updates, dispatching events if required.
func (w *LogWatcher) Poll() {
	w.pollMu.Lock()
	defer w.pollMu.Unlock()
	glog.V(2).Info("Polling watched files.")
	w.watchedMu.RLock()
	for n, watch := range w.watched {
		w.watchedMu.RUnlock()
		w.pollWatchedPath(n, watch)
		w.watchedMu.RLock()
	}
	w.watchedMu.RUnlock()
}

// pollWatchedPathLocked polls an already-watched path for updates.
func (w *LogWatcher) pollWatchedPath(pathname string, watched *watch) {
	glog.V(2).Infof("Stat %q", pathname)
	fi, err := os.Stat(pathname)
	if err != nil {
		if os.IsNotExist(err) {
			glog.V(2).Infof("sending delete for %s", pathname)
			w.sendWatchedEvent(watched, Event{Delete, pathname})
			// Need to remove the watch for any subsequent create to be sent.
			w.watchedMu.Lock()
			delete(w.watched, pathname)
			w.watchedMu.Unlock()
		} else {
			glog.V(1).Info(err)
		}
{'fraction_non_alphanumeric': 0.1295867365597952, 'fraction_numerical': 0.00548579787882482, 'mean_word_length': 3.5909345271404587, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1760815', 'n_tokens_mistral': 3169, 'n_tokens_neox': 2987, 'n_words': 859}