Plancia/Console/uModbusWorker.pas

unit uModbusWorker;

{
  Il bus Modbus in un thread suo, cosi' il thread principale non si ferma mai:
  una richiesta che va in timeout costa mezzo secondo al thread del bus, non
  alla finestra, che resta trascinabile e ridisegnabile.

  Un thread solo, non uno per le letture e uno per le scritture. Su RS485 il
  filo e' uno: due richieste insieme si sovrappongono, e le risposte arrivano
  mescolate. Chi comanda deve comunque aspettare che la richiesta in corso
  finisca, quindi un secondo thread aggiungerebbe un lucchetto attorno alla
  porta senza far guadagnare niente. La reattivita' dei comandi si ottiene
  invece dando loro la precedenza: la coda delle scritture viene svuotata
  prima di ogni lettura, cosi' un pulsante premuto parte al massimo dopo la
  richiesta in corso.

  Chi usa questa classe non chiama mai il bus: mette le scritture in coda e
  ritira le letture quando gli fa comodo (TakeReadings). Nessuna callback
  cross-thread: il form legge lo stato con il suo timer, e il thread non
  tocca ne' controlli ne' elementi.
}

interface

uses
  Winapi.Windows, System.SysUtils, System.Classes, System.SyncObjs,
  System.Generics.Collections,
  uModbusRTU;

type
  /// Che funzione Modbus usa una richiesta del piano.
  TRequestKind = (rqCoils, rqInputs, rqRegisters);

  /// Una richiesta del piano di polling: copre canali contigui di uno slave.
  TPollRequest = record
    Kind: TRequestKind;
    Slave: Integer;
    First: Integer;
    Count: Integer;
  end;

  /// Esito di una richiesta, con i dati se e' andata bene.
  TPollReading = record
    Request: TPollRequest;
    /// Istante in cui la richiesta e' partita sul filo. Serve a capire se una
    /// risposta e' piu' vecchia di un comando appena dato.
    Sent: UInt64;
    Ok: Boolean;
    Error: string;
    Bits: TArray<Boolean>;
    Regs: TArray<Word>;
  end;

  TWriteCommand = record
    Slave: Integer;
    Channel: Integer;
    Value: Boolean;
    /// Etichetta dell'elemento, per i messaggi in barra di stato.
    Caption: string;
  end;

  TWriteOutcome = record
    Caption: string;
    Value: Boolean;
    Ok: Boolean;
    Error: string;
  end;

  /// Quanto uno slave sta rispondendo male, per non inseguirlo a ogni giro.
  TSlaveState = record
    Failures: Integer;
    NextTry: UInt64;
  end;

  TModbusWorker = class(TThread)
  private
    FPort: string;
    FBaud: Integer;
    FTimeoutMs: Integer;
    FPollMs: Integer;
    FLock: TCriticalSection;
    /// Sveglia il thread: nuova scrittura, piano cambiato, o chiusura.
    FWake: TEvent;
    // Tutto quello che segue si tocca solo con FLock preso.
    FPlan: TArray<TPollRequest>;
    /// Piano appena cambiato: gli slave vanno riprovati subito, perche' fra i
    /// canali nuovi puo' esserci un modulo appena montato.
    FPlanFresh: Boolean;
    FWrites: TQueue<TWriteCommand>;
    FReadings: TArray<TPollReading>;
    FReadingsFresh: Boolean;
    FOutcomes: TList<TWriteOutcome>;
    FConnected: Boolean;
    FStatus: string;
    FStatusOk: Boolean;
    /// Stato di ogni RICHIESTA del piano: quante volte di fila e' fallita e
    /// quando riprovarla. Per richiesta e non per slave: sullo stesso modulo
    /// le bobine possono rispondere benissimo mentre gli ingressi, che non
    /// ha, danno errore, e mettere in pausa tutto il modulo per colpa loro
    /// lascerebbe la plancia cieca su quello che invece si legge.
    FBackoff: TDictionary<string, TSlaveState>;
    /// A chi tocca, fra le richieste rimandate, il tentativo di questo giro.
    FRetryTurn: Integer;
    procedure SetState(AConnected: Boolean; const AMsg: string; AOk: Boolean);
    function CurrentPlan: TArray<TPollRequest>;
    function TakeWrite(out ACmd: TWriteCommand): Boolean;
    procedure AddOutcome(const ACmd: TWriteCommand; AOk: Boolean;
      const AError: string);
    procedure PublishReadings(const AReadings: TArray<TPollReading>);
    /// Svuota la coda dei comandi. False se la porta e' caduta.
    function FlushWrites(AModbus: TModbusRTU): Boolean;
    function ReadOne(AModbus: TModbusRTU;
      const ARequest: TPollRequest): TPollReading;
    /// True se lo slave va interrogato adesso; ASecondsLeft dice quanto manca
    /// al prossimo tentativo quando e' in pausa.
    function RequestDue(const ARequest: TPollRequest;
      out ASecondsLeft: Integer): Boolean;
    /// Vero se la richiesta non ha ancora sbagliato: quelle che sbagliano si
    /// rimandano in fondo al giro.
    function FirstTry(const ARequest: TPollRequest): Boolean;
    procedure NoteRequest(const ARequest: TPollRequest; AOk: Boolean);
  protected
    procedure Execute; override;
  public
    constructor Create(const APort: string; ABaud, ATimeoutMs,
      APollMs: Integer);
    destructor Destroy; override;
    /// Sostituisce il piano di polling; il giro dopo usa questo.
    procedure SetPlan(const APlan: TArray<TPollRequest>);
    /// Mette un comando in coda e torna subito.
    procedure EnqueueWrite(ASlave, AChannel: Integer; AValue: Boolean;
      const ACaption: string);
    /// Letture dell'ultimo giro completo, una volta sola: False se non ce ne
    /// sono di nuove da quando sono state ritirate.
    function TakeReadings(out AReadings: TArray<TPollReading>): Boolean;
    /// Esiti delle scritture eseguite da quando sono stati ritirati.
    function TakeOutcomes: TArray<TWriteOutcome>;
    function IsBusConnected: Boolean;
    procedure CurrentStatus(out AMsg: string; out AOk: Boolean);
    /// Chiede la chiusura e sveglia il thread; non attende.
    procedure Stop;
  end;

