Scrierea handlerelor de întrerupere

Pe hardware adecvat, MicroPython oferă posibilitatea de a scrie handlere de întrerupere în Python. Handlerele de întrerupere - cunoscute și ca rutine de serviciu pentru întreruperi (ISR-uri) - sunt definite ca funcții de retroapelare (callback). Acestea sunt executate ca răspuns la un eveniment precum declanșarea unui temporizator sau o variație de tensiune pe un pin. Astfel de evenimente pot apărea în orice punct al execuției codului programului. Acest lucru are consecințe semnificative, unele specifice limbajului MicroPython. Altele sunt comune tuturor sistemelor capabile să răspundă la evenimente în timp real. Acest document tratează mai întâi aspectele specifice limbajului, urmate de o scurtă introducere în programarea în timp real pentru cei care nu sunt familiarizați cu ea.

Această introducere folosește termeni vagi precum „lent” sau „cât mai repede posibil”. Acest lucru este intenționat, deoarece vitezele depind de aplicație. Duratele acceptabile pentru un ISR depind de rata la care apar întreruperile, de natura programului principal și de prezența altor evenimente concurente.

Probleme specifice MicroPython

Tamponul de excepții de urgență

Dacă apare o eroare într-un ISR, MicroPython nu poate genera un raport de eroare decât dacă este creat un tampon special în acest scop. Depanarea este simplificată dacă următorul cod este inclus în orice program care folosește întreruperi.

import micropython

micropython.alloc_emergency_exception_buf(100)

Tamponul de excepții de urgență poate reține doar o singură urmă de stivă pentru excepții. Aceasta înseamnă că, dacă o a doua excepție este aruncată în timpul tratării unei excepții, cât timp heap-ul este blocat, urma de stivă a celei de-a doua excepții o va înlocui pe cea originală - chiar dacă a doua excepție este tratată corect. Acest lucru poate duce la mesaje de excepție derutante dacă tamponul este afișat ulterior.

Simplitate

Din mai multe motive, este important să păstrați codul ISR cât mai scurt și mai simplu posibil. Acesta ar trebui să facă doar ceea ce trebuie făcut imediat după evenimentul care l-a provocat: operațiunile care pot fi amânate ar trebui delegate buclei programului principal. De obicei, un ISR se ocupă de dispozitivul hardware care a cauzat întreruperea, pregătindu-l pentru următoarea întrerupere. Acesta va comunica cu bucla principală prin actualizarea datelor partajate pentru a indica faptul că întreruperea a avut loc, apoi va returna. Un ISR ar trebui să returneze controlul buclei principale cât mai repede posibil. Aceasta nu este o problemă specifică MicroPython, așa că este tratată mai detaliat mai jos.

Comunicarea între un ISR și programul principal

În mod normal, un ISR trebuie să comunice cu programul principal. Cel mai simplu mod de a face acest lucru este prin intermediul unuia sau mai multor obiecte de date partajate, declarate fie ca globale, fie partajate printr-o clasă (vezi mai jos). Există diverse restricții și pericole legate de acest lucru, care sunt tratate mai detaliat mai jos. Numerele întregi, obiectele bytes și bytearray sunt utilizate frecvent în acest scop, împreună cu tablourile (din modulul array), care pot stoca diverse tipuri de date.

Utilizarea metodelor obiectelor ca funcții de retroapelare

MicroPython acceptă această tehnică puternică, care permite unui ISR să partajeze variabile de instanță cu codul subiacent. De asemenea, permite unei clase care implementează un driver de dispozitiv să suporte mai multe instanțe de dispozitiv. Exemplul următor face ca două LED-uri să clipească la rate diferite.

import machine
import micropython

micropython.alloc_emergency_exception_buf(100)


class Foo(object):
    def __init__(self, freq, led):
        self.led = led
        self.timer = machine.Timer(-1, freq=freq, callback=self.cb, hard=True)

    def cb(self, tim):
        self.led.toggle()


red = Foo(1, machine.LED("LED_RED"))
green = Foo(0.8, machine.LED("LED_GREEN"))

