Implementieren von Sub-Millisekunden-Verarbeitung in Python ohne busywait

Question

Wie würde ich die Verarbeitung eines Arrays mit Millisekunden-Genauigkeit mit Python unter Linux implementieren (auf einem Single-Core-Raspberry Pi ausgeführt).

Ich versuche, Informationen aus einer MIDI-Datei zu analysieren, die zu einem Array vorverarbeitet wurde, wo ich jede Millisekunde überprüfe, ob das Array Einträge am aktuellen Zeitstempel hat und einige Funktionen auslöst.

Derzeit verwende ich time.time() und beschäftigt beschäftigt warten (wie abgeschlossen here). Dies verschlingt die gesamte CPU, daher entscheide ich mich für eine bessere Lösung.

# 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() 

wo last die Millisekunde des letzten Ereignisses ist (bekannt aus der Vorverarbeitung)

Dies ist keine Frage über time() vs clock() mehr über sleep vs busy wait.

Ich kann wirklich nicht schlafen in der "füllen Rest der Millisekunde" -Schleife, wegen der too lowaccuracy of sleep(). Wenn ich ctypes verwenden würde, wie würde ich es richtig machen?

Gibt es eine Timer-Bibliothek, die jeden Millisekunde zuverlässig einen Callback aufruft?

Meine aktuelle Implementierung ist auf GitHub. Mit diesem Ansatz bekomme ich einen Skew von ungefähr 4 oder 5ms auf dem drum_sample, was 3,7s insgesamt ist (mit Mock, also keine echte Hardware angeschlossen). Bei einem Sample von 30.7s liegt der Skew bei 32ms (also zumindest nicht linear!).

Ich habe versucht, mit time.sleep() und nanosleep() via ctypes mit dem folgenden Code

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)) 

Das Ergebnis auf meinem i5 ist

FIN: 0,000165443

während auf dem RPI, es ist

FIN: 0,000578617

, die eine Schlafperiode von etwa 0,1 oder 0,5 Millisekunden vorschlagen, mit dem gegebenen Jitter (neigt dazu, länger zu schlafen), dass ich höchstens hilft der Last ein wenig zu reduzieren.

Answer 1

Eine mögliche Lösung, mit der sched Modul:

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() 

Ergebnis:

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 

Neben einer Konstanten von etwa 2 Millisekunden-Offset (die Sie kalibrieren könnten), scheint der Jitter zu sein in der Größenordnung von 1 oder 2 Millisekunden (wie von time.time selbst berichtet). Nicht sicher, ob das für Ihre Anwendung gut genug ist.

Wenn Sie in der Zwischenzeit einige nützliche Arbeiten ausführen müssen, sollten Sie sich mit Multi-Threading oder Multi-Processing beschäftigen.

Hinweis: Eine Standard-Linux-Distribution, die auf einem RPi ausgeführt wird, ist kein hartes Echtzeitbetriebssystem. Auch Python kann ein nicht-deterministisches Timing zeigen, z. wenn es eine Speicherbereinigung startet. So könnte Ihr Code gut laufen mit geringem Jitter die meiste Zeit, aber Sie haben gelegentlich 'hickups', wo es ein bisschen Verzögerung gibt.