Repository: nsqio/nsq
File: nsq/reader.go (Go)

Code before:
func (q *Reader) sendRDY(c *nsqConn, count int64) error {
	var buf bytes.Buffer

	if count == 0 && atomic.LoadInt64(&c.lastRdyCount) == 0 {
		// no need to send. It's already that RDY count
		return nil
	}

	atomic.AddInt64(&q.totalRdyCount, atomic.LoadInt64(&c.rdyCount)*-1+count)
	atomic.StoreInt64(&c.rdyCount, count)
	atomic.StoreInt64(&c.lastRdyCount, count)
	err := c.sendCommand(&buf, Ready(int(count)))
	if err != nil {
		handleError(q, c, fmt.Sprintf("[%s] error sending RDY %d - %s", c, count, err.Error()))
		return err
	}
	return nil
}

func (q *Reader) redistributeRdyState() {
	for {
		time.Sleep(5 * time.Second)

		q.RLock()
		l := len(q.nsqConnections)
		q.RUnlock()
		if l <= q.maxInFlight {
			continue
		}

		log.Printf("redistributing ready state (%d conns > %d max_in_flight)", l, q.maxInFlight)
		q.RLock()
		possibleConns := make([]*nsqConn, 0, len(q.nsqConnections))
		for _, c := range q.nsqConnections {
			lastMsgTimestamp := atomic.LoadInt64(&c.lastMsgTimestamp)
			lastMsgDuration := time.Now().Sub(time.Unix(0, lastMsgTimestamp))
			rdyCount := atomic.LoadInt64(&c.rdyCount)
			if q.VerboseLogging {
				log.Printf("[%s] rdy: %d (last message received %s)", c, rdyCount, lastMsgDuration)
			}
			if rdyCount > 0 && lastMsgDuration > q.LowRdyIdleTimeout {
				log.Printf("[%s] idle connection, giving up RDY count", c)
				q.updateRDY(c, 0)
			}
			possibleConns = append(possibleConns, c)
		}
		availableMaxInFlight := int64(q.maxInFlight) - atomic.LoadInt64(&q.totalRdyCount)
		for len(possibleConns) > 0 && availableMaxInFlight > 0 {
			availableMaxInFlight--
			i := rand.Int() % len(possibleConns)
			c := possibleConns[i]

Review comment (bug):
I ended up removing this from the `pynsq` side, I found it difficult to reason about for tests (edge case).

This is slightly different in that it only skips a dupe `RDY 0`, but is it necessary?


Code after:
	if atomic.LoadInt32(&c.stopFlag) != 0 {
		return nil
	}

	// never exceed the nsqd's configured max RDY count
	if count > c.maxRdyCount {
		count = c.maxRdyCount
	}

	// never exceed our global max in flight. truncate if possible.
	// this could help a new connection get partial max-in-flight
	maxPossibleRdy := int64(q.maxInFlight) - atomic.LoadInt64(&q.totalRdyCount)
	if maxPossibleRdy > 0 && maxPossibleRdy < count {
		count = maxPossibleRdy
	}
	if maxPossibleRdy <= 0 && count > 0 {
		return ErrOverMaxInFlight
	}

	return q.sendRDY(c, count)
}

func (q *Reader) sendRDY(c *nsqConn, count int64) error {
	var buf bytes.Buffer

	if count == 0 && atomic.LoadInt64(&c.lastRdyCount) == 0 {
		// no need to send. It's already that RDY count
		return nil
	}

	atomic.AddInt64(&q.totalRdyCount, atomic.LoadInt64(&c.rdyCount)*-1+count)
	atomic.StoreInt64(&c.rdyCount, count)
	atomic.StoreInt64(&c.lastRdyCount, count)
	err := c.sendCommand(&buf, Ready(int(count)))
	if err != nil {
		handleError(q, c, fmt.Sprintf("[%s] error sending RDY %d - %s", c, count, err.Error()))
		return err
	}
	return nil
}

func (q *Reader) redistributeRdyState() {
	for {
		time.Sleep(5 * time.Second)

		q.RLock()
		l := len(q.nsqConnections)
		q.RUnlock()
		if l <= q.maxInFlight {
			continue
		}