În acest exemplu, instanța red controlează LED-ul roșu de la un temporizator virtual de 1 Hz: de fiecare dată când temporizatorul se declanșează, se apelează red.cb(), comutând LED-ul roșu. Instanța green funcționează similar, cu un temporizator de 0,8 Hz care comută LED-ul verde. Utilizarea metodelor de instanță conferă două avantaje. În primul rând, o singură clasă permite partajarea codului între mai multe instanțe hardware. În al doilea rând, fiind o metodă legată, primul argument al funcției de retroapelare este self. Acest lucru permite funcției de retroapelare să acceseze datele de instanță și să salveze starea între apeluri succesive. De exemplu, dacă clasa de mai sus ar avea o variabilă self.count setată la zero în constructor, cb() ar putea incrementa contorul. Instanțele red și green ar menține apoi numărări independente ale numărului de dăți în care fiecare LED și-a schimbat starea.

Crearea obiectelor Python

ISR-urile nu pot crea instanțe ale obiectelor Python. Acest lucru se datorează faptului că MicroPython trebuie să aloce memorie pentru obiect dintr-un depozit de blocuri de memorie liberă numit heap. Acest lucru nu este permis într-un handler de întrerupere, deoarece alocarea pe heap nu este reintrantă. Cu alte cuvinte, întreruperea ar putea apărea în timp ce programul principal este în mijlocul efectuării unei alocări - pentru a menține integritatea heap-ului, interpretorul interzice alocările de memorie în codul ISR.

O consecință a acestui fapt este că ISR-urile nu pot folosi aritmetica în virgulă mobilă; aceasta se datorează faptului că numerele în virgulă mobilă sunt obiecte Python. În mod similar, un ISR nu poate adăuga un element la o listă. În practică, poate fi dificil de determinat exact care construcții de cod vor încerca să efectueze alocarea de memorie și să provoace un mesaj de eroare: un alt motiv pentru a păstra codul ISR scurt și simplu.

O modalitate de a evita această problemă este ca ISR-ul să folosească tampoane prealocate. De exemplu, un constructor de clasă creează o instanță bytearray și un indicator boolean. Metoda ISR atribuie date unor locații din tampon și setează indicatorul. Alocarea de memorie are loc în codul programului principal atunci când obiectul este instanțiat, mai degrabă decât în ISR.

Metodele de I/O din biblioteca MicroPython oferă de obicei o opțiune de utilizare a unui tampon prealocat. De exemplu, machine.I2C.readfrom_into() citește într-un tampon mutabil furnizat de apelant: acest lucru permite utilizarea sa într-un ISR.

O modalitate de creare a unui obiect fără a folosi o clasă sau variabile globale este următoarea:

def set_volume(t, buf=bytearray(3)):
    buf[0] = 0xa5
    buf[1] = t >> 4
    buf[2] = 0x5a
    return buf

Compilatorul instanțiază argumentul implicit buf atunci când funcția este încărcată pentru prima dată (de obicei când modulul în care se află este importat).

O instanță de creare de obiect apare atunci când este creată o referință către o metodă legată. Aceasta înseamnă că un ISR nu poate transmite o metodă legată unei funcții. O soluție este crearea unei referințe către metoda legată în constructorul clasei și transmiterea acelei referințe în ISR. De exemplu:

class Foo():
    def __init__(self):
        self.bar_ref = self.bar  # Allocation occurs here
        self.x = 0.1
        self.tim = machine.Timer(-1, freq=2, callback=self.cb, hard=True)

    def bar(self, _):
        self.x *= 1.2
        print(self.x)

    def cb(self, t):
        # Passing self.bar would cause allocation.
        micropython.schedule(self.bar_ref, 0)

Alte tehnici sunt definirea și instanțierea metodei în constructor sau transmiterea Foo.bar() cu argumentul self.

Utilizarea obiectelor Python

O restricție suplimentară privind obiectele apare din cauza modului în care funcționează Python. Când o instrucțiune import este executată, codul Python este compilat în bytecode, o linie de cod corespunzând de obicei mai multor bytecoduri. Când codul rulează, interpretorul citește fiecare bytecode și îl execută ca o serie de instrucțiuni de cod mașină. Având în vedere că o întrerupere poate apărea oricând între instrucțiunile de cod mașină, linia originală de cod Python poate fi executată doar parțial. În consecință, un obiect Python precum un set, o listă sau un dicționar modificat în bucla principală poate să nu aibă consistență internă în momentul în care apare întreruperea.

