unix: create wrappers for solaris/illumos Event Ports

This work is in support of a cleanup of fsnotify/fsnotify#263

Change-Id: Ibd7500d20322765bfd50aa18333eb43ee7b659d7
Reviewed-on: https://go-review.googlesource.com/c/sys/+/324630
Run-TryBot: Ian Lance Taylor <iant@golang.org>
TryBot-Result: Go Bot <gobot@golang.org>
Reviewed-by: Ian Lance Taylor <iant@golang.org>
Reviewed-by: Tobias Klauser <tobias.klauser@gmail.com>

unix/syscall_solaris.go

@@ -13,7 +13,10 @@
 package unix
 
 import (
+	"fmt"
+	"os"
 	"runtime"
+	"sync"
 	"syscall"
 	"unsafe"
 )
@@ -744,3 +747,240 @@ func Mmap(fd int, offset int64, length int, prot int, flags int) (data []byte, e
 func Munmap(b []byte) (err error) {
 	return mapper.Munmap(b)
 }
+
+// Event Ports
+
+type fileObjCookie struct {
+	fobj   *fileObj
+	cookie interface{}
+}
+
+// EventPort provides a safe abstraction on top of Solaris/illumos Event Ports.
+type EventPort struct {
+	port  int
+	mu    sync.Mutex
+	fds   map[uintptr]interface{}
+	paths map[string]*fileObjCookie
+}
+
+// PortEvent is an abstraction of the port_event C struct.
+// Compare Source against PORT_SOURCE_FILE or PORT_SOURCE_FD
+// to see if Path or Fd was the event source. The other will be
+// uninitialized.
+type PortEvent struct {
+	Cookie interface{}
+	Events int32
+	Fd     uintptr
+	Path   string
+	Source uint16
+	fobj   *fileObj
+}
+
+// NewEventPort creates a new EventPort including the
+// underlying call to port_create(3c).
+func NewEventPort() (*EventPort, error) {
+	port, err := port_create()
+	if err != nil {
+		return nil, err
+	}
+	e := &EventPort{
+		port:  port,
+		fds:   make(map[uintptr]interface{}),
+		paths: make(map[string]*fileObjCookie),
+	}
+	return e, nil
+}
+
+//sys	port_create() (n int, err error)
+//sys	port_associate(port int, source int, object uintptr, events int, user *byte) (n int, err error)
+//sys	port_dissociate(port int, source int, object uintptr) (n int, err error)
+//sys	port_get(port int, pe *portEvent, timeout *Timespec) (n int, err error)
+//sys	port_getn(port int, pe *portEvent, max uint32, nget *uint32, timeout *Timespec) (n int, err error)
+
+// Close closes the event port.
+func (e *EventPort) Close() error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.fds = nil
+	e.paths = nil
+	return Close(e.port)
+}
+
+// PathIsWatched checks to see if path is associated with this EventPort.
+func (e *EventPort) PathIsWatched(path string) bool {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	_, found := e.paths[path]
+	return found
+}
+
+// FdIsWatched checks to see if fd is associated with this EventPort.
+func (e *EventPort) FdIsWatched(fd uintptr) bool {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	_, found := e.fds[fd]
+	return found
+}
+
+// AssociatePath wraps port_associate(3c) for a filesystem path including
+// creating the necessary file_obj from the provided stat information.
+func (e *EventPort) AssociatePath(path string, stat os.FileInfo, events int, cookie interface{}) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if _, found := e.paths[path]; found {
+		return fmt.Errorf("%v is already associated with this Event Port", path)
+	}
+	fobj, err := createFileObj(path, stat)
+	if err != nil {
+		return err
+	}
+	fCookie := &fileObjCookie{fobj, cookie}
+	_, err = port_associate(e.port, PORT_SOURCE_FILE, uintptr(unsafe.Pointer(fobj)), events, (*byte)(unsafe.Pointer(&fCookie.cookie)))
+	if err != nil {
+		return err
+	}
+	e.paths[path] = fCookie
+	return nil
+}
+
+// DissociatePath wraps port_dissociate(3c) for a filesystem path.
+func (e *EventPort) DissociatePath(path string) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	f, ok := e.paths[path]
+	if !ok {
+		return fmt.Errorf("%v is not associated with this Event Port", path)
+	}
+	_, err := port_dissociate(e.port, PORT_SOURCE_FILE, uintptr(unsafe.Pointer(f.fobj)))
+	if err != nil {
+		return err
+	}
+	delete(e.paths, path)
+	return nil
+}
+
+// AssociateFd wraps calls to port_associate(3c) on file descriptors.
+func (e *EventPort) AssociateFd(fd uintptr, events int, cookie interface{}) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if _, found := e.fds[fd]; found {
+		return fmt.Errorf("%v is already associated with this Event Port", fd)
+	}
+	pcookie := &cookie
+	_, err := port_associate(e.port, PORT_SOURCE_FD, fd, events, (*byte)(unsafe.Pointer(pcookie)))
+	if err != nil {
+		return err
+	}
+	e.fds[fd] = pcookie
+	return nil
+}
+
+// DissociateFd wraps calls to port_dissociate(3c) on file descriptors.
+func (e *EventPort) DissociateFd(fd uintptr) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	_, ok := e.fds[fd]
+	if !ok {
+		return fmt.Errorf("%v is not associated with this Event Port", fd)
+	}
+	_, err := port_dissociate(e.port, PORT_SOURCE_FD, fd)
+	if err != nil {
+		return err
+	}
+	delete(e.fds, fd)
+	return nil
+}
+
+func createFileObj(name string, stat os.FileInfo) (*fileObj, error) {
+	fobj := new(fileObj)
+	bs, err := ByteSliceFromString(name)
+	if err != nil {
+		return nil, err
+	}
+	fobj.Name = (*int8)(unsafe.Pointer(&bs[0]))
+	s := stat.Sys().(*syscall.Stat_t)
+	fobj.Atim.Sec = s.Atim.Sec
+	fobj.Atim.Nsec = s.Atim.Nsec
+	fobj.Mtim.Sec = s.Mtim.Sec
+	fobj.Mtim.Nsec = s.Mtim.Nsec
+	fobj.Ctim.Sec = s.Ctim.Sec
+	fobj.Ctim.Nsec = s.Ctim.Nsec
+	return fobj, nil
+}
+
+// GetOne wraps port_get(3c) and returns a single PortEvent.
+func (e *EventPort) GetOne(t *Timespec) (*PortEvent, error) {
+	pe := new(portEvent)
+	_, err := port_get(e.port, pe, t)
+	if err != nil {
+		return nil, err
+	}
+	p := new(PortEvent)
+	p.Events = pe.Events
+	p.Source = pe.Source
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	switch pe.Source {
+	case PORT_SOURCE_FD:
+		p.Fd = uintptr(pe.Object)
+		cookie := (*interface{})(unsafe.Pointer(pe.User))
+		p.Cookie = *cookie
+		delete(e.fds, p.Fd)
+	case PORT_SOURCE_FILE:
+		p.fobj = (*fileObj)(unsafe.Pointer(uintptr(pe.Object)))
+		p.Path = BytePtrToString((*byte)(unsafe.Pointer(p.fobj.Name)))
+		cookie := (*interface{})(unsafe.Pointer(pe.User))
+		p.Cookie = *cookie
+		delete(e.paths, p.Path)
+	}
+	return p, nil
+}
+
+// Pending wraps port_getn(3c) and returns how many events are pending.
+func (e *EventPort) Pending() (int, error) {
+	var n uint32 = 0
+	_, err := port_getn(e.port, nil, 0, &n, nil)
+	return int(n), err
+}
+
+// Get wraps port_getn(3c) and fills a slice of PortEvent.
+// It will block until either min events have been received
+// or the timeout has been exceeded. It will return how many
+// events were actually received along with any error information.
+func (e *EventPort) Get(s []PortEvent, min int, timeout *Timespec) (int, error) {
+	if min == 0 {
+		return 0, fmt.Errorf("need to request at least one event or use Pending() instead")
+	}
+	if len(s) < min {
+		return 0, fmt.Errorf("len(s) (%d) is less than min events requested (%d)", len(s), min)
+	}
+	got := uint32(min)
+	max := uint32(len(s))
+	var err error
+	ps := make([]portEvent, max, max)
+	_, err = port_getn(e.port, &ps[0], max, &got, timeout)
+	// got will be trustworthy with ETIME, but not any other error.
+	if err != nil && err != ETIME {
+		return 0, err
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	for i := 0; i < int(got); i++ {
+		s[i].Events = ps[i].Events
+		s[i].Source = ps[i].Source
+		switch ps[i].Source {
+		case PORT_SOURCE_FD:
+			s[i].Fd = uintptr(ps[i].Object)
+			cookie := (*interface{})(unsafe.Pointer(ps[i].User))
+			s[i].Cookie = *cookie
+			delete(e.fds, s[i].Fd)
+		case PORT_SOURCE_FILE:
+			s[i].fobj = (*fileObj)(unsafe.Pointer(uintptr(ps[i].Object)))
+			s[i].Path = BytePtrToString((*byte)(unsafe.Pointer(s[i].fobj.Name)))
+			cookie := (*interface{})(unsafe.Pointer(ps[i].User))
+			s[i].Cookie = *cookie
+			delete(e.paths, s[i].Path)
+		}
+	}
+	return int(got), err
+}

unix/syscall_solaris_test.go

@@ -8,7 +8,11 @@
 package unix_test
 
 import (
+	"fmt"
+	"io/ioutil"
+	"os"
 	"os/exec"
+	"runtime"
 	"testing"
 
 	"golang.org/x/sys/unix"
@@ -41,3 +45,290 @@ func TestSysconf(t *testing.T) {
 	}
 	t.Logf("Sysconf(SC_CLK_TCK) = %d", n)
 }
+
+// Event Ports
+
+func TestBasicEventPort(t *testing.T) {
+	tmpfile, err := ioutil.TempFile("", "eventport")
+	if err != nil {
+		t.Fatalf("unable to create a tempfile: %v", err)
+	}
+	path := tmpfile.Name()
+	defer os.Remove(path)
+
+	stat, err := os.Stat(path)
+	if err != nil {
+		t.Fatalf("Failed to stat %s: %v", path, err)
+	}
+	port, err := unix.NewEventPort()
+	if err != nil {
+		t.Fatalf("NewEventPort failed: %v", err)
+	}
+	defer port.Close()
+	cookie := stat.Mode()
+	err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, cookie)
+	if err != nil {
+		t.Errorf("AssociatePath failed: %v", err)
+	}
+	if !port.PathIsWatched(path) {
+		t.Errorf("PathIsWatched unexpectedly returned false")
+	}
+	err = port.DissociatePath(path)
+	if err != nil {
+		t.Errorf("DissociatePath failed: %v", err)
+	}
+	err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, cookie)
+	if err != nil {
+		t.Errorf("AssociatePath failed: %v", err)
+	}
+	bs := []byte{42}
+	tmpfile.Write(bs)
+	timeout := new(unix.Timespec)
+	timeout.Sec = 1
+	pevent, err := port.GetOne(timeout)
+	if err == unix.ETIME {
+		t.Errorf("GetOne timed out: %v", err)
+	}
+	if err != nil {
+		t.Errorf("GetOne failed: %v", err)
+	}
+	if pevent.Path != path {
+		t.Errorf("Path mismatch: %v != %v", pevent.Path, path)
+	}
+	err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, cookie)
+	if err != nil {
+		t.Errorf("AssociatePath failed: %v", err)
+	}
+	err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, cookie)
+	if err == nil {
+		t.Errorf("Unexpected success associating already associated path")
+	}
+}
+
+func TestEventPortFds(t *testing.T) {
+	_, path, _, _ := runtime.Caller(0)
+	stat, err := os.Stat(path)
+	cookie := stat.Mode()
+	port, err := unix.NewEventPort()
+	if err != nil {
+		t.Errorf("NewEventPort failed: %v", err)
+	}
+	defer port.Close()
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Errorf("unable to create a pipe: %v", err)
+	}
+	defer w.Close()
+	defer r.Close()
+	fd := r.Fd()
+
+	port.AssociateFd(fd, unix.POLLIN, cookie)
+	if !port.FdIsWatched(fd) {
+		t.Errorf("FdIsWatched unexpectedly returned false")
+	}
+	err = port.DissociateFd(fd)
+	err = port.AssociateFd(fd, unix.POLLIN, cookie)
+	bs := []byte{42}
+	w.Write(bs)
+	n, err := port.Pending()
+	if n != 1 {
+		t.Errorf("Pending() failed: %v, %v", n, err)
+	}
+	timeout := new(unix.Timespec)
+	timeout.Sec = 1
+	pevent, err := port.GetOne(timeout)
+	if err == unix.ETIME {
+		t.Errorf("GetOne timed out: %v", err)
+	}
+	if err != nil {
+		t.Errorf("GetOne failed: %v", err)
+	}
+	if pevent.Fd != fd {
+		t.Errorf("Fd mismatch: %v != %v", pevent.Fd, fd)
+	}
+	var c = pevent.Cookie
+	if c == nil {
+		t.Errorf("Cookie missing: %v != %v", cookie, c)
+		return
+	}
+	if c != cookie {
+		t.Errorf("Cookie mismatch: %v != %v", cookie, c)
+	}
+	port.AssociateFd(fd, unix.POLLIN, cookie)
+	err = port.AssociateFd(fd, unix.POLLIN, cookie)
+	if err == nil {
+		t.Errorf("unexpected success associating already associated fd")
+	}
+}
+
+func TestEventPortErrors(t *testing.T) {
+	tmpfile, err := ioutil.TempFile("", "eventport")
+	if err != nil {
+		t.Errorf("unable to create a tempfile: %v", err)
+	}
+	path := tmpfile.Name()
+	stat, _ := os.Stat(path)
+	os.Remove(path)
+	port, _ := unix.NewEventPort()
+	defer port.Close()
+	err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, nil)
+	if err == nil {
+		t.Errorf("unexpected success associating nonexistant file")
+	}
+	err = port.DissociatePath(path)
+	if err == nil {
+		t.Errorf("unexpected success dissociating unassociated path")
+	}
+	timeout := new(unix.Timespec)
+	timeout.Nsec = 1
+	_, err = port.GetOne(timeout)
+	if err != unix.ETIME {
+		t.Errorf("unexpected lack of timeout")
+	}
+	err = port.DissociateFd(uintptr(0))
+	if err == nil {
+		t.Errorf("unexpected success dissociating unassociated fd")
+	}
+	events := make([]unix.PortEvent, 4, 4)
+	_, err = port.Get(events, 5, nil)
+	if err == nil {
+		t.Errorf("unexpected success calling Get with min greater than len of slice")
+	}
+	_, err = port.Get(nil, 1, nil)
+	if err == nil {
+		t.Errorf("unexpected success calling Get with nil slice")
+	}
+	_, err = port.Get(nil, 0, nil)
+	if err == nil {
+		t.Errorf("unexpected success calling Get with nil slice")
+	}
+}
+
+func ExamplePortEvent() {
+	type MyCookie struct {
+		Name string
+	}
+	cookie := MyCookie{"Cookie Monster"}
+	port, err := unix.NewEventPort()
+	if err != nil {
+		fmt.Printf("NewEventPort failed: %v\n", err)
+		return
+	}
+	defer port.Close()
+	r, w, err := os.Pipe()
+	if err != nil {
+		fmt.Printf("os.Pipe() failed: %v\n", err)
+		return
+	}
+	defer w.Close()
+	defer r.Close()
+	fd := r.Fd()
+
+	port.AssociateFd(fd, unix.POLLIN, cookie)
+
+	bs := []byte{42}
+	w.Write(bs)
+	timeout := new(unix.Timespec)
+	timeout.Sec = 1
+	pevent, err := port.GetOne(timeout)
+	if err != nil {
+		fmt.Printf("didn't get the expected event: %v\n", err)
+	}
+
+	// Use a type assertion to convert the received cookie back to its original type
+	c := pevent.Cookie.(MyCookie)
+	fmt.Printf("%s", c.Name)
+	//Output: Cookie Monster
+}
+
+func TestPortEventSlices(t *testing.T) {
+	port, err := unix.NewEventPort()
+	if err != nil {
+		t.Fatalf("NewEventPort failed: %v", err)
+	}
+	// Create, associate, and delete 6 files
+	for i := 0; i < 6; i++ {
+		tmpfile, err := ioutil.TempFile("", "eventport")
+		if err != nil {
+			t.Fatalf("unable to create tempfile: %v", err)
+		}
+		path := tmpfile.Name()
+		stat, err := os.Stat(path)
+		if err != nil {
+			t.Fatalf("unable to stat tempfile: %v", err)
+		}
+		err = port.AssociatePath(path, stat, unix.FILE_MODIFIED, nil)
+		if err != nil {
+			t.Fatalf("unable to AssociatePath tempfile: %v", err)
+		}
+		err = os.Remove(path)
+		if err != nil {
+			t.Fatalf("unable to Remove tempfile: %v", err)
+		}
+	}
+	n, err := port.Pending()
+	if err != nil {
+		t.Errorf("Pending failed: %v", err)
+	}
+	if n != 6 {
+		t.Errorf("expected 6 pending events, got %d", n)
+	}
+	timeout := new(unix.Timespec)
+	timeout.Nsec = 1
+	events := make([]unix.PortEvent, 4, 4)
+	n, err = port.Get(events, 3, timeout)
+	if err != nil {
+		t.Errorf("Get failed: %v", err)
+	}
+	if n != 4 {
+		t.Errorf("expected 4 events, got %d", n)
+	}
+	e := events[:n]
+	for _, p := range e {
+		if p.Events != unix.FILE_DELETE {
+			t.Errorf("unexpected event. got %v, expected %v", p.Events, unix.FILE_DELETE)
+		}
+	}
+	n, err = port.Get(events, 3, timeout)
+	if err != unix.ETIME {
+		t.Errorf("unexpected error. got %v, expected %v", err, unix.ETIME)
+	}
+	if n != 2 {
+		t.Errorf("expected 2 events, got %d", n)
+	}
+	e = events[:n]
+	for _, p := range e {
+		if p.Events != unix.FILE_DELETE {
+			t.Errorf("unexpected event. got %v, expected %v", p.Events, unix.FILE_DELETE)
+		}
+	}
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("unable to create a pipe: %v", err)
+	}
+	port.AssociateFd(r.Fd(), unix.POLLIN, nil)
+	port.AssociateFd(w.Fd(), unix.POLLOUT, nil)
+	bs := []byte{41}
+	w.Write(bs)
+
+	n, err = port.Get(events, 1, timeout)
+	if err != nil {
+		t.Errorf("Get failed: %v", err)
+	}
+	if n != 2 {
+		t.Errorf("expected 2 events, got %d", n)
+	}
+	err = w.Close()
+	if err != nil {
+		t.Errorf("w.Close() failed: %v", err)
+	}
+	err = r.Close()
+	if err != nil {
+		t.Errorf("r.Close() failed: %v", err)
+	}
+	err = port.Close()
+	if err != nil {
+		t.Errorf("port.Close() failed: %v", err)
+	}
+}

