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

Code before:
				select {
				case q.lookupdRecheckChan <- 1:
				default:
				}
			}
			return
		}

		if msg.Success {
			if q.VerboseLogging {
				log.Printf("[%s] finishing %s", c, msg.Id)
			}
			err := c.sendCommand(Finish(msg.Id))
			if err != nil {
				log.Printf("[%s] error finishing %s - %s", c, msg.Id, err.Error())
				c.stopFinishLoop()
				continue
			}
			atomic.AddUint64(&c.messagesFinished, 1)
			atomic.AddUint64(&q.MessagesFinished, 1)
		} else {
			if q.VerboseLogging {
				log.Printf("[%s] requeuing %s", c, msg.Id)
			}
			err := c.sendCommand(Requeue(msg.Id, msg.RequeueDelayMs))
			if err != nil {
				log.Printf("[%s] error requeueing %s - %s", c, msg.Id, err.Error())
				c.stopFinishLoop()
				continue
			}
			atomic.AddUint64(&c.messagesRequeued, 1)
			atomic.AddUint64(&q.MessagesRequeued, 1)
		}

		atomic.AddInt64(&q.messagesInFlight, -1)
		if atomic.AddInt64(&c.messagesInFlight, -1) == 0 &&
			(atomic.LoadInt32(&c.stopFlag) == 1 || atomic.LoadInt32(&q.stopFlag) == 1) {
			c.stopFinishLoop()
			continue
		}
	}
}

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

	remain := atomic.LoadInt64(&c.rdyCount)
	s := q.ConnectionMaxInFlight()
	// refill when at 1, or at 25% whichever comes first

Review comment (bug):
Yeah, though there are probably multiple ways around it. If we get an error sending to `nsqd` from within the `finishLoop` (see, for example, line 460-463), we try to call `stopFinishLoop` and then `continue` to the next loop iteration. The problem is that if that channel is unbuffered, then `stopFinishLoop` is blocked, waiting for execution to continue in `finishLoop` - but we are blocked in `finishLoop` by the call to `stopFinishLoop` - it's a guaranteed deadlock. Now, I do see your point though, and I think a related race condition would still be present with the way I had organized the atomic reads and decrements prior to my last commit. 

One thing that might be able to help things would be for the draining goroutine to be selecting between `finishedMessages` and a timer, so that after a reasonable amount of time it will no longer wait for anything else on finishedMessages. Not a great solution, but might be an improvement?


Code after:
					log.Printf("[%s] requeuing %s", c, msg.Id)
				}
				err := c.sendCommand(Requeue(msg.Id, msg.RequeueDelayMs))
				if err != nil {
					log.Printf("[%s] error requeueing %s - %s", c, msg.Id, err.Error())
					q.stopFinishLoop(c)
					continue
				}
				atomic.AddUint64(&c.messagesRequeued, 1)
				atomic.AddUint64(&q.MessagesRequeued, 1)
			}

			if atomic.LoadInt64(&c.messagesInFlight) == 0 &&
				(atomic.LoadInt32(&c.stopFlag) == 1 || atomic.LoadInt32(&q.stopFlag) == 1) {
				q.stopFinishLoop(c)
				continue
			}
		}
	}
}

func (q *Reader) stopFinishLoop(c *nsqConn) {
	c.stopper.Do(func() {
		log.Printf("[%s] beginning stopFinishLoop logic", c)
		// This doesn't block because dying has buffer of 1
		c.dying <- struct{}{}

		// Drain the finishedMessages channel
		go func() {
			<-c.drainReady
			for atomic.AddInt64(&c.messagesInFlight, -1) >= 0 {
				<-c.finishedMessages
			}
		}()
		c.Close()
		delete(q.nsqConnections, c.String())

		log.Printf("there are %d connections left alive", len(q.nsqConnections))

		if len(q.nsqConnections) == 0 && len(q.lookupdHTTPAddrs) == 0 {
			// no lookupd entry means no reconnection
			if atomic.LoadInt32(&q.stopFlag) == 1 {
				q.stopHandlers()
			}
			return
		}

		// ie: we were the last one, and stopping
		if len(q.nsqConnections) == 0 && atomic.LoadInt32(&q.stopFlag) == 1 {
			q.stopHandlers()
		}