Использование Delphi + Jedi, потеря USB-данных, когда устройство отправляет их "слишком быстро"

Question

С помощью Delphi XE2 а также TJvHidDevice класс от Jedi библиотека, мне удалось успешно связаться с USB-устройством (pic32mx7 доска, с моим кодом, работающим на нем). Обычный способ "отправить запрос, дождаться одного ответа" работает.

Проблема с командой, которая приводит к большему количеству последовательных ответов. Если эти ответы отправляются устройством как можно быстрее - или даже если я добавляю небольшую задержку между ними, например 5 мс, - я теряю пакеты (отчеты? Кадры?). OnDeviceData событие просто не похоже на огонь для всех из них. Если я добавлю большие задержки в коде устройства, проблема исчезнет.

я использовал USBPcap программа для захвата данных USB и вывода их в файл, который, как только я открою его в WireShark, содержит все данные, отправленные устройством (я отправляю 255 пакетов в качестве теста, все нули и одна "1" смещают свое место на 1 позиция в каждом пакете). Итак, я думаю, что и устройство, и Windows делают свою работу.

Чтобы убедиться, что мой код Delphi не является неисправным, я попробовал пример проекта Jedi "DevReader" (здесь код main.pas), который выводит данные на экран, а также пропускает пакеты.

Я чувствую, что в сети должно быть больше информации о классах USB джедаев, но у меня возникают проблемы с их поиском.

Возможно, я смогу избежать этой проблемы, агрегируя / уплотняя ответы устройства, но все же хотел бы знать, что происходит.

Редактировать:

1. Пробовал из консольного приложения: пакеты больше не терялись.
2. Модифицировано демо-приложение Jedi для подсчета только полученных пакетов и обновления метки счетчика на экране (без принудительной перерисовки окна) - нет потерянных пакетов.
3. Добавлен sleep(1) в событии OnData - нет потерянных пакетов.
4. Добавлен sleep(2) в событии OnData - потеря пакетов снова.

Это похоже на то, что поток джедаев, который читает данные, не должен задерживаться какой-либо обработкой - не должна ли происходить некоторая буферизация данных (в Windows?), Которая допускала бы такой тип обработки? Судя по "шаблону" потери пакетов, создается впечатление, что буферизация существует, но этого недостаточно, потому что я могу получить, например, 30 пакетов, затем потерять 5, затем получить еще 20 и т. Д.

Я изменю свой код, чтобы скопировать данные и как можно быстрее выйти из события OnData, чтобы у потока было минимальное время простоя, и я сообщу о результате.

Answer 1

Поскольку причина проблемы, по-видимому, связана с периодом времени, когда поток чтения USB блокируется Synchronise, т. е. обработку данных, выполняемую основным потоком, я внес в код потока, (класс TJvHidDeviceReadThread, блок JvHidControllerClass.pas). Любой код, который использовал этот модуль и содержащиеся в нем классы, должен работать без каких-либо изменений, ничего публичного не изменилось.

Новое поведение: каждый раз, когда данные читаются, они помещаются в потокобезопасный список. Вместо синхронизации теперь используется Queue, но только если он еще не поставлен в очередь. Метод Queued читает из списка безопасных потоков, пока он не станет пустым. Он запускает событие (то же, что и в старом коде) для каждого буферизованного отчета в списке. Как только список пуст, флаг "Очередь" сбрасывается, и при следующем чтении очередь снова становится.

В тестах до сих пор я не встречал потерянных пакетов.

Класс потока был расширен:

  TJvHidDeviceReadThread = class(TJvCustomThread)
  private
    FErr: DWORD;

    // start of additions
    ReceivedReports : TThreadList;
    Queued: boolean;
    procedure PushReceivedReport(const bytes: array of byte; const NumBytesRead: cardinal);
    function PopReceivedReport(var ReportID: byte; var ReportBytes: TBytes): boolean;
    procedure FlushBuffer;
    // end of additions

    procedure DoData;
    procedure DoDataError;
    constructor CtlCreate(const Dev: TJvHidDevice);
  protected
    procedure Execute; override;
  public
    Device: TJvHidDevice;
    NumBytesRead: Cardinal;
    Report: array of Byte;
    constructor Create(CreateSuspended: Boolean);
    //added destructor:
    destructor Destroy; override;
  end;

В разделе реализации было изменено следующее:

constructor TJvHidDeviceReadThread.CtlCreate(const Dev: TJvHidDevice);
begin
  inherited Create(False);
  // start of changes
  ReceivedReports := TThreadList.Create; 
  // end of changes
  Device := Dev;
  NumBytesRead := 0;
  SetLength(Report, Dev.Caps.InputReportByteLength);
end;

procedure TJvHidDeviceReadThread.Execute;
...
...
...
    //replaced: Synchronize(DoData); with:
    PushReceivedReport (Report, NumBytesRead);
...

И было добавлено следующее:

type

TReport = class
  ID: byte;
  Bytes: TBytes;
end;

destructor TJvHidDeviceReadThread.Destroy;
var
  l: TList;
begin
  RemoveQueuedEvents (self);
  try
    l := ReceivedReports.LockList;
    while l.Count>0 do
      begin
        TReport(l[0]).Free;
        l.Delete(0);
      end;
  finally
    ReceivedReports.UnlockList;
    FreeAndNil (ReceivedReports);
  end;

  inherited;
end;

procedure TJvHidDeviceReadThread.FlushBuffer;
var
  ReportID: byte;
  ReportBytes: TBytes;
begin
  while PopReceivedReport (ReportID, ReportBytes) do
        Device.OnData(Device, ReportID, ReportBytes, length(ReportBytes));
end;

function TJvHidDeviceReadThread.PopReceivedReport(var ReportID: byte; var ReportBytes: TBytes): boolean;
var
  l: TList;
  rep: TReport;
begin
  l := ReceivedReports.LockList;
  rep := nil;
  try
    result := l.Count>0;
    if result
      then
        begin
          rep := l[0];
          l.Delete(0);
        end
      else Queued := false;
  finally
    ReceivedReports.UnlockList;
  end;

  if result then
    begin
      ReportID := rep.ID;
      SetLength(ReportBytes, length(rep.Bytes));
      System.move (rep.Bytes[0], ReportBytes[0], length(rep.Bytes));
      rep.Free;
    end;
end;

procedure TJvHidDeviceReadThread.PushReceivedReport(const bytes: array of byte; const NumBytesRead: cardinal);
var
  rep: TReport;
begin
  rep := TReport.Create;
  setlength (rep.Bytes, NumBytesRead-1);
  rep.ID := Bytes[0];
  System.move (Bytes[1], rep.Bytes[0], NumBytesRead-1);

  // explicitely lock the list just to provide a locking mechanism for the Queue flag as well
  ReceivedReports.LockList;
  try
    if not Queued then
      begin
        Queued := true;
        Queue (FlushBuffer);
      end;
    ReceivedReports.Add(rep);
  finally
    ReceivedReports.UnlockList;
  end;
end;