src/syscall/zx/dispatch/dispatcher.go - third_party/go - Git at Google

 // Copyright 2018 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 // +build fuchsia

 package dispatch

 import (
 	"strconv"
 	"sync"
 	"syscall/zx"
 )

 // WaitResult represents a wait result that is returned by the callback.
 // WaitResult determines whether the wait should be re-queued.
 type WaitResult int

 const (
 	WaitFinished WaitResult = iota
 	WaitAgain
 )

 // WaitID is a monotonically increasing ID which corresponds to a particular
 // call to BeginWait. Note that it may become invalidated if a wait is dequeued.
 type WaitID uint64

 const (
 	shutdownKey uint64 = 1
 )

 // Handler is the callback that will be called when the wait is complete.
 type Handler func(error, *zx.PacketSignal) WaitResult

 // waitContext is a bookkeeping structure for in-progress waits.
 type waitContext struct {
 	object   zx.Handle
 	callback Handler
 	trigger  zx.Signals
 }

 // Dispatcher can read messages from a handle and assign them to a callback.
 type Dispatcher struct {
 	// port is the underlying port used to wait on signals and dispatch.
 	port zx.Port

 	mu struct {
 		sync.RWMutex
 		// objects is a map that manages wait contexts.
 		objects    map[WaitID]*waitContext
 		nextWaitID WaitID
 		// serveLoops is the count of calls to dispatcher.Serve outstanding.
 		serveLoops int
 		// closed is whether Close has already been called.
 		closed bool
 	}
 }

 // NewDispatcher creates a new dispatcher.
 func NewDispatcher() (*Dispatcher, error) {
 	port, err := zx.NewPort(0)
 	if err != nil {
 		return nil, err
 	}
 	d := Dispatcher{
 		port: port,
 	}
 	d.mu.objects = make(map[WaitID]*waitContext)
 	return &d, nil
 }

 func assertWaitResult(result WaitResult, err error) {
 	if !(result == WaitFinished || (result == WaitAgain && err == nil)) {
 		panic("expected " + strconv.Itoa(int(result)) + " for status " + err.Error())
 	}
 }

 // Close closes a dispatcher, shutting down all handles added to the dispatcher
 // and all Serve loops.
 func (d *Dispatcher) Close() {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.mu.closed {
 		return
 	}

 	for i := 0; i < d.mu.serveLoops; i++ {
 		if err := d.ShutdownOne(); err != nil {
 			if err, ok := err.(*zx.Error); ok {
 				switch err.Status {
 				case zx.ErrShouldWait:
 					// //docs/reference/syscalls/port_queue.md@e2c644: The port has too
 					// many pending packets. This may be because Serve loops are stuck in
 					// a dispatch callback and can't dequeue packets. Such a loop can't
 					// be shut down otherwise - panic so that these Serve loops can be
 					// killed and the system can restart the offending process.
 					panic("dispatcher.Close(): zx_port_queue: too many pending packets to send shutdown packet")
 				case zx.ErrBadHandle:
 					// Port was closed out from under us; this shouldn't be possible
 					// since we checked shutdown above.
 					panic("dispatcher.Close(): attempted to zx_port_queue on closed port")
 				default:
 					panic("dispatcher.Close(): zx_port_queue unexpected error: " + err.Error())
 				}
 			}
 		}
 	}

 	d.mu.objects = make(map[WaitID]*waitContext)
 	d.mu.closed = true
 }

 // BeginWait creates a new wait on the handle h for signals t. The new wait
 // will call the handler c on wait completion.
 func (d *Dispatcher) BeginWait(h zx.Handle, t zx.Signals, c Handler) (WaitID, error) {
 	d.mu.Lock()
 	defer d.mu.Unlock()

 	if d.mu.closed {
 		return 0, &zx.Error{Status: zx.ErrBadState}
 	}

 	// Find the next unused ID.
 	for {
 		if _, ok := d.mu.objects[d.mu.nextWaitID]; !ok {
 			break
 		}
 		d.mu.nextWaitID++
 	}
 	id := d.mu.nextWaitID

 	if err := d.port.WaitAsync(h, uint64(id), t, zx.PortWaitAsyncOnce); err != nil {
 		return id, err
 	}

 	d.mu.objects[id] = &waitContext{
 		object:   h,
 		callback: c,
 		trigger:  t,
 	}

 	return id, nil
 }

 // CancelWait cancels the wait with the given WaitID.
 func (d *Dispatcher) CancelWait(id WaitID) error {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if wc, ok := d.mu.objects[id]; ok {
 		delete(d.mu.objects, id)
 		// Note this must be done under lock to avoid another wait occupying the
 		// same slot and being cancelled accidentally.
 		return d.port.Cancel(wc.object, uint64(id))
 	}
 	return &zx.Error{Status: zx.ErrNotFound}
 }

 func (d *Dispatcher) dispatch(id WaitID, wc *waitContext, signals *zx.PacketSignal) {
 	// Call the handler.
 	result := wc.callback(nil, signals)
 	assertWaitResult(result, nil)
 	if result == WaitAgain {
 		// If we fail to re-arm, notify the handler of what happened.
 		if err := d.port.WaitAsync(wc.object, uint64(id), wc.trigger, zx.PortWaitAsyncOnce); err != nil {
 			assertWaitResult(wc.callback(err, nil), err)
 		}
 	}
 }

 // ShutdownOne signals that one Serve loop should exit.
 //
 // error reflects failure to queue the shutdown signal.
 func (d *Dispatcher) ShutdownOne() error {
 	return d.port.Queue(&zx.Packet{
 		Hdr: zx.PacketHeader{
 			Key:    shutdownKey,
 			Type:   zx.PortPacketTypeUser,
 			Status: zx.ErrOk,
 		},
 	})
 }

 // Serve runs until the dispatcher is closed, waiting for the port to return a
 // packet and dispatches the relevant handlers.
 //
 // Serve exits when the Dispatcher is closed.
 func (d *Dispatcher) Serve() {
 	d.mu.Lock()
 	closed := d.mu.closed
 	if !closed {
 		d.mu.serveLoops++
 	}
 	d.mu.Unlock()
 	if closed {
 		return
 	}
 	defer func() {
 		d.mu.Lock()
 		d.mu.serveLoops--
 		if d.mu.serveLoops == 0 {
 			d.port.Close()
 		}
 		d.mu.Unlock()
 	}()
 	for {
 		var packet zx.Packet
 		if err := d.port.Wait(&packet, zx.TimensecInfinite); err != nil {
 			// Possible errors:
 			//
 			// ZX_ERR_BAD_HANDLE: indicates the port has been closed. Should never
 			// happen because the port is not closed until all loops have been shut
 			// down.
 			//
 			// ZX_ERR_INVALID_ARGS: indicates packet is not a valid pointer.
 			//
 			// ZX_ERR_ACCESS_DENIED: indicates missing rights.
 			//
 			// ZX_ERR_TIMED_OUT: indicates a timeout. Should never happen because we
 			// use an infinite timeout.
 			//
 			// These errors all indicate bugs in the implementation. Panic.
 			panic("dispatcher.Serve(): zx_port_wait unexpected error: " + err.Error())
 		}

 		switch typ := packet.Hdr.Type; typ {
 		case zx.PortPacketTypeUser:
 			switch key := packet.Hdr.Key; key {
 			case shutdownKey:
 				return
 			default:
 				panic("unhandled user packet key=" + strconv.FormatUint(key, 10))
 			}
 		case zx.PortPacketTypeSignalOne:
 			id := WaitID(packet.Hdr.Key)
 			d.mu.RLock()
 			wc, ok := d.mu.objects[id]
 			d.mu.RUnlock()
 			if ok {
 				d.dispatch(id, wc, packet.Signal())
 			}
 		default:
 			panic("unhandled packet type=" + strconv.FormatUint(uint64(typ), 10))
 		}

 		d.mu.RLock()
 		closed := d.mu.closed
 		d.mu.RUnlock()
 		if closed {
 			return
 		}
 	}
 }
	// Copyright 2018 The Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	// +build fuchsia

	package dispatch

	import (
	"strconv"
	"sync"
	"syscall/zx"
	)

	// WaitResult represents a wait result that is returned by the callback.
	// WaitResult determines whether the wait should be re-queued.
	type WaitResult int

	const (
	WaitFinished WaitResult = iota
	WaitAgain
	)

	// WaitID is a monotonically increasing ID which corresponds to a particular
	// call to BeginWait. Note that it may become invalidated if a wait is dequeued.
	type WaitID uint64

	const (
	shutdownKey uint64 = 1
	)

	// Handler is the callback that will be called when the wait is complete.
	type Handler func(error, *zx.PacketSignal) WaitResult

	// waitContext is a bookkeeping structure for in-progress waits.
	type waitContext struct {
	object zx.Handle
	callback Handler
	trigger zx.Signals
	}

	// Dispatcher can read messages from a handle and assign them to a callback.
	type Dispatcher struct {
	// port is the underlying port used to wait on signals and dispatch.
	port zx.Port

	mu struct {
	sync.RWMutex
	// objects is a map that manages wait contexts.
	objects map[WaitID]*waitContext
	nextWaitID WaitID
	// serveLoops is the count of calls to dispatcher.Serve outstanding.
	serveLoops int
	// closed is whether Close has already been called.
	closed bool
	}
	}

	// NewDispatcher creates a new dispatcher.
	func NewDispatcher() (*Dispatcher, error) {
	port, err := zx.NewPort(0)
	if err != nil {
	return nil, err
	}
	d := Dispatcher{
	port: port,
	}
	d.mu.objects = make(map[WaitID]*waitContext)
	return &d, nil
	}

	func assertWaitResult(result WaitResult, err error) {
	if !(result == WaitFinished \|\| (result == WaitAgain && err == nil)) {
	panic("expected " + strconv.Itoa(int(result)) + " for status " + err.Error())
	}
	}

	// Close closes a dispatcher, shutting down all handles added to the dispatcher
	// and all Serve loops.
	func (d *Dispatcher) Close() {
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.mu.closed {
	return
	}

	for i := 0; i < d.mu.serveLoops; i++ {
	if err := d.ShutdownOne(); err != nil {
	if err, ok := err.(*zx.Error); ok {
	switch err.Status {
	case zx.ErrShouldWait:
	// //docs/reference/syscalls/port_queue.md@e2c644: The port has too
	// many pending packets. This may be because Serve loops are stuck in
	// a dispatch callback and can't dequeue packets. Such a loop can't
	// be shut down otherwise - panic so that these Serve loops can be
	// killed and the system can restart the offending process.
	panic("dispatcher.Close(): zx_port_queue: too many pending packets to send shutdown packet")
	case zx.ErrBadHandle:
	// Port was closed out from under us; this shouldn't be possible
	// since we checked shutdown above.
	panic("dispatcher.Close(): attempted to zx_port_queue on closed port")
	default:
	panic("dispatcher.Close(): zx_port_queue unexpected error: " + err.Error())
	}
	}
	}
	}

	d.mu.objects = make(map[WaitID]*waitContext)
	d.mu.closed = true
	}

	// BeginWait creates a new wait on the handle h for signals t. The new wait
	// will call the handler c on wait completion.
	func (d *Dispatcher) BeginWait(h zx.Handle, t zx.Signals, c Handler) (WaitID, error) {
	d.mu.Lock()
	defer d.mu.Unlock()

	if d.mu.closed {
	return 0, &zx.Error{Status: zx.ErrBadState}
	}

	// Find the next unused ID.
	for {
	if _, ok := d.mu.objects[d.mu.nextWaitID]; !ok {
	break
	}
	d.mu.nextWaitID++
	}
	id := d.mu.nextWaitID

	if err := d.port.WaitAsync(h, uint64(id), t, zx.PortWaitAsyncOnce); err != nil {
	return id, err
	}

	d.mu.objects[id] = &waitContext{
	object: h,
	callback: c,
	trigger: t,
	}

	return id, nil
	}

	// CancelWait cancels the wait with the given WaitID.
	func (d *Dispatcher) CancelWait(id WaitID) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	if wc, ok := d.mu.objects[id]; ok {
	delete(d.mu.objects, id)
	// Note this must be done under lock to avoid another wait occupying the
	// same slot and being cancelled accidentally.
	return d.port.Cancel(wc.object, uint64(id))
	}
	return &zx.Error{Status: zx.ErrNotFound}
	}

	func (d Dispatcher) dispatch(id WaitID, wc waitContext, signals *zx.PacketSignal) {
	// Call the handler.
	result := wc.callback(nil, signals)
	assertWaitResult(result, nil)
	if result == WaitAgain {
	// If we fail to re-arm, notify the handler of what happened.
	if err := d.port.WaitAsync(wc.object, uint64(id), wc.trigger, zx.PortWaitAsyncOnce); err != nil {
	assertWaitResult(wc.callback(err, nil), err)
	}
	}
	}

	// ShutdownOne signals that one Serve loop should exit.
	//
	// error reflects failure to queue the shutdown signal.
	func (d *Dispatcher) ShutdownOne() error {
	return d.port.Queue(&zx.Packet{
	Hdr: zx.PacketHeader{
	Key: shutdownKey,
	Type: zx.PortPacketTypeUser,
	Status: zx.ErrOk,
	},
	})
	}

	// Serve runs until the dispatcher is closed, waiting for the port to return a
	// packet and dispatches the relevant handlers.
	//
	// Serve exits when the Dispatcher is closed.
	func (d *Dispatcher) Serve() {
	d.mu.Lock()
	closed := d.mu.closed
	if !closed {
	d.mu.serveLoops++
	}
	d.mu.Unlock()
	if closed {
	return
	}
	defer func() {
	d.mu.Lock()
	d.mu.serveLoops--
	if d.mu.serveLoops == 0 {
	d.port.Close()
	}
	d.mu.Unlock()
	}()
	for {
	var packet zx.Packet
	if err := d.port.Wait(&packet, zx.TimensecInfinite); err != nil {
	// Possible errors:
	//
	// ZX_ERR_BAD_HANDLE: indicates the port has been closed. Should never
	// happen because the port is not closed until all loops have been shut
	// down.
	//
	// ZX_ERR_INVALID_ARGS: indicates packet is not a valid pointer.
	//
	// ZX_ERR_ACCESS_DENIED: indicates missing rights.
	//
	// ZX_ERR_TIMED_OUT: indicates a timeout. Should never happen because we
	// use an infinite timeout.
	//
	// These errors all indicate bugs in the implementation. Panic.
	panic("dispatcher.Serve(): zx_port_wait unexpected error: " + err.Error())
	}

	switch typ := packet.Hdr.Type; typ {
	case zx.PortPacketTypeUser:
	switch key := packet.Hdr.Key; key {
	case shutdownKey:
	return
	default:
	panic("unhandled user packet key=" + strconv.FormatUint(key, 10))
	}
	case zx.PortPacketTypeSignalOne:
	id := WaitID(packet.Hdr.Key)
	d.mu.RLock()
	wc, ok := d.mu.objects[id]
	d.mu.RUnlock()
	if ok {
	d.dispatch(id, wc, packet.Signal())
	}
	default:
	panic("unhandled packet type=" + strconv.FormatUint(uint64(typ), 10))
	}

	d.mu.RLock()
	closed := d.mu.closed
	d.mu.RUnlock()
	if closed {
	return
	}
	}
	}