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

Code before:
			message, ok := <-q.incomingMessages
			if !ok {
				log.Printf("closing Handler (after self.incomingMessages closed)")
				if atomic.AddInt32(&q.runningHandlers, -1) == 0 {
					q.ExitChan <- 1
				}
				break
			}

			err := handler.HandleMessage(message.Message)
			if err != nil {
				log.Printf("ERR: handler returned %s for msg %s %s", err.Error(), message.Id, message.Body)
			}

			// message passed the max number of attempts
			if err != nil && q.MaxAttemptCount > 0 && message.Attempts > q.MaxAttemptCount {
				log.Printf("WARNING: msg attempted %d times. giving up %s %s", message.Attempts, message.Id, message.Body)
				logger, ok := handler.(FailedMessageLogger)
				if ok {
					logger.LogFailedMessage(message.Message)
				}
				message.responseChannel <- &FinishedMessage{message.Id, 0, true}
				continue
			}

			// linear delay
			requeueDelay := q.DefaultRequeueDelay * time.Duration(message.Attempts)
			// bound the requeueDelay to configured max
			if requeueDelay > q.MaxRequeueDelay {
				requeueDelay = q.MaxRequeueDelay
			}

			message.responseChannel <- &FinishedMessage{message.Id, int(requeueDelay / time.Millisecond), err == nil}
		}
	}()
}

// AddAsyncHandler adds an AsyncHandler for messages received by this Reader.
//
// See AsyncHandler for details on implementing this interface.
//
// It's ok to start more than one handler simultaneously, they
// are concurrently executed in goroutines.
func (q *Reader) AddAsyncHandler(handler AsyncHandler) {
	atomic.AddInt32(&q.runningHandlers, 1)
	log.Println("starting AsyncHandler go-routine")
	go func() {
		for {
			message, ok := <-q.incomingMessages
			if !ok {
				log.Printf("closing AsyncHandler (after self.incomingMessages closed)")

Review comment (bug):
i would like to see `updateRDY` updated (no pun intended) so that after recovering from a backoff it only sets RDY to a max of 1. This avoids the situation where you recover from a 120sec backoff w/ a RDY 50, succeed all of them and then drain your backoff to 0 immediately. In order for the exponential recovery to happen on the time scale expected only a single message should be RDY'd after a failure. 

The same sort of situation exists on the initial backoff side. if you are at RDY 50, and you fail all 50 at the same instance are you instantly at a delay of 120sec? One way to avoid that could be to calculate the delay time in finishLoop, and set a timestmp = time.Now().Add(backoffDelay), and ignore any success/failures till that time. (so that you are guaranteed the first 2sec backoff)

Also I've been thinking about the value of skipping backoff on the first failure, and only backing off after two contiguous failures.


Code after:

			// send ready immediately
			if backoffCounter == 0 {
				q.updateRDY(c)
				continue
			}

			backoffDuration := q.backoffDuration(backoffCounter)
			backoffTimer = time.NewTimer(backoffDuration)
			backoffTimerChan = backoffTimer.C
			readyChan = nil

			log.Printf("[%s] backing off for %.02f seconds", c, backoffDuration.Seconds())
		case <-c.exitChan:
			if backoffTimer != nil {
				backoffTimer.Stop()
			}
			goto exit
		}
	}

exit:
	log.Printf("[%s] rdyLoop exiting", c)
}

func (q *Reader) updateRDY(c *nsqConn) error {
	if atomic.LoadInt32(&c.stopFlag) != 0 {
		return nil
	}

	remain := atomic.LoadInt64(&c.rdyCount)
	mif := q.ConnectionMaxInFlight()
	// refill when at 1, or at 25% whichever comes first
	if remain <= 1 || remain < (int64(mif)/int64(4)) {
		if q.VerboseLogging {
			log.Printf("[%s] sending RDY %d (%d remain)", c, mif, remain)
		}
		q.sendRDY(c, mif)
	} else {
		if q.VerboseLogging {
			log.Printf("[%s] skip sending RDY (%d remain out of %d)", c, remain, mif)
		}
	}

	return nil
}

func (q *Reader) sendRDY(c *nsqConn, count int) error {
	var buf bytes.Buffer
	atomic.StoreInt64(&c.rdyCount, int64(count))
	err := c.sendCommand(&buf, Ready(count))