implementation

const
  /// Porta assente o occupata: si riprova, senza martellare.
  RETRY_MS = 2000;
  /// Errori di fila dopo i quali uno slave e' considerato assente. Tre, non
  /// uno: un disturbo sulla linea non deve mettere in pausa un modulo che c'e'.
  SLAVE_TOLERANCE = 3;
  /// Prima pausa di una richiesta che non risponde. Una plancia puo' essere disegnata
  /// prima che i moduli siano montati: le sue richieste non devono rubare mezzo
  /// secondo di timeout a ogni giro a quelle dei moduli che ci sono. Appena il
  /// modulo viene collegato riparte da solo, entro questa pausa.
  SLAVE_BACKOFF_MS = 5000;
  /// La pausa si allunga a ogni tentativo andato a vuoto, fino a questo
  /// massimo: un modulo che non c'e' costa un timeout al minuto invece di uno
  /// ogni cinque secondi, e il giro resta veloce per i moduli che ci sono.
  MAX_BACKOFF_MS = 60000;

{ TModbusWorker }

constructor TModbusWorker.Create(const APort: string; ABaud, ATimeoutMs,
  APollMs: Integer);
begin
  FPort := APort;
  FBaud := ABaud;
  FTimeoutMs := ATimeoutMs;
  FPollMs := APollMs;
  if FPollMs < 20 then
    FPollMs := 20;
  FLock := TCriticalSection.Create;
  FWake := TEvent.Create(nil, False, False, '');
  FWrites := TQueue<TWriteCommand>.Create;
  FOutcomes := TList<TWriteOutcome>.Create;
  FBackoff := TDictionary<string, TSlaveState>.Create;
  FStatus := 'Apertura porta...';
  FStatusOk := True;
  // Il thread parte solo quando tutto e' pronto.
  inherited Create(False);
end;

destructor TModbusWorker.Destroy;
begin
  // Prima si sveglia, poi si aspetta: l'attesa dura al massimo quanto la
  // richiesta in corso.
  Stop;
  inherited;
  FBackoff.Free;
  FOutcomes.Free;
  FWrites.Free;
  FWake.Free;
  FLock.Free;
