Асинхронная работа с libusb 1.009.03.2021 17:17

Поэтому, во-первых, было бы полезно видеть, какой их объём уже прошёл в буфер, чтобы представлять, работает система или нет. Ну, и во-вторых, если данных не предвидится, а всё интересное уже попало к нам в память, надо иметь возможность прекратить приём и начать разбор того, что уже накопилось. Ни то, ни другое невозможно при использовании функций, которые были рассмотрены в той статье. По крайней мере, со стороны PC. Без читов, добавленных в «прошивку» ПЛИС.

Сегодня мы научимся обращаться к библиотеке libusb асинхронным методом. Это позволит и грубо отслеживать объём уже пришедших данных, и прерывать работу в любой момент, и даже повысить общую производительность системы. Причём всё это будет сделано только за счёт вызова штатных функций libusb. Код для FX3 и ПЛИС мы для этого дорабатывать не будем. Итак, приступаем.

Предыдущие статьи цикла:

1. Начинаем опыты с интерфейсом USB 3.0 через контроллер семейства FX3 фирмы Cypress
2. Дорабатываем прошивку USB 3.0, используя анализатор SignalTap, встроенный в среду разработки Quartus
3. Учимся работать с USB-устройством и испытываем систему, сделанную на базе контроллера FX3
4. Боремся с таймаутами при использовании USB 3.0 через контроллер FX3, возникающими при определенных условиях
5. Добавляем поддержку Vendor-команд к USB3.0 устройству на базе FX3
6. Делаем блок SPI to AVALON_MM для USB-устройства на базе FX3

1 Зачем всё это

int res = libusb_bulk_transfer(tester.m_hUsb,0x81,(uint8_t*)pData,bytesCnt,&actualLength,10000);

В противовес такой работе (её ещё называют синхронной), в любой уважающей себя USB-библиотеке должна быть ещё и асинхронная. Там мы вызываем функцию чтения, и управление немедленно возвращается нам. Что будет дальше, зависит от конкретной библиотеки. Где-то в нужный момент будет переведён в активное состояние объект «Событие». В libusb будет вызвана Callback-функция. Короче, нам поступит сообщение о том, что наш запрос выполнен, как на вступительном рисунке с галочкой.

И никто не запрещает нам в любой момент вызвать функцию, отменяющую процесс обмена. При этом, если часть данных уже прошла по каналу, они будут нам всё равно выданы. Ничего не пропадёт.

Скоро мы рассмотрим, как это работает, но сначала я объясню, почему не стал рассказывать об асинхронной работе немедленно после синхронной. Дело в том, что при проверке асинхронной работы, мы должны провоцировать ситуации с таймаутами. А значит, наш таймер должен не молотить от начала жизни и до бесконечности, а выдавать строго дозированные порции данных. Чтобы это делать, мы должны управлять им. Собственно, в предыдущих двух статьях мы и осваивали этот процесс. Теперь мы вполне можем посылать запросы на шину AVALON_MM. Поэтому, подключив таймер к этой шине, мы сможем управлять и им. Приступаем.

2 Дорабатываем таймер

Таким образом, наша программа сможет вырабатывать строго заданное количество 16-разрядных слов, которые уйдут в шину AVALON_ST, дальше — в FIFO, а из него — в FX3 и USB 3.0.

Под катом — получившийся Verilog код таймера.

module Timer_ST (
  input              clk,
  input              reset,

  input [2:0]   avalon_mm_address,
  input [31:0]  avalon_mm_data_in,
  input         avalon_mm_wr,
	
  input  logic       source_ready,
  output logic       source_valid,
  output logic[15:0] source_data
	
);
    logic [31:0] cnt = 0;
    logic [31:0] counter = 0;

    always @ (posedge clk)
    begin 
        // На том конце очередь переполнена
        // Значит, когда она освободится - начнём
        // слать данные не сразу...
        if (reset == 1)
        begin
            counter <= 0;
            cnt <= 0;
        end else
        begin

            if (avalon_mm_wr)
            begin 
               cnt <= avalon_mm_data_in;
               counter <= 0;
            end else
            begin
               counter <= counter + 1;
               if ((source_ready==1)&&(cnt != 0))
               begin
                  cnt <= cnt - 1;
               end 
            end
        end
    end

    assign source_valid = (source_ready!=0) && (cnt != 0);
    assign source_data [15:0] = counter [15:0];

