consumer: redistribute RDY when connections are active

[mreiferson]

fixes #179

[mreiferson]

RFR @jehiah @judwhite @ploxiln

This is one simple way to address #179 by bounding the time a consumer will stay pinned to any producing nsqd, regardless of last message delivery.

[mreiferson]

This should probably never be lower than LowRdyIdleTimeout, I can enforce this...

[ploxiln]

Looks good to me.

I noticed this part below seems less than optimal (abbreviated):

 	for len(possibleConns) > 0 && availableMaxInFlight > 0 {
 		availableMaxInFlight--
 		i := r.rng.Int() % len(possibleConns)
 		c := possibleConns[i]
 		possibleConns = append(possibleConns[:i], possibleConns[i+1:]...)
 		r.updateRDY(c, 1)
}

All conns are added to possibleConns, so it's quite possible that some of them already have non-zero ready count. So some of availableMaxInFlight will end up not actually distributed. (But this problem will probably fix itself in later redistributions. And it's not new.)

This could be fixed by appending to possibleConns only if the conn has a zero ready count.

[mreiferson]

All conns are added to possibleConns, so it's quite possible that some of them already have non-zero ready count. So some of availableMaxInFlight will end up not actually distributed.

Hmm, yea, can probably be improved, but it's at least randomized.

[mreiferson]

@ploxiln want to gimme a 👍 — also, there are probably a few lessons to learn from nsqio/pynsq#179 and nsqio/pynsq#176

[ploxiln]

👍 to these additions.

... but the existing rdyCount logic is a bit tricky, and I'm not really familiar with it, and it's taking a while to digest. Anyway, I'm confident that this is strictly an improvement. fwiw :)