Python线程/进程间数据传递峰值延迟过高的排查与优化问询

我正在开发一个音频应用，要求“音频循环”里所有函数的执行时间远小于1ms。我知道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(num):
    starttime = time.perf_counter()
    # put
    with shared_array.get_lock():
      np.copyto(np.frombuffer(shared_array.get_obj(), dtype=np.float32), myarray1)
    endtime = time.perf_counter()
    put.append(endtime-starttime)

    starttime = time.perf_counter()
    # get
    with shared_array.get_lock():
      np.copyto(myarray2, np.frombuffer(shared_array.get_obj(), dtype=np.float32))
    endtime = time.perf_counter()
    get.append(endtime-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(num):
    
    starttime = time.perf_counter()
    # put
    free.acquire()
    np.copyto(transfer, myarray1)
    used.release()
    endtime = time.perf_counter()
    put.append(endtime-starttime)

    starttime = time.perf_counter()
    # get
    used.acquire()
    np.copyto(myarray2, transfer)
    free.release()
    endtime = time.perf_counter()
    get.append(endtime-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)
    endtime = time.perf_counter()
    put.append(endtime-starttime)
    
    starttime = time.perf_counter()
    # get
    myarray2 = q.get()
    endtime = time.perf_counter()
    get.append(endtime-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

平均时间完全符合我的应用要求，但峰值延迟却让我很头疼——它们都超过或接近1ms了。除了test_queue的例子，我的代码都没有分配新内存。

我有几个问题想请教：

• 你知道为什么会出现这种峰值延迟吗？
• 我该如何修复或加速这段代码？
• 有没有通用的Python设置可以避免这种情况？
• 你会怎么调试这个问题？

备注：内容来源于stack exchange，提问作者helixfoo