С многопоточностью я получаю 9 «попаданий», без многопоточности я получаю 214. Что происходит?

Основная цель моего скрипта — отфильтровать диапазон чисел (скажем, 5000), числа valid сохраняются в списке под названием hit_list. Реальный диапазон, который я просматриваю, намного больше 5000, поэтому мне нужен параллелизм, чтобы сделать время управляемым.

Я не знаю долю допустимых чисел в любом заданном диапазоне, поэтому, когда мой (потоковый) скрипт вернул 9 чисел в hit_list, я не задавался этим вопросом. Однако в качестве окончательной проверки я запустил скрипт без потоков, как обычный скрипт. Он вернул 214 чисел в hit_list!

РЕДАКТИРОВАТЬ: Чтобы быть ясным, проблема заключается в том, что числа не находятся правильно, а не сохраняются правильно.

Мне очень щедро помогли с созданием этой программы, как на SO, здесь и Reddit, здесь.

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

import concurrent.futures as cf
import requests
from bs4 import BeautifulSoup
import time
from datetime import datetime
import xlwt

hit_list =[]
print('-List Created')
startrange= 100000000
end_range = 100005000
startTime = datetime.now()
print(datetime.now())
url = 'https://ndber.seai.ie/pass/ber/search.aspx'

#print('Working...')

def id_filter(_range):

    with requests.session() as s:
        s.headers.update({
            'user-agent': 'For more information on this data collection please contact #########'
        })

        r = s.get(url)
        time.sleep(.5)


        soup = BeautifulSoup(r.content, 'html.parser')
        viewstate    = soup.find('input', {'name': '__VIEWSTATE'          }).get('value')
        viewstategen = soup.find('input', {'name': '__VIEWSTATEGENERATOR' }).get('value')
        validation   = soup.find('input', {'name': '__EVENTVALIDATION'    }).get('value')

        for ber in _range:            


            data = {
            'ctl00$DefaultContent$BERSearch$dfSearch$txtBERNumber': ber,
            'ctl00$DefaultContent$BERSearch$dfSearch$Bottomsearch': 'Search',
            '__VIEWSTATE'                                         : viewstate,
            '__VIEWSTATEGENERATOR'                                : viewstategen,
            '__EVENTVALIDATION'                                   : validation,
        }

            y = s.post(url, data=data)

            if 'No results found' in y.text:
                #print('Invalid ID Number')
                pass
            else:
                #print('Valid ID Number')
                hit_list.append(ber)




if __name__=='__main__': #not 100% clear on what exactly this does, but that's a lesson for another day.

#Using threads to call the function    
workers = 20
with cf.ThreadPoolExecutor(max_workers=workers) as e:

    IDs = range(startrange,end_range)
    cs = 20 
    ranges = [IDs[i+1 :i+cs] for i in range(-1, len(IDs), cs)]
    results = e.map(id_filter, ranges)
#below is code for saving the data to an excel file, I've left it out for parsimony.

person SeánMcK    schedule 25.05.2016    source источник
comment
Хиты найдены неправильно или неправильно сохранены? Можете ли вы воспроизвести проблему без использования каких-либо HTTP-запросов, а вместо этого использовать макеты? (Подсказка: тогда у вас нет минимального примера!) Кстати: код имеет неправильный отступ.   -  person Ulrich Eckhardt    schedule 25.05.2016
comment
Чтобы расширить комментарий @UlrichEckhardt: сначала проверьте, при необходимости, вручную, каков правильный ответ. Затем убедитесь, что ваша однопоточная версия дает этот ответ. Затем запустите свою многопоточную версию. Я предполагаю, что любые видимые различия будут вызваны плохой обработкой доступа с несколькими записями к глобальному.   -  person Peter Rowell    schedule 25.05.2016
comment
@UlrichEckhardt, спасибо за ваше предложение. Я отредактировал свой вопрос, чтобы прояснить проблему: это действительно проблема с поиском, а не с сохранением попаданий. Я проверил и подтвердил это, следуя совету Питера Роуэлла вручную проверить, является ли однопоточная версия (я делал это более 200 раз... это заняло некоторое время). Когда вы говорите манекены, вы имеете в виду multiprocessing.dummy? Я посмотрю на это сейчас. Я также исправил проблему с отступами.   -  person SeánMcK    schedule 26.05.2016
comment
@PeterRowell, я последовал вашему предложению и вручную ввел более 200 итераций. Проблема с поиском чисел, многопоточная версия их почему-то не находит. Я думаю, это говорит о том, что ваша теория неверна, как если бы я правильно понял, вы думали, что это может быть проблема с несколькими потоками, обращающимися к hit_list одновременно?   -  person SeánMcK    schedule 26.05.2016
comment
@sean_raven: я не тратил время на глубокий анализ вашего кода. Я знаю, что если версия А дает правильные результаты, а версия Б нет, то проблема кроется где-то в различиях между А и Б. Если с 1 потоком id_filter работает нормально (но медленно) на всем файле, но используя больше чем 1 поток дает неправильный результат, то либо фрагменты, которые вы передаете id_filter, не являются тем, чем вы их считаете, или есть какой-то другой аспект среды потоков, вызывающий проблемы. Работает лучше или хуже с 2 работниками, чем с 1? С 3, а не с 2? Образец появится.   -  person Peter Rowell    schedule 27.05.2016
comment
Привет, Питер, я ценю, что ты вернулся ко мне по этому поводу. Вчера я провел весь день, перестраивая код с нуля. Думаю, мне удалось получить рабочую версию. Сегодня я потрачу несколько часов на тестирование, надеюсь, это сработает! Еще раз спасибо (я, очевидно, опубликую здесь правильный код, как только подтвержу)   -  person SeánMcK    schedule 27.05.2016
comment
Что должно сообщать hit_list основной программе? map обычно работает с функциями, которые что-то возвращают. 9/214≈1/20 попаданий с 20 воркерами — это примерный показатель для одного из воркеров, работающих в основном контексте. Кстати, именно поэтому if __name__=='__main__' необходимо; остальные рабочие импортируют модуль на win32 (на posix они форкаются).   -  person Yann Vernier    schedule 27.05.2016
comment
@YannVernier, как ни странно, вчера я читал о том, что нужно что-то return делать. Но мой ответ ниже, кажется, может нормально работать, ничего не возвращая. Мысли?   -  person SeánMcK    schedule 27.05.2016