end;

procedure TModbusWorker.Stop;
begin
  Terminate;
  FWake.SetEvent;
end;

procedure TModbusWorker.SetState(AConnected: Boolean; const AMsg: string;
  AOk: Boolean);
begin
  FLock.Enter;
  try
    FConnected := AConnected;
    FStatus := AMsg;
    FStatusOk := AOk;
  finally
    FLock.Leave;
  end;
end;

function TModbusWorker.IsBusConnected: Boolean;
begin
  FLock.Enter;
  try
    Result := FConnected;
  finally
    FLock.Leave;
  end;
end;

procedure TModbusWorker.CurrentStatus(out AMsg: string; out AOk: Boolean);
begin
  FLock.Enter;
  try
    AMsg := FStatus;
    AOk := FStatusOk;
  finally
    FLock.Leave;
  end;
end;

procedure TModbusWorker.SetPlan(const APlan: TArray<TPollRequest>);
begin
  FLock.Enter;
  try
    FPlan := Copy(APlan);
    FPlanFresh := True;
  finally
    FLock.Leave;
  end;
  FWake.SetEvent;
end;

function TModbusWorker.CurrentPlan: TArray<TPollRequest>;
begin
  FLock.Enter;
  try
    Result := Copy(FPlan);
    if FPlanFresh then
    begin
      FPlanFresh := False;
      // Piano nuovo: nessuno slave resta in pausa per colpa del piano vecchio.
      FBackoff.Clear;
    end;
  finally
    FLock.Leave;
  end;
end;

procedure TModbusWorker.EnqueueWrite(ASlave, AChannel: Integer;
  AValue: Boolean; const ACaption: string);
var
  Cmd: TWriteCommand;
begin
  Cmd.Slave := ASlave;
  Cmd.Channel := AChannel;
  Cmd.Value := AValue;
  Cmd.Caption := ACaption;
  FLock.Enter;
  try
    FWrites.Enqueue(Cmd);
  finally
    FLock.Leave;
  end;
  // Non aspetta il prossimo giro di polling: il comando parte appena la
  // richiesta in corso e' finita.
  FWake.SetEvent;
end;

function TModbusWorker.TakeWrite(out ACmd: TWriteCommand): Boolean;
begin
  FLock.Enter;
  try
    Result := FWrites.Count > 0;
    if Result then
      ACmd := FWrites.Dequeue;
  finally
    FLock.Leave;
  end;
end;

procedure TModbusWorker.AddOutcome(const ACmd: TWriteCommand; AOk: Boolean;
  const AError: string);
var
  O: TWriteOutcome;
begin
  O.Caption := ACmd.Caption;
  O.Value := ACmd.Value;
  O.Ok := AOk;
  O.Error := AError;
  FLock.Enter;
  try
    // Se nessuno li ritira (finestra occupata) non devono crescere all'infinito.
    while FOutcomes.Count > 200 do
      FOutcomes.Delete(0);
    FOutcomes.Add(O);
  finally
    FLock.Leave;
  end;
end;

function TModbusWorker.TakeOutcomes: TArray<TWriteOutcome>;
begin
  FLock.Enter;
  try
    Result := FOutcomes.ToArray;
    FOutcomes.Clear;
  finally
    FLock.Leave;
  end;
end;

procedure TModbusWorker.PublishReadings(const AReadings: TArray<TPollReading>);
begin
  FLock.Enter;
  try
    FReadings := AReadings;
    FReadingsFresh := True;
  finally
    FLock.Leave;
  end;
end;

function TModbusWorker.TakeReadings(out AReadings: TArray<TPollReading>): Boolean;
begin
  FLock.Enter;
  try
    Result := FReadingsFresh;
    if Result then
    begin
      AReadings := FReadings;
      FReadingsFresh := False;
    end;
  finally
    FLock.Leave;
  end;
end;

function TModbusWorker.FlushWrites(AModbus: TModbusRTU): Boolean;
var
  Cmd: TWriteCommand;
