async def connect(
        self, req: "ClientRequest", traces: List["Trace"], timeout: "ClientTimeout"
    ) -> Connection:
        """Get from pool or create new connection."""
        key = req.connection_key
        available = self._available_connections(key)

        # Wait if there are no available connections or if there are/were
        # waiters (i.e. don't steal connection from a waiter about to wake up)
        if available <= 0 or key in self._waiters:
            fut = self._loop.create_future()

            # This connection will now count towards the limit.
            self._waiters[key].append(fut)

            if traces:
                for trace in traces:
                    await trace.send_connection_queued_start()

            try:
                await fut
            except BaseException as e:
                if key in self._waiters:
                    # remove a waiter even if it was cancelled, normally it's
                    #  removed when it's notified
                    try:
                        self._waiters[key].remove(fut)
                    except ValueError:  # fut may no longer be in list
                        pass

                raise e
            finally:
                if key in self._waiters and not self._waiters[key]:
                    del self._waiters[key]

            if traces:
                for trace in traces:
                    await trace.send_connection_queued_end()

        proto = self._get(key)
        if proto is None:
            placeholder = cast(ResponseHandler, _TransportPlaceholder())
            self._acquired.add(placeholder)
            self._acquired_per_host[key].add(placeholder)

            if traces:
                for trace in traces:
                    await trace.send_connection_create_start()

            try:
                proto = await self._create_connection(req, traces, timeout)
                if self._closed:
                    proto.close()
                    raise ClientConnectionError("Connector is closed.")
            except BaseException:
                if not self._closed:
                    self._acquired.remove(placeholder)
                    self._drop_acquired_per_host(key, placeholder)
                    self._release_waiter()
                raise
            else:
                if not self._closed:
                    self._acquired.remove(placeholder)
                    self._drop_acquired_per_host(key, placeholder)

            if traces:
                for trace in traces:
                    await trace.send_connection_create_end()
        else:
            if traces:
                # Acquire the connection to prevent race conditions with limits
                placeholder = cast(ResponseHandler, _TransportPlaceholder())
                self._acquired.add(placeholder)
                self._acquired_per_host[key].add(placeholder)
                for trace in traces:
                    await trace.send_connection_reuseconn()
                self._acquired.remove(placeholder)
                self._drop_acquired_per_host(key, placeholder)

        self._acquired.add(proto)
        self._acquired_per_host[key].add(proto)
        return Connection(self, key, proto, self._loop)
Get from pool or create new connection.
lib/aiohttp/connector.py