Это должен быть мой третий и последний вопрос, касающийся моих попыток повысить производительность по некоторому статистическому анализу, который я делаю с python. У меня есть 2 версии моего кода (одноядерный или многопроцессорный), я ожидал получить производительность, используя несколько ядер, поскольку я ожидаю, что мой код распакует/распаковывает несколько двоичных строк, к сожалению, я заметил, что производительность фактически уменьшилась за счет использования нескольких ядра.
Мне интересно, есть ли у кого-нибудь возможное объяснение того, что я наблюдаю (прокрутить вниз до обновления 16 апреля для получения дополнительной информации)?
Ключевой частью программы является функция numpy_array (+ декодирование при многопроцессорной обработке), фрагмент кода ниже (полный код, доступный через pastebin, далее):
def numpy_array(data, peaks):
rt_counter=0
for x in peaks:
if rt_counter %(len(peaks)/20) == 0:
update_progress()
peak_counter=0
data_buff=base64.b64decode(x)
buff_size=len(data_buff)/4
unpack_format=">%dL" % buff_size
index=0
for y in struct.unpack(unpack_format,data_buff):
buff1=struct.pack("I",y)
buff2=struct.unpack("f",buff1)[0]
if (index % 2 == 0):
data[rt_counter][1][peak_counter][0]=float(buff2)
else:
data[rt_counter][1][peak_counter][1]=float(buff2)
peak_counter+=1
index+=1
rt_counter+=1
Многопроцессорная версия выполняет это с помощью набора функций, я покажу ключ 2 ниже:
def tonumpyarray(mp_arr):
return np.frombuffer(mp_arr.get_obj())
def numpy_array(shared_arr,peaks):
processors=mp.cpu_count()
with contextlib.closing(mp.Pool(processes=processors,
initializer=pool_init,
initargs=(shared_arr, ))) as pool:
chunk_size=int(len(peaks)/processors)
map_parameters=[]
for i in range(processors):
counter = i*chunk_size
chunk=peaks[i*chunk_size:(i+1)*chunk_size]
map_parameters.append((chunk, counter))
pool.map(decode,map_parameters)
def decode ((chunk, counter)):
data=tonumpyarray(shared_arr).view(
[('f0','<f4'), ('f1','<f4',(250000,2))])
for x in chunk:
peak_counter=0
data_buff=base64.b64decode(x)
buff_size=len(data_buff)/4
unpack_format=">%dL" % buff_size
index=0
for y in struct.unpack(unpack_format,data_buff):
buff1=struct.pack("I",y)
buff2=struct.unpack("f",buff1)[0]
#with shared_arr.get_lock():
if (index % 2 == 0):
data[counter][1][peak_counter][0]=float(buff2)
else:
data[counter][1][peak_counter][1]=float(buff2)
peak_counter+=1
index+=1
counter+=1
Доступ к полным программным кодам можно получить через эти ссылки pastebin
Pastebin для одноядерной версии
Pastebin для многопроцессорной версии
Производительность, которую я наблюдаю с файлом, содержащим 239 временных точек и парами измерений ~ 180 тыс. за каждую точку времени, составляет ~ 2,5 м для одного ядра и ~ 3,5 для многопроцессорной обработки.
PS: два предыдущих вопроса (из моих первых попыток паралеллизации):
- 16 апреля -
Я профилировал свою программу с помощью библиотеки cProfile (с cProfile.run('main()') в __main__, которая показывает, что есть 1 шаг, который замедляет все:
ncalls tottime percall cumtime percall filename:lineno(function)
23 85.859 3.733 85.859 3.733 {method 'acquire' of 'thread.lock' objects}
То, что я не понимаю здесь, состоит в том, что thread.lock объекты используются в threading (насколько мне известно), но не должны использоваться при многопроцессорности, поскольку каждое ядро должно запускать один поток (кроме того, у него есть собственный механизм блокировки), так как это происходит, и почему один вызов занимает 3,7 секунды?
