3313 payload works with io asynchronously (#3478)

* I/O based payload works with I/O asynchronously

* I/O based payload works with I/O asynchronously

CHANGES/3313.feature

@@ -1 +1,2 @@
 FileResponse from web_fileresponse.py uses a ThreadPoolExecutor to work with files asynchronously.
+I/O based payloads from payload.py uses a ThreadPoolExecutor to work with I/O objects asynchronously.

aiohttp/payload.py

@@ -1,3 +1,4 @@
+import asyncio
 import enum
 import io
 import json
@@ -294,13 +295,18 @@ def __init__(self,
             self.set_content_disposition(disposition, filename=self._filename)
 
     async def write(self, writer: AbstractStreamWriter) -> None:
+        loop = asyncio.get_event_loop()
         try:
-            chunk = self._value.read(DEFAULT_LIMIT)
+            chunk = await loop.run_in_executor(
+                None, self._value.read, DEFAULT_LIMIT
+            )
             while chunk:
                 await writer.write(chunk)
-                chunk = self._value.read(DEFAULT_LIMIT)
+                chunk = await loop.run_in_executor(
+                    None, self._value.read, DEFAULT_LIMIT
+                )
         finally:
-            self._value.close()
+            await loop.run_in_executor(None, self._value.close)
 
 
 class TextIOPayload(IOBasePayload):
@@ -339,13 +345,18 @@ def size(self) -> Optional[float]:
             return None
 
     async def write(self, writer: AbstractStreamWriter) -> None:
+        loop = asyncio.get_event_loop()
         try:
-            chunk = self._value.read(DEFAULT_LIMIT)
+            chunk = await loop.run_in_executor(
+                None, self._value.read, DEFAULT_LIMIT
+            )
             while chunk:
                 await writer.write(chunk.encode(self._encoding))
-                chunk = self._value.read(DEFAULT_LIMIT)
+                chunk = await loop.run_in_executor(
+                    None, self._value.read, DEFAULT_LIMIT
+                )
         finally:
-            self._value.close()
+            await loop.run_in_executor(None, self._value.close)
 
 
 class BytesIOPayload(IOBasePayload):