endmodule

3 Первый эксперимент

3.1 Подготовка

Чтобы сделать простейший асинхронный запрос, надо проделать следующий путь:

1. Выделить память для структуры libusb_transfer, для чего имеется функция libusb_alloc_transfer ();
2. Заполнить поля созданной структуры. Причём не надо мучиться с ручным заполнением. В зависимости от типа транзакции мы можем использовать функции libusb_fill_control_setup (), libusb_fill_control_transfer (), libusb_fill_bulk_transfer (), libusb_fill_interrupt_transfer () или libusb_fill_iso_transfer (). Есть ещё заполнение потоковой структуры, но врать не буду, я с нею не разобрался.
3. На шаге 2 заполняется указатель на функцию обратного вызова. Эту функцию надо написать.
4. Запустить передачу в работу, вызвав libusb_submit_transfer ().

3.2 Работа

Если транзакция была не последняя, после завершения работы обработки текущей можно, прямо находясь в функции обратного вызова, снова обратиться к функции libusb_submit_transfer (), чтобы инициировать новую транзакцию. Поэтому получаем такое рабочее колесо (фраза «функция обратного вызова» заняла бы на нём много места, поэтому она заменена на «CallBack-функция»):

То есть, после отработки libusb_submit_transfer (), Callback-функция, получив управление, под конец своей работы снова вызывает libusb_submit_transfer () и завершается. Через некоторое время она снова получает управление, снова вызывает libusb_submit_transfer () и т.д.

При этом функция libusb_submit_transfer () возвращает управление немедленно, а следующий вызов CallBаck-функции произойдёт через существенный промежуток времени. Таким образом, всё это время основная программа может продолжать своё выполнение. Мало того, CallBack-функция по своей сути похожа на обработчик прерывания микроконтроллера. Она совершенно не изменяет ход работы основной программы. Появились данные — функция была вызвана. Функция отработала — управление вернулось в то место, где программа была в момент прихода данных. А как эта функция взаимодействует с основной программой — всё в руках программиста.

3.3 Очень важный момент

Вызывать эту функцию можно и в текущем потоке, но можно создать для этого отдельный поток.
В общем, рисунок изменяется так:

3.4 Окончание работы

3.5 Практический пример

Вот такую я сделал функцию обратного вызова. Она не выполняет никаких полезных действий, я просто ставил там точки останова и изучал в отладчике структуры пришедших данных. Ну, и статистику она мне строит. Тест мне был нужен, чтобы убедиться, что я понимаю принципы работы. В конце, как я уже и говорил, вызывается libusb_submit_transfer () для запуска приёма новой посылки. Объёмы изохронного трафика от микрофона (а ловил я именно их) таковы, что заботиться о скорости мне не требовалось. Там буквально килобайты в секунду идут.

