poll.go

package srtgo

/*
#cgo LDFLAGS: -lsrt
#include <srt/srt.h>
*/
import "C"
import (
	"math"
	"sync"
	"sync/atomic"
	"time"
)

// noDeadline arms a timer effectively forever (~292 years). A timer created
// with a zero or negative duration fires immediately and leaves a stale tick
// in its channel, which wait() would observe as a spurious deadline expiry.
const noDeadline = time.Duration(math.MaxInt64)

const (
	pollDefault = int32(iota)
	pollReady   = int32(iota)
	pollWait    = int32(iota)
)

type PollMode int

const (
	ModeRead = PollMode(iota)
	ModeWrite
	// ModeReadWrite addresses the read and the write side at once. It must be
	// its own value: ModeRead is 0, so the expression ModeRead+ModeWrite is
	// just ModeWrite and cannot be distinguished from it.
	ModeReadWrite
)

/*
	pollDesc contains the polling state for the associated SrtSocket
	closing: socket is closing, reject all poll operations
	pollErr: an error occured on the socket, indicates it's not useable anymore. 0 = no error, 1 = error.
	unblockRd: is used to unblock the poller when the socket becomes ready for io
	rdState: polling state for read operations
	rdDeadline: deadline in NS before poll operation times out, -1 means timedout (needs to be cleared), 0 is without timeout
	rdSeq: sequence number protects against spurious signalling of timeouts when timer is reset.
	rdTimer: timer used to enforce deadline.

	Concurrency note: pollErr, rdState and wrState are accessed by the pollServer
	goroutine from unblock(), which deliberately holds no lock. They must
	therefore be touched with sync/atomic ONLY -- a plain load or store here
	is a data race against unblock()'s atomic.SwapInt32, even where pd.lock
	or rdLock/wrLock is held, because unblock() takes neither.
*/
type pollDesc struct {
	lock       sync.Mutex
	closing    bool
	fd         C.SRTSOCKET
	pollErr    int32
	unblockRd  chan interface{}
	rdState    int32
	rdLock     sync.Mutex
	rdDeadline int64
	rdSeq      int64
	rdTimer    *time.Timer
	rtSeq      int64
	unblockWr  chan interface{}
	wrState    int32
	wrLock     sync.Mutex
	wdDeadline int64
	wdSeq      int64
	wdTimer    *time.Timer
	wtSeq      int64
	pollS      *pollServer
}

var pdPool = sync.Pool{
	New: func() interface{} {
		return &pollDesc{
			unblockRd: make(chan interface{}, 1),
			unblockWr: make(chan interface{}, 1),
			rdTimer:   time.NewTimer(noDeadline),
			wdTimer:   time.NewTimer(noDeadline),
		}
	},
}

// stopTimer stops t and drains any tick already delivered to its channel, so
// that a subsequent Reset cannot be observed as an immediate expiry. It must
// not be called while a goroutine is selecting on t.C for this pollDesc.
func stopTimer(t *time.Timer) {
	if !t.Stop() {
		select {
		case <-t.C:
		default:
		}
	}
}

func pollDescInit(s C.SRTSOCKET) *pollDesc {
	pd := pdPool.Get().(*pollDesc)
	pd.lock.Lock()
	pd.fd = s
	pd.rdState = pollDefault
	pd.wrState = pollDefault
	pd.pollS = pollServerCtx()
	pd.closing = false
	atomic.StoreInt32(&pd.pollErr, 0)
	pd.rdSeq++
	pd.wdSeq++
	pd.lock.Unlock()
	//This is defensive, the lock order inversion between pollDesc.lock and
	//pollServer.pollDescLock is fixed for this side by the other change in
	//pollServer.pollClose.
	pd.pollS.pollOpen(pd)
	return pd
}

func (pd *pollDesc) release() {
	pd.lock.Lock()
	defer pd.lock.Unlock()
	if !pd.closing || pd.rdState == pollWait || pd.wrState == pollWait {
		panic("returning open or blocked upon pollDesc")
	}
	//Timers are stopped and drained before the pollDesc goes back into the
	//pool, so a leftover tick cannot be seen by the next socket to reuse it.
	stopTimer(pd.rdTimer)
	stopTimer(pd.wdTimer)
	pd.fd = 0
	pdPool.Put(pd)
}