Ответы (1)


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

Сценарий, похоже, работает хорошо — он подбирает магическое число 214 hits в цикле 5000.

Хотя он отвечает на мой вопрос, код далек от совершенства, и я рад принять ответы, которые строят/улучшают его.

Не будучи уверенным в том, что такое этикет SO для задавания вопросов в ответах, я называю следующие необязательные темы для обсуждения (а не вопросы!):

  1. Мысли о requests надстройке: кажется, это гораздо более простая версия того, что я пытался сделать?
  2. Лучший способ оптимизировать эффективность потоков? Существуют ли какие-либо эмпирические правила относительно соотношения рабочих и чанков, или это просто метод проб и ошибок?
  3. Лучший способ raise exceptions при использовании requsts и threading?

Рабочее (но несовершенное) решение

import concurrent.futures as cf
import requests
from bs4 import BeautifulSoup
from datetime import datetime



hit_list =[]
processed_list=[]
startrange= 100000000
end_range = 100005000

loop_size=range(startrange,end_range)
workers= 5
chunks= 20
startTime = datetime.now()



def id_filter(_range):
    url = 'https://ndber.seai.ie/pass/ber/search.aspx'
    with requests.session() as s:
        s.headers.update({
            'user-agent': 'For more information on this data collection please contact ######'
        })    
        r = s.get(url)   

        soup = BeautifulSoup(r.content, 'html.parser')
        viewstate    = soup.find('input', {'name': '__VIEWSTATE'          }).get('value')
        viewstategen = soup.find('input', {'name': '__VIEWSTATEGENERATOR' }).get('value')
        validation   = soup.find('input', {'name': '__EVENTVALIDATION'    }).get('value')

        for ber in _range:            
            data = {
                'ctl00$DefaultContent$BERSearch$dfSearch$txtBERNumber': ber,
                'ctl00$DefaultContent$BERSearch$dfSearch$Bottomsearch': 'Search',
                '__VIEWSTATE'                                         : viewstate,
                '__VIEWSTATEGENERATOR'                                : viewstategen,
                '__EVENTVALIDATION'                                   : validation,
            }        
            y = s.post(url, data=data)            
            if 'No results found' in y.text:
                processed_list.append(ber)
                #print('Invalid ID', ber)
            else:
                hit_list.append(ber)
                processed_list.append(ber)
                print('Valid ID',ber)

if __name__ == '__main__':   
    with cf.ThreadPoolExecutor(max_workers=workers) as boss:
        jobs= [loop_size[x: x + chunks] for x in range(0, len(loop_size), chunks)]        
        boss.map(id_filter, jobs)

#note below is a simplified version of what I'm doing in real-life. In reality, I print out details similar to these and then save them in a .txt file. I left out all that for parsimony's sake.
print('Details')
run_time = datetime.now() - startTime
print('Start Range',startrange)
print('End Range', end_range)
print('Run Time', run_time) 
print('Number of Hits', len(hit_list))
print('Number of processed IDs', len(processed_list)) 
print('Number of Workers', workers)
print('Job Size:',jobs)
person SeánMcK    schedule 27.05.2016
comment
Я ожидаю, что это работает, потому что вы используете ThreadPoolExecutor, а не ProcessPoolExecutor (что больше похоже на то, что делает многопроцессорность). id_filter по-прежнему действует как побочный эффект, заполняя hit_list иprocess_list вместо того, чтобы возвращать значения, которые собирала бы карта. TheadPoolExecutor.map даже собирает и передает исключения как часть возвращаемой последовательности. - person Yann Vernier; 27.05.2016
comment
@YannVernier Не уверен, что понимаю. 1) Насколько я понимаю, мне нужна многопоточность, а не многопроцессорность, потому что это программа ввода-вывода, а не программа, привязанная к процессору. 2) Вы говорите, что это просто работает, потому что ThreadPoolExecutor.map собирает исключения, которые в данном случае будут hit_list и processed_list? Если да, то есть ли что-то принципиально неправильное в использовании этой гибкости? - person SeánMcK; 27.05.2016
comment
1) Многопоточность работает, потому что она разделяет пространство имен Python, поэтому каждый из ваших вызовов id_filter обращается к одному и тому же hit_list. При многопроцессорности у каждого воркера будет свой. Вы также привязаны к вводу-выводу, поэтому многопоточность не повредит, несмотря на глобальную блокировку интерпретатора Python (которая специально существует, чтобы потоки могли обновлять общие вещи без нарушения). 2) Нет, набор исключений — это просто бонус для отладки, когда вы можете перебирать pool.map и получать те же исключения, что и с картой. - person Yann Vernier; 27.05.2016