Задача APScheduler не срабатывает из-за eventlet monkey_patch

Question

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

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

2016-12-26 02:30:30 UTC (+0000): Закончено Скачать Pass
2016-12-26 2:46:07 (UTC + 0000): ВЫИГРЫВАНИЕ из-за контроля-C или другого сигнала выхода
Место работы по умолчанию:
Время активации (триггер: интервал [0:05:00], следующий прогон: 2016-12-25 18:35:00 PST) 2016-12-26 02:46:07 UTC (+0000): 1

(18:35 PST = 2:35 UTC) ... так что это должно было сожжено 11 минут, прежде чем я нажал Ctrl + 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, указывая, что есть либо проблема с импортом патчер и GreenPool или с инструкция monkey_patch. Комментируя patcher.monkey_patch(all = True), он снова «работает».

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

Answer 1

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

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

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

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