begin
  Result := True;
  while not Terminated and TakeWrite(Cmd) do
  begin
    try
      AModbus.WriteSingleCoil(Cmd.Slave, Cmd.Channel, Cmd.Value);
      AddOutcome(Cmd, True, '');
    except
      on E: Exception do
      begin
        AddOutcome(Cmd, False, E.Message);
        // La porta non c'e' piu' (chiavetta staccata): si riapre da capo.
        if not AModbus.IsConnected then
          Exit(False);
      end;
    end;
  end;
end;

/// Chiave di una richiesta nel registro delle pause.
function RequestKey(const ARequest: TPollRequest): string;
begin
  Result := Format('%d:%d:%d:%d', [ARequest.Slave, Ord(ARequest.Kind),
    ARequest.First, ARequest.Count]);
end;

function TModbusWorker.RequestDue(const ARequest: TPollRequest;
  out ASecondsLeft: Integer): Boolean;
var
  S: TSlaveState;
  Now: UInt64;
begin
  ASecondsLeft := 0;
  if not FBackoff.TryGetValue(RequestKey(ARequest), S) then
    Exit(True);
  if S.Failures < SLAVE_TOLERANCE then
    Exit(True);
  Now := GetTickCount64;
  Result := Now >= S.NextTry;
  if not Result then
    ASecondsLeft := Integer((S.NextTry - Now + 999) div 1000);
end;

function TModbusWorker.FirstTry(const ARequest: TPollRequest): Boolean;
var
  S: TSlaveState;
begin
  Result := not FBackoff.TryGetValue(RequestKey(ARequest), S) or
    (S.Failures = 0);
end;

procedure TModbusWorker.NoteRequest(const ARequest: TPollRequest;
  AOk: Boolean);
var
  S: TSlaveState;
  Attesa: Integer;
begin
  if not FBackoff.TryGetValue(RequestKey(ARequest), S) then
  begin
    S.Failures := 0;
    S.NextTry := 0;
  end;
  if AOk then
  begin
    // Ha risposto: torna una richiesta normale, fatta a ogni giro.
    S.Failures := 0;
    S.NextTry := 0;
  end
  else
  begin
    Inc(S.Failures);
    if S.Failures >= SLAVE_TOLERANCE then
    begin
      Attesa := SLAVE_BACKOFF_MS * (S.Failures - SLAVE_TOLERANCE + 1);
      if Attesa > MAX_BACKOFF_MS then
        Attesa := MAX_BACKOFF_MS;
      S.NextTry := GetTickCount64 + UInt64(Attesa);
    end;
  end;
  FBackoff.AddOrSetValue(RequestKey(ARequest), S);
end;

function TModbusWorker.ReadOne(AModbus: TModbusRTU;
  const ARequest: TPollRequest): TPollReading;
begin
  Result := Default(TPollReading);
  Result.Request := ARequest;
  Result.Sent := GetTickCount64;
  try
    case ARequest.Kind of
      rqCoils:
        Result.Bits := AModbus.ReadCoils(ARequest.Slave, ARequest.First,
          ARequest.Count);
      rqInputs:
        Result.Bits := AModbus.ReadDiscreteInputs(ARequest.Slave,
          ARequest.First, ARequest.Count);
      rqRegisters:
        Result.Regs := AModbus.ReadHoldingRegisters(ARequest.Slave,
          ARequest.First, ARequest.Count);
    end;
    Result.Ok := True;
  except
    on E: Exception do
    begin
      Result.Ok := False;
      Result.Error := E.Message;
    end;
  end;
end;

procedure TModbusWorker.Execute;
var
  Modbus: TModbusRTU;
  Plan: TArray<TPollRequest>;
  Rimandate: TArray<TPollRequest>;
  Readings: TArray<TPollReading>;
  Reading: TPollReading;
  I, J, Wait, Left: Integer;
  Started: UInt64;