Un rezultat tipic este următorul. În ocazii rare, ISR-ul va rula exact în momentul în care obiectul este actualizat parțial. Când ISR-ul încearcă să citească obiectul, rezultă o blocare. Deoarece astfel de probleme apar de obicei în ocazii rare și aleatorii, ele pot fi greu de diagnosticat. Există modalități de a ocoli această problemă, descrise în Secțiuni critice mai jos.

Este important să fie clar ce constituie modificarea unui obiect. Alterarea conținutului unui tablou sau bytearray este sigură. Aceasta se datorează faptului că octeții sau cuvintele sunt scrise ca o singură instrucțiune de cod mașină, care nu poate fi întreruptă: în terminologia programării în timp real, scrierea este atomică. Același lucru este valabil și pentru actualizarea unui element de dicționar, deoarece elementele sunt cuvinte mașină, fiind numere întregi sau pointeri către obiecte. Un obiect definit de utilizator ar putea instanția un tablou sau bytearray. Este valid ca atât bucla principală, cât și ISR-ul să altereze conținutul acestora.

Pericolul apare atunci când structura unui obiect este alterată, în special în cazul dicționarelor. Adăugarea sau ștergerea de chei poate declanșa o rehashare. Dacă un ISR hard rulează în timp ce o rehashare este în curs și încearcă să acceseze un element, poate apărea o blocare. La nivel intern, variabilele globale sunt implementate ca un dicționar. În consecință, programul principal ar trebui să creeze toate variabilele globale necesare înainte de a porni un proces care generează întreruperi hard. Codul aplicației ar trebui, de asemenea, să evite ștergerea variabilelor globale.

MicroPython acceptă numere întregi de precizie arbitrară. Valorile dintre 230 -1 și -230 vor fi stocate într-un singur cuvânt mașină. Valorile mai mari sunt stocate ca obiecte Python. În consecință, modificările aduse numerelor întregi lungi nu pot fi considerate atomice. Utilizarea numerelor întregi lungi în ISR-uri este nesigură, deoarece se poate încerca alocarea de memorie pe măsură ce valoarea variabilei se modifică.

Depășirea limitării privind numerele în virgulă mobilă

În general, este preferabil să evitați utilizarea numerelor în virgulă mobilă în codul ISR: dispozitivele hardware gestionează în mod normal numere întregi, iar conversia la numere în virgulă mobilă se face de obicei în bucla principală. Cu toate acestea, există câțiva algoritmi DSP care necesită virgulă mobilă. Pe platformele cu virgulă mobilă hardware (cum ar fi camerele OpenMV Cam bazate pe STM32), asamblorul inline ARM Thumb poate fi folosit pentru a ocoli această limitare. Aceasta se datorează faptului că procesorul stochează valorile în virgulă mobilă într-un cuvânt mașină; valorile pot fi partajate între ISR și codul programului principal printr-un tablou de numere în virgulă mobilă.

Utilizarea micropython.schedule

Această funcție permite unui ISR să programeze o funcție de retroapelare pentru execuție „foarte curând”. Funcția de retroapelare este pusă în coadă pentru execuție, care va avea loc într-un moment în care heap-ul nu este blocat. Prin urmare, ea poate crea obiecte Python și poate folosi numere în virgulă mobilă. De asemenea, este garantat că funcția de retroapelare va rula într-un moment în care programul principal a finalizat orice actualizare a obiectelor Python, astfel încât funcția de retroapelare nu va întâlni obiecte actualizate parțial.

Utilizarea tipică este pentru gestionarea hardware-ului de senzori. ISR-ul achiziționează date de la hardware și îi permite să emită o întrerupere ulterioară. Apoi programează o funcție de retroapelare pentru a procesa datele.

Funcțiile de retroapelare programate ar trebui să respecte principiile de proiectare a handlerelor de întrerupere prezentate mai jos. Acest lucru este pentru a evita problemele rezultate din activitatea de I/O și din modificarea datelor partajate, care pot apărea în orice cod care preîntâmpină bucla programului principal.

