MB-14833: fix large memory consumption when performing cbbackup using…

… DCP. Problem was pump_dcp.py was using an unlimited Queue for buffering DCP messages. This fix limits queue length to 1000, by default, but can be configured with the option: '-x dcp_consumer_queue_length=<queue length>'

Change-Id: I978e5473ad06e40e430c944363541502fb2121c8
Reviewed-on: http://review.couchbase.org/51658
Reviewed-by: Dave Finlay <dave.finlay@couchbase.com>
Tested-by: Dave Finlay <dave.finlay@couchbase.com>

pump_dcp.py

@@ -64,8 +64,9 @@ def __init__(self, opts, spec, source_bucket, source_node,
         self.flow_control = int(opts.extra.get("flow_control", 1))
         self.vbucket_list = getattr(opts, "vbucket_list", None)
         self.r=random.Random()
-        self.queue = PumpQueue()
-        self.response = PumpQueue()
+        self.queue = PumpQueue()# self.queue does not appear to be used anywhere, remove in future
+        self.queue_size = int(opts.extra.get("dcp_consumer_queue_length", 1000))
+        self.response = PumpQueue(self.queue_size)
         self.running = False
         self.stream_list = {}
         self.unack_size = 0

pump_transfer.py

@@ -224,6 +224,7 @@ def opt_extra_defaults(self):
             "uncompress":      (0,      "For value 1, restore data in uncompressed mode"),
             "backoff_cap":     (10,     "Max backoff time during rebalance period"),
             "flow_control":    (1,      "For value 0, disable flow control to improve throughput"),
+            "dcp_consumer_queue_length": (1000,"A DCP client needs a queue for incoming documents/messages. A large length is more efficient, but memory proportional to length*avg. doc size. Below length 150, performance degrades significantly."),
             }
 
     def find_handlers(self, opts, source, sink):