3.20. Protokoły szeregowe, ramkowanie i sumy CRC

UART w kodzie przenosił bajty między dwoma końcami łącza. Samo w sobie to nie wystarczy, aby zbudować niezawodne połączenie. Trzy problemy ujawniają się w chwili, gdy po drugiej stronie przewodu znajduje się rzeczywiste urządzenie:

• Gdzie zaczyna się i kończy wiadomość? Bajty docierają w postaci strumienia bez wbudowanego ogranicznika. Jeśli odbiornik pominie pierwszy bajt (został włączony po nadajniku; krótkie zakłócenie elektryczne na linii), każdy kolejny bajt jest przesunięty o jeden, dopóki odbiornik nie znajdzie nowego punktu resynchronizacji.

• Jak długa jest każda wiadomość? 32-bajtowy odczyt sensora i 4-bajtowa odpowiedź o statusie wyglądają identycznie na poziomie bajtów. Odbiornik potrzebuje sposobu, aby wiedzieć, ile bajtów należy do bieżącej wiadomości.

• Czy bajty dotarły nienaruszone? Szumy mogą odwracać pojedyncze bity. Bez kontroli odbiornik beztrosko podejmie działania na uszkodzonych danych.

Standardową odpowiedzią na wszystkie trzy problemy jest opakowanie danych w ramkę pakietu: znana sekwencja bajtów na początku, pole długości, sam ładunek oraz suma kontrolna na końcu.

3.20.1. Ramkowanie pakietów

Typowy format ramkowania:

Ramkowany pakiet z osobnymi sumami CRC nagłówka i danych: nagłówek (bajty magiczne), polecenie, długość, CRC nagłówka, ładunek, CRC danych.

Każde pole wykonuje jedno zadanie:

• Nagłówek (bajty magiczne). Stała, nietypowa sekwencja bajtów – często dwa bajty, jak 0xAA 0x55 – której odbiornik szuka w napływającym strumieniu. Gdy znajdzie tę sekwencję, wie, że rozpoczyna się nowy pakiet, i może odrzucić wszelkie śmieci, które pojawiły się wcześniej.

• Polecenie. Pojedynczy bajt, który mówi, czym jest pakiet. Różne wartości poleceń używają różnych formatów ładunku – jedno polecenie może oznaczać „ustaw kąt serwa” z dwoma bajtami ładunku, inne może oznaczać „odczytaj sensor” bez ładunku, a jeszcze inne „zapisz komunikat w dzienniku” z łańcuchem znaków. Odbiornik rozdziela działanie na podstawie bajtu polecenia, aby wiedzieć, jak interpretować resztę pakietu.

• Długość. Dwa bajty podające rozmiar ładunku w bajtach (tutaj w porządku little-endian), pozwalające na ładunki o rozmiarze do około 64 KiB. Odbiornik odczytuje dokładnie tyle bajtów, gdy CRC nagłówka zostanie zweryfikowany.

• CRC nagłówka. Jednobajtowa suma kontrolna obejmująca pola HEADER, CMD i LEN. Odbiornik sprawdza ją przed odczytaniem jakiegokolwiek ładunku, dzięki czemu uszkodzone pole LEN zostaje wykryte już po kilku bajtach (zobacz poniższą sekcję o CRC, aby dowiedzieć się, dlaczego to ważne).

• Ładunek. Dane aplikacji specyficzne dla danego polecenia, o długości dokładnie LEN bajtów. Format jest określany przez bajt polecenia: rekord o stałej szerokości pól spakowany za pomocą struct, łańcuch znaków, surowa pamięć – cokolwiek obie strony uzgodnią dla danego polecenia.

• CRC danych. Czterobajtowy CRC obejmujący bajty ładunku. Odbiornik ponownie oblicza go z właśnie odczytanych bajtów i odrzuca pakiet, jeśli wartości się nie zgadzają.

