use individual ports, rather than full urls in connection files

IPython/parallel/apps/ipcontrollerapp.py

@@ -211,19 +211,15 b' class IPControllerApp(BaseParallelApplication):'
         c = self.config
         url = cdict['registration']
         location = cdict['location']
+        
         if not location:
             try:
-                proto,ip,port = split_url(url)
-            except AssertionError:
-                pass
-            else:
-                try:
-                    location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
-                except (socket.gaierror, IndexError):
-                    self.log.warn("Could not identify this machine's IP, assuming 127.0.0.1."
-                    " You may need to specify '--location=<external_ip_address>' to help"
-                    " IPython decide when to connect via loopback.")
-                    location = '127.0.0.1'
+                location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
+            except (socket.gaierror, IndexError):
+                self.log.warn("Could not identify this machine's IP, assuming 127.0.0.1."
+                " You may need to specify '--location=<external_ip_address>' to help"
+                " IPython decide when to connect via loopback.")
+                location = '127.0.0.1'
             cdict['location'] = location
         fname = os.path.join(self.profile_dir.security_dir, fname)
         self.log.info("writing connection info to %s", fname)
@@ -235,35 +231,51 b' class IPControllerApp(BaseParallelApplication):'
         """load config from existing json connector files."""
         c = self.config
         self.log.debug("loading config from JSON")
-        # load from engine config
+        
+        # load engine config
+        
         fname = os.path.join(self.profile_dir.security_dir, self.engine_json_file)
         self.log.info("loading connection info from %s", fname)
         with open(fname) as f:
-            cfg = json.loads(f.read())
-        key = cfg['exec_key']
+            ecfg = json.loads(f.read())
+        
         # json gives unicode, Session.key wants bytes
-        c.Session.key = key.encode('ascii')
-        xport,addr = cfg['url'].split('://')
-        c.HubFactory.engine_transport = xport
-        ip,ports = addr.split(':')
+        c.Session.key = ecfg['exec_key'].encode('ascii')
+        
+        xport,ip = ecfg['interface'].split('://')
+        
         c.HubFactory.engine_ip = ip
-        c.HubFactory.regport = int(ports)
-        self.location = cfg['location']
+        c.HubFactory.engine_transport = xport
+        
+        self.location = ecfg['location']
         if not self.engine_ssh_server:
-            self.engine_ssh_server = cfg['ssh']
+            self.engine_ssh_server = ecfg['ssh']
+        
         # load client config
+        
         fname = os.path.join(self.profile_dir.security_dir, self.client_json_file)
         self.log.info("loading connection info from %s", fname)
         with open(fname) as f:
-            cfg = json.loads(f.read())
-        assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
-        xport,addr = cfg['url'].split('://')
+            ccfg = json.loads(f.read())
+        
+        for key in ('exec_key', 'registration', 'pack', 'unpack'):
+            assert ccfg[key] == ecfg[key], "mismatch between engine and client info: %r" % key
+        
+        xport,addr = ccfg['interface'].split('://')
+        
         c.HubFactory.client_transport = xport
-        ip,ports = addr.split(':')
         c.HubFactory.client_ip = ip
         if not self.ssh_server:
-            self.ssh_server = cfg['ssh']
-        assert int(ports) == c.HubFactory.regport, "regport mismatch"
+            self.ssh_server = ccfg['ssh']
+        
+        # load port config:
+        c.HubFactory.regport = ecfg['registration']
+        c.HubFactory.hb = (ecfg['hb_ping'], ecfg['hb_pong'])
+        c.HubFactory.control = (ccfg['control'], ecfg['control'])
+        c.HubFactory.mux = (ccfg['mux'], ecfg['mux'])
+        c.HubFactory.task = (ccfg['task'], ecfg['task'])
+        c.HubFactory.iopub = (ccfg['iopub'], ecfg['iopub'])
+        c.HubFactory.notifier_port = ccfg['notification']
     
     def cleanup_connection_files(self):
         if self.reuse_files:
@@ -335,14 +347,14 b' class IPControllerApp(BaseParallelApplication):'
         children = self.children
         mq = import_item(str(self.mq_class))
         
-        hub = self.factory
+        f = self.factory
         # disambiguate url, in case of *
-        monitor_url = disambiguate_url(hub.monitor_url)
+        monitor_url = disambiguate_url(f.monitor_url)
         # maybe_inproc = 'inproc://monitor' if self.use_threads else monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, b'N/A',b'iopub')
-        q.bind_in(hub.client_info['iopub'])
-        q.bind_out(hub.engine_info['iopub'])
+        q.bind_in(f.client_url('iopub'))
+        q.bind_out(f.engine_url('iopub'))
         q.setsockopt_out(zmq.SUBSCRIBE, b'')
         q.connect_mon(monitor_url)
         q.daemon=True