void LIBUSB_CALL CIsoWriteTest::cb_xfr(struct libusb_transfer *xfr)
{
    CIsoWriteTest* pForm = (CIsoWriteTest*) xfr->user_data;
    if (xfr->status == LIBUSB_TRANSFER_COMPLETED)
    {
        uint minBlockSize = 100000;
        uint maxBlockSize = 0;
        uint total = 0;
        int nBlocks = 0;
        for (int i=0;inum_iso_packets;i++)
        {
            if (xfr->iso_packet_desc[i].status == LIBUSB_TRANSFER_COMPLETED)
            {
                if (xfr->iso_packet_desc[i].actual_length > maxBlockSize)
                {
                    maxBlockSize = xfr->iso_packet_desc[i].actual_length;
                }
                if (xfr->iso_packet_desc[i].actual_length < minBlockSize)
                {
                    minBlockSize = xfr->iso_packet_desc[i].actual_length;
                }
                nBlocks += 1;
                total += xfr->iso_packet_desc[i].actual_length;
            }
        }
        pForm->ui->m_lblSize->setText(QString ("%1 Byttes transfered in %2 blocks").arg(total).arg(nBlocks));
        pForm->ui->m_lblMinBlockSize->setText(QString ("Min Block Size = %1 bytes").arg(minBlockSize));
        pForm->ui->m_lblMaxBlockSize->setText(QString ("Max Block Size = %1 bytes").arg(maxBlockSize));
        pForm->ui->m_lblBytesPerSec->setText("***");
    } else
    {

        pForm->ui->m_lblSize->setText(DecodeTransferStatus(xfr->status));
        pForm->ui->m_lblMinBlockSize->setText("***");
        pForm->ui->m_lblMaxBlockSize->setText("***");
        pForm->ui->m_lblBytesPerSec->setText("***");
    }
    if (libusb_submit_transfer(xfr) < 0)
    {
        // todo Catch Errors
    }
}

    // Allocate transfer
    m_xfr = libusb_alloc_transfer(TEST_NUM_PACKETS);
    if (m_xfr == 0)
    {
        QMessageBox::critical (this,"Error","Cannot Allocate Transfer");
        return;
    }

    m_buffer.resize(m_epParams.epMmaxPacketSize * TEST_NUM_PACKETS);

    libusb_fill_iso_transfer(m_xfr, m_epParams.hDev, m_epParams.nEndPoint,(unsigned char*) m_buffer.constData(),
            m_buffer.size(), TEST_NUM_PACKETS, cb_xfr, this, 1000);
    libusb_set_iso_packet_lengths(m_xfr, m_epParams.epMmaxPacketSize);

#define TEST_NUM_PACKETS 128

    int res = libusb_submit_transfer(m_xfr);
    if (res != 0)
    {
        QMessageBox::critical(this,"Error",libusb_error_name(res));
    }

    m_thread.start();

void CMonitorIsoTransactionThread::run()
{
    while (!isInterruptionRequested())
    {
        libusb_handle_events(NULL);
    }
}

4 Пользуемся очередью запросов

4.1 Работа с буфером

Но мой опыт говорит, что при скоростной работе с огромными объёмами данных функции копирования памяти отнимают большой процент времени. Поэтому я пошёл другим путём. Я нарезаю буфер на маленькие кусочки (размером с одну транзакцию) и настраиваю структуру libusb_transfer на приём сразу туда, куда нужно мне. Поэтому дополнительные копирования уже не требуются. Функция обратного вызова просто перенаправляет указатель на очередной участок буфера и снова ставит его в очередь.

4.2 Функция обратного вызова

Как я уже говорил, эта функция как-то должна взаимодействовать с основной программой. Для этого нужна кучка переменных. Я предпочёл объединить их в единую структуру:

    struct asyncParams
    {
        // Указатель на большой буфер, в который
        // мы принимаем массив. Он может быть размером
        // в сотни мегабайт
        uint8_t* pData;

        // Каждая транзакция принимает маленький кусочек.
        // Например, 64 килобайта. В этой переменной лежит
        // текущее смещение. Поставили транзакцию в очередь - 
        // сдвинулись в буфере. Короче, это указатель 
        // запрошенной части буфера.
        int dataOffset;

        // Размер буфера в байтах. Полезен, чтобы знать,
        // когда следует прекращать работу. Если смещение
        // дошло до этой величины - функция обратного вызова
        // перестанет создавать новые запросы на передачу
        int dataSizeInBytes;

        // Размер одной передачи
        int transferLen;

        // Используется для отображения заполненности буфера,
        // увеличивается по факту прихода новой порции данных.
        // То есть, это указатель фактически заполненного буфера
        int actualTranfered;
    };

