Use dunder writer internally in ClientResponse

In #7815 the writer was wrapped with a setter to make
sure it was always reset on completion and maintain
compat for subclasses. This added a tiny bit of overhead
since we access self._writer in a lot of places. We can
access self.__writer instead since its all inside the class
which has less overhead.

aiohttp/client_reqrep.py

@@ -677,7 +677,7 @@ async def send(self, conn: "Connection") -> "ClientResponse":
         self.response = response_class(
             self.method,
             self.original_url,
-            writer=self._writer,
+            writer=self.__writer,
             continue100=self._continue,
             timer=self._timer,
             request_info=self.request_info,
@@ -688,9 +688,9 @@ async def send(self, conn: "Connection") -> "ClientResponse":
         return self.response
 
     async def close(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)
@@ -700,11 +700,11 @@ async def close(self) -> None:
                     raise
 
     def terminate(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             if not self.loop.is_closed():
-                self._writer.cancel()
-            self._writer.remove_done_callback(self.__reset_writer)
-            self._writer = None
+                self.__writer.cancel()
+            self.__writer.remove_done_callback(self.__reset_writer)
+            self.__writer = None
 
     async def _on_chunk_request_sent(self, method: str, url: URL, chunk: bytes) -> None:
         for trace in self._traces:
@@ -761,7 +761,7 @@ def __init__(
         self._real_url = url
         self._url = url.with_fragment(None)
         self._body: Optional[bytes] = None
-        self._writer: Optional[asyncio.Task[None]] = writer
+        self._writer = writer
         self._continue = continue100  # None by default
         self._closed = True
         self._history: Tuple[ClientResponse, ...] = ()
@@ -789,10 +789,16 @@ def __reset_writer(self, _: object = None) -> None:
 
     @property
     def _writer(self) -> Optional["asyncio.Task[None]"]:
+        """The writer task for streaming data.
+
+        _writer is only provided for backwards compatibility
+        for subclasses that may need to access it.
+        """
         return self.__writer
 
     @_writer.setter
     def _writer(self, writer: Optional["asyncio.Task[None]"]) -> None:
+        """Set the writer task for streaming data."""
         if self.__writer is not None:
             self.__writer.remove_done_callback(self.__reset_writer)
         self.__writer = writer
@@ -1038,16 +1044,16 @@ def raise_for_status(self) -> None:
 
     def _release_connection(self) -> None:
         if self._connection is not None:
-            if self._writer is None:
+            if self.__writer is None:
                 self._connection.release()
                 self._connection = None
             else:
-                self._writer.add_done_callback(lambda f: self._release_connection())
+                self.__writer.add_done_callback(lambda f: self._release_connection())
 
     async def _wait_released(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)
@@ -1058,8 +1064,8 @@ async def _wait_released(self) -> None:
         self._release_connection()
 
     def _cleanup_writer(self) -> None:
-        if self._writer is not None:
-            self._writer.cancel()
+        if self.__writer is not None:
+            self.__writer.cancel()
         self._session = None
 
     def _notify_content(self) -> None:
@@ -1070,9 +1076,9 @@ def _notify_content(self) -> None:
         self._released = True
 
     async def wait_for_close(self) -> None:
-        if self._writer is not None:
+        if self.__writer is not None:
             try:
-                await self._writer
+                await self.__writer
             except asyncio.CancelledError:
                 if (
                     sys.version_info >= (3, 11)