@@ -350,18 +362,18 b' class IPControllerApp(BaseParallelApplication):'
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.ROUTER, zmq.ROUTER, zmq.PUB, b'in', b'out')
-        q.bind_in(hub.client_info['mux'])
+        q.bind_in(f.client_url('mux'))
         q.setsockopt_in(zmq.IDENTITY, b'mux')
-        q.bind_out(hub.engine_info['mux'])
+        q.bind_out(f.engine_url('mux'))
         q.connect_mon(monitor_url)
         q.daemon=True
         children.append(q)
 
         # Control Queue (in a Process)
         q = mq(zmq.ROUTER, zmq.ROUTER, zmq.PUB, b'incontrol', b'outcontrol')
-        q.bind_in(hub.client_info['control'])
+        q.bind_in(f.client_url('control'))
         q.setsockopt_in(zmq.IDENTITY, b'control')
-        q.bind_out(hub.engine_info['control'])
+        q.bind_out(f.engine_url('control'))
         q.connect_mon(monitor_url)
         q.daemon=True
         children.append(q)
@@ -374,9 +386,9 b' class IPControllerApp(BaseParallelApplication):'
             self.log.warn("task::using pure DEALER Task scheduler")
             q = mq(zmq.ROUTER, zmq.DEALER, zmq.PUB, b'intask', b'outtask')
             # q.setsockopt_out(zmq.HWM, hub.hwm)
-            q.bind_in(hub.client_info['task'][1])
+            q.bind_in(f.client_url('task'))
             q.setsockopt_in(zmq.IDENTITY, b'task')
-            q.bind_out(hub.engine_info['task'])
+            q.bind_out(f.engine_url('task'))
             q.connect_mon(monitor_url)
             q.daemon=True
             children.append(q)
@@ -385,8 +397,8 b' class IPControllerApp(BaseParallelApplication):'
 
         else:
             self.log.info("task::using Python %s Task scheduler"%scheme)
-            sargs = (hub.client_info['task'], hub.engine_info['task'],
-                                monitor_url, disambiguate_url(hub.client_info['notification']))
+            sargs = (f.client_url('task'), f.engine_url('task'),
+                                monitor_url, disambiguate_url(f.client_url('notification')))
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
             if 'Process' in self.mq_class:

IPython/parallel/apps/ipengineapp.py

@@ -45,7 +45,7 b' from IPython.zmq.session import ('
 from IPython.config.configurable import Configurable
 
 from IPython.parallel.engine.engine import EngineFactory
-from IPython.parallel.util import disambiguate_url
+from IPython.parallel.util import disambiguate_ip_address
 
 from IPython.utils.importstring import import_item
 from IPython.utils.py3compat import cast_bytes
@@ -225,14 +225,15 b' class IPEngineApp(BaseParallelApplication):'
         
         location = config.EngineFactory.location
         
-        for key in ('registration', 'hb_ping', 'hb_pong', 'mux', 'task', 'control'):
-            d[key] = disambiguate_url(d[key], location)
+        proto, ip = d['interface'].split('://')
+        ip = disambiguate_ip_address(ip)
+        d['interface'] = '%s://%s' % (proto, ip)
         
         # DO NOT allow override of basic URLs, serialization, or exec_key
         # JSON file takes top priority there
-        config.Session.key = asbytes(d['exec_key'])
+        config.Session.key = cast_bytes(d['exec_key'])
         
-        config.EngineFactory.url = d['registration']
+        config.EngineFactory.url = d['interface'] + ':%i' % d['registration']
         
         config.Session.packer = d['pack']
         config.Session.unpacker = d['unpack']

IPython/parallel/client/client.py

@@ -403,10 +403,17 b' class Client(HasTraits):'
             cfg['ssh'] = sshserver
 
         location = cfg.setdefault('location', None)
-        for key in ('control', 'task', 'mux', 'notification', 'registration'):
-            cfg[key] = util.disambiguate_url(cfg[key], location)
+        
+        proto,addr = cfg['interface'].split('://')
+        addr = util.disambiguate_ip_address(addr)
+        cfg['interface'] = "%s://%s" % (proto, addr)
+        
+        # turn interface,port into full urls:
+        for key in ('control', 'task', 'mux', 'iopub', 'notification', 'registration'):
+            cfg[key] = cfg['interface'] + ':%i' % cfg[key]
+        
         url = cfg['registration']
-        proto,addr,port = util.split_url(url)
+        
         if location is not None and addr == '127.0.0.1':
             # location specified, and connection is expected to be local
             if location not in LOCAL_IPS and not sshserver:
