Задание APScheduler не запускается из-за события

Question

У меня есть некоторый код Python, в котором задание APScheduler не запускается. В качестве контекста у меня также есть обработчик, который ищет каталог для изменений файла, кроме того, используя eventlet/GreenPool для многопоточности. Основываясь на некотором устранении неполадок, кажется, что существует какой-то конфликт между APScheduler и eventlet.

Мой вывод выглядит следующим образом:

2016-12-26 02:30:30 UTC (+0000): Закончено Скачать Пропуск
2016-12-26 02:46:07 UTC (+0000): ВЫХОД из-за контроля-C или другого сигнала выхода
Jobstore default:
Загрузка с активацией по времени (триггер: интервал [0:05:00], следующий запуск в: 2016-12-25 18:35:00 PST) 2016-12-26 02:46:07 UTC (+0000): 1

(18:35 PST = 02:35 UTC)... так что он должен был выстрелить за 11 минут до того, как я нажал Control-C

from apscheduler import events ## pip install apscheduler
from apscheduler.schedulers.background import BackgroundScheduler

# Threading
from eventlet import patcher, GreenPool ## pip install eventlet
patcher.monkey_patch(all = True)

def setSchedule(scheduler, cfg, minutes = 60*2, hours = 0):
  """Set up the schedule of how frequently a download should be attempted.
  scheduler object must already be declared.
  will accept either minutes or hours for the period between downloads"""
  if hours > 0:
    minutes = 60*hours if minutes == 60 else 60*hours+minutes
  handle = scheduler.add_job(processAllQueues,
                            trigger='interval',
                            kwargs={'cfg': cfg},
                            id='RQmain',
                            name='Time-Activated Download',
                            coalesce=True,
                            max_instances=1,
                            minutes=minutes,
                            start_date=dt.datetime.strptime('2016-10-10 00:15:00', '%Y-%m-%d %H:%M:%S') # computer's local time
  )
  return handle

def processAllQueues(cfg):
  SQSpool = GreenPool(size=int(cfg.get('GLOBAL','Max_AWS_Connections')))
  FHpool = GreenPool(size=int(cfg.get('GLOBAL','Max_Raw_File_Process')))
  arSects = []
  dGlobal = dict(cfg.items('GLOBAL'))
  for sect in filter(lambda x: iz.notEqualz(x,'GLOBAL','RUNTIME'),cfg.sections()):
    dSect = dict(cfg.items(sect)) # changes all key names to lowercase
    n = dSect['sqs_queue_name']
    nn = dSect['node_name']
    fnbase = "{}_{}".format(nn,n)
    dSect["no_ext_file_name"] = os.path.normpath(os.path.join(cfg.get('RUNTIME','Data_Directory'),fnbase))
    arSects.append(mergeTwoDicts(dGlobal,dSect)) # section overrides global
  arRes = []
  for (que_data,spec_section) in SQSpool.imap(doQueueDownload,arSects): 
    if que_data: fileResult = FHpool.spawn(outputQueueToFiles,spec_section,que_data).wait()
    else: fileResult = (False,spec_section['sqs_queue_name'])
    arRes.append(fileResult)
  SQSpool.waitall()
  FHpool.waitall()
  pr.ts_print("Finished Download Pass")
  return None

def main():
  cfgglob = readConfigs(cfgdir, datdir)
  sched = BackgroundScheduler()
  cron_job = setSchedule(sched, cfgglob, 5)  
  sched.start(paused=True)
  try:
    change_handle = win32file.FindFirstChangeNotification(cfgdir, 0, win32con.FILE_NOTIFY_CHANGE_FILE_NAME | win32con.FILE_NOTIFY_CHANGE_LAST_WRITE)
    processAllQueues(cfgglob)
    sched.resume() # turn the scheduler back on and monitor both wallclock and config directory.
    cron_job.resume()
    while 1:
      SkipDownload = False
      result = win32event.WaitForSingleObject(change_handle, 500)
      if result == win32con.WAIT_OBJECT_0: # If the WaitForSO returned because of a notification rather than error/timing out
        sched.pause() # make sure we don't run the job as a result of timestamp AND file modification
        while 1:
          try:
            win32file.FindNextChangeNotification(change_handle) # rearm - done at start because of the loop structure here
            cfgglob = None
            cfgglob = readConfigs(cfgdir,datdir)
            cron_job.modify(kwargs={'cfg': cfgglob}) # job_id="RQmain",
            change_handle = win32file.FindFirstChangeNotification(cfgdir, 0, win32con.FILE_NOTIFY_CHANGE_FILE_NAME | win32con.FILE_NOTIFY_CHANGE_LAST_WRITE) # refresh handle
            if not SkipDownload: processAllQueues(cfgglob)
            sched.resume()
            cron_job.resume()
            break
  except KeyboardInterrupt:
    if VERBOSE | DEBUG: pr.ts_print("EXITING due to control-C or other exit signal")
  finally:
    sched.print_jobs()
    pr.ts_print(sched.state)
    sched.shutdown(wait=False)

Если я закомментирую большую часть функции processAllQueues вместе с включенным в начало eventlet, он срабатывает соответствующим образом. Если я сохраню

from eventlet import patcher, GreenPool ## pip install eventlet
patcher.monkey_patch(all = True)

но закомментируйте processAllQueues вплоть до строки печати от второй до последней строки, он не запускает APScheduler, указывая, что существует проблема либо с импортом patcher и GreenPool, либо с оператором monkey_patch. Комментируя patcher.monkey_patch(all = True) заставляет это "работать" снова.

Кто-нибудь знает, какой альтернативный оператор monkey_patch будет работать в моих обстоятельствах?

Answer 1

У вас есть явный цикл событий, отслеживающий изменения файла. Это блокирует запуск цикла событий Eventlet. У вас есть два варианта:

• Блокировка вызовов (например, win32event.WaitForSingleObject()) в eventlet.tpool.execute()
• Бежать eventlet.sleep() до / после блокировки вызовов и убедитесь, что вы не блокируете слишком долго.

eventlet.monkey_patch(thread=False) более короткая альтернатива перечислению всех остальных модулей как истинных. Вообще хочешь thread=True при использовании блокировок, локального потока или API потоков для порождения зеленых потоков. Вы можете хотеть thread=False если вы действительно используете потоки ОС, как для забавных GUI-сред.

Вы не должны рассматривать Eventlet на Windows для запуска важных проектов. Производительность намного ниже, чем у POSIX. Я не запускал тесты на Windows с 0.17. Это скорее для простоты разработки на популярной настольной платформе.