在没有busywait的情况下在python中实现亚毫秒处理

Question

我将如何使用linux下的python(在单核Raspberry Pi上运行)实现毫秒精度的数组处理.

我正在尝试解析MIDI文件中的信息,该文件已被预处理到一个数组,其中每毫秒我检查数组是否在当前时间戳处有条目并触发某些功能(如果有).

目前我正在使用time.time()并使用繁忙的等待(如此处所述).这会占用所有CPU,因此我选择了更好的解决方案.

# iterate through all milliseconds
for current_ms in xrange(0, last+1):
  start = time()

  # check if events are to be processed
  try:
    events = allEvents[current_ms]
    # iterate over all events for this millisecond
    for event in events:
      # check if event contains note information
      if 'note' in event:
        # check if mapping to pin exists
        if event['note'] in mapping:
          pin = mapping[event['note']]
          # check if event contains on/off information
          if 'mode' in event:
            if event['mode'] == 0:
              pin_off(pin)
            elif event['mode'] == 1:
              pin_on(pin)
            else:
              debug("unknown mode in event:"+event)
          else:
            debug("no mapping for note:" + event['note'])
  except:
    pass

  end = time()

  # fill the rest of the millisecond
  while (end-start) < (1.0/(1000.0)):
    end = time()

哪个last是最后一个事件的毫秒(从预处理中得知)

这不是关于time()vs clock()的问题,更多关于睡眠与忙碌等待的问题.

由于睡眠精度太低 (),我无法真正睡在"毫秒剩余"循环中.如果我使用ctypes,我将如何正确地解决它？

是否有一些Timer库可以每毫秒可靠地调用一次回调？

我目前的实现是在GitHub上.使用这种方法,我在drum_sample上得到大约4或5ms的偏差,总共为3.7s(使用模拟,因此没有附加真正的硬件).在30.7s样本上,偏差约为32ms(因此它至少不是线性的!).

我已经尝试使用time.sleep()和了nanosleep()经由ctypes的用下面的代码

import time
import timeit
import ctypes
libc = ctypes.CDLL('libc.so.6')

class Timespec(ctypes.Structure):
  """ timespec struct for nanosleep, see:
      http://linux.die.net/man/2/nanosleep """
  _fields_ = [('tv_sec', ctypes.c_long),
              ('tv_nsec', ctypes.c_long)]

libc.nanosleep.argtypes = [ctypes.POINTER(Timespec),
                           ctypes.POINTER(Timespec)]
nanosleep_req = Timespec()
nanosleep_rem = Timespec()

def nsleep(us):
  #print('nsleep: {0:.9f}'.format(us)) 
  """ Delay microseconds with libc nanosleep() using ctypes. """
  if (us >= 1000000):
    sec = us/1000000
    us %= 1000000
  else: sec = 0
  nanosleep_req.tv_sec = sec
  nanosleep_req.tv_nsec = int(us * 1000)

  libc.nanosleep(nanosleep_req, nanosleep_rem)

LOOPS = 10000

def do_sleep(min_sleep):
  #print('try: {0:.9f}'.format(min_sleep))
  total = 0.0
  for i in xrange(0, LOOPS):
    start = timeit.default_timer()
    nsleep(min_sleep*1000*1000)
    #time.sleep(min_sleep)
    end = timeit.default_timer()
    total += end - start
  return (total / LOOPS)

iterations = 5
iteration = 1
min_sleep = 0.001
result = None
while True:
    result = do_sleep(min_sleep)
    #print('res: {0:.9f}'.format(result))
    if result > 1.5 * min_sleep:
      if iteration > iterations:
        break
      else:
        min_sleep = result
        iteration += 1
    else:
      min_sleep /= 2.0

print('FIN: {0:.9f}'.format(result))

我的i5的结果是

FIN:0.000165443

而在RPi上它是

FIN:0.000578617

这表明睡眠时间约为0.1或0.5毫秒,给定的抖动(倾向于睡得更久)最多可以帮助我减少负荷.

Answer 1

一种可能的解决方案是使用sched模块：

import sched
import time

def f(t0):
    print 'Time elapsed since t0:', time.time() - t0
s = sched.scheduler(time.time, time.sleep)

for i in range(10):
    s.enterabs(t0 + 10 + i, 0, f, (t0,))
s.run()

结果：

Time elapsed since t0: 10.0058200359
Time elapsed since t0: 11.0022959709
Time elapsed since t0: 12.0017120838
Time elapsed since t0: 13.0022599697
Time elapsed since t0: 14.0022521019
Time elapsed since t0: 15.0015859604
Time elapsed since t0: 16.0023040771
Time elapsed since t0: 17.0023028851
Time elapsed since t0: 18.0023078918
Time elapsed since t0: 19.002286911

除了大约 2 毫秒的恒定偏移（您可以校准）之外，抖动似乎约为 1 或 2 毫秒（如其自身报告的time.time）。不确定这是否足以满足您的应用程序。

如果您需要同时做一些有用的工作，您应该考虑多线程或多处理。

注意：在 RPi 上运行的标准 Linux 发行版不是硬实时操作系统。Python 还可以显示非确定性计时，例如，当它启动垃圾收集时。因此，您的代码在大多数情况下可能运行良好且抖动较低，但您可能偶尔会出现“卡顿”，即存在一点延迟。