// sort timeout data and find expired
sort timeout data and find expired
// sort timeout data and find expired
func (q *TimeoutQueue) sort() {
	q.mu.Lock()
	defer q.mu.Unlock()
	if len(q.data) == 0 {
		return
	}
	sort.Sort(q)
	now := time.Now().UTC().UnixNano()
	expiredCount := 0
	for i, c := range q.data {
		if c.expireAt-now < int64(time.Millisecond) {
			expiredCount = i + 1
			continue
		}
		q.nextSort = c.expireAt
		select {
		case <-q.stop:
			return
		case q.resetCh <- c.expireAt - now:
		}
		break
	}
	if expiredCount > 0 {
		select {
		case <-q.stop:
			return
		default:
			for atomic.LoadUint32(&q.expireNotified) == 1 {
				runtime.Gosched()
			}
		}
		q.expiredData = q.data[:expiredCount]
		q.data = q.data[expiredCount:]
		for _, d := range q.expiredData {
			delete(q.index, d.Key)
		}
		for i, d := range q.data {
			q.index[d.Key] = i
		}
		atomic.StoreUint32(&q.expireNotified, 1)
		close(q.hasExpiredData)
	}
}
func ( q * TimeoutQueue ) sort ( ) { q . mu . Lock ( ) defer q . mu . Unlock ( ) if len ( q . data ) == 0 { return } sort . Sort ( q ) now := time . Now ( ) . UTC ( ) . UnixNano ( ) expiredCount := 0 for i , c := range q . data { if c . expireAt - now < int64 ( time . Millisecond ) { expiredCount = i + 1 continue } q . nextSort = c . expireAt select { case <- q . stop : return case q . resetCh <- c . expireAt - now : } break } if expiredCount > 0 { select { case <- q . stop : return default : for atomic . LoadUint32 ( & q . expireNotified ) == 1 { runtime . Gosched ( ) } } q . expiredData = q . data [ : expiredCount ] q . data = q . data [ expiredCount : ] for _ , d := range q . expiredData { delete ( q . index , d . Key ) } for i , d := range q . data { q . index [ d . Key ] = i } atomic . StoreUint32 ( & q . expireNotified , 1 ) close ( q . hasExpiredData ) } }
sort timeout data and find expired
sort timeout data and find expired
// make sure the timeout data is sorted // not expired, stop iteration and reset timer for the next // has expired data, signal to send // wait for last expired data sent // rebuild index