reijii567 17 янв 2021 в 22:37

Многопоточное скачивание файлов с ftp python-скриптом

6 мин

16K

Python*

Из песочницы

Комментарии 24

Dair_Targ 17 янв 2021 в 23:07

https://github.com/aio-libs/aioftp
Для подобных операций в питоне asyncio зачастую на порядок быстрее многопоточности при существенно меньшем потреблении ресурсов.

Sun_123 18 янв 2021 в 09:01

Несмотря на утверждение автора «увы, ничего готового работающего так же быстро для моих условий найти не удалось» у меня есть приложение, написанное на асинкио (поскольку он лучше подходит для большого числа I/O_операций, да): github.com/Sunlight-Rim/FTPSearcher
Оно индексирует файлы и сохраняет в список, но там парой строчек автор мог бы добавить функцию скачивания. Однако статья энивей познавательная.

Sun_123 29 янв 2021 в 08:37

Кстати, обновил, добавив возможность рекурсивного выкачивания содержимого.

Dasdy 17 янв 2021 в 23:13

Если хочется всё-таки написать что-то своё, то есть несколько замечаний:

1) Не смотрели в сторону ThreadPoolExecutor? Вручную управлять полноценными потоками — неблагодарное занятие, довольно много граблей. Должно упростить код и избавить от некоторых проблем, которые могут внезапно сейчас возникнуть (например, глобальный доступ всех потоков к переменной thread_count, еще и каждый поток её менять может*)

2) Так как у вас написана обёртка FTP, вы не сможете её использовать как контекстный менеджер. Это в целом еще ладно, но с таким подходом надо следить за правильным закрытием клиента — например, если при получении файла вылетит любая ошибка, кроме error_perm

3) Обычно принято называть методы get_* тогда, когда они что-нибудь да возвращают. В целом, логичнее было бы его вообще спрятать внутри класса, и вызывать только при вызове get_next_file (который можно сделать генератором)

* Не используйте __del__, по возможности, никогда. Вы не контролируете момент, когда обьект будет удаляться.

a-l-e-x 17 янв 2021 в 23:17

я не очень большой специалист…
что произойдёт, если соединение (или много соединений) разрывается (по любой причине) в процессе скачивания?
можно ли один (например большой) файл скачивать параллельно (одновременно в несколько потоков)?
что произойдёт, в случае ошибки в основном скрипте (цикле), а не в скачивающем потоке?

gecube 17 янв 2021 в 23:19

Да тут даже веселее — если в процессе скачивания появляются новые файлы или изменяются уже существующие, то приплыли?

reijii567 18 янв 2021 в 07:56

Глобально вы правы, но конкретно в моей ситуации такое не может произойти, потому не предусмотрено.

Sergey_datex 18 янв 2021 в 00:02

По логике многопоточное копирование всегда медленней однопоточного (в масштабах, к примеру, одного ПК). При многопотоке дисковой подсистеме приходится одновременно отрабатывать запись/чтение из разных мест носителя, что суммарно медленней последовательного копирования тех же файлов. В чем преимущество многопоточности в FTP? Изначально ограничения в скорости одного потока? Тогда нужно копать в сторону устранения этого узкого места.

pansa 18 янв 2021 в 00:14

Во всех актуальных ОС (почти)весь io кэшируется в памяти.
ssd решает =)

Sergey_datex 18 янв 2021 в 00:21

1) вы все равно упретесь в дисковую подсистему. Сколько не кешируй, но данные нужно будет сбрасывать на диск. Если пишешь в 10 потоков — проиграешь в сумме.
2) в цепочке vps-комп автора, да и с оглядкой на объем бекапа (500 гиг) с большой вероятностью есть обычный HDD.

-1

gecube 18 янв 2021 в 00:28

Не согласен. Во многих случаях не дисковая подсистема является узким местом, а сам протокол фтп или сеть.

pansa 18 янв 2021 в 01:10

Ну, понятно, можно специально собрать из барахла очень тормозной массив, а сеть поднять 10GE, но мы же говорим о вменяемых конфигурациях. Если сервер удаленный, а не в локалке, то наиболее вероятно именно затыки сети будут батлнеком.
"обычный" современный hdd выдаст 80-100мб/с линейной записи, к которой как раз будет стремиться процесс сброса кэша. Я был бы рад, если бы мой бэкап с удалённого хоста упирался не в сеть, а в 80мб/с локальной записи =)

ilyakruchinin 18 янв 2021 в 01:58

lftp и не нужно никакое велосипедостроение
Документация: https://lftp.yar.ru/lftp-man.html

S0mbre 18 янв 2021 в 03:50

Уже отмечали, но все же — мои «5 копеек»:
1) Гораздо лучше использовать вместо самописного пула потоков готовое решение на python: самое простое — это Pool из пакета multiprocessing. По умолчанию, многозадачность реализована за счет параллельных процессов, что лучше, чем потоки, т.к. не накладывает ограничений на ядра процессора. А еще лучше — Dask, который позволяет делать намного больше!
2) Зачем использовать FTP.retrbinary, передавая в callback метод write файлового потока, когда гораздо эффективнее сразу использовать FTP.storbinary, который сразу сохраняет блоки данных в указанный поток? Сигнатура метода:

