Python Уведомлять, когда все файлы были переданы

Я использую API «сторожевого пса», чтобы постоянно проверять изменения в папке в моей файловой системе. Какие бы файлы ни изменялись в этой папке, я передаю их определенной функции, которая запускает потоки для каждого файла, который я им передаю.

Но сторожевой таймер или любой другой API-интерфейс наблюдателя за файловой системой (насколько мне известно) уведомляет пользователей файл за файлом, т.е. по мере поступления файлов они уведомляют пользователя. Но я хотел бы, чтобы он одновременно уведомлял меня о целой куче файлов, чтобы я мог передать этот список своей функции и использовать многопоточность. В настоящее время, когда я использую «сторожевой таймер», он уведомляет меня по одному файлу за раз, и я могу передать только этот файл своей функции. Я хочу передать ему много файлов за раз, чтобы иметь многопоточность.

Мне приходит в голову одно решение: вы видите, что когда вы копируете кучу файлов в папку, ОС показывает вам индикатор выполнения. Если бы я мог быть уведомлен, когда этот индикатор выполнения будет готов, то это было бы идеальным решением для моего вопроса. Но я не знаю, возможно ли это.

Также я знаю, что сторожевой таймер — это API опроса, и идеальным API для наблюдения за файловой системой будет API, управляемый прерываниями, такой как pyinotify. Но я не нашел никакого API, который был бы управляемым с помощью прерываний, а также кросс-платформенным. iWatch хорош, но только для Linux, а я хочу что-то для всех ОС. Итак, если у вас есть предложения по любому другому API, дайте мне знать.

Спасибо.


python-3.x filesystemwatcher python-watchdog
person Rash    schedule 04.10.2014    source источник
comment
Почему бы не создать пул воркеров, которые получают задачи из общей очереди, а затем создать сторожевой поток, который помещает задачи в очередь? Тогда, вроде бы, не пришлось бы ждать, пока наберется куча файлов, прежде чем ваши воркеры начнут работать.   -  person unutbu    schedule 04.10.2014
comment
Хм... на самом деле классное решение. Я должен был подумать, что .. спасибо человек. Я попробую это. Но на общем узле мне было интересно, можем ли мы получить уведомление, когда индикатор выполнения ОС завершится, как я упоминал в пункте 3 ??   -  person Rash    schedule 04.10.2014
comment
Индикатор выполнения зависит от ОС. Когда я копирую что-то с помощью /usr/bin/cp в Linux, индикатора выполнения нет. Таким образом, зависимость от индикатора выполнения не будет надежным кросс-платформенным решением.   -  person unutbu    schedule 04.10.2014
comment
хорошо, это имеет смысл. Мне нравится ваше решение. Но если я создам очередь, мне придется продолжать проверять, была ли внесена какая-либо новая запись в очередь или нет. Так что я все еще был бы в бесконечном цикле, чтобы продолжать проверять вещи. Я, очевидно, могу придумать некоторые обходные пути, но знаете ли вы какой-нибудь хороший способ, с помощью которого я могу инициировать событие, когда что-то добавляется в мою очередь. Я новичок в python и особо не искал.   -  person Rash    schedule 04.10.2014
comment
Метод очереди get блокируется до тех пор, пока не появится элемент, который нужно получить. Дуг Хеллман написал отличный набор руководств, которые помогут вам начать работу: по использованию Queue, модуль потоковой обработки, как настроить и использовать пул рабочих процессов, как настроить пул рабочих потоков.   -  person unutbu    schedule 04.10.2014
comment
Хорошо... спасибо, чувак. Если вы хотите, вы можете опубликовать резюме нашего обсуждения в ответе, и я отмечу его как правильный. Это меньшее, что я могу сделать, чтобы отблагодарить вас. :)   -  person Rash    schedule 04.10.2014

Ответы (1)


arrow_upward
2
arrow_downward

Вместо накопления событий файловой системы вы можете создать пул рабочих потоков, которые получают задачи из общей очереди. Затем сторожевой поток может помещать задачи в очередь по мере возникновения событий файловой системы. Таким образом, рабочий поток может начать работу, как только произойдет событие.

Например,

import logging
import Queue
import threading
import time
import watchdog.observers as observers
import watchdog.events as events

logger = logging.getLogger(__name__)

SENTINEL = None

class MyEventHandler(events.FileSystemEventHandler):
    def on_any_event(self, event):
        super(MyEventHandler, self).on_any_event(event)            
        queue.put(event)
    def __init__(self, queue):
        self.queue = queue

def process(queue):
    while True:
        event = queue.get()
        logger.info(event)


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG,
                        format='[%(asctime)s %(threadName)s] %(message)s',
                        datefmt='%H:%M:%S')

    queue = Queue.Queue()
    num_workers = 4
    pool = [threading.Thread(target=process, args=(queue,)) for i in range(num_workers)]
    for t in pool:
        t.daemon = True
        t.start()

    event_handler = MyEventHandler(queue)
    observer = observers.Observer()
    observer.schedule(
        event_handler,
        path='/tmp/testdir',
        recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

Бег

% mkdir /tmp/testdir
% script.py

дает вывод, как

[14:48:31 Thread-1] <FileCreatedEvent: src_path=/tmp/testdir/.#foo>
[14:48:32 Thread-2] <FileModifiedEvent: src_path=/tmp/testdir/foo>
[14:48:32 Thread-3] <FileModifiedEvent: src_path=/tmp/testdir/foo>
[14:48:32 Thread-4] <FileDeletedEvent: src_path=/tmp/testdir/.#foo>
[14:48:42 Thread-1] <FileDeletedEvent: src_path=/tmp/testdir/foo>
[14:48:47 Thread-2] <FileCreatedEvent: src_path=/tmp/testdir/.#bar>
[14:48:49 Thread-4] <FileCreatedEvent: src_path=/tmp/testdir/bar>
[14:48:49 Thread-4] <FileModifiedEvent: src_path=/tmp/testdir/bar>
[14:48:49 Thread-1] <FileDeletedEvent: src_path=/tmp/testdir/.#bar>
[14:48:54 Thread-2] <FileDeletedEvent: src_path=/tmp/testdir/bar>

Дуг Хеллман написал отличный набор руководств (которые теперь отредактированы в книга), которая поможет вам начать работу:

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

person unutbu    schedule 04.10.2014
comment
OMG ... это потрясающий ответ. Вы также дали мне советы и примеры кода, чтобы я мог более четко понять весь процесс. Огромное спасибо!! - person Rash; 04.10.2014