Try to read and write to sockets.

        This method will also attempt to complete node connections, refresh
        stale metadata, and run previously-scheduled tasks.

        Arguments:
            timeout_ms (int, optional): maximum amount of time to wait (in ms)
                for at least one response. Must be non-negative. The actual
                timeout will be the minimum of timeout, request timeout and
                metadata timeout. Default: request_timeout_ms
            future (Future, optional): if provided, blocks until future.is_done

        Returns:
            list: responses received (can be empty)

def poll(self, timeout_ms=None, future=None):
        """Try to read and write to sockets.

        This method will also attempt to complete node connections, refresh
        stale metadata, and run previously-scheduled tasks.

        Arguments:
            timeout_ms (int, optional): maximum amount of time to wait (in ms)
                for at least one response. Must be non-negative. The actual
                timeout will be the minimum of timeout, request timeout and
                metadata timeout. Default: request_timeout_ms
            future (Future, optional): if provided, blocks until future.is_done

        Returns:
            list: responses received (can be empty)
        """
        if future is not None:
            timeout_ms = 100
        elif timeout_ms is None:
            timeout_ms = self.config['request_timeout_ms']
        elif not isinstance(timeout_ms, (int, float)):
            raise TypeError('Invalid type for timeout: %s' % type(timeout_ms))

        # Loop for futures, break after first loop if None
        responses = []
        while True:
            with self._lock:
                if self._closed:
                    break

                # Attempt to complete pending connections
                for node_id in list(self._connecting):
                    self._maybe_connect(node_id)

                # Send a metadata request if needed
                metadata_timeout_ms = self._maybe_refresh_metadata()

                # If we got a future that is already done, don't block in _poll
                if future is not None and future.is_done:
                    timeout = 0
                else:
                    idle_connection_timeout_ms = self._idle_expiry_manager.next_check_ms()
                    timeout = min(
                        timeout_ms,
                        metadata_timeout_ms,
                        idle_connection_timeout_ms,
                        self.config['request_timeout_ms'])
                    timeout = max(0, timeout / 1000)  # avoid negative timeouts

                self._poll(timeout)

                responses.extend(self._fire_pending_completed_requests())

            # If all we had was a timeout (future is None) - only do one poll
            # If we do have a future, we keep looping until it is done
            if future is None or future.is_done:
                break

        return responses