FTP.storbinary(cmd, fp, blocksize=8192, callback=None, rest=None)

Т.е. вы можете в callback теперь передавать свою процедуру для отслеживания самого процесса скачивания (например, перерисовка progressbar в каком-то интерфейсе или возможность ранней отмены ненужной закачки). А то у вас только можно отследить начало и конец скачивания файла.
3) Ну и наконец… А нужен ли python для этого? Я имею в виду, чисто скачивать с сервера файлы и писать лог. Честно, я бы поискал и нашел какой-то готовый клиент с поддержкой нужного функционала.

VD42 18 янв 2021 в 09:02

STOR — это закачать файл на сервер.

S0mbre 20 янв 2021 в 03:13

Спасибо, да, попутал! Надо все-таки использовать RETR, т.е. FTP.retrbinary.
Откорректировал коммент:
В callback при вызове FTP.retrbinary лучше использовать не file.write, а самописную функцию, позволяющую отслеживать прогресс скачивания и прерывать его при необходимости, например:

def download_callback(self, data):
    # пишем файл (self.fstream - член класса, объект файлового потока)
    self.fstream.write(data)
    # сколько закачано?
    self.downloaded_sz = self.fstream.tell()
    # вызываем метод отображения прогресса
    # и выбрасываем собственное исключение если юзер хочет прервать
    if not self.show_progress(): raise UserStopDownload(self) 

def show_progress(self):
    # отображение прогресса закачки
    # пользуемся переменными класса (filename, downloaded_sz, FTP.size до закачки)
    return True # False если хотим отменить

def run(self):
    # ...
    try:
        # ...       
        self.ftp.retrbinary(self.command + self.filename, self.download_callback)
        # ... 
    except:
        # остановить закачку...

reijii567 18 янв 2021 в 14:27

Под ограничением на ядра Вы имеете ввиду ограничения в связи с одним GIL?

S0mbre 20 янв 2021 в 03:09

И это, и факт ограниченного количества ядер, которое предполагает ограничение реально одновременно выполняемых потоков.

ALexhha 18 янв 2021 в 12:49

А чем не устроил axel?

reijii567 18 янв 2021 в 14:07

Если я не ошибся, то он умеет примерно тоже, что и wget — скачать указанный файл. Т.е. к axel пришлось бы точно так же писать скрипт, который будет получать список файлов с сервера и запускать загрузку каждого из них. В чём тогда профит от его использования, если он заменяет собой одну команду в этом скрипте?

ALexhha 18 янв 2021 в 14:43

Он умеет многопоточное скачивание, wget вроде такое не умеет

$ axel -a ftp://speedtest.tele2.net/10GB.zip
Initializing download: ftp://speedtest.tele2.net/10GB.zip
File size: 10737418240 bytes
Opening output file 10GB.zip
Starting download

Connection 1 finished                                                          ]
Connection 2 finished                                                          ]
Connection 0 finished                                                          ]
[100%] [..................................................] [ 255.4MB/s] [00:00]

Downloaded 10.0 Gigabyte in 40 seconds. (261544.98 KB/s)

$ axel -n 5 -a ftp://speedtest.tele2.net/10GB.zip
Initializing download: ftp://speedtest.tele2.net/10GB.zip
File size: 10737418240 bytes
Opening output file 10GB.zip
Starting download

Connection 2 finished                                                          ]
Connection 1 finished                                                          ]
Connection 3 finished                                                          ]
Connection 4 finished                                                          ]
Connection 0 finished                                                          ]

Downloaded 10.0 Gigabyte in 31 seconds. (330568.19 KB/s)

Так же есть aria2

$ aria2c -x4 -i list.txt

Т.е. в чем ценность написания скрипта на питоне, когда уже есть готовые и проверенные решения?

Tishka17 19 янв 2021 в 11:36

ВСЕ bмпорты переместить наверх соответствующих файлов
Логирование настраивать в main один раз. Код, который выполняет основную работу не должен знать куда там пишутся логи, он просто использует один раз созданный для него logger (один из немногих случаев, когда можно сделать глобальную переменную)
Выкинуть MyLogger, разобраться с иерархией логгеров, хэндлерами и прочим, что уже есть в logging
Вероятно выкинуть self.ftp.__class__.encoding = sys.getfilesystemencoding(). Это оооочень подозрительный код. Может сломать работу другого кода, использующего ftp библиотеку
Убрать наследование от Thread. Это нужно бывает когда вы делаете свою логику многозадачности (таймеры, универсальные воркеры). Бизнес логику в треде запускают через задание target
Выкинуть global. Константы и так прекрасно ищутся в глобальном скоупе.
Конфиг грузит из конфигурационного файла или переменных окружения. Когда вы упакуете программу как положено, пользователь не сможет править её код, он будет просто её устанавливать.
Для ограничения числа одновременных закачек воспользоваться ThreadPoolExecutor