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):
Spent a little more time thinking about this.... want to hear your thoughts...

I think we _need_ to synchronize (ie not buffer this write) on the fact that the finish loop above is exiting so we can be sure to eliminate any possibility of a race race condition between lines 437 to 441 (which pull a message off the channel and update counters, non-atomically) and the code below that fires the goroutine to purge the `finishedMessages` channel.

If it was the last message in flight and the goroutine below won the race, it would decrement first (before line 441), pass its logic statement, and never be able to read anything off of `finishedMessages`... essentially orphaning the goroutine.

Was there a specific reason that I'm not accounting for that required it to be buffered?


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()
		}