지정된 시간 후 루프 위치 다시 시도

Question

루프의 일부가 하위 프로세스 인 지정된 시간 후에 완료되지 않은 경우 루프를 다시 시작하는 방법이 있습니까?

subprocess.Popen()을 사용하여 다른 작업자 스크립트의 X 인스턴스를 생성하는 main.py 스크립트가 있습니다. 각 "작업자"는 본질적으로 작업 (각기 다른 기능과 작업을 제공하는 작업자와 대기열)에 대해 Azure에서 호스팅되는 각각의 대기열을 확인합니다.

문제는 andy 작업자 (andy.py)가 가끔 함수를 호출하는 while 루프의 특정 부분에서 멈추는 경향이 있다는 것입니다. 나는 SIGALRM을 사용하여 어떤 일을 저지하려고 시도했는데, 이는 단순히 pass을 호출하는 예외를 발생시킨다. signal.alarm()은 시도를 중단하고 while 루프에 있기 때문에 다시 검색을 시도합니다.

알람은 알람이 시작될 때와 마찬가지로 실행중인 완전히 다른 하위 프로세스에도 영향을 미치며 때로는 실행중인 작업을 방해합니다. 내가 원하는 것은 함수가 완료하는 데 X 초 이상 걸리는 경우 함수를 다시 실행하려고하는 것입니다. 여기

코드가 어떻게 생겼는지의 예입니다 (사람을위한 runable이다 부여 기능은 코드로 대체되었고, 나는 andy 이외의 근로자 실행하는 모든 기능을 제거했습니다) :

을 의 main.py

import subprocess as sp 
import sys 
import time 
import datetime 
import thread 


max_workers = {'andy': 10} 


def check(): 
    workers = {'andy': {}} 

    while True: 
     for worker, instances in workers.items(): 
      while len(instances) < max_workers[worker]: 
       process = sp.Popen(['python', 'workers/%s.py' % worker], shell=False) 
       workers[worker][process] = process.pid 
     for worker, instances in workers.items(): 
      for process, pid in instances.items(): 
       if process.poll() is not None: 
        del workers[worker][process] 


def time_check(): 
    global max_workers 
    start = datetime.time(hour=07, minute=05) 
    end = datetime.time(hour=23, minute=00) 
    while 1: 
     now = datetime.datetime.now().time().replace(second=0, microsecond=0) 
     if now == start: 
      time.sleep(60) 
      max_workers['andy'] = 7 
     elif now == end: 
      time.sleep(60) 
      max_workers['andy'] = 0 
     else: 
      time.sleep(1) 


if __name__ == "__main__": 
    while 1: 
     try: 
      thread.start_new_thread(check,()) 
      thread.start_new_thread(time_check,()) 
     except KeyboardInterrupt: 
      sys.exit(0) 

andy.py

import datetime 
import otas 
import json 
import time 
import signal 


def alarm_handler(): 
    pass 


def start(): 

    resort_ids = 'Los Angeles', 'New York', 'Chicago', 'Miami' 
    start_date = datetime.datetime.now() 
    end_date = start_date + datetime.timedelta(days=10) 
    ota = otas.Expedia(headless=False) 
    signal.signal(signal.SIGALRM, alarm_handler) 
    for resort_id in resort_ids: 
     search_date = start_date 
     while search_date < end_date: 
      signal.alarm(15) 
      try: 
       data = ota.search_by_date(resort=resort_id, checkin=search_date) 
      except: 
       pass 
      else: 
       try: 
        print data 
       except TypeError: 
        pass 
       search_date += datetime.timedelta(days=1) 


if __name__ == '__main__': 
    start() 

otas.py

from selenium import webdriver 
import datetime 


class Expedia: 
    def __init__(self, headless=True): 
     if headless is True: 
      self.driver = webdriver.PhantomJS() 
     else: 
      self.driver = webdriver.Firefox() 


    def search_by_date(self, resort, checkin, flexibility=4, nights=3): 
     driver = self.driver 
     try: 
      driver.get(
       'http://www.expedia.com/Hotel-Search?#&destination={0}&startDate={1}&endDate={2}'.format(
        resort, checkin.strftime("%m/%d/%Y"), (checkin + datetime.timedelta(days=1)).strftime("%m/%d/%Y") 
       ) 
      ) 
      return driver.page_source 
     except Exception, e: 
      return e 

EDIT4 : 재 작성 질문 코드는 사용자를 재현 할 수 있으며, 더 명확합니다.

Answer 1

나는 프로세스 특정 코드가없는 -하지만 여기에 내가 뭘하려하고 그것을 작동하는 것 같다 :

sp_worker.py

import time 
import signal 
try: 
    time.sleep(60) 
    print("finished") 
except: 
    print("got interupted") 

이 __main__

import subprocess as sp 
import signal 
proc1 = sp.Popen(['python', 'sp_worker.py'], stdout=sp.PIPE) 
proc2 = sp.Popen(['python', 'sp_worker.py'], stdout=sp.PIPE) 
proc1.send_signal(signal.SIGALRM) 
proc1.communicate() 
(b'got interupted\n', None) 
proc2.communicate() #(blocks) 

Answer 2

여러 문제가 있습니다 귀하의 코드입니다. 무한 수의 스레드를 시작하는 단편은 다음과 같습니다.

#XXX BROKEN, DO NOT DO IT 
while 1: 
    try: 
     thread.start_new_thread(check,()) 
     thread.start_new_thread(time_check,()) 
    except KeyboardInterrupt: 
     sys.exit(0) 

OS 리소스가 유한합니다. 메모리가 다른 리소스보다 먼저 소모 된 경우 (각 스레드가 스택을 필요로 함); 일부 OS는 관련없는 프로세스 (OOM Killer)를 죽일 수도 있습니다.

귀하의 의도는 같은 시간에 반복적으로 두 가지 기능을 실행하는 데 아마도 :

from multiprocessing.pool import ThreadPool 

pool = ThreadPool(2) 
while True: 
    try: 
     r = pool.apply_async(check) 
     pool.apply(time_check) # block until time_check() returns 
     r.get() # block until check() returns 
    except KeyboardInterrupt: 
     break 

랩 check(), time_check() 기능 잡아 모든 예외를 기록 할 수 있습니다.