void SpiToAvalonDemo::ReadDataTranfserCallback(libusb_transfer *transfer)
{
    SpiToAvalonDemo* pClass = (SpiToAvalonDemo*) transfer->user_data;
    switch (transfer->status )
    {
    case LIBUSB_TRANSFER_COMPLETED:
        // Отметили, что принят очередной блок данных
        pClass->m_asyncParams.actualTranfered += transfer->length;

        // Если буфер принят ещё не весь
        if (pClass->m_asyncParams.dataOffset < pClass->m_asyncParams.dataSizeInBytes)
        {
            // Новая пачка ляжет вот сюда
            transfer->buffer = pClass->m_asyncParams.pData+pClass->m_asyncParams.dataOffset;

            // Сдвигаем указатель на следующий блок в буфере
            pClass->m_asyncParams.dataOffset += pClass->m_asyncParams.transferLen;

            // Запустили передачу
            libusb_submit_transfer(transfer);
        }
        break;
    case LIBUSB_TRANSFER_CANCELLED:
    {
        pClass->m_cancelCnt -= 1;
    }
        break;
    default:
        break;
    }
}

4.3 Запуск процесса и моторная часть

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

Сначала я заполняю поля структуры, которая используется для связи с функцией обратного вызова:

    m_asyncParams.pData = (uint8_t*) pData;
    m_asyncParams.dataOffset = 0;
    m_asyncParams.dataSizeInBytes = bytesCnt;
    m_asyncParams.transferLen = transferSize;

    m_asyncParams.actualTranfered = 0;

    for (int i=0;i

    QElapsedTimer timer;
    timer.start();

    // Separated loop for more careful time checking
    for (int i=0;i

   while (m_asyncParams.actualTranfered

    for (int i=0;i

bool SpiToAvalonDemo::AsyncStep(uint16_t *pData, const int bytesCnt, const int transferSize, const int nTransfersInParallel)
{
    QElapsedTimer timer;

    m_asyncParams.pData = (uint8_t*) pData;
    m_asyncParams.dataOffset = 0;
    m_asyncParams.dataSizeInBytes = bytesCnt;
    m_asyncParams.transferLen = transferSize;

    m_asyncParams.actualTranfered = 0;

    // Allocate Transfers
    for (int i=0;i

void SpiToAvalonDemo::on_m_btnAsync128M_clicked()
{
    static const int len = 128 * 1024 * 1024;
    // Set Up Timer to 128 megabytes
    QByteArray ar1;
    ar1.resize(len);

    // Заполнили буфера FX3, иначе подвиснет
    m_tester.WriteDword(0,0x10000);

    for (int i=1;i<16;i++)
    {
        // Set Transfer Size for timer
        m_tester.WriteDword(0,len/2);
        AsyncStep ((uint16_t*)ar1.constData(),len,65536,i);
    }
}

void SpiToAvalonDemo::on_m_btnAsync128M_clicked()
{
    static const int len = 128 * 1024 * 1024;
    // Set Up Timer to 128 megabytes
    QByteArray ar1;
    ar1.resize(len);

    // Заполнили буфера FX3, иначе подвиснет
    m_tester.WriteDword(0,0x10000);

    qDebug()<<"";
    qDebug()<<"";
    qDebug()<<"";
//    for (int i=1024;i<512*1024;i*=2)
    for (int i=1024;i<32*1024;i+=1024)
    {
        // Set Transfer Size for timer
        m_tester.WriteDword(0,len);
        AsyncStep ((uint16_t*)ar1.constData(),len-128*1024,i,1);
    }
    qDebug()<<"";
    qDebug()<<"";
    qDebug()<<"";
    //    for (int i=1024;i<512*1024;i*=2)
    for (int i=1024;i<32*1024;i+=1024)
    {
        // Set Transfer Size for timer
        m_tester.WriteDword(0,len);
        AsyncStep ((uint16_t*)ar1.constData(),len-128*1024,i,2);
    }
    qDebug()<<"";
    qDebug()<<"";
    qDebug()<<"";
    //    for (int i=1024;i<512*1024;i*=2)
    for (int i=1024;i<32*1024;i+=1024)
    {
        // Set Transfer Size for timer
        m_tester.WriteDword(0,len);
        AsyncStep ((uint16_t*)ar1.constData(),len-128*1024,i,4);
    }
}

void SpiToAvalonDemo::UsbLoop(uint32_t timeIn_ms)
{
    timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = 500000;

    QElapsedTimer timer;
    timer.start();
    while (timer.elapsed()

    UsbLoop (500);
    m_tester.WriteDword(0,len/4);
    UsbLoop (500);
    m_tester.WriteDword(0,(len/4)-2);
    UsbLoop (1500);
    m_tester.WriteDword(0,2);

void SpiToAvalonDemo::on_m_btnWithLatencyProblem_clicked()
{
    static const int len = 0x10000;

    // Set Up Timer to 128 megabytes
    QByteArray ar1;
    ar1.resize(0x100000);

    m_asyncParams.pData = (uint8_t*)ar1.constData();
    m_asyncParams.dataOffset = 0;
    m_asyncParams.dataSizeInBytes = len;
    m_asyncParams.transferLen = len;
    m_asyncParams.actualTranfered = 0;

    libusb_transfer* transfer = libusb_alloc_transfer(0);
    libusb_fill_bulk_transfer (transfer,m_tester.m_hUsb,0x81,
                               m_asyncParams.pData+m_asyncParams.dataOffset,len,ReadDataTranfserCallback,
                               this,60000);
    m_asyncParams.dataOffset += len;

    libusb_submit_transfer(transfer);

    UsbLoop (500);
    m_tester.WriteDword(0,len/4);
    UsbLoop (500);
    m_tester.WriteDword(0,(len/4)-2);
    UsbLoop (1500);
    m_tester.WriteDword(0,2);

    while (m_asyncParams.actualTranfered

void SpiToAvalonDemo::timerEvent(QTimerEvent *event)
{
    Q_UNUSED(event)
    static int prevPercent = 0;
    int percent = (int)(((int64_t)m_asyncParams.actualTranfered * 100LL)/(int64_t)m_asyncParams.dataSizeInBytes);
    if (prevPercent != percent)
    {
        ui->m_progressForCancel->setValue(percent);
        prevPercent = percent;
    }
    if (percent == 100)
    {
        StopProcess();
    }
}

void SpiToAvalonDemo::StopProcess()
{
    // Всё, больше мотор не нужен! Глушим его!
    m_transactionsThread.requestInterruption();
    while (m_transactionsThread.isRunning())
    {
        QThread::msleep(10);
    }
    // Освободили память
    for (int i=0;im_progressForCancel->setValue(0);
    // Таймер, собственно, тоже стал не нужен
    killTimer(m_timeId);
}

void CTransactionThread::run()
{
    timeval tv;
    tv.tv_sec = 0;
    tv.tv_usec = 500000;

    while (!isInterruptionRequested())
    {
        libusb_handle_events_timeout(m_libusb_ctx,&tv);
    }
}

    m_cancelCnt = 0;
    for (int i=0;i= 0)
        {
            m_cancelCnt += 1;
        }
    }

    QAtomicInt m_cancelCnt;

    case LIBUSB_TRANSFER_CANCELLED:
    {
        pClass->m_cancelCnt -= 1;
    }

    // Ждём фактической отмены пересылки
    while (m_cancelCnt != 0)
    {
        QThread::msleep(100);
    }

    // Остановили работу.
    StopProcess();

void SpiToAvalonDemo::on_m_btnCancel_clicked()
{
    m_cancelCnt = 0;
    for (int i=0;i= 0)
        {
            m_cancelCnt += 1;
        }
    }
    // Ждём фактической отмены пересылки
    while (m_cancelCnt != 0)
    {
        QThread::msleep(100);
    }

    // Остановили работу.
    StopProcess();

}

    m_analyzerData.resize(120*1024*1024);

    m_asyncParams.pData = (uint8_t*)m_analyzerData.constData();
    m_asyncParams.dataOffset = 0;
    m_asyncParams.dataSizeInBytes = m_analyzerData.size()*sizeof(uint16_t);
    m_asyncParams.transferLen = 0x20000;
    m_asyncParams.actualTranfered = 0;

    for (int i=0;i

    for (int i=0;i