Avoid passing client writer to response when already finished (#9485)

(cherry picked from commit da0099d)

CHANGES/9485.misc.rst

@@ -0,0 +1 @@
+Improved performance of sending client requests when the writer can finish synchronously -- by :user:`bdraco`.

aiohttp/client_reqrep.py

@@ -355,17 +355,11 @@ def _writer(self) -> Optional["asyncio.Task[None]"]:
         return self.__writer
 
     @_writer.setter
-    def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
+    def _writer(self, writer: "asyncio.Task[None]") -> None:
         if self.__writer is not None:
             self.__writer.remove_done_callback(self.__reset_writer)
         self.__writer = writer
-        if writer is None:
-            return
-        if writer.done():
-            # The writer is already done, so we can reset it immediately.
-            self.__reset_writer()
-        else:
-            writer.add_done_callback(self.__reset_writer)
+        writer.add_done_callback(self.__reset_writer)
 
     def is_ssl(self) -> bool:
         return self.url.scheme in ("https", "wss")
@@ -779,6 +773,7 @@ async def send(self, conn: "Connection") -> "ClientResponse":
         await writer.write_headers(status_line, self.headers)
         coro = self.write_bytes(writer, conn)
 
+        task: Optional["asyncio.Task[None]"]
         if sys.version_info >= (3, 12):
             # Optimization for Python 3.12, try to write
             # bytes immediately to avoid having to schedule
@@ -787,7 +782,11 @@ async def send(self, conn: "Connection") -> "ClientResponse":
         else:
             task = self.loop.create_task(coro)
 
-        self._writer = task
+        if task.done():
+            task = None
+        else:
+            self._writer = task
+
         response_class = self.response_class
         assert response_class is not None
         self.response = response_class(
@@ -864,7 +863,7 @@ def __init__(
         method: str,
         url: URL,
         *,
-        writer: "asyncio.Task[None]",
+        writer: "Optional[asyncio.Task[None]]",
         continue100: Optional["asyncio.Future[bool]"],
         timer: BaseTimerContext,
         request_info: RequestInfo,
@@ -880,7 +879,8 @@ def __init__(
         self._real_url = url
         self._url = url.with_fragment(None) if url.raw_fragment else url
         self._body: Optional[bytes] = None
-        self._writer = writer
+        if writer is not None:
+            self._writer = writer
         self._continue = continue100  # None by default
         self._closed = True
         self._history: Tuple[ClientResponse, ...] = ()
@@ -924,8 +924,8 @@ def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
         if writer is None:
             return
         if writer.done():
-            # The writer is already done, so we can reset it immediately.
-            self.__reset_writer()
+            # The writer is already done, so we can clear it immediately.
+            self.__writer = None
         else:
             writer.add_done_callback(self.__reset_writer)