Почему ThreadPool не двигается рядом после TimeoutError

Question

Моя цель:

• Чтобы пройти через список веб-сайтов, чтобы проверить их с помощью запросов. Это делается в apply_job.

Моей проблема:

• Когда job_pool.next называется, несколько веб-сайтов по ошибке, и вместо того, чтобы давать ошибку, они просто стоят и даже не дают TimeoutError. Вот почему я использую тайм-аут в функции next с 10 таймаутом. Этот тайм-аут работает хорошо, но когда возникает исключение TimeoutError, функция next в следующий раз продолжает поднимать исключение, даже если следующие веб-сайты хороши. Мне кажется, что он не переходит к следующему элементу и просто петлю над одним и тем же.
• Я пробовал с imap и imap_unordered, никакой разницы в этом.

Мой код здесь:

def run_check(websites): 
     """ Run check on the given websites """ 
     import multiprocessing 
     from multiprocessing.pool import ThreadPool 

     pool = ThreadPool(processes=JOB_POOL_SIZE) 

     try: 
      job_pool = pool.imap_unordered(apply_job, websites) 

      try: 
       while True: 
        try: 
         res = job_pool.next(10) 
        except multiprocessing.TimeoutError: 
         logging.error("Timeout Error") 
         res = 'No Res' 

        csv_callback(res) 

      except StopIteration: 
       pass 

      pool.terminate() 
     except Exception, e: 
      logging.error("Run_check Error: %s"%e) 
      raise 

Я использую res = requests.get(url, timeout=10) для проверки веб-сайтов. Этот тайм-аут не работает для этой проблемы.

Чтобы проверить, вот те сайты, которые создают проблему (не каждый раз, но очень часто): , http://www.railcar.netfirms.com.

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

Если у кого-то есть идея, было бы признательно, что я застрял на этом в течение нескольких дней. Я даже попробовал future и async, но они не поднимают исключение, которое мне нужно.

Спасибо, ребята!

Answer 1

Я нашел решение для своей проблемы, я использовал eventlet и его функцию Timeout.

def apply_job(account_info): 
    """ Job for the Thread """ 
    try: 
     account_id = account_info['id'] 
     account_website = account_info['website'] 
     url = account_website 
     result = "ERROR: GreenPool Timeout" 
     with Timeout(TIMEOUT*2, False): 
      url, result = tpool.execute(website.try_url, account_website) 

     return (account_id, account_website, url, result) 

    except Exception, e: 
     logging.error("Apply_job Error: %s"%e) 

def start_db(res): 
    update_db(res) 
    csv_file.csv_callback(res) 

def spawn_callback(result): 
    res = result.wait() 
    tpool.execute(start_db, res) 

def run_check(websites): 
    """ Run check on the given websites """ 
    print str(len(websites)) + " items found\n" 

    pool = eventlet.GreenPool(100) 
    for i, account_website in enumerate(websites): 
     res = pool.spawn(apply_job, account_website) 
     res.link(spawn_callback) 

    pool.waitall() 

Это решение работает хорошо, потому что он разы-над всем выполнением функции website.try_url в команде url, result = tpool.execute(website.try_url, account_website).

Answer 2

Ваша интуиция, что переход на тайм-аут на next прервал бы работу, неверен. Он просто прерывает ожидание, но конкретная работа продолжает работать. В следующий раз, когда вы ждете, вы ждете той же работы. Чтобы достичь тайм-аута на реальных рабочих местах, вы должны посмотреть на . Обратите внимание, что нет надежного способа завершения другого потока. Поэтому, если вы абсолютно не можете прекратить работу в течение разумного периода времени, вы можете переключиться на пул, основанный на процессе, и принудительно завершить процессы (например, используя signal.alarm).