refactor inotify implementation

internal/fswatcher/event.go

@@ -1,75 +1,22 @@
 package fswatcher
 
 import (
-	"bytes"
 	"fmt"
-	"strconv"
-	"unsafe"
 
 	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
-	"golang.org/x/sys/unix"
 )
 
-var InotifyEvents = map[uint32]string{
-	unix.IN_ACCESS:                          "ACCESS",
-	unix.IN_ATTRIB:                          "ATTRIB",
-	unix.IN_CLOSE_NOWRITE:                   "CLOSE_NOWRITE",
-	unix.IN_CLOSE_WRITE:                     "CLOSE_WRITE",
-	unix.IN_CREATE:                          "CREATE",
-	unix.IN_DELETE:                          "DELETE",
-	unix.IN_DELETE_SELF:                     "DELETE_SELF",
-	unix.IN_MODIFY:                          "MODIFY",
-	unix.IN_MOVED_FROM:                      "MOVED_FROM",
-	unix.IN_MOVED_TO:                        "MOVED_TO",
-	unix.IN_MOVE_SELF:                       "MOVE_SELF",
-	unix.IN_OPEN:                            "OPEN",
-	(unix.IN_ACCESS | unix.IN_ISDIR):        "ACCESS DIR",
-	(unix.IN_ATTRIB | unix.IN_ISDIR):        "ATTRIB DIR",
-	(unix.IN_CLOSE_NOWRITE | unix.IN_ISDIR): "CLOSE_NOWRITE DIR",
-	(unix.IN_CLOSE_WRITE | unix.IN_ISDIR):   "CLOSE_WRITE DIR",
-	(unix.IN_CREATE | unix.IN_ISDIR):        "CREATE DIR",
-	(unix.IN_DELETE | unix.IN_ISDIR):        "DELETE DIR",
-	(unix.IN_DELETE_SELF | unix.IN_ISDIR):   "DELETE_SELF DIR",
-	(unix.IN_MODIFY | unix.IN_ISDIR):        "MODIFY DIR",
-	(unix.IN_MOVED_FROM | unix.IN_ISDIR):    "MOVED_FROM DIR",
-	(unix.IN_MOVE_SELF | unix.IN_ISDIR):     "MODE_SELF DIR",
-	(unix.IN_OPEN | unix.IN_ISDIR):          "OPEN DIR",
-}
-
 func parseEvents(i *inotify.Inotify, dataCh chan []byte, eventCh chan string, errCh chan error) {
 	for buf := range dataCh {
-		n := len(buf)
-		if n < unix.SizeofInotifyEvent {
-			errCh <- fmt.Errorf("Inotify event parser: incomplete read: n=%d", n)
-			continue
-		}
-
 		var ptr uint32
-		var name string
+		for len(buf[ptr:]) > 0 {
-		for ptr <= uint32(n-unix.SizeofInotifyEvent) {
+			event, size, err := i.ParseNextEvent(buf[ptr:])
-			sys := (*unix.InotifyEvent)(unsafe.Pointer(&buf[ptr]))
+			if err != nil {
-			ptr += unix.SizeofInotifyEvent
+				errCh <- fmt.Errorf("parsing events: %v", err)
-
-			watcher, ok := i.Watchers[int(sys.Wd)]
-			if !ok {
-				errCh <- fmt.Errorf("Inotify event parser: unknown watcher ID: %d", sys.Wd)
 				continue
 			}
-			name = watcher.Dir + "/"
+			ptr += size
-			if sys.Len > 0 && len(buf) >= int(ptr+sys.Len) {
+			eventCh <- fmt.Sprintf("%20s | %s", event.Op, event.Name)
-				name += string(bytes.TrimRight(buf[ptr:ptr+sys.Len], "\x00"))
-				ptr += sys.Len
-			}
-
-			eventCh <- formatEvent(name, sys.Mask)
 		}
 	}
 }
-
-func formatEvent(name string, mask uint32) string {
-	op, ok := InotifyEvents[mask]
-	if !ok {
-		op = strconv.FormatInt(int64(mask), 2)
-	}
-	return fmt.Sprintf("%20s | %s", op, name)
-}

internal/fswatcher/inotify/event.go

@@ -0,0 +1,29 @@
+package inotify
+
+import "golang.org/x/sys/unix"
+
+var InotifyEvents = map[uint32]string{
+	unix.IN_ACCESS:                          "ACCESS",
+	unix.IN_ATTRIB:                          "ATTRIB",
+	unix.IN_CLOSE_NOWRITE:                   "CLOSE_NOWRITE",
+	unix.IN_CLOSE_WRITE:                     "CLOSE_WRITE",
+	unix.IN_CREATE:                          "CREATE",
+	unix.IN_DELETE:                          "DELETE",
+	unix.IN_DELETE_SELF:                     "DELETE_SELF",
+	unix.IN_MODIFY:                          "MODIFY",
+	unix.IN_MOVED_FROM:                      "MOVED_FROM",
+	unix.IN_MOVED_TO:                        "MOVED_TO",
+	unix.IN_MOVE_SELF:                       "MOVE_SELF",
+	unix.IN_OPEN:                            "OPEN",
+	(unix.IN_ACCESS | unix.IN_ISDIR):        "ACCESS DIR",
+	(unix.IN_ATTRIB | unix.IN_ISDIR):        "ATTRIB DIR",
+	(unix.IN_CLOSE_NOWRITE | unix.IN_ISDIR): "CLOSE_NOWRITE DIR",
+	(unix.IN_CLOSE_WRITE | unix.IN_ISDIR):   "CLOSE_WRITE DIR",
+	(unix.IN_CREATE | unix.IN_ISDIR):        "CREATE DIR",
+	(unix.IN_DELETE | unix.IN_ISDIR):        "DELETE DIR",
+	(unix.IN_DELETE_SELF | unix.IN_ISDIR):   "DELETE_SELF DIR",
+	(unix.IN_MODIFY | unix.IN_ISDIR):        "MODIFY DIR",
+	(unix.IN_MOVED_FROM | unix.IN_ISDIR):    "MOVED_FROM DIR",
+	(unix.IN_MOVE_SELF | unix.IN_ISDIR):     "MODE_SELF DIR",
+	(unix.IN_OPEN | unix.IN_ISDIR):          "OPEN DIR",
+}

internal/fswatcher/inotify/inotify.go

@@ -1,40 +1,50 @@
 package inotify
 
 import (
+	"bytes"
 	"fmt"
+	"io/ioutil"
+	"strconv"
+	"strings"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
 )
 
-type InotifySyscalls interface {
+const maximumWatchersFile = "/proc/sys/fs/inotify/max_user_watches"
-	Init() (int, error)
-	AddWatch(int, string) (int, error)
-	Close(int) error
-}
 
 type Inotify struct {
 	FD       int
 	Watchers map[int]*Watcher
-	sys      InotifySyscalls
 }
 
-func NewInotify(isys InotifySyscalls) (*Inotify, error) {
+type Watcher struct {
-	fd, err := isys.Init()
+	WD  int
-	if err != nil {
+	Dir string
-		return nil, fmt.Errorf("initializing inotify: %v", err)
+}
+
+type Event struct {
+	Name string
+	Op   string
+}
+
+func NewInotify() (*Inotify, error) {
+	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
+	if fd < 0 {
+		return nil, fmt.Errorf("initializing inotify: errno: %d", errno)
 	}
 
 	i := &Inotify{
 		FD:       fd,
 		Watchers: make(map[int]*Watcher),
-		sys:      isys,
 	}
-
 	return i, nil
 }
 
 func (i *Inotify) Watch(dir string) error {
-	wd, err := i.sys.AddWatch(i.FD, dir)
+	wd, errno := unix.InotifyAddWatch(i.FD, dir, unix.IN_ALL_EVENTS)
-	if err != nil {
+	if wd < 0 {
-		return fmt.Errorf("adding watcher on %s: %v", dir, err)
+		return fmt.Errorf("adding watch: errno: %d", errno)
 	}
 	i.Watchers[wd] = &Watcher{
 		WD:  wd,
@@ -43,9 +53,45 @@ func (i *Inotify) Watch(dir string) error {
 	return nil
 }
 
+func (i *Inotify) Read(buf []byte) (int, error) {
+	n, errno := unix.Read(i.FD, buf)
+	if n < 0 {
+		return n, fmt.Errorf("reading from inotify fd %d: errno: %d", i.FD, errno)
+	}
+	return n, nil
+}
+
+func (i *Inotify) ParseNextEvent(buf []byte) (*Event, uint32, error) {
+	n := len(buf)
+	if n < unix.SizeofInotifyEvent {
+		return nil, uint32(n), fmt.Errorf("incomplete read: n=%d", n)
+	}
+	sys := (*unix.InotifyEvent)(unsafe.Pointer(&buf[0]))
+	offset := unix.SizeofInotifyEvent + sys.Len
+
+	watcher, ok := i.Watchers[int(sys.Wd)]
+	if !ok {
+		return nil, offset, fmt.Errorf("unknown watcher ID: %d", sys.Wd)
+	}
+
+	name := watcher.Dir + "/"
+	if sys.Len > 0 && len(buf) >= int(offset) {
+		name += string(bytes.TrimRight(buf[unix.SizeofInotifyEvent:offset], "\x00"))
+	}
+	op, ok := InotifyEvents[sys.Mask]
+	if !ok {
+		op = strconv.FormatInt(int64(sys.Mask), 2)
+	}
+
+	return &Event{
+		Name: name,
+		Op:   op,
+	}, offset, nil
+}
+
 func (i *Inotify) Close() error {
-	if err := i.sys.Close(i.FD); err != nil {
+	if err := unix.Close(i.FD); err != nil {
-		return fmt.Errorf("closing inotify file descriptor: %v", err)
+		return fmt.Errorf("closing inotify fd: %v", err)
 	}
 	return nil
 }
@@ -65,3 +111,18 @@ func (i *Inotify) String() string {
 		return fmt.Sprintf("Watching %d directories", len(i.Watchers))
 	}
 }
+
+func GetMaxWatchers() (int, error) {
+	b, err := ioutil.ReadFile(maximumWatchersFile)
+	if err != nil {
+		return 0, fmt.Errorf("reading from %s: %v", maximumWatchersFile, err)
+	}
+
+	s := strings.TrimSpace(string(b))
+	m, err := strconv.Atoi(s)
+	if err != nil {
+		return 0, fmt.Errorf("converting to integer: %v", err)
+	}
+
+	return m, nil
+}

internal/fswatcher/inotify/inotify_test.go

@@ -1,49 +1,46 @@
 package inotify
 
 import (
-	"errors"
+	"io/ioutil"
+	"os"
 	"testing"
+
+	"golang.org/x/sys/unix"
 )
 
-func TestNewInotify(t *testing.T) {
+func TestInotify(t *testing.T) {
-	mis := &MockInotifySyscalls{fd: 1}
+	i, err := NewInotify()
+	expectNoError(t, err)
 
-	i, err := NewInotify(mis)
+	err = i.Watch("testdata/folder")
+	expectNoError(t, err)
+
+	err = ioutil.WriteFile("testdata/folder/f1", []byte("file content"), 0644)
+	expectNoError(t, err)
+	defer os.Remove("testdata/folder/f1")
+
+	buf := make([]byte, 5*unix.SizeofInotifyEvent)
+	_, err = i.Read(buf)
+	expectNoError(t, err)
+
+	e, offset, err := i.ParseNextEvent(buf[0:])
+	expectNoError(t, err)
+	if e.Name != "testdata/folder/f1" {
+		t.Fatalf("Wrong event name: %s", e.Name)
+	}
+	if e.Op != "CREATE" {
+		t.Fatalf("Wrong op: %s", e.Op)
+	}
+	if offset != 32 {
+		t.Fatalf("Wrong offset: %d", offset)
+	}
+
+	err = i.Close()
+	expectNoError(t, err)
+}
+
+func expectNoError(t *testing.T, err error) {
 	if err != nil {
-		t.Fatalf("Unexpected error")
+		t.Fatalf("Unexpected error: %v", err)
-	}
-	if i.FD != mis.fd {
-		t.Fatalf("Did not set FD of inotify object")
 	}
 }
-
-func TestNewInotifyError(t *testing.T) {
-	mis := &MockInotifySyscalls{fd: -1}
-
-	_, err := NewInotify(mis)
-	if err == nil || err.Error() != "initializing inotify: syscall error" {
-		t.Fatalf("Expected syscall error but did not get: %v", err)
-	}
-}
-
-// mock
-
-type MockInotifySyscalls struct {
-	fd int
-}
-
-func (mis *MockInotifySyscalls) Init() (int, error) {
-	if mis.fd >= 0 {
-		return mis.fd, nil
-	} else {
-		return -1, errors.New("syscall error")
-	}
-}
-
-func (mis *MockInotifySyscalls) AddWatch(fd int, dir string) (int, error) {
-	return 2, nil
-}
-
-func (mis *MockInotifySyscalls) Close(fd int) error {
-	return nil
-}

internal/fswatcher/inotify/sys/syscalls.go

@@ -1,36 +0,0 @@
-// +build linux
-
-package sys
-
-import (
-	"fmt"
-
-	"golang.org/x/sys/unix"
-)
-
-const events = unix.IN_ALL_EVENTS
-
-type InotifySyscallsUNIX struct{}
-
-func (isu *InotifySyscallsUNIX) Init() (int, error) {
-	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
-	if fd < 0 {
-		return fd, fmt.Errorf("errno: %d", errno)
-	}
-	return fd, nil
-}
-
-func (isu *InotifySyscallsUNIX) AddWatch(fd int, dir string) (int, error) {
-	wd, errno := unix.InotifyAddWatch(fd, dir, events)
-	if wd < 0 {
-		return wd, fmt.Errorf("errno: %d", errno)
-	}
-	return wd, nil
-}
-
-func (isu *InotifySyscallsUNIX) Close(fd int) error {
-	if err := unix.Close(fd); err != nil {
-		return err
-	}
-	return nil
-}

internal/fswatcher/inotify/sys/syscalls_test.go

@@ -1,45 +0,0 @@
-// +build linux
-
-package sys
-
-import (
-	"testing"
-)
-
-func TestSyscalls(t *testing.T) {
-	is := &InotifySyscallsUNIX{}
-
-	fd, err := is.Init()
-	if err != nil {
-		t.Fatalf("Unexpected error for inotify init: %v", err)
-	}
-
-	_, err = is.AddWatch(fd, "testdata")
-	if err != nil {
-		t.Fatalf("Unexpected error adding watch to dir 'testdata': %v", err)
-	}
-
-	err = is.Close(fd)
-	if err != nil {
-		t.Fatalf("Unexpected error closing inotify: %v", err)
-	}
-}
-
-func TestSyscallsError(t *testing.T) {
-	is := &InotifySyscallsUNIX{}
-
-	fd, err := is.Init()
-	if err != nil {
-		t.Fatalf("Unexpected error for inotify init: %v", err)
-	}
-
-	_, err = is.AddWatch(fd, "non-existing-dir")
-	if err == nil || err.Error() != "errno: 2" {
-		t.Fatalf("Expected errno 2 for non-existing-dir but got: %v", err)
-	}
-
-	err = is.Close(fd)
-	if err != nil {
-		t.Fatalf("Unexpected error closing inotify: %v", err)
-	}
-}

internal/fswatcher/inotify/watcher.go

@@ -1,30 +0,0 @@
-package inotify
-
-import (
-	"fmt"
-	"io/ioutil"
-	"strconv"
-	"strings"
-)
-
-const maximumWatchersFile = "/proc/sys/fs/inotify/max_user_watches"
-
-type Watcher struct {
-	WD  int
-	Dir string
-}
-
-func GetLimit() (int, error) {
-	b, err := ioutil.ReadFile(maximumWatchersFile)
-	if err != nil {
-		return 0, fmt.Errorf("reading from %s: %v", maximumWatchersFile, err)
-	}
-
-	s := strings.TrimSpace(string(b))
-	m, err := strconv.Atoi(s)
-	if err != nil {
-		return 0, fmt.Errorf("converting to integer: %v", err)
-	}
-
-	return m, nil
-}

internal/fswatcher/observer.go

@@ -1,8 +1,6 @@
 package fswatcher
 
 import (
-	"fmt"
-
 	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
 	"golang.org/x/sys/unix"
 )
@@ -11,10 +9,9 @@ func Observe(i *inotify.Inotify, triggerCh chan struct{}, dataCh chan []byte, er
 	buf := make([]byte, 5*unix.SizeofInotifyEvent)
 
 	for {
-		n, errno := unix.Read(i.FD, buf)
+		n, err := i.Read(buf)
-		if n == -1 {
+		if err != nil {
-			errCh <- fmt.Errorf("reading from inotify fd: errno: %d", errno)
+			errCh <- err
-			return
 		}
 		triggerCh <- struct{}{}
 		bufCopy := make([]byte, n)

internal/fswatcher/setup.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 
 	"github.com/dominicbreuker/pspy/internal/fswatcher/inotify"
-	isys "github.com/dominicbreuker/pspy/internal/fswatcher/inotify/sys"
 	"github.com/dominicbreuker/pspy/internal/fswatcher/walker"
 )
 
@@ -17,7 +16,7 @@ func (iw *InotifyWatcher) Close() {
 }
 
 func NewInotifyWatcher() (*InotifyWatcher, error) {
-	i, err := inotify.NewInotify(&isys.InotifySyscallsUNIX{})
+	i, err := inotify.NewInotify()
 	if err != nil {
 		return nil, fmt.Errorf("setting up inotify: %v", err)
 	}
@@ -27,7 +26,7 @@ func NewInotifyWatcher() (*InotifyWatcher, error) {
 }
 
 func (iw *InotifyWatcher) Setup(rdirs, dirs []string, errCh chan error) (chan struct{}, chan string, error) {
-	maxWatchers, err := getLimit()
+	maxWatchers, err := inotify.GetMaxWatchers()
 	if err != nil {
 		errCh <- fmt.Errorf("Can't get inotify watcher limit...: %v\n", err)
 		maxWatchers = -1