Python 3 asyncio กับ aioboto3 ดูเหมือนเป็นลำดับ

Question

ฉันกำลังโอนสคริปต์ python 3 อย่างง่ายไปยัง AWS Lambda สคริปต์นั้นเรียบง่าย: รวบรวมข้อมูลจากอ็อบเจ็กต์ S3 จำนวนโหลและส่งคืนผลลัพธ์

สคริปต์ที่ใช้multiprocessing.Poolในการรวบรวมไฟล์ทั้งหมดแบบขนาน แม้ว่าmultiprocessingจะไม่สามารถใช้ในสภาพแวดล้อม AWS Lambda ได้เนื่องจาก/dev/shmไม่มี ดังนั้นฉันจึงคิดว่าแทนที่จะเขียนสกปรกmultiprocessing.Process/ multiprocessing.Queueแทนที่ฉันจะลองasyncioแทน

ฉันใช้aioboto3(8.0.5) เวอร์ชันล่าสุดบน Python 3.8

ปัญหาของฉันคือดูเหมือนว่าฉันจะไม่ได้รับการปรับปรุงใด ๆ ระหว่างการดาวน์โหลดไฟล์ตามลำดับที่ไร้เดียงสาและการมัลติเพล็กซ์เหตุการณ์แบบ asyncio ซึ่งทำการดาวน์โหลด

นี่คือรหัสสองเวอร์ชันของฉัน

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())

เวลาสำหรับทั้งสองrun_sequential()และrun_concurrent()ค่อนข้างใกล้เคียงกัน (ประมาณ 3 วินาทีสำหรับไฟล์ 10MB โหล) ฉันเชื่อว่าเวอร์ชันพร้อมกันไม่ได้เกิดจากหลายสาเหตุ:

• ฉันพยายามเปลี่ยนไปใช้Process/ThreadPoolExecutorและฉันเกิดกระบวนการ / เธรดในช่วงระยะเวลาของฟังก์ชันแม้ว่าพวกเขาจะไม่ทำอะไรเลย
• เวลาระหว่างลำดับและการทำงานพร้อมกันนั้นใกล้เคียงกันมากแม้ว่าอินเทอร์เฟซเครือข่ายของฉันจะไม่อิ่มตัวอย่างแน่นอนและ CPU ก็ไม่ได้ถูกผูกไว้ด้วย
• เวลาที่ใช้ในเวอร์ชันพร้อมกันจะเพิ่มขึ้นตามจำนวนไฟล์ในเชิงเส้น

ฉันแน่ใจว่ามีบางอย่างขาดหายไป แต่ฉันก็ไม่สามารถคาดเดาอะไรได้

ความคิดใด ๆ ?

Answer 1

หลังจากเสียเวลาไปหลายชั่วโมงเพื่อพยายามทำความเข้าใจวิธีใช้aioboto3อย่างถูกต้องฉันตัดสินใจเปลี่ยนไปใช้โซลูชันสำรอง ฉันลงเอยด้วยการเปิดตัวเวอร์ชันไร้เดียงสาของตัวเองmultiprocessing.Poolเพื่อใช้ภายในสภาพแวดล้อม AWS lambda

หากมีคนสะดุดเจอกระทู้นี้ในอนาคตนี่คือ มันยังห่างไกลจากความสมบูรณ์แบบ แต่ก็ง่ายพอที่จะแทนที่multiprocessing.Poolได้สำหรับเคสง่ายๆของฉัน

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