Timpul de execuție trebuie luat în considerare în raport cu frecvența cu care pot apărea întreruperile. Dacă o întrerupere apare în timp ce funcția de retroapelare anterioară se execută, o nouă instanță a funcției de retroapelare va fi pusă în coadă pentru execuție; aceasta va rula după ce instanța curentă s-a finalizat. O rată ridicată și susținută de repetiție a întreruperilor implică, prin urmare, riscul unei creșteri necontrolate a cozii și al unei eventuale eșuări cu un RuntimeError.

Dacă funcția de retroapelare ce urmează a fi transmisă către schedule() este o metodă legată, luați în considerare nota din secțiunea „Crearea obiectelor Python”.

Excepții

Dacă un ISR ridică o excepție, aceasta nu se va propaga către bucla principală. Întreruperea va fi dezactivată dacă excepția nu este tratată de codul ISR.

Interconectarea cu asyncio

Când un ISR rulează, acesta poate preîntâmpina planificatorul asyncio. Dacă ISR-ul efectuează o operațiune asyncio, funcționarea planificatorului poate fi perturbată. Acest lucru se aplică indiferent dacă întreruperea este hard sau soft și se aplică, de asemenea, dacă ISR-ul a transferat execuția către o altă funcție prin micropython.schedule. În special, crearea sau anularea de task-uri este invalidă într-un context ISR. Modul sigur de a interacționa cu asyncio este de a implementa o corutină cu sincronizarea efectuată de asyncio.ThreadSafeFlag. Fragmentul următor ilustrează crearea unui task ca răspuns la o întrerupere:

tsf = asyncio.ThreadSafeFlag()


def isr(_):  # Interrupt handler
    tsf.set()


async def foo():
    while True:
        await tsf.wait()
        asyncio.create_task(bar())

În acest exemplu, va exista o cantitate variabilă de latență între execuția ISR-ului și execuția foo(). Aceasta este inerentă planificării cooperative. Latența maximă depinde de aplicație și de platformă, dar poate fi de obicei măsurată în zeci de ms.

Probleme generale

Aceasta este doar o scurtă introducere în subiectul programării în timp real. Începătorii ar trebui să rețină că erorile de proiectare în programele în timp real pot duce la defecțiuni deosebit de greu de diagnosticat. Aceasta se datorează faptului că ele pot apărea rar și la intervale care sunt în esență aleatorii. Este crucial să obțineți o proiectare inițială corectă și să anticipați problemele înainte ca acestea să apară. Atât handlerele de întrerupere, cât și programul principal trebuie proiectate cu o înțelegere a următoarelor probleme.

Proiectarea handlerelor de întrerupere

După cum s-a menționat mai sus, ISR-urile ar trebui proiectate să fie cât mai simple posibil. Ele ar trebui să returneze întotdeauna într-o perioadă de timp scurtă și previzibilă. Acest lucru este important deoarece, atunci când ISR-ul rulează, bucla principală nu rulează: în mod inevitabil, bucla principală suferă pauze în execuția sa în puncte aleatorii ale codului. Astfel de pauze pot fi o sursă de erori greu de diagnosticat, în special dacă durata lor este lungă sau variabilă. Pentru a înțelege implicațiile timpului de execuție al unui ISR, este necesară o înțelegere de bază a priorităților întreruperilor.

Întreruperile sunt organizate conform unei scheme de priorități. Codul ISR poate fi el însuși întrerupt de o întrerupere cu prioritate mai mare. Acest lucru are implicații dacă cele două întreruperi partajează date (vezi secțiunile critice mai jos). Dacă apare o astfel de întrerupere, ea introduce o întârziere în codul ISR. Dacă o întrerupere cu prioritate mai mică apare în timp ce ISR-ul rulează, ea va fi întârziată până la finalizarea ISR-ului: dacă întârzierea este prea lungă, întreruperea cu prioritate mai mică poate eșua. O altă problemă cu ISR-urile lente este cazul în care o a doua întrerupere de același tip apare în timpul execuției sale. A doua întrerupere va fi tratată la finalizarea celei dintâi. Cu toate acestea, dacă rata întreruperilor sosite depășește în mod constant capacitatea ISR-ului de a le deservi, rezultatul nu va fi unul fericit.