unix/zsyscall_solaris_amd64.go

@@ -141,6 +141,11 @@ import (
 //go:cgo_import_dynamic libc_getpeername getpeername "libsocket.so"
 //go:cgo_import_dynamic libc_setsockopt setsockopt "libsocket.so"
 //go:cgo_import_dynamic libc_recvfrom recvfrom "libsocket.so"
+//go:cgo_import_dynamic libc_port_create port_create "libc.so"
+//go:cgo_import_dynamic libc_port_associate port_associate "libc.so"
+//go:cgo_import_dynamic libc_port_dissociate port_dissociate "libc.so"
+//go:cgo_import_dynamic libc_port_get port_get "libc.so"
+//go:cgo_import_dynamic libc_port_getn port_getn "libc.so"
 
 //go:linkname procpipe libc_pipe
 //go:linkname procpipe2 libc_pipe2
@@ -272,6 +277,11 @@ import (
 //go:linkname procgetpeername libc_getpeername
 //go:linkname procsetsockopt libc_setsockopt
 //go:linkname procrecvfrom libc_recvfrom
+//go:linkname procport_create libc_port_create
+//go:linkname procport_associate libc_port_associate
+//go:linkname procport_dissociate libc_port_dissociate
+//go:linkname procport_get libc_port_get
+//go:linkname procport_getn libc_port_getn
 
 var (
 	procpipe,
@@ -403,7 +413,12 @@ var (
 	proc__xnet_getsockopt,
 	procgetpeername,
 	procsetsockopt,
-	procrecvfrom syscallFunc
+	procrecvfrom,
+	procport_create,
+	procport_associate,
+	procport_dissociate,
+	procport_get,
+	procport_getn syscallFunc
 )
 
 // THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
@@ -1981,3 +1996,58 @@ func recvfrom(fd int, p []byte, flags int, from *RawSockaddrAny, fromlen *_Sockl
 	}
 	return
 }
+
+// THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
+
+func port_create() (n int, err error) {
+	r0, _, e1 := sysvicall6(uintptr(unsafe.Pointer(&procport_create)), 0, 0, 0, 0, 0, 0, 0)
+	n = int(r0)
+	if e1 != 0 {
+		err = e1
+	}
+	return
+}
+
+// THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
+
+func port_associate(port int, source int, object uintptr, events int, user *byte) (n int, err error) {
+	r0, _, e1 := sysvicall6(uintptr(unsafe.Pointer(&procport_associate)), 5, uintptr(port), uintptr(source), uintptr(object), uintptr(events), uintptr(unsafe.Pointer(user)), 0)
+	n = int(r0)
+	if e1 != 0 {
+		err = e1
+	}
+	return
+}
+
+// THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
+
+func port_dissociate(port int, source int, object uintptr) (n int, err error) {
+	r0, _, e1 := sysvicall6(uintptr(unsafe.Pointer(&procport_dissociate)), 3, uintptr(port), uintptr(source), uintptr(object), 0, 0, 0)
+	n = int(r0)
+	if e1 != 0 {
+		err = e1
+	}
+	return
+}
+
+// THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
+
+func port_get(port int, pe *portEvent, timeout *Timespec) (n int, err error) {
+	r0, _, e1 := sysvicall6(uintptr(unsafe.Pointer(&procport_get)), 3, uintptr(port), uintptr(unsafe.Pointer(pe)), uintptr(unsafe.Pointer(timeout)), 0, 0, 0)
+	n = int(r0)
+	if e1 != 0 {
+		err = e1
+	}
+	return
+}
+
+// THIS FILE IS GENERATED BY THE COMMAND AT THE TOP; DO NOT EDIT
+
+func port_getn(port int, pe *portEvent, max uint32, nget *uint32, timeout *Timespec) (n int, err error) {
+	r0, _, e1 := sysvicall6(uintptr(unsafe.Pointer(&procport_getn)), 5, uintptr(port), uintptr(unsafe.Pointer(pe)), uintptr(max), uintptr(unsafe.Pointer(nget)), uintptr(unsafe.Pointer(timeout)), 0)
+	n = int(r0)
+	if e1 != 0 {
+		err = e1
+	}
+	return
+}