def acquire_direct(self, address):
if self.closed():<EOL><INDENT>raise ServiceUnavailable("<STR_LIT>")<EOL><DEDENT>with self.lock:<EOL><INDENT>try:<EOL><INDENT>connections = self.connections[address]<EOL><DEDENT>except KeyError:<EOL><INDENT>connections = self.connections[address] = deque()<EOL><DEDENT>connection_acquisition_start_timestamp = perf_counter()<EOL>while True:<EOL><INDENT>for connection in list(connections):<EOL><INDENT>if connection.closed() or connection.defunct() or connection.timedout():<EOL><INDENT>connections.remove(connection)<EOL>continue<EOL><DEDENT>if not connection.in_use:<EOL><INDENT>connection.in_use = True<EOL>return connection<EOL><DEDENT><DEDENT>infinite_connection_pool = (self._max_connection_pool_size < <NUM_LIT:0> or<EOL>self._max_connection_pool_size == float("<STR_LIT>"))<EOL>can_create_new_connection = infinite_connection_pool or len(connections) < self._max_connection_pool_size<EOL>if can_create_new_connection:<EOL><INDENT>try:<EOL><INDENT>connection = self.connector(address)<EOL><DEDENT>except ServiceUnavailable:<EOL><INDENT>self.remove(address)<EOL>raise<EOL><DEDENT>else:<EOL><INDENT>connection.pool = self<EOL>connection.in_use = True<EOL>connections.append(connection)<EOL>return connection<EOL><DEDENT><DEDENT>span_timeout = self._connection_acquisition_timeout - (perf_counter() - connection_acquisition_start_timestamp)<EOL>if span_timeout > <NUM_LIT:0>:<EOL><INDENT>self.cond.wait(span_timeout)<EOL>if self._connection_acquisition_timeout <= (perf_counter() - connection_acquisition_start_timestamp):<EOL><INDENT>raise ClientError("<STR_LIT>".format(<EOL>self._connection_acquisition_timeout))<EOL><DEDENT><DEDENT>else:<EOL><INDENT>raise ClientError("<STR_LIT>".format(self._connection_acquisition_timeout))<EOL><DEDENT><DEDENT><DEDENT>
Acquire a connection to a given address from the pool.
        The address supplied should always be an IP address, not
        a host name.

        This method is thread safe.