Быстрая связь между потоками Python ⇐ Python

[Anonymous]

Я работаю над аудиоприложением, в котором все функции, выполняемые в «аудиоцикле», должны выполняться быстрее 1 мс. Я осознаю, что Python не является подходящим языком программирования для этой задачи, однако Python прошел долгий путь, и я уверен, что с помощью правильных советов и настроек я смогу заставить его работать.
В настоящее время я изучаю, как передавать данные между потоками, и обнаружил некоторые странные результаты.
Я запускаю следующую программу тестирования, чтобы протестировать различные методы:

Код: Выделить всё

import multiprocessing
import threading
import queue
import numpy as np
import time

SIZE = 2048
myarray1 = np.ones(SIZE)
myarray2 = np.ones(SIZE)

def test_multiprocessing(num: int, put: list, get: list):
# init
shared_array = multiprocessing.Array('f', SIZE, lock=True)
for _ in range(nums):
starttime = time.perf_counter()
# put
with shared_array.get_lock():
np.copyto(np.frombuffer(shared_array.get_obj(), dtype=np.float32), myarray1)
enddtime = time.perf_counter()
put.append(enddtime-starttime)

starttime = time.perf_counter()
# get
with shared_array.get_lock():
np.copyto(myarray2, np.frombuffer(shared_array.get_obj(), dtype=np.float32))
enddtime = time.perf_counter()
get.append(enddtime-starttime)

def test_threading_copy(num: int, put: list, get: list):
# init
free = threading.Semaphore(value=1)
used = threading.Semaphore(value=0)
transfer = np.empty(SIZE)
for _ in range(nums):

starttime = time.perf_counter()
# put
free.acquire()
np.copyto(transfer, myarray1)
used.release()
enddtime = time.perf_counter()
put.append(enddtime-starttime)

starttime = time.perf_counter()
# get
used.acquire()
np.copyto(myarray2, transfer)
free.release()
enddtime = time.perf_counter()
get.append(enddtime-starttime)

def test_queue(num: int, put: list, get: list):
# init
q = queue.Queue(maxsize=1)
for _ in range(num):
starttime = time.perf_counter()
# put
q.put(myarray1)
enddtime = time.perf_counter()
put.append(enddtime-starttime)

starttime = time.perf_counter()
# get
myarray2 = q.get()
enddtime = time.perf_counter()
get.append(enddtime-starttime)

if __name__ == "__main__":
nums = int(1e6)
for test in [test_multiprocessing, test_threading_copy, test_queue]:
put = []; get = []
test(nums, put, get)
print("results:")
print(f"\tput_avg = {sum(put) / len(put)}")
print(f"\tput_max = {max(put)}")
print(f"\tget_avg = {sum(get) / len(get)}")
print(f"\tget_max = {max(get)}")

Результаты на моей машине выглядят примерно так:

Код: Выделить всё

results:
put_avg = 3.930823400122108e-06
put_max = 0.002819699999918157
get_avg = 3.895689899812624e-06
get_max = 0.0016344000000572123
results:
put_avg = 3.603283000182273e-06
put_max = 0.007975700000088182
get_avg = 3.501774700153874e-06
get_max = 0.010190099999817903
results:
put_avg = 1.4336647000006905e-06
put_max = 0.0008001000001058856
get_avg = 1.2777225000797898e-06
get_max = 0.00023200000009637733

Среднее время вполне подходит для моего приложения, но максимальные значения меня беспокоят, поскольку они составляют >1 мс. За исключением примера test_queue, ни один из моих кодов не выделяет новую память.

• Знаете, почему это происходит?
• Как я могу это исправить или ускорить работу кода?
• Существуют ли какие-то общие настройки Python, позволяющие предотвратить такое поведение?

Подробнее здесь: https://stackoverflow.com/questions/792 ... on-threads