Блог

Я пытаюсь выяснить, почему приведенный ниже код выполняется в одно и то же время, будь то однопоточный или с использованием ThreadPoolExecutor. Моя лямбда-функция извлекает несколько файлов JSON из S3, все они размером примерно 2 кб. Для моего теста я использую 100 файлов, и на это уходит более 2 секунд, независимо от того, использую ли я ThreadPoolExecutor или однопоточный код. Количество рабочих потоков также не имеет никакого значения, поскольку я пробовал 10 и 25 с тем же результатом.

Это многопоточная версия:

     s3_data = {}
    logger.debug('Creating boto3 S3 client')
    # Make sure s3 client has a large enough connection pool
    s3_client = boto3.client('s3', config=botocore.config.Config(max_pool_connections=max_s3_threads))
    # This assumes boto3 clients are thread safe
    logger.debug(f'Starting parallel S3 data retrieval, max threads={max_s3_threads}')
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_s3_threads) as executor:
        s3_threads = {
            executor.submit(get_s3_data, s3_client, row['s3Key']): row['id']
            for row in rows.values() 
        }
        for s3_thread in concurrent.futures.as_completed(s3_threads):
            id = s3_threads[s3_thread]
            s3_data[ad_id] = s3_thread.result()
    logger.debug(f'Finished parallel S3 data retrieval, threads completed={len(s3_threads)}')
  

И это однопоточная версия:

     logger.debug('Starting sequential S3 data retrieval')
    s3_client = boto3.client('s3')
    for row in unique_artworks.values():
        s3_data[row['artworkId']] = get_s3_data(s3_client, row['s3Key'])
    logger.debug(f'Finished sequential S3 data retrieval')
  

get_s3_data Функция просто вызывает s3_client.get_object имя корзины, которое она получает из переменной среды и переданного ключа, и возвращает JSON в виде dict. Не очень сложно.

Этот код должен быть привязан к вводу-выводу, а не к процессору, поэтому я не думаю, что GIL мешает, основываясь на том, что я прочитал об этом. Один и тот же экземпляр клиентского объекта S3 используется всеми потоками, но предположительно это безопасно (я не вижу никаких шатких результатов в своих выходных данных). На всякий случай я попытался создать клиент S3 в вызываемой функции, но это еще медленнее. Я ожидал бы увидеть некоторую выгоду от использования ThreadPoolExecutor, но меня озадачивает, почему я этого не делаю.

Возможно, в моем коде что-то не так, или, может быть, где-то мне не хватает какого-то параметра. Я прогуглил массу сообщений, блогов и статей и до сих пор не нашел решения, поэтому я надеюсь, что кто-нибудь здесь сможет дать некоторое представление.