3.20.2. Sumy CRC

Najprostszą „sumą kontrolną” jest suma wszystkich bajtów, modulo 256 lub 65536. Wykrywa większość przekłamań pojedynczego bitu, ale pomija wiele błędów wielobitowych i ignoruje kolejność bajtów.

Cykliczny kod nadmiarowy (CRC) to standardowe ulepszenie. Traktuje wejście jako jedną długą liczbę binarną i dzieli je (w specjalny sposób) przez ustalony wielomian; resztą z tego dzielenia jest CRC. Różne wielomiany wykrywają różne klasy błędów; popularne wielomiany 8-, 16- i 32-bitowe wykrywają każdą serię błędów krótszą niż ich szerokość oraz dużą część dłuższych serii.

3.20.2.1. Dlaczego dwa CRC

Diagram pakietu powyżej niesie dwie osobne sumy CRC – jedną obejmującą nagłówek (HEADER, CMD, LEN) i drugą obejmującą ładunek. Tego właśnie potrzebuje solidne ramkowanie, ze względu na to, jak pojedynczy końcowy CRC zawodzi, gdy samo pole LEN ulegnie uszkodzeniu podczas transmisji:

• Odbiornik działa na podstawie uszkodzonego pola LEN i odczytuje z linii tyle bajtów – prawdopodobnie znacznie więcej, niż zamierzał nadajnik.

• Dopiero końcowy CRC ostatecznie informuje odbiornik, że coś poszło nie tak, i to dopiero po pochłonięciu wszystkich tych bajtów.

• Podczas gdy parser tkwi w oczekiwaniu na niewłaściwą liczbę bajtów, prawdziwe pakiety nadchodzące za uszkodzonym zostają wchłonięte jako ładunek, a odbiornik traci kilka pakietów zamiast tylko jednego.

Podział CRC rozwiązuje ten problem:

• CRC nagłówka obejmuje HEADER, CMD i LEN. Odbiornik sprawdza go przed odczytaniem jakiegokolwiek ładunku, więc uszkodzone pole LEN zostaje wykryte po kilku bajtach, a parser natychmiast się resynchronizuje, eliminując tylko ten jeden błędny pakiet.

• CRC danych obejmuje ładunek. Gdy CRC nagłówka przejdzie pomyślnie, odbiornik wie, że może zaufać polu LEN, odczytuje dokładnie tyle bajtów ładunku i weryfikuje je względem CRC danych.

Typowy rozmiar – i taki, jakiego używa ta strona – to jeden bajt na CRC nagłówka (CRC-8 w zupełności wystarcza dla pięciobajtowego nagłówka) oraz cztery bajty na CRC danych (CRC-32 obejmuje wiele kilobajtów ładunku przy znikomo niskim współczynniku kolizji).

3.20.2.2. Funkcje pomocnicze

MicroPython dostarcza binascii.crc32() bezpośrednio dla czterobajtowego CRC. W przypadku jednobajtowego CRC nagłówka mała funkcja pomocnicza wykorzystująca wielomian używany przez urządzenia 1-wire firmy Maxim (0x8C w postaci odbitej) jest na tyle krótka, że można ją napisać w jednej linii:

def crc8(data: bytes) -> int:
    crc = 0
    for byte in data:
        crc ^= byte
        for _ in range(8):
            crc = (crc >> 1) ^ 0x8C if crc & 1 else crc >> 1
    return crc & 0xFF

Kompletny enkoder łączy oba CRC w jednej funkcji:

import binascii
import struct

def encode_packet(cmd: int, payload: bytes) -> bytes:
    header = b"\xAA\x55" + bytes([cmd]) + struct.pack("<H", len(payload))
    hcrc   = crc8(header)
    dcrc   = binascii.crc32(payload)
    return header + bytes([hcrc]) + payload + struct.pack("<I", dcrc)