func (pd *pollDesc) wait(mode PollMode) error {
	defer pd.reset(mode)
	if err := pd.checkPollErr(mode); err != nil {
		return err
	}
	state := &pd.rdState
	unblockChan := pd.unblockRd
	expiryChan := pd.rdTimer.C
	timerSeq := int64(0)
	pd.lock.Lock()
	if mode == ModeRead {
		timerSeq = pd.rtSeq
		pd.rdLock.Lock()
		defer pd.rdLock.Unlock()
	} else if mode == ModeWrite {
		timerSeq = pd.wtSeq
		state = &pd.wrState
		unblockChan = pd.unblockWr
		expiryChan = pd.wdTimer.C
		pd.wrLock.Lock()
		defer pd.wrLock.Unlock()
	}

	for {
		old := atomic.LoadInt32(state)
		if old == pollReady {
			atomic.StoreInt32(state, pollDefault)
			pd.lock.Unlock()
			return nil
		}
		if atomic.CompareAndSwapInt32(state, pollDefault, pollWait) {
			break
		}
	}
	pd.lock.Unlock()

wait:
	for {
		select {
		case <-unblockChan:
			break wait
		case <-expiryChan:
			pd.lock.Lock()
			if mode == ModeRead {
				if timerSeq == pd.rdSeq {
					pd.rdDeadline = -1
					pd.lock.Unlock()
					break wait
				}
				timerSeq = pd.rtSeq
			}
			if mode == ModeWrite {
				if timerSeq == pd.wdSeq {
					pd.wdDeadline = -1
					pd.lock.Unlock()
					break wait
				}
				timerSeq = pd.wtSeq
			}
			pd.lock.Unlock()
		}
	}
	err := pd.checkPollErr(mode)
	return err
}

func (pd *pollDesc) close() {
	pd.lock.Lock()
	if pd.closing {
		pd.lock.Unlock()
		return
	}
	pd.closing = true
	//Lock order rule: pollClose holds pollDescLock while there is a chance
	//that pollServer.run is waiting on pd.lock while holding pollDescLock
	pd.lock.Unlock()
	pd.pollS.pollClose(pd)
}

func (pd *pollDesc) checkPollErr(mode PollMode) error {
	pd.lock.Lock()
	defer pd.lock.Unlock()
	if pd.closing {
		return &SrtSocketClosed{}
	}

	if mode == ModeRead && pd.rdDeadline < 0 || mode == ModeWrite && pd.wdDeadline < 0 {
		return &SrtEpollTimeout{}
	}

	if atomic.LoadInt32(&pd.pollErr) != 0 {
		return &SrtSocketClosed{}
	}

	return nil
}

func (pd *pollDesc) setDeadline(t time.Time, mode PollMode) {
	pd.lock.Lock()
	defer pd.lock.Unlock()
	var d int64
	if !t.IsZero() {
		d = int64(time.Until(t))
		if d == 0 {
			d = -1
		}
	}
	if mode == ModeRead || mode == ModeReadWrite {
		pd.rdSeq++
		pd.rtSeq = pd.rdSeq
		stopTimer(pd.rdTimer)
		pd.rdDeadline = d
		if d > 0 {
			pd.rdTimer.Reset(time.Duration(d))
		}
		if d < 0 {
			pd.unblock(ModeRead, false, false)
		}
	}
	if mode == ModeWrite || mode == ModeReadWrite {
		pd.wdSeq++
		pd.wtSeq = pd.wdSeq
		stopTimer(pd.wdTimer)
		pd.wdDeadline = d
		if d > 0 {
			pd.wdTimer.Reset(time.Duration(d))
		}
		if d < 0 {
			pd.unblock(ModeWrite, false, false)
		}
	}
}

// unblock takes no pollDesc lock, on any path. setDeadline calls it while
// holding pd.lock, so acquiring pd.lock here self-deadlocks on a non-reentrant
// mutex -- the bug PR #72 fixed for the state update and left in the pollErr
// branch. pollErr is therefore atomic, like rdState and wrState, and callers
// are free to hold pd.lock, pollDescLock, or nothing at all.
func (pd *pollDesc) unblock(mode PollMode, pollerr, ioready bool) {
	if pollerr {
		atomic.StoreInt32(&pd.pollErr, 1)
	}
	state := &pd.rdState
	unblockChan := pd.unblockRd
	if mode == ModeWrite {
		state = &pd.wrState
		unblockChan = pd.unblockWr
	}
	var old int32
	if ioready {
		old = atomic.SwapInt32(state, pollReady)
	} else {
		old = atomic.LoadInt32(state)
	}
	if old == pollWait {
		//make sure we never block here
		select {
		case unblockChan <- struct{}{}:
			//
		default:
			//
		}
	}
}

func (pd *pollDesc) reset(mode PollMode) {
	pd.lock.Lock()
	defer pd.lock.Unlock()
	if mode == ModeRead {
		pd.rdLock.Lock()
		atomic.StoreInt32(&pd.rdState, pollDefault)
		pd.rdLock.Unlock()
	} else if mode == ModeWrite {
		pd.wrLock.Lock()
		atomic.StoreInt32(&pd.wrState, pollDefault)
		pd.wrLock.Unlock()
	}
}