În consecință, construcțiile de tip buclă ar trebui evitate sau reduse la minimum. Operațiunile de I/O către alte dispozitive decât dispozitivul care întrerupe ar trebui în mod normal evitate: operațiunile de I/O precum accesul la disc, instrucțiunile print și accesul UART sunt relativ lente, iar durata lor poate varia. O altă problemă aici este că funcțiile sistemului de fișiere nu sunt reintrante: utilizarea I/O a sistemului de fișiere într-un ISR și în programul principal ar fi periculoasă. În mod crucial, codul ISR nu ar trebui să aștepte un eveniment. Operațiunile de I/O sunt acceptabile dacă se poate garanta că codul returnează într-o perioadă previzibilă, de exemplu comutarea unui pin sau LED. Accesarea dispozitivului care întrerupe prin I2C sau SPI poate fi necesară, dar timpul necesar pentru astfel de accesări ar trebui calculat sau măsurat, iar impactul său asupra aplicației ar trebui evaluat.

De obicei, există nevoia de a partaja date între ISR și bucla principală. Acest lucru se poate face fie prin variabile globale, fie prin variabile de clasă sau de instanță. Variabilele sunt de obicei de tip întreg sau boolean, sau tablouri de numere întregi sau octeți (un tablou de numere întregi prealocat oferă acces mai rapid decât o listă). Când mai multe valori sunt modificate de ISR, este necesar să luați în considerare cazul în care întreruperea apare într-un moment în care programul principal a accesat unele, dar nu toate valorile. Acest lucru poate duce la inconsecvențe.

Luați în considerare următoarea proiectare. Un ISR stochează datele sosite într-un bytearray, apoi adaugă numărul de octeți primiți la un număr întreg care reprezintă totalul de octeți pregătiți pentru procesare. Programul principal citește numărul de octeți, procesează octeții, apoi resetează numărul de octeți pregătiți. Acest lucru va funcționa până când o întrerupere apare imediat după ce programul principal a citit numărul de octeți. ISR-ul pune datele adăugate în tampon și actualizează numărul primit, dar programul principal a citit deja numărul, așa că procesează datele primite inițial. Octeții nou sosiți sunt pierduți.

Există diverse modalități de a evita acest pericol, cea mai simplă fiind utilizarea unui tampon circular. Dacă nu este posibil să se folosească o structură cu siguranță inerentă pentru fire de execuție, alte modalități sunt descrise mai jos.

Reintranță

Un potențial pericol poate apărea dacă o funcție sau o metodă este partajată între programul principal și unul sau mai multe ISR-uri sau între mai multe ISR-uri. Problema aici este că funcția poate fi ea însăși întreruptă, iar o nouă instanță a acelei funcții să ruleze. Pentru ca acest lucru să se întâmple, funcția trebuie proiectată să fie reintrantă. Modul în care se realizează acest lucru este un subiect avansat care depășește domeniul de aplicare al acestui tutorial.

Secțiuni critice

Un exemplu de secțiune critică de cod este una care accesează mai mult de o variabilă care poate fi afectată de un ISR. Dacă întreruperea se întâmplă să apară între accesările variabilelor individuale, valorile lor vor fi inconsistente. Aceasta este o instanță a unui pericol cunoscut sub numele de condiție de cursă: ISR-ul și bucla programului principal concurează pentru a altera variabilele. Pentru a evita inconsistența, trebuie utilizat un mijloc prin care să se asigure că ISR-ul nu alterează valorile pe durata secțiunii critice. O modalitate de a realiza acest lucru este de a emite machine.disable_irq() înainte de începutul secțiunii și machine.enable_irq() la sfârșit. Iată un exemplu al acestei abordări:

import machine
import micropython
import array
import random
import time

micropython.alloc_emergency_exception_buf(100)


class BoundsException(Exception):
    pass


ARRAYSIZE = const(20)
index = 0
data = array.array('i', [0] * ARRAYSIZE)


def callback1(t):
    global data, index
    for x in range(5):
        data[index] = random.getrandbits(30)  # simulate input
        index += 1
        if index >= ARRAYSIZE:
            raise BoundsException('Array bounds exceeded')