Funkcja odwrotna odzyskuje polecenie i ładunek z kompletnego pakietu lub zwraca None, jeśli którakolwiek z kontroli CRC zawiedzie:

def decode_packet(packet: bytes):
    # Layout: HEADER(2) + CMD(1) + LEN(2) + HCRC(1) + PAYLOAD(LEN) + DCRC(4)
    if len(packet) < 10 or packet[0:2] != b"\xAA\x55":
        return None

    header = packet[0:5]
    if crc8(header) != packet[5]:
        return None                       # header CRC mismatch

    cmd    = packet[2]
    length = struct.unpack("<H", packet[3:5])[0]
    if len(packet) != 6 + length + 4:
        return None                       # truncated or oversized

    payload       = packet[6:6 + length]
    received_dcrc = struct.unpack("<I", packet[6 + length:])[0]
    if binascii.crc32(payload) != received_dcrc:
        return None                       # data CRC mismatch

    return cmd, bytes(payload)

W praktyce odbiornik nie otrzymuje całego pakietu naraz – bajty docierają pojedynczo przez UART, a nadajnika, który robi pauzę w środku pakietu (lub zaszumionej linii, która gubi bajt), nie można po prostu odczytać przez read() do bufora o właściwym rozmiarze. Następna sekcja uruchamia tę samą logikę dekodowania bajt po bajcie jako maszynę stanów.

3.20.3. Odbiornik jako maszyna stanów

Odbiornik nie może po prostu wywołać uart.read(N) dla pewnego stałego N – nie wie, ile bajtów będzie miał następny pakiet, a wszelkie śmieci na linii zaburzają wyrównanie. Rozwiązaniem jest mała maszyna stanów, która pochłania bajty pojedynczo i reaguje w zależności od tego, w którym miejscu pakietu się znajduje. Główna pętla odpytuje any(), aby sprawdzić, ile bajtów jest zbuforowanych, opróżnia je w jednym wywołaniu read() i przekazuje każdy bajt przez maszynę stanów:

import time
import binascii
import struct
from machine import UART

HEADER = b"\xAA\x55"
HEADER_LEN = len(HEADER)

# States, in the order the receiver walks through them per packet.
HUNT_FOR_HEADER = 0
READ_COMMAND    = 1
READ_LENGTH     = 2
READ_HEADER_CRC = 3
READ_PAYLOAD    = 4
READ_DATA_CRC   = 5

uart = UART(3, baudrate=115200)

# Receiver state plus partial-field buffers.
state        = HUNT_FOR_HEADER
matched      = 0              # bytes of HEADER matched so far
cmd          = 0              # CMD captured in READ_COMMAND
length_bytes = bytearray()    # raw LEN bytes (kept for the header CRC)
length       = 0              # unpacked LEN
payload      = bytearray()    # payload bytes accumulated in READ_PAYLOAD
crc_bytes    = bytearray()    # DCRC bytes accumulated in READ_DATA_CRC

def handle_packet(cmd: int, payload: bytes) -> None:
    print("cmd", cmd, "payload", payload)

