reduce duplicate rdy update requests

[andyxning]

This PR will reduct duplicate rdy update requests.

[andyxning]

@ploxiln @mreiferson

[ploxiln]

I'm not convinced about the benefit of this change. How often does the overall "ready" algorithm generate redundant RDY messages, and is this the simplest way to improve it? Maybe, but I'm not convinced.

[andyxning]

I'm not convinced about the benefit of this change. How often does the overall "ready" algorithm generate redundant RDY messages, and is this the simplest way to improve it? Maybe, but I'm not convinced.

When:

• the max_in_flight setting to 4000
• the nsqd instances is about 1500

When consumer connects to every nsqd, it needs to rebalance the rdy around all the existing connections, at this time, a connection may send continuous duplicate rdy value.

[andyxning]

@ploxiln You can just take a look at the go-nsq change. Its change is simpler. nsqio/go-nsq#282

[andyxning]

@ploxiln Code has been cleaned in the correct way. PTAL.

[andyxning]

@ploxiln This PR's code has been cleaned. PTAL.~

[ploxiln]

here's an idea for an even simpler test change:

--- a/tests/test_backoff.py
+++ b/tests/test_backoff.py
@@ -260,9 +260,15 @@ def test_backoff_many_conns(mock_ioloop_current):
     for i in range(num_conns):
         conn = get_conn(r)
         conn.expected_args = [b'SUB test test\n', b'RDY 1\n']
+        conn.last_exp_rdy = b'RDY 1\n'
         conn.fails = 0
         conns.append(conn)
 
+    def add_exp_rdy(conn, rdy):
+        if conn.last_exp_rdy != rdy:
+            conn.expected_args.append(rdy)
+            conn.last_exp_rdy = rdy
+
     fail = True
     total_fails = 0
     last_timeout_time = 0
@@ -271,7 +277,7 @@ def test_backoff_many_conns(mock_ioloop_current):
         msg = send_message(conn)
 
         if r.backoff_timer.get_interval() == 0:
-            conn.expected_args.append(b'RDY 1\n')
+            add_exp_rdy(conn, b'RDY 1\n')
 
         if fail or not conn.fails:
             msg.trigger(event.REQUEUE, message=msg)
@@ -279,7 +285,7 @@ def test_backoff_many_conns(mock_ioloop_current):
             conn.fails += 1
 
             for c in conns:
-                c.expected_args.append(b'RDY 0\n')
+                add_exp_rdy(c, b'RDY 0\n')
             conn.expected_args.append(b'REQ 1234 0\n')
         else:
             msg.trigger(event.FINISH, message=msg)
@@ -287,7 +293,7 @@ def test_backoff_many_conns(mock_ioloop_current):
             conn.fails -= 1
 
             for c in conns:
-                c.expected_args.append(b'RDY 0\n')
+                add_exp_rdy(c, b'RDY 0\n')
             conn.expected_args.append(b'FIN 1234\n')
...

[andyxning]

@ploxiln code format is done. PTAL.

[ploxiln]

cool. thanks.

[andyxning]

@ploxiln Thanks for your review.