Backport PR #4074: close Client sockets if connection fails

avoids runaway FDs in cases of frequent client failure.

also cleanup Client.close a little bit, allowing setting `linger` on sockets at shutdown time,
which is helpful for preventing FD growth on connection failure.

Should be considered for backport to 1.1.

IPython/parallel/client/client.py

@@ -489,7 +489,12 b' class Client(HasTraits):'
                                     }
         self._queue_handlers = {'execute_reply' : self._handle_execute_reply,
                                 'apply_reply' : self._handle_apply_reply}
-        self._connect(sshserver, ssh_kwargs, timeout)
+
+        try:
+            self._connect(sshserver, ssh_kwargs, timeout)
+        except:
+            self.close(linger=0)
+            raise
         
         # last step: setup magics, if we are in IPython:
         
@@ -593,7 +598,6 b' class Client(HasTraits):'
         self._connected=True
 
         def connect_socket(s, url):
-            # url = util.disambiguate_url(url, self._config['location'])
             if self._ssh:
                 return tunnel.tunnel_connection(s, url, sshserver, **ssh_kwargs)
             else:
@@ -950,14 +954,23 b' class Client(HasTraits):'
         view.activate(suffix)
         return view
 
-    def close(self):
+    def close(self, linger=None):
+        """Close my zmq Sockets
+
+        If `linger`, set the zmq LINGER socket option,
+        which allows discarding of messages.
+        """
         if self._closed:
             return
         self.stop_spin_thread()
-        snames = filter(lambda n: n.endswith('socket'), dir(self))
-        for socket in map(lambda name: getattr(self, name), snames):
-            if isinstance(socket, zmq.Socket) and not socket.closed:
-                socket.close()
+        snames = [ trait for trait in self.trait_names() if trait.endswith("socket") ]
+        for name in snames:
+            socket = getattr(self, name)
+            if socket is not None and not socket.closed:
+                if linger is not None:
+                    socket.close(linger=linger)
+                else:
+                    socket.close()
         self._closed = True
 
     def _spin_every(self, interval=1):