while True:
    # Drain whatever bytes have arrived since the last poll. Idle
    # briefly when the line is quiet so the loop is not a busy spin.
    n = uart.any()
    if not n:
        time.sleep_ms(1)
        continue

    for b in uart.read(n):
        if state == HUNT_FOR_HEADER:
            # Walk the magic bytes. On a mismatch, back off by one
            # so a stray HEADER[0] in the noise still counts as a
            # possible start.
            if b == HEADER[matched]:
                matched += 1
                if matched == HEADER_LEN:
                    state = READ_COMMAND
                    matched = 0
            else:
                matched = 1 if b == HEADER[0] else 0

        elif state == READ_COMMAND:
            cmd = b
            length_bytes = bytearray()
            state = READ_LENGTH

        elif state == READ_LENGTH:
            # LEN is two bytes little-endian.
            length_bytes.append(b)
            if len(length_bytes) == 2:
                length = struct.unpack("<H", length_bytes)[0]
                state = READ_HEADER_CRC

        elif state == READ_HEADER_CRC:
            # Verify the CRC over HEADER + CMD + LEN before
            # committing to read LEN payload bytes. A mismatch
            # aborts here, after just five header bytes -- the
            # next valid header re-syncs quickly.
            expected = crc8(HEADER + bytes([cmd]) + length_bytes)
            if b == expected:
                payload = bytearray()
                state = READ_PAYLOAD
            else:
                state = HUNT_FOR_HEADER

        elif state == READ_PAYLOAD:
            payload.append(b)
            if len(payload) == length:
                crc_bytes = bytearray()
                state = READ_DATA_CRC

        elif state == READ_DATA_CRC:
            # Verify the CRC over the payload and either deliver
            # the packet or drop it. Either way, go back to
            # looking for the next header.
            crc_bytes.append(b)
            if len(crc_bytes) == 4:
                expected = binascii.crc32(payload)
                received = struct.unpack("<I", crc_bytes)[0]
                if expected == received:
                    handle_packet(cmd, bytes(payload))
                state = HUNT_FOR_HEADER

Każdy bajt przesuwa maszynę stanów o jeden krok lub powoduje powrót do HUNT_FOR_HEADER po kompletnym pakiecie, błędnym CRC nagłówka albo błędnym CRC danych. Śmieci na linii, które nie pasują do nagłówka, są po cichu odrzucane; następny prawidłowy nagłówek ponownie synchronizuje odbiornik. Kluczowa właściwość bezpieczeństwa wynika z CRC nagłówka: jeśli pole LEN jest uszkodzone, parser wychwyci to po sprawdzeniu CRC nagłówka (kilka bajtów), a nie po zobowiązaniu się do odczytania szalenie błędnej liczby bajtów ładunku.

3.20.4. Poza poziomem podstawowym

Powyższe ramkowanie to minimum, jakiego potrzebuje łącze szeregowe, aby odzyskać sprawność po szumach na linii: magiczny nagłówek, długość, polecenie i dwie sumy CRC. Wykrywa uszkodzenia i resynchronizuje się po przekłamanych bajtach, ale rezygnuje z uszkodzonych pakietów, zamiast je przepuścić, i pozostawia nadajnik bez wiedzy o tym, co odbiornik faktycznie usłyszał.

Rzeczywiste protokoły szeregowe nadbudowują funkcje ponad ten poziom podstawowy. Nie każde łącze wbudowane potrzebuje ich wszystkich – wybierz to, czego naprawdę wymaga aplikacja:

• Numery sekwencyjne. Mały licznik, który zwiększa się przy każdym wysłaniu. Odbiornik wykrywa luki (pakiet został utracony), duplikaty (nadajnik retransmitował, ale odbiornik już zaakceptował pierwszą kopię) oraz – tam, gdzie kanał może zmieniać kolejność – przybycia poza kolejnością.

• Potwierdzenia. Dedykowany pakiet ACK (lub dołączony bit w odpowiedzi), który odbiornik odsyła, aby potwierdzić każdy pakiet. Bez potwierdzeń ACK nadajnik nie ma sposobu, aby wiedzieć, że jego dane dotarły.

• Potwierdzenia negatywne. NACK wysyłany, gdy odbiornik wykryje błąd CRC lub lukę w sekwencji. Nadajnik natychmiast retransmituje, zamiast czekać na upływ czasu oczekiwania na ACK.

• Retransmisja. Nadajnik przechowuje każdy niepotwierdzony pakiet w małej kolejce i wysyła go ponownie po upływie czasu oczekiwania (lub po otrzymaniu NACK). Limit prób oraz pewne opóźnienie między próbami zapobiegają temu, aby trwale uszkodzone łącze zapętlało się w nieskończoność.