@@ -431,7 +438,7 b' class Client(HasTraits):'
         self._ssh = bool(sshserver or sshkey or password)
         if self._ssh and sshserver is None:
             # default to ssh via localhost
-            sshserver = url.split('://')[1].split(':')[0]
+            sshserver = addr
         if self._ssh and password is None:
             if tunnel.try_passwordless_ssh(sshserver, sshkey, paramiko):
                 password=False
@@ -449,9 +456,9 b' class Client(HasTraits):'
         self._query_socket = self._context.socket(zmq.DEALER)
 
         if self._ssh:
-            tunnel.tunnel_connection(self._query_socket, url, sshserver, **ssh_kwargs)
+            tunnel.tunnel_connection(self._query_socket, cfg['registration'], sshserver, **ssh_kwargs)
         else:
-            self._query_socket.connect(url)
+            self._query_socket.connect(cfg['registration'])
 
         self.session.debug = self.debug
 

IPython/parallel/controller/hub.py

@@ -131,29 +131,29 b' class HubFactory(RegistrationFactory):'
 
     # port-pairs for monitoredqueues:
     hb = Tuple(Integer,Integer,config=True,
-        help="""DEALER/SUB Port pair for Engine heartbeats""")
+        help="""PUB/ROUTER Port pair for Engine heartbeats""")
     def _hb_default(self):
         return tuple(util.select_random_ports(2))
 
     mux = Tuple(Integer,Integer,config=True,
-        help="""Engine/Client Port pair for MUX queue""")
+        help="""Client/Engine Port pair for MUX queue""")
 
     def _mux_default(self):
         return tuple(util.select_random_ports(2))
 
     task = Tuple(Integer,Integer,config=True,
-        help="""Engine/Client Port pair for Task queue""")
+        help="""Client/Engine Port pair for Task queue""")
     def _task_default(self):
         return tuple(util.select_random_ports(2))
 
     control = Tuple(Integer,Integer,config=True,
-        help="""Engine/Client Port pair for Control queue""")
+        help="""Client/Engine Port pair for Control queue""")
 
     def _control_default(self):
         return tuple(util.select_random_ports(2))
 
     iopub = Tuple(Integer,Integer,config=True,
-        help="""Engine/Client Port pair for IOPub relay""")
+        help="""Client/Engine Port pair for IOPub relay""")
 
     def _iopub_default(self):
         return tuple(util.select_random_ports(2))
@@ -231,10 +231,16 b' class HubFactory(RegistrationFactory):'
         self.heartmonitor.start()
         self.log.info("Heartmonitor started")
 
+    def client_url(self, channel):
+        """return full zmq url for a named client channel"""
+        return "%s://%s:%i" % (self.client_transport, self.client_ip, self.client_info[channel])
+    
+    def engine_url(self, channel):
+        """return full zmq url for a named engine channel"""
+        return "%s://%s:%i" % (self.engine_transport, self.engine_ip, self.engine_info[channel])
+    
     def init_hub(self):
-        """construct"""
-        client_iface = "%s://%s:" % (self.client_transport, self.client_ip) + "%i"
-        engine_iface = "%s://%s:" % (self.engine_transport, self.engine_ip) + "%i"
+        """construct Hub object"""
 
         ctx = self.context
         loop = self.loop
@@ -247,23 +253,25 b' class HubFactory(RegistrationFactory):'
         
         # build connection dicts
         engine = self.engine_info = {
-            'registration'  : engine_iface % self.regport,
-            'control'       : engine_iface % self.control[1],
-            'mux'           : engine_iface % self.mux[1],
-            'hb_ping'       : engine_iface % self.hb[0],
-            'hb_pong'       : engine_iface % self.hb[1],
-            'task'          : engine_iface % self.task[1],
-            'iopub'         : engine_iface % self.iopub[1],
+            'interface'     : "%s://%s" % (self.engine_transport, self.engine_ip),
+            'registration'  : self.regport,
+            'control'       : self.control[1],
+            'mux'           : self.mux[1],
+            'hb_ping'       : self.hb[0],
+            'hb_pong'       : self.hb[1],
+            'task'          : self.task[1],
+            'iopub'         : self.iopub[1],
             }
 
         client = self.client_info = {
-            'registration'  : client_iface % self.regport,
-            'control'       : client_iface % self.control[0],
-            'mux'           : client_iface % self.mux[0],
-            'task'          : client_iface % self.task[0],
+            'interface'     : "%s://%s" % (self.client_transport, self.client_ip),
+            'registration'  : self.regport,
+            'control'       : self.control[0],
+            'mux'           : self.mux[0],
+            'task'          : self.task[0],
             'task_scheme'   : scheme,
-            'iopub'         : client_iface % self.iopub[0],
-            'notification'  : client_iface % self.notifier_port,
+            'iopub'         : self.iopub[0],
+            'notification'  : self.notifier_port,
             }
         
         self.log.debug("Hub engine addrs: %s", self.engine_info)
