Я использую concurrent.futures.ProcessPoolExecutor, чтобы найти вхождение числа из диапазона чисел. Цель состоит в том, чтобы исследовать степень повышения производительности за счет параллелизма. Для оценки производительности у меня есть элемент управления — серийный код для выполнения указанной задачи (показан ниже). Я написал 2 параллельных кода, один с использованием concurrent.futures.ProcessPoolExecutor.submit(), а другой с использованием concurrent.futures.ProcessPoolExecutor.map() для выполнения одной и той же задачи. Они показаны ниже. Советы по составлению первого и второго вариантов можно найти здесь и здесь соответственно.
Задача, выданная всем трем кодам, состояла в том, чтобы найти количество вхождений числа 5 в диапазоне чисел от 0 до 1E8. И .submit(), и .map() были назначены по 6 рабочих, а .map() имел размер фрагмента 10 000. Способ дискретизации рабочей нагрузки был идентичен в параллельных кодах. Однако функция, используемая для поиска вхождений в обоих кодах, была разной. Это произошло потому, что способ передачи аргументов в функцию, вызываемую .submit() и .map(), был разным.
Все 3 кода сообщили об одном и том же количестве случаев, то есть 56 953 279 раз. Однако время, затраченное на выполнение задачи, было очень разным. .submit() выполнялось в 2 раза быстрее, чем контроль, а .map() потребовалось в два раза больше времени, чем контроль, чтобы выполнить свою задачу.
Вопросы:
- Я хотел бы знать, является ли низкая производительность
.map()артефактом моего кодирования или она изначально медленная?» Если первое, как я могу ее улучшить. много стимулов для его использования. - Мне хотелось бы знать, есть ли способ заставить код
.submit()работать еще быстрее. У меня есть условие, что функция_concurrent_submit()должна возвращать итерацию с числами/вхождениями, содержащими число 5.
Результаты тестирования
concurrent.futures.ProcessPoolExecutor.submit()
#!/usr/bin/python3.5
# -*- coding: utf-8 -*-
import concurrent.futures as cf
from time import time
from traceback import print_exc
def _findmatch(nmin, nmax, number):
'''Function to find the occurrence of number in range nmin to nmax and return
the found occurrences in a list.'''
print('\n def _findmatch', nmin, nmax, number)
start = time()
match=[]
for n in range(nmin, nmax):
if number in str(n):
match.append(n)
end = time() - start
print("found {0} in {1:.4f}sec".format(len(match),end))
return match
def _concurrent_submit(nmax, number, workers):
'''Function that utilises concurrent.futures.ProcessPoolExecutor.submit to
find the occurences of a given number in a number range in a parallelised
manner.'''
# 1. Local variables
start = time()
chunk = nmax // workers
futures = []
found =[]
#2. Parallelization
with cf.ProcessPoolExecutor(max_workers=workers) as executor:
# 2.1. Discretise workload and submit to worker pool
for i in range(workers):
cstart = chunk * i
cstop = chunk * (i + 1) if i != workers - 1 else nmax
futures.append(executor.submit(_findmatch, cstart, cstop, number))
# 2.2. Instruct workers to process results as they come, when all are
# completed or .....
cf.as_completed(futures) # faster than cf.wait()
# 2.3. Consolidate result as a list and return this list.
for future in futures:
for f in future.result():
try:
found.append(f)
except:
print_exc()
foundsize = len(found)
end = time() - start
print('within statement of def _concurrent_submit():')
print("found {0} in {1:.4f}sec".format(foundsize, end))
return found
if __name__ == '__main__':
nmax = int(1E8) # Number range maximum.
number = str(5) # Number to be found in number range.
workers = 6 # Pool of workers
start = time()
a = _concurrent_submit(nmax, number, workers)
end = time() - start
print('\n main')
print('workers = ', workers)
print("found {0} in {1:.4f}sec".format(len(a),end))
concurrent.futures.ProcessPoolExecutor.map()
#!/usr/bin/python3.5
# -*- coding: utf-8 -*-
import concurrent.futures as cf
import itertools
from time import time
from traceback import print_exc
def _findmatch(listnumber, number):
'''Function to find the occurrence of number in another number and return
a string value.'''
#print('def _findmatch(listnumber, number):')
#print('listnumber = {0} and ref = {1}'.format(listnumber, number))
if number in str(listnumber):
x = listnumber
#print('x = {0}'.format(x))
return x
def _concurrent_map(nmax, number, workers):
'''Function that utilises concurrent.futures.ProcessPoolExecutor.map to
find the occurrences of a given number in a number range in a parallelised
manner.'''
# 1. Local variables
start = time()
chunk = nmax // workers
futures = []
found =[]
#2. Parallelization
with cf.ProcessPoolExecutor(max_workers=workers) as executor:
# 2.1. Discretise workload and submit to worker pool
for i in range(workers):
cstart = chunk * i
cstop = chunk * (i + 1) if i != workers - 1 else nmax
numberlist = range(cstart, cstop)
futures.append(executor.map(_findmatch, numberlist,
itertools.repeat(number),
chunksize=10000))
# 2.3. Consolidate result as a list and return this list.
for future in futures:
for f in future:
if f:
try:
found.append(f)
except:
print_exc()
foundsize = len(found)
end = time() - start
print('within statement of def _concurrent(nmax, number):')
print("found {0} in {1:.4f}sec".format(foundsize, end))
return found
if __name__ == '__main__':
nmax = int(1E8) # Number range maximum.
number = str(5) # Number to be found in number range.
workers = 6 # Pool of workers
start = time()
a = _concurrent_map(nmax, number, workers)
end = time() - start
print('\n main')
print('workers = ', workers)
print("found {0} in {1:.4f}sec".format(len(a),end))
Серийный код:
#!/usr/bin/python3.5
# -*- coding: utf-8 -*-
from time import time
def _serial(nmax, number):
start = time()
match=[]
nlist = range(nmax)
for n in nlist:
if number in str(n):match.append(n)
end=time()-start
print("found {0} in {1:.4f}sec".format(len(match),end))
return match
if __name__ == '__main__':
nmax = int(1E8) # Number range maximum.
number = str(5) # Number to be found in number range.
start = time()
a = _serial(nmax, number)
end = time() - start
print('\n main')
print("found {0} in {1:.4f}sec".format(len(a),end))
Обновление от 13 февраля 2017 г.
В дополнение к ответу @niemmi я предоставил ответ после некоторых личных исследований, чтобы показать:
- как еще больше ускорить решения @niemmi
.map()и.submit(), и - когда
ProcessPoolExecutor.map()может привести к большему ускорению, чемProcessPoolExecutor.submit().
