poll_race_bsd.go

// Copyright 2021 CloudWeGo Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//go:build (darwin || netbsd || freebsd || openbsd || dragonfly) && race
// +build darwin netbsd freebsd openbsd dragonfly
// +build race

package netpoll

import (
	"log"
	"sync"
	"sync/atomic"
	"syscall"
)

// mock no race poll
func openPoll() Poll {
	return openDefaultPoll()
}

func openDefaultPoll() *defaultPoll {
	l := new(defaultPoll)
	p, err := syscall.Kqueue()
	if err != nil {
		panic(err)
	}
	l.fd = p
	_, err = syscall.Kevent(l.fd, []syscall.Kevent_t{{
		Ident:  0,
		Filter: syscall.EVFILT_USER,
		Flags:  syscall.EV_ADD | syscall.EV_CLEAR,
	}}, nil, nil)
	if err != nil {
		panic(err)
	}
	return l
}

type defaultPoll struct {
	fd      int
	trigger uint32
	m       sync.Map
	hups    []func(p Poll) error
}

// Wait implements Poll.
func (p *defaultPoll) Wait() error {
	// init
	var size, caps = 1024, barriercap
	var events, barriers = make([]syscall.Kevent_t, size), make([]barrier, size)
	for i := range barriers {
		barriers[i].bs = make([][]byte, caps)
		barriers[i].ivs = make([]syscall.Iovec, caps)
	}
	// wait
	for {
		n, err := syscall.Kevent(p.fd, nil, events, nil)
		if err != nil && err != syscall.EINTR {
			// exit gracefully
			if err == syscall.EBADF {
				return nil
			}
			return err
		}
		for i := 0; i < n; i++ {
			var fd = int(events[i].Ident)
			// trigger
			if fd == 0 {
				// clean trigger
				atomic.StoreUint32(&p.trigger, 0)
				continue
			}
			tmp, ok := p.m.Load(fd)
			if !ok {
				continue
			}
			operator := tmp.(*FDOperator)
			if !operator.do() {
				continue
			}

			// check poll in
			if events[i].Filter == syscall.EVFILT_READ && events[i].Flags&syscall.EV_ENABLE != 0 {
				if operator.OnRead != nil {
					// for non-connection
					operator.OnRead(p)
				} else {
					// only for connection
					var bs = operator.Inputs(barriers[i].bs)
					if len(bs) > 0 {
						var n, err = readv(operator.FD, bs, barriers[i].ivs)
						operator.InputAck(n)
						if err != nil && err != syscall.EAGAIN && err != syscall.EINTR {
							log.Printf("readv(fd=%d) failed: %s", operator.FD, err.Error())
							p.appendHup(operator)
							continue
						}
					}
				}
			}

			// check hup
			if events[i].Flags&syscall.EV_EOF != 0 {
				p.appendHup(operator)
				continue
			}

			// check poll out
			if events[i].Filter == syscall.EVFILT_WRITE && events[i].Flags&syscall.EV_ENABLE != 0 {
				if operator.OnWrite != nil {
					// for non-connection
					operator.OnWrite(p)
				} else {
					// only for connection
					var bs, supportZeroCopy = operator.Outputs(barriers[i].bs)
					if len(bs) > 0 {
						// TODO: Let the upper layer pass in whether to use ZeroCopy.
						var n, err = sendmsg(operator.FD, bs, barriers[i].ivs, false && supportZeroCopy)
						operator.OutputAck(n)
						if err != nil && err != syscall.EAGAIN {
							log.Printf("sendmsg(fd=%d) failed: %s", operator.FD, err.Error())
							p.appendHup(operator)
							continue
						}
					}
				}
			}
			operator.done()
		}
		// hup conns together to avoid blocking the poll.
		p.detaches()
	}
}

// TODO: Close will bad file descriptor here
func (p *defaultPoll) Close() error {
	var err = syscall.Close(p.fd)
	// delete all *FDOperator
	p.m.Range(func(key, value interface{}) bool {
		var operator, _ = value.(*FDOperator)
		if operator.OnHup != nil {
			operator.OnHup(p)
		}
		return true
	})
	return err
}

// Trigger implements Poll.
func (p *defaultPoll) Trigger() error {
	if atomic.AddUint32(&p.trigger, 1) > 1 {
		return nil
	}
	_, err := syscall.Kevent(p.fd, []syscall.Kevent_t{{
		Ident:  0,
		Filter: syscall.EVFILT_USER,
		Fflags: syscall.NOTE_TRIGGER,
	}}, nil, nil)
	return err
}

// Control implements Poll.
func (p *defaultPoll) Control(operator *FDOperator, event PollEvent) error {
	var evs = make([]syscall.Kevent_t, 1)
	evs[0].Ident = uint64(operator.FD)
	switch event {
	case PollReadable, PollModReadable:
		operator.inuse()
		p.m.Store(operator.FD, operator)
		evs[0].Filter, evs[0].Flags = syscall.EVFILT_READ, syscall.EV_ADD|syscall.EV_ENABLE
	case PollDetach:
		p.m.Delete(operator.FD)
		evs[0].Filter, evs[0].Flags = syscall.EVFILT_READ, syscall.EV_DELETE|syscall.EV_ONESHOT
	case PollWritable:
		operator.inuse()
		p.m.Store(operator.FD, operator)
		evs[0].Filter, evs[0].Flags = syscall.EVFILT_WRITE, syscall.EV_ADD|syscall.EV_ENABLE|syscall.EV_ONESHOT
	case PollR2RW:
		evs[0].Filter, evs[0].Flags = syscall.EVFILT_WRITE, syscall.EV_ADD|syscall.EV_ENABLE
	case PollRW2R:
		evs[0].Filter, evs[0].Flags = syscall.EVFILT_WRITE, syscall.EV_DELETE|syscall.EV_ONESHOT
	}
	_, err := syscall.Kevent(p.fd, evs, nil, nil)
	return err
}

func (p *defaultPoll) appendHup(operator *FDOperator) {
	p.hups = append(p.hups, operator.OnHup)
	operator.Control(PollDetach)
	operator.done()
}

func (p *defaultPoll) detaches() {
	if len(p.hups) == 0 {
		return
	}
	hups := p.hups
	p.hups = nil
	go func(onhups []func(p Poll) error) {
		for i := range onhups {
			if onhups[i] != nil {
				onhups[i](p)
			}
		}
	}(hups)
}