@@ -271,19 +279,19 b' class HubFactory(RegistrationFactory):'
         
         # Registrar socket
         q = ZMQStream(ctx.socket(zmq.ROUTER), loop)
-        q.bind(client['registration'])
-        self.log.info("Hub listening on %s for registration.", client_iface % self.regport)
+        q.bind(self.client_url('registration'))
+        self.log.info("Hub listening on %s for registration.", self.client_url('registration'))
         if self.client_ip != self.engine_ip:
-            q.bind(engine['registration'])
-            self.log.info("Hub listening on %s for registration.", engine_iface % self.regport)
+            q.bind(self.engine_url('registration'))
+            self.log.info("Hub listening on %s for registration.", self.engine_url('registration'))
 
         ### Engine connections ###
 
         # heartbeat
         hpub = ctx.socket(zmq.PUB)
-        hpub.bind(engine['hb_ping'])
+        hpub.bind(self.engine_url('hb_ping'))
         hrep = ctx.socket(zmq.ROUTER)
-        hrep.bind(engine['hb_pong'])
+        hrep.bind(self.engine_url('hb_pong'))
         self.heartmonitor = HeartMonitor(loop=loop, config=self.config, log=self.log,
                                 pingstream=ZMQStream(hpub,loop),
                                 pongstream=ZMQStream(hrep,loop)
@@ -293,7 +301,7 b' class HubFactory(RegistrationFactory):'
         
         # Notifier socket
         n = ZMQStream(ctx.socket(zmq.PUB), loop)
-        n.bind(client['notification'])
+        n.bind(self.client_url('notification'))
 
         ### build and launch the queues ###
 
@@ -313,7 +321,7 b' class HubFactory(RegistrationFactory):'
 
         # resubmit stream
         r = ZMQStream(ctx.socket(zmq.DEALER), loop)
-        url = util.disambiguate_url(self.client_info['task'])
+        url = util.disambiguate_url(self.client_url('task'))
         r.connect(url)
 
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
@@ -388,15 +396,6 b' class Hub(SessionFactory):'
         super(Hub, self).__init__(**kwargs)
         self.registration_timeout = max(5000, 2*self.heartmonitor.period)
 
-        # validate connection dicts:
-        for k,v in self.client_info.iteritems():
-            if k == 'task_scheme':
-                continue
-            else:
-                util.validate_url_container(v)
-        # util.validate_url_container(self.client_info)
-        util.validate_url_container(self.engine_info)
-
         # register our callbacks
         self.query.on_recv(self.dispatch_query)
         self.monitor.on_recv(self.dispatch_monitor_traffic)

IPython/parallel/engine/engine.py

@@ -145,19 +145,23 b' class EngineFactory(RegistrationFactory):'
         content = msg['content']
         info = self.connection_info
         
+        def url(key):
+            """get zmq url for given channel"""
+            return str(info["interface"] + ":%i" % info[key])
+        
         if content['status'] == 'ok':
             self.id = int(content['id'])
 
             # launch heartbeat
             # possibly forward hb ports with tunnels
-            hb_ping = maybe_tunnel(info['hb_ping'])
-            hb_pong = maybe_tunnel(info['hb_pong'])
+            hb_ping = maybe_tunnel(url('hb_ping'))
+            hb_pong = maybe_tunnel(url('hb_pong'))
 
             heart = Heart(hb_ping, hb_pong, heart_id=identity)
             heart.start()
 
             # create Shell Connections (MUX, Task, etc.):
-            shell_addrs = map(str, [info['mux'], info['task']])
+            shell_addrs = url('mux'), url('task')
 
             # Use only one shell stream for mux and tasks
             stream = zmqstream.ZMQStream(ctx.socket(zmq.ROUTER), loop)
@@ -167,13 +171,13 b' class EngineFactory(RegistrationFactory):'
                 connect(stream, addr)
 
             # control stream:
-            control_addr = str(info['control'])
+            control_addr = url('control')
             control_stream = zmqstream.ZMQStream(ctx.socket(zmq.ROUTER), loop)
             control_stream.setsockopt(zmq.IDENTITY, identity)
             connect(control_stream, control_addr)
 
             # create iopub stream:
-            iopub_addr = info['iopub']
+            iopub_addr = url('iopub')
             iopub_socket = ctx.socket(zmq.PUB)
             iopub_socket.setsockopt(zmq.IDENTITY, identity)
             connect(iopub_socket, iopub_addr)