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

[Dair_Targ]

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

[Sun_123]

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

[Sun_123]

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

[Dasdy]

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

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

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

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

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

[a-l-e-x]

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

[gecube]

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

[reijii567]

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

[Sergey_datex]

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

[pansa]

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

[Sergey_datex]

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

[gecube]

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

[pansa]

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

[ilyakruchinin]

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

[S0mbre]

Уже отмечали, но все же — мои «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]

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

[S0mbre]

Спасибо, да, попутал! Надо все-таки использовать 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]

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

[S0mbre]

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

[ALexhha]

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

[reijii567]

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

[ALexhha]

Он умеет многопоточное скачивание, 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]

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

В общем, 90% кода показывать другим людям не стоит.

[stitrace]

А разве питоновский Global Interpreter Lock (GIL) не помножит все эти потуги на ноль?

[reijii567]

Замедлит, да. Но профит все равно есть. Несколько файлов качаются одновременно.