begin
  NameThreadForDebugging('ModbusBus');
  Modbus := nil;
  try
    while not Terminated do
    begin
      // 1. porta aperta? altrimenti si riprova ogni RETRY_MS, senza rumore
      //    sul bus e senza bloccare nessuno.
      if Modbus = nil then
      begin
        try
          Modbus := TModbusRTU.Create(FPort, FBaud, FTimeoutMs);
          Modbus.Connect;
          // Porta riaperta: si riparte interrogando tutti, anche quelli che
          // prima erano muti.
          FBackoff.Clear;
          SetState(True, 'Connesso.', True);
        except
          on E: Exception do
          begin
            FreeAndNil(Modbus);
            SetState(False, 'Connessione fallita: ' + E.Message, False);
            FWake.WaitFor(RETRY_MS);
            Continue;
          end;
        end;
      end;

      Started := GetTickCount64;

      // 2. i comandi passano avanti alle letture.
      if not FlushWrites(Modbus) then
      begin
        FreeAndNil(Modbus);
        SetState(False, 'Porta caduta, riapertura...', False);
        Continue;
      end;

      // 3. un giro di letture, una richiesta per gruppo. Fra una richiesta e
      //    l'altra si guarda di nuovo la coda dei comandi: un pulsante premuto
      //    non deve aspettare la fine del giro.
      Plan := CurrentPlan;
      SetLength(Readings, 0);
      SetLength(Rimandate, 0);
      for I := 0 to High(Plan) do
      begin
        if Terminated then
          Break;
        if not FlushWrites(Modbus) then
          Break;
        if Modbus = nil then
          Break;
        // Una richiesta che non risponde da un po' si interroga di rado: una
        // plancia disegnata prima dei moduli resta comunque scorrevole, e
        // quello che c'e' viene letto alla velocita' giusta.
        if not RequestDue(Plan[I], Left) then
        begin
          Reading := Default(TPollReading);
          Reading.Request := Plan[I];
          Reading.Sent := GetTickCount64;
          Reading.Ok := False;
          Reading.Error := Format('non risponde, riprovo fra %d s', [Left]);
          Readings := Readings + [Reading];
          Continue;
        end;
        // Le richieste gia' in difficolta' si rimandano in fondo al giro, e
        // se ne riprova UNA SOLA per giro: ogni tentativo a vuoto costa un
        // timeout, e otto timeout davanti alle spie voleva dire vedere un
        // allarme quattro secondi dopo che era arrivato.
        if not FirstTry(Plan[I]) then
        begin
          Rimandate := Rimandate + [Plan[I]];
          Continue;
        end;
        Reading := ReadOne(Modbus, Plan[I]);
        NoteRequest(Plan[I], Reading.Ok);
        Readings := Readings + [Reading];
      end;

      // Le rimandate: una per giro, a turno, cosi' un modulo che torna viene
      // ritrovato senza rallentare gli altri.
      if (Length(Rimandate) > 0) and not Terminated and (Modbus <> nil) then
      begin
        if FRetryTurn >= Length(Rimandate) then
          FRetryTurn := 0;
        Reading := ReadOne(Modbus, Rimandate[FRetryTurn]);
        NoteRequest(Rimandate[FRetryTurn], Reading.Ok);
        Readings := Readings + [Reading];
        Inc(FRetryTurn);
        // Le altre rimandate: si dice che non sono state fatte adesso, senza
        // spacciarle per fallite.
        for J := 0 to High(Rimandate) do
          if J <> FRetryTurn - 1 then
          begin
            Reading := Default(TPollReading);
            Reading.Request := Rimandate[J];
            Reading.Sent := GetTickCount64;
            Reading.Ok := False;
            Reading.Error := 'non risponde, riprovata a turno';
            Readings := Readings + [Reading];
          end;
      end;
      if Length(Readings) > 0 then
        PublishReadings(Readings);
      if Terminated then
        Break;
      if (Modbus <> nil) and not Modbus.IsConnected then
      begin
        FreeAndNil(Modbus);
        SetState(False, 'Porta caduta, riapertura...', False);
        Continue;
      end;

      // 4. riposo fino al prossimo giro, interrompibile da un comando o
      //    dalla chiusura.
      Wait := FPollMs - Integer(GetTickCount64 - Started);
      if Wait < 1 then
        Wait := 1;
      FWake.WaitFor(Wait);
    end;
  finally
    if Modbus <> nil then
    begin
      Modbus.Disconnect;
      Modbus.Free;
    end;
    SetState(False, 'Bus chiuso.', True);
  end;
end;

end.