• Okna przesuwne. Dopuszczenie kilku pakietów w locie przed wymaganiem ACK utrzymuje wysoką przepustowość na łączach, gdzie czas obiegu jest długi w porównaniu z czasem wysyłania pojedynczego pakietu. Kosztem jest większy stan po stronie nadajnika – jedno miejsce na każdy pakiet w locie.

• Kontrola przepływu. Sygnał od odbiornika nakazujący nadajnikowi zwolnić lub wstrzymać się, gdy jego bufor się zapełnia. Implementacje są różne – jawne bajty XON / XOFF, mechanizmy oparte na kredytach, gdzie odbiornik udziela licencji na N kolejnych pakietów naraz, albo sprzętowe linie RTS / CTS na samym przewodzie. Bez kontroli przepływu szybki nadajnik w końcu przeciąża wolny odbiornik i pakiety są gubione.

• Wersja protokołu. Pole wersji umieszczone na początku pakietu pozwala formatowi ewoluować. Każda strona może podczas uruchamiania wynegocjować najwyższą wersję obsługiwaną przez obie, albo odrzucać pakiety od niezgodnych partnerów.

• Fragmentacja i ponowne składanie. Dwubajtowe pole LEN ogranicza pakiet do 64 KiB; wiadomości większe niż to są dzielone na wiele pakietów i ponownie składane po drugiej stronie. Metadane fragmentacji (indeks fragmentu, łączna liczba lub flaga „więcej fragmentów”) znajdują się wewnątrz ładunku.

• Sygnały podtrzymania (heartbeat). Mały okresowy pakiet, który mówi „wciąż tu jestem”. Druga strona zauważa, gdy sygnały podtrzymania ustają, i ponawia połączenie (lub zgłasza głośną awarię), zamiast cicho zawisnąć.

• Kanały. Identyfikator kanału lub strumienia w nagłówku, dzięki czemu jedno fizyczne łącze niesie kilka logicznych strumieni – kanał sterowania, kanał telemetrii, kanał dziennika – rozróżnianych wyłącznie na podstawie tego pola.

• Uwierzytelnianie. Krótki znacznik obliczony na podstawie ładunku i tajnej wartości, którą znają tylko uprawniony nadajnik i odbiornik. Odbiornik ponownie oblicza znacznik z otrzymanych bajtów i odrzuca pakiet, jeśli oba się nie zgadzają. Wychwytuje to zarówno manipulację (atakujący zmodyfikował bajty), jak i – jeśli numer sekwencyjny lub znacznik czasu jest częścią tego, co obejmuje znacznik – atak typu replay, w którym atakujący nagrywa prawdziwy pakiet z linii i wysyła go później ponownie, aby odbiornik zadziałał na nim dwukrotnie.

• Szyfrowanie. Mieszanie bajtów ładunku za pomocą współdzielonego tajnego klucza, tak aby każdy odczytujący linię bez tego klucza widział jedynie szum. Zwykle łączone z opisanym powyżej znacznikiem uwierzytelniania – bez niego atakujący może podsyłać śmieci, które przypadkiem przejdą kontrolę CRC, a odbiornik marnuje cykle, próbując odszyfrować bezsens.

Typowy „dobry” protokół dla sprzętu przemysłowego ostatecznie obejmuje ramkowanie, podwójny CRC, numery sekwencyjne, ACK / NACK z retransmisją oraz sygnały podtrzymania. Warte uwagi przykłady ze świata rzeczywistego: MAVLink (telemetria dronów, z numerami sekwencyjnymi, identyfikatorami systemu / komponentu oraz opcjonalnymi podpisami pakietów), Modbus (przemysłowe sterowniki PLC, z kodami funkcji i CRC) oraz NMEA 0183 (protokół ASCII, którym posługuje się każdy konsumencki odbiornik GPS – wiadomości oparte na liniach z sumą kontrolną po ograniczniku w postaci gwiazdki).