tim = machine.Timer(-1, freq=100, callback=callback1, hard=True)

for loop in range(1000):
    if index > 0:
        irq_state = machine.disable_irq()  # Start of critical section
        for x in range(index):
            print(data[x])
        index = 0
        machine.enable_irq(irq_state)  # End of critical section
        print('loop {}'.format(loop))
    time.sleep_ms(1)

tim.deinit()

O secțiune critică poate cuprinde o singură linie de cod și o singură variabilă. Luați în considerare următorul fragment de cod.

count = 0


def cb(): # An interrupt callback
    count += 1


def main():
    # Code to set up the interrupt callback omitted
    while True:
        count += 1

Acest exemplu ilustrează o sursă subtilă de erori. Linia count += 1 din bucla principală implică un pericol specific de condiție de cursă cunoscut sub numele de citire-modificare-scriere. Aceasta este o cauză clasică de erori în sistemele în timp real. În bucla principală, MicroPython citește valoarea lui count, îi adaugă 1 și o scrie înapoi. În ocazii rare, întreruperea apare după citire și înainte de scriere. Întreruperea modifică count, dar modificarea sa este suprascrisă de bucla principală când ISR-ul returnează. Într-un sistem real, acest lucru ar putea duce la defecțiuni rare și imprevizibile.

După cum s-a menționat mai sus, ar trebui acordată atenție dacă o instanță a unui tip încorporat Python este modificată în codul principal, iar acea instanță este accesată într-un ISR. Codul care efectuează modificarea ar trebui considerat o secțiune critică pentru a asigura că instanța se află într-o stare validă când ISR-ul rulează.

Trebuie acordată o atenție deosebită dacă un set de date este partajat între diferite ISR-uri. Pericolul aici este că întreruperea cu prioritate mai mare poate apărea când cea cu prioritate mai mică a actualizat parțial datele partajate. Gestionarea acestei situații este un subiect avansat care depășește domeniul de aplicare al acestei introduceri, în afară de menționarea faptului că obiectele mutex descrise mai jos pot fi uneori utilizate.

Dezactivarea întreruperilor pe durata unei secțiuni critice este modul obișnuit și cel mai simplu de a proceda, dar dezactivează toate întreruperile, nu doar pe cea care are potențialul de a cauza probleme. În general, nu este de dorit să se dezactiveze o întrerupere pentru mult timp. În cazul întreruperilor de temporizator, aceasta introduce variabilitate în momentul în care apare o funcție de retroapelare. În cazul întreruperilor de dispozitiv, poate duce la deservirea prea târzie a dispozitivului, cu posibila pierdere de date sau erori de depășire în hardware-ul dispozitivului. La fel ca ISR-urile, o secțiune critică din codul principal ar trebui să aibă o durată scurtă și previzibilă.

O abordare pentru gestionarea secțiunilor critice care reduce radical timpul pentru care întreruperile sunt dezactivate este utilizarea unui obiect numit mutex (nume derivat din noțiunea de excludere mutuală). Programul principal blochează mutexul înainte de a rula secțiunea critică și îl deblochează la sfârșit. ISR-ul testează dacă mutexul este blocat. Dacă este, evită secțiunea critică și returnează. Provocarea de proiectare este definirea a ceea ce ar trebui să facă ISR-ul în cazul în care accesul la variabilele critice este refuzat. Un exemplu simplu de mutex poate fi găsit aici. Rețineți că codul mutexului dezactivează întreruperile, dar numai pe durata a opt instrucțiuni mașină: avantajul acestei abordări este că alte întreruperi sunt practic neafectate.

Întreruperile și REPL

Handlerele de întrerupere, cum ar fi cele asociate cu temporizatoarele, pot continua să ruleze după terminarea unui program. Acest lucru poate produce rezultate neașteptate, acolo unde v-ați fi putut aștepta ca obiectul care ridică funcția de retroapelare să fi ieșit din domeniul de vizibilitate. De exemplu, pe o cameră OpenMV Cam:

def bar():
    foo = machine.Timer(-1, freq=4, callback=lambda t: print('.', end=''), hard=True)

bar()

Aceasta continuă să ruleze până când temporizatorul este dezactivat explicit sau placa este resetată cu Ctrl-D.