From f216d709363e5f6d9b6b6acd69addee08319da77 2011-04-23 00:36:33
From: MinRK <benjaminrk@gmail.com>
Date: 2011-04-23 00:36:33
Subject: [PATCH] expose TaskScheduler.hwm on command-line for ipcontroller

also removed ControllerFactory.hwm that didn't actually do what it claimed.
---

diff --git a/IPython/parallel/apps/ipcontrollerapp.py b/IPython/parallel/apps/ipcontrollerapp.py
index 5e583a4..0cc69ac 100755
--- a/IPython/parallel/apps/ipcontrollerapp.py
+++ b/IPython/parallel/apps/ipcontrollerapp.py
@@ -205,9 +205,9 @@ class IPControllerAppConfigLoader(ClusterDirConfigLoader):
             help='Use threads instead of processes for the schedulers',
             )
         paa('--hwm',
-            dest='ControllerFactory.hwm', type=int,
-            help='specify the High Water Mark (HWM) for the downstream '
-            'socket in the pure ZMQ scheduler. This is the maximum number '
+            dest='TaskScheduler.hwm', type=int,
+            help='specify the High Water Mark (HWM) '
+            'in the Python scheduler. This is the maximum number '
             'of allowed outstanding tasks on each engine.',
             )
         
diff --git a/IPython/parallel/controller/controller.py b/IPython/parallel/controller/controller.py
index 3295706..e19ed97 100755
--- a/IPython/parallel/controller/controller.py
+++ b/IPython/parallel/controller/controller.py
@@ -35,8 +35,6 @@ class ControllerFactory(HubFactory):
     """Configurable for setting up a Hub and Schedulers."""
     
     usethreads = Bool(False, config=True)
-    # pure-zmq downstream HWM
-    hwm = Int(0, config=True)
     
     # internal
     children = List()
@@ -97,7 +95,6 @@ class ControllerFactory(HubFactory):
         if self.scheme == 'pure':
             self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
-            q.setsockopt_out(zmq.HWM, self.hwm)
             q.bind_in(self.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, 'task')
             q.bind_out(self.engine_info['task'])