Python 3 asyncio với aioboto3 có vẻ tuần tự

Question

Tôi đang chuyển một tập lệnh python 3 đơn giản sang AWS Lambda. Tập lệnh rất đơn giản: nó thu thập thông tin từ hàng chục đối tượng S3 và trả về kết quả.

Tập lệnh được sử dụng multiprocessing.Poolđể tập hợp tất cả các tệp song song. Mặc dù multiprocessingkhông thể sử dụng trong môi trường AWS Lambda vì /dev/shmbị thiếu. Vì vậy, tôi nghĩ thay vì viết bẩn multiprocessing.Process/ multiprocessing.Queuethay thế, tôi sẽ thử asynciothay thế.

Tôi đang sử dụng phiên bản mới nhất của aioboto3(8.0.5) trên Python 3.8.

Vấn đề của tôi là dường như tôi không thể đạt được bất kỳ cải thiện nào giữa việc tải xuống tuần tự các tệp một cách ngây thơ và một vòng lặp sự kiện asyncio ghép các tệp đã tải xuống.

Đây là hai phiên bản mã của tôi.

import sys
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

import boto3
import aioboto3

BUCKET = 'some-bucket'
KEYS = [
    'some/key/1',
    [...]
    'some/key/10',
]

async def download_aio():
    """Concurrent download of all objects from S3"""
    async with aioboto3.client('s3') as s3:
        objects = [s3.get_object(Bucket=BUCKET, Key=k) for k in KEYS]
        objects = await asyncio.gather(*objects)
        buffers = await asyncio.gather(*[o['Body'].read() for o in objects])

def download():
    """Sequentially download all objects from S3"""
    s3 = boto3.client('s3')
    for key in KEYS:
        object = s3.get_object(Bucket=BUCKET, Key=key)
        object['Body'].read()

def run_sequential():
    download()

def run_concurrent():
    loop = asyncio.get_event_loop()
    #loop.set_default_executor(ProcessPoolExecutor(10))
    #loop.set_default_executor(ThreadPoolExecutor(10))
    loop.run_until_complete(download_aio())

Thời gian cho cả hai run_sequential()và run_concurrent()khá giống nhau (~ 3 giây cho hàng chục tệp 10MB). Tôi tin rằng phiên bản đồng thời không phải, vì nhiều lý do:

• Tôi đã thử chuyển sang Process/ThreadPoolExecutorvà tôi các tiến trình / luồng sinh ra trong suốt thời gian của hàm, mặc dù chúng không làm gì cả
• Thời gian giữa tuần tự và đồng thời rất gần giống nhau, mặc dù giao diện mạng của tôi chắc chắn không bão hòa và CPU cũng không bị ràng buộc
• Thời gian thực hiện của phiên bản đồng thời tăng tuyến tính với số lượng tệp.

Tôi chắc chắn cái gì đó còn thiếu, nhưng tôi không thể quấn lấy cái gì.

Có ý kiến ​​gì không?

Answer 1

Sau khi mất vài giờ để cố gắng hiểu cách sử dụng aioboto3chính xác, tôi quyết định chuyển sang giải pháp sao lưu của mình. Tôi đã kết thúc phiên bản ngây thơ của riêng mình multiprocessing.Poolđể sử dụng trong môi trường AWS lambda.

Nếu ai đó tình cờ gặp chủ đề này trong tương lai, nó đây. Nó còn lâu mới hoàn hảo, nhưng đủ dễ dàng để thay thế multiprocessing.Poolđối với các trường hợp đơn giản của tôi.

from multiprocessing import Process, Pipe
from multiprocessing.connection import wait


class Pool:
    """Naive implementation of a process pool with mp.Pool API.

    This is useful since multiprocessing.Pool uses a Queue in /dev/shm, which
    is not mounted in an AWS Lambda environment.
    """

    def __init__(self, process_count=1):
        assert process_count >= 1
        self.process_count = process_count

    @staticmethod
    def wrap_pipe(pipe, index, func):
        def wrapper(args):
            try:
                result = func(args)
            except Exception as exc:  # pylint: disable=broad-except
                result = exc
            pipe.send((index, result))
        return wrapper

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, exc_traceback):
        pass

    def map(self, function, arguments):
        pending = list(enumerate(arguments))
        running = []
        finished = [None] * len(pending)
        while pending or running:
            # Fill the running queue with new jobs
            while len(running) < self.process_count:
                if not pending:
                    break
                index, args = pending.pop(0)
                pipe_parent, pipe_child = Pipe(False)
                process = Process(
                    target=Pool.wrap_pipe(pipe_child, index, function),
                    args=(args, ))
                process.start()
                running.append((index, process, pipe_parent))
            # Wait for jobs to finish
            for pipe in wait(list(map(lambda t: t[2], running))):
                index, result = pipe.recv()
                # Remove the finished job from the running list
                running = list(filter(lambda x: x[0] != index, running))
                # Add the result to the finished list
                finished[index] = result

        return finished