Петля через изменяющийся набор данных с inlineCallbacks/yield (python-twisted)

Question

У меня есть функция defer.inlineCallback для постепенного обновления большого (> 1k) списка за один раз. Этот список может измениться в любое время, и я получаю ошибки из-за этого поведения.

Самое простое представление о том, что я делаю это: -

@defer.inlineCallbacks 
def _get_details(self, dt=None): 
    data = self.data 
    for e in data: 
     if needs_update(e): 
      more_detail = yield get_more_detail(e) 
      do_the_update(e, more_detail) 
    schedule_future(self._get_details) 

self.data список словарей, который первоначально заполняется основной информацией (например, имя и ID) при запуске приложения. _get_details будет запускаться всякий раз, когда разрешается реактором, чтобы получить более подробную информацию по каждому элементу данных, обновив элемент по мере его продвижения.

Это хорошо работает, когда self.data не изменяется, но после его изменения (может быть в любой точке) цикл, очевидно, ссылается на неверную информацию. Фактически в этой ситуации было бы лучше просто полностью остановить цикл.

Я могу установить флаг в моем классе (который inlineCallback может проверить) при изменении данных.

1. Куда должна проводиться эта проверка?
2. Как выполняется код inlineCallback по сравнению с обычным deferred (и даже с обычным генератором питона).
3. Выполняется ли выполнение кода каждый раз, когда он встречает yield (т. Е. Могу ли я полагаться на этот код между одним yield и рядом с атомарным)?
4. В случае ненадежных больших списков, должен ли я даже перебирать данные (for e in data), или есть лучший способ?

Answer 1

@defer.inlineCallback 
def _get_details(self, dt=None): 
    data = self.data 
    i = 0 
    while i < len(data): 
     e = data[i] 
     if needs_update(e): 
      more_detail = yield get_more_detail(e) 
      if i < len(data) or data[i] != e: 
       break 
      do_the_update(e, more_detail) 
     i += 1 
    schedule_future(self._get_details) 

Основываясь на более тестирования, следующие мои наблюдения.

1. for e in data перебирает элементы, с элементом еще существует, даже если сами данные не как до, так и после yield заявления.

2. Насколько я могу судить, выполнение является атомарным между одним yield и следующим.

3. Проникновение данных осуществляется более прозрачно с помощью счетчика. Это также позволяет проверить, изменились ли данные. Проверка может быть выполнена в любое время после yield, потому что все изменения должны были произойти до того, как возвращается yield. Это приводит к приведенному выше коду.

Answer 2

Закрученный реактор никогда не упреждает ваш код во время его выполнения - вы должны добровольно уступить реактору, вернув значение. Вот почему так страшно писать Twisted-код, который блокирует операции ввода-вывода, потому что реактор не может планировать какие-либо задачи, пока вы ждете своего диска.

Итак, короткий ответ: да, выполнение является атомарным между выходами.

Без @inlineCallbacks функция _get_details возвращает генератор.Аннотации @inlineCallbacks просто обертывают генератор в Отложенном, который обходит генератор, пока не достигнет исключения StopIteration или исключения defer.returnValue. Когда любое из этих условий достигнуто, inlineCallbacks запускает свою Отсрочку. На самом деле это очень умно.

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

Answer 3

self.data список словарей ..., когда она меняется (может быть в любой момент) цикл, очевидно, относится к неправильной информации

Если вы изменяете список, пока вы итерацию он, как сказал бы Раймонд Хеттингер «Ты живешь в земле греха, и ты заслуживаешь всего, что с тобой происходит». :) Сценарии, подобные этому, следует избегать или список должен быть неизменным. Чтобы обойти эту проблему, вы можете использовать объект self.data.pop() или DeferredQueue для хранения данных. Таким образом, вы можете добавлять и удалять элементы в любое время, не вызывая побочных эффектов. Пример со списком:

@defer.inlineCallbacks 
def _get_details(self, dt=None): 
    try: 
     data = yield self.data.pop() 
    except IndexError: 
     schedule_future(self._get_details) 
     defer.returnValue(None)   # exit function 

    if needs_update(e): 
     more_detail = yield get_more_detail(data) 
     do_the_update(data, more_detail) 

    schedule_future(self._get_details) 

Посмотрите на DeferredQueue потому Deferred возвращается, когда функция get() называется, который вы можете цепи обратных вызовов для обработки каждого элемента вы выскочить из очереди.

Answer 4

Вам необходимо защитить доступ к общему ресурсу (self.data). Вы можете сделать это с помощью: twisted.internet.defer.DeferredLock.

http://twistedmatrix.com/documents/current/api/twisted.internet.defer.DeferredLock.html

Метод acquire

Попытка получить блокировку. Возвращает отложенную, которая срабатывает при блокировке с использованием значения DeferredLock. Если блокировка заблокирована, , то Отсрочка помещается в конец списка ожидания.

Метод release

Снимите блокировку. Если есть список ожидания, то первый Отсроченный в этом списке ожидания будет вызван обратно.