c - 使用异步 I/O 的 Libusb 竞争条件

Question

我正在使用 libusb 与配置为 USB 设备的飞利浦 ISP1362 进行通信。我能够使用同步 I/O 成功环回数据，没有任何问题。由于某种原因，在使用异步 I/O 时似乎存在竞争条件。

我正在使用背靠背 OUT-IN 传输来传输 64 字节数据包。有时，当我运行我的程序时，libusb 会抛出超时错误并且一些环回数据会丢失。在使用我的 Beagle 12 分析 USB 总线时，我可以看到 OUT-IN 事务在应该是(OUT-IN-OUT-IN)时出现了乱序(即 OUT-OUT-IN-TIMEOUT)。

更新 传输在回调函数中出现乱序，这很奇怪，因为它们与总线分析器上的实际情况不一致。

示例 1:(进-出-进-出)

main(): submitting transfer 0, endpoint 1
main(): submitting transfer 1, endpoint 82
main(): submitting transfer 2, endpoint 1
main(): submitting transfer 3, endpoint 82
xfr_cb(): count 0, status = 0, endpoint = 82, actual_length = 64, completed = 0
xfr_cb(): count 1, status = 0, endpoint = 1, actual_length = 64, completed = 0
xfr_cb(): count 2, status = 0, endpoint = 82, actual_length = 64, completed = 0
xfr_cb(): count 3, status = 0, endpoint = 1, actual_length = 64, completed = 0
completed

示例 2:(出-入-出)

main(): submitting transfer 0, endpoint 1
main(): submitting transfer 1, endpoint 82
main(): submitting transfer 2, endpoint 1
main(): submitting transfer 3, endpoint 82
xfr_cb(): count 0, status = 0, endpoint = 1, actual_length = 64, completed = 0
xfr_cb(): count 1, status = 0, endpoint = 82, actual_length = 64, completed = 0
xfr_cb(): count 2, status = 0, endpoint = 82, actual_length = 64, completed = 0
xfr_cb(): count 3, status = 0, endpoint = 1, actual_length = 64, completed = 0
completed

下面是分析器的截图:

代码如下:

#include <stdlib.h>
#include <stdio.h>

#include <libusb-1.0/libusb.h>

/* Specify VENDOR_ID and PRODUCT_ID for device */
#define VENDOR_ID   0x0471
#define PRODUCT_ID  0x3630

/* Define number of bytes to transfer */
#define EP_SIZE 64               // bytes
#define TRANSFERS 4              // number of transfers
#define BYTES EP_SIZE*TRANSFERS
#define TIMEOUT 3*1000           // milliseconds

/* Use a global variable to keep the device handle */
static struct libusb_device_handle *devh = NULL;

/* use a global variable to keep the context */
static struct libusb_context *usb_context = NULL;

/* count variable */
int count = 0;

/* The Endpoint addresses are hard-coded.  You should use libusb -v to find
 * the values corresponding to device
 */
static int ep_in  = 0x82;
static int ep_out = 0x01;

void xfr_cb(struct libusb_transfer *transfer )
{
  int *completed = transfer->user_data;

  /* callback - This is called after the transfer has been received by libusb */
  fprintf(stderr, "xfr_cb(): count %d, status = %d, endpoint = %x, actual_length = %d, completed = %d\n",
          count,
          transfer->status,
          transfer->endpoint,
          transfer->actual_length,
          *completed);
  if (transfer->status != LIBUSB_TRANSFER_COMPLETED)
  {
    /* Error! */
    fprintf(stderr, "Error: %s\n", libusb_error_name((int)transfer->status));
  }

  if (count == TRANSFERS-1)
    *completed = 1;
  count++;
}

int main(int argc, char **argv)
{
  int ep_addr;
  int completed = 0;
  unsigned char *buf;
  size_t length = 64;
  int n;
  int i;
  int rc;

  /* Initialize libusb */
  rc = libusb_init(NULL);
  if (rc < 0)
  {
    fprintf(stderr, "Error Initializing libusb: %s\n", libusb_error_name(rc));
    exit(1);
  }

  /* Set debugging output to max level */
  libusb_set_debug(NULL, 3);

  /* Look for a specific device and open it */
  devh = libusb_open_device_with_vid_pid(NULL, VENDOR_ID, PRODUCT_ID);
  if (!devh)
  {
    fprintf(stderr, "Error finding USB device\n");
    goto out;
  }

  /* allocate memory */
  buf = malloc(length);

  /* start with OUT transfer */
  ep_addr = ep_out;

  /* queue up alternating OUT-IN transfers */
  for (i = 0; i < TRANSFERS; i++)
  {
    /* fill the buffer with incrementing data */
    for (n = 0; n < EP_SIZE; n++)
    {
      buf[n] = i+n;
    }

    /* Set up the transfer object */
    struct libusb_transfer *transfer;
    transfer = libusb_alloc_transfer(0);
    libusb_fill_bulk_transfer(transfer, devh, ep_addr, buf, EP_SIZE, xfr_cb, &completed, TIMEOUT); /* callback data = &completed */

    /* Submit the transfer object */
    libusb_submit_transfer(transfer);
    fprintf(stderr, "main(): submitting transfer %d, endpoint %x\n", i, ep_addr);

    /* alternate writing and reading for loopback */
    ep_addr = (ep_addr == ep_out) ? ep_in : ep_out;
  }

  /* Handle Events */
  while (!completed)
  {
    rc = libusb_handle_events_completed(NULL, &completed);
    if (rc < 0)
    {
      if (rc == LIBUSB_ERROR_INTERRUPTED)
        continue;
      fprintf(stderr, "Transfer Error: %s", libusb_error_name(rc));
      continue;
    }
  }

  fprintf(stderr, "completed\n");

  /* Release the interface */
  libusb_release_interface(devh, 0);


  /* Close the device handle */
  if (devh)
    libusb_close(devh);


out:
  if (devh)
  {
    libusb_close(devh);
  }
  libusb_exit(NULL);

  return rc;
}

更新 2 我成功消除了超时。 libusb 超时的原因是主机在总线上间歇性地发送两个连续的 OUT 事务。

分析器截图:

以下是工作代码(无超时)。毫无问题地运行了数千次

static void LIBUSB_CALL xfr_cb(struct libusb_transfer *transfer )
{
  int *completed = transfer->user_data;
  unsigned char *wbuf, *rbuf;
  size_t length = 64;

  fprintf(stderr, "xfr_cb(): status = %d, endpoint = %x, actual_length = %d\n",
          transfer->status,
          transfer->endpoint,
          transfer->actual_length);

  *completed = 1;
}

int main(int argc, char **argv)
{
  const struct libusb_version *version;
  int ep_addr;
  int completed = 0;
  unsigned char *buf, *wbuf1, *wbuf2, *rbuf1, *rbuf2;
  size_t length = 64;
  int n;
  int m;
  int i;
  int rc;

  /* Get libusb version */
  version = libusb_get_version();
  fprintf(stderr, "libusb version: %d.%d.%d.%d\n", version->major, version->minor, version->micro, version->nano);

  /* Initialize libusb */
  rc = libusb_init(NULL);
  if (rc < 0)
  {
    fprintf(stderr, "Error Initializing libusb: %s\n", libusb_error_name(rc));
    exit(1);
  }

  /* Set debugging output to max level */
  libusb_set_debug(NULL, 3);

  /* Look for a specific device and open it */
  handle = libusb_open_device_with_vid_pid(NULL, VENDOR_ID, PRODUCT_ID);
  if (!handle)
  {
    fprintf(stderr, "Error finding USB device\n");
    goto out;
  }

  /* claim interface */
  rc = libusb_claim_interface(handle, 0);
  if (rc < 0)
  {
    fprintf(stderr, "Error claiming interface.\n");
    goto out;
  }

  /* allocate memory */
  wbuf1 = malloc(length);
  wbuf2 = malloc(length);
  rbuf1 = malloc(length);
  rbuf2 = malloc(length);

  /* fill the buffer with incrementing data */
  for (n = 0; n < EP_SIZE; n++)
    wbuf1[n] = n;

  for (m = 0; m < EP_SIZE; m++)
    wbuf2[m] = m+1;

  struct libusb_transfer *transfer1;
  struct libusb_transfer *transfer2;
  struct libusb_transfer *transfer3;
  struct libusb_transfer *transfer4;

  /* Set up the transfer object */
  transfer1 = libusb_alloc_transfer(0);
  transfer2 = libusb_alloc_transfer(0);
  transfer3 = libusb_alloc_transfer(0);
  transfer4 = libusb_alloc_transfer(0);
  libusb_fill_bulk_transfer(transfer1, handle, ep_out, wbuf1, EP_SIZE, xfr_cb, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer2, handle, ep_in, rbuf1, EP_SIZE,  xfr_cb, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer3, handle, ep_out, wbuf2, EP_SIZE, xfr_cb, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer4, handle, ep_in, rbuf2, EP_SIZE,  xfr_cb, &completed, TIMEOUT); /* callback data = &completed */

  /* Submit the transfers */
  libusb_submit_transfer(transfer1);
  libusb_submit_transfer(transfer2);
  libusb_submit_transfer(transfer3);
  libusb_submit_transfer(transfer4);

  /* Handle Events */
  while (!completed)
  {
    rc = libusb_handle_events_completed(NULL, &completed);
    if (rc != LIBUSB_SUCCESS)
    {
      fprintf(stderr, "Transfer Error: %s\n", libusb_error_name(rc));
      break;
    }
  }

  fprintf(stderr, "completed\n");

  //* Release the interface */
  libusb_release_interface(handle, 0);

  /* Close the device handle */
  if (handle)
    libusb_close(handle);

out:
  if (handle)
  {
    libusb_close(handle);
  }
  libusb_exit(NULL);

  return rc;
}

如下更改代码(即 callback = NULL for transfer 1-3)重新创建间歇性重复事务，如屏幕截图所示。

  libusb_fill_bulk_transfer(transfer1, handle, ep_out, wbuf1, EP_SIZE, NULL, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer2, handle, ep_in, rbuf1, EP_SIZE,  NULL, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer3, handle, ep_out, wbuf2, EP_SIZE, NULL, NULL, TIMEOUT);
  libusb_fill_bulk_transfer(transfer4, handle, ep_in, rbuf2, EP_SIZE,  xfr_cb, &completed, TIMEOUT); /* callback data = &completed */

老实说，根据他们的文档和示例，我不明白为什么循环会导致竞争条件。在 libusb 示例 (sam3u_benchmark.c) 之一中实际上建议对多个传输进行排队，并且还在以下 .pdf 中演示(使用循环)。

查看异步 I/O 部分:

https://www.socallinuxexpo.org/sites/default/files/presentations/scale_2017_usb.pdf http://www.signal11.us/oss/elc2014/elc_2014_usb_0.pdf

根据我的理解，使用 libusb_handle_events_completed(NULL, &completed) 应该可以解决同步问题。我误会了什么吗？从多线程中查看libusb_handle_events() http://libusb.sourceforge.net/api-1.0/libusb_mtasync.html

-“这就是为什么 libusb-1.0.9 引入了新的 libusb_handle_events_timeout_completed() 和 libusb_handle_events_completed() 函数，它们在获得锁后为您完成完成检查:”

他们需要的是在这种情况下如何使用他们的 API 的清晰示例。

我可以添加更多事件检查，但这里似乎有些不对。

更新 3:查看已接受的答案。

Answer 1

我开始阅读 libusb 源代码中的文档并了解发生了什么。

特别是关于 libusb 如何处理数据包大小的部分: http://libusb.sourceforge.net/api-1.0/libusb_packetoverflow.html

阅读后我很受用，我发现了两种使用异步 I/O 完成大数据环回测试的方法。

第一种方式是连续提交两次传输，传输->缓冲区包含整个数据结构(即发送和接收的总字节数)。第二种方式是使用包含 wMaxPacketSize(例如 64 字节)的 transfer->buffer 提交两个传输，并让 out 和 in 回调函数提交额外的传输以收发其余数据。

对于第二种情况，需要添加额外的代码来跟踪传输次数并在完成时设置完成信号。 OUT-IN 数据包交错由 libusb 和操作系统处理——这是我没有意识到的部分。换句话说，并非每个 OUT-IN 传输都需要单独指定和排队。

这是异步代码以及到我的 USB 设备 (ISP1362) 的传输速率。我的 USB 设备 Controller 是用纯 SystemVerilog 编码的 FPGA。

注意:关于传输速率，我只在 BULK_EP_IN 上启用了双缓冲。我假设如果在 BULK_EP_OUT 上启用双缓冲，IN-NAK (# POLL) 和传输速率将在第二种方法中提高。因此，由于设备配置的原因，这可能不是一个公平的比较。

第一种方法:~1.161 MB/s(~9.288 Mb/s)

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include </usr/include/libusb-1.0/libusb.h>

/* Specify VENDOR_ID and PRODUCT_ID for device */
#define VENDOR_ID   0x0471
#define PRODUCT_ID  0x3630

/* Define number of bytes to transfer */
#define EP_SIZE 64                       // bytes
#define TRANSFERS 1024*768*3/EP_SIZE     // number of transfers
#define TIMEOUT 10*1000                  // milliseconds

/* Use a global variable to keep the device handle */
static struct libusb_device_handle *handle = NULL;

/* count variables */
unsigned int count = 0;
unsigned int count_in = 0;
unsigned int count_out = 0;

/* The Endpoint addresses are hard-coded.  You should use libusb -v to find
 * the values corresponding to device
 */
static int ep_in  = 0x82;
static int ep_out = 0x01;

/* Write and Read buffers */
unsigned char wbuf[EP_SIZE*TRANSFERS];
unsigned char wbuf_tmp[EP_SIZE*TRANSFERS];
unsigned char rbuf[EP_SIZE*TRANSFERS];
unsigned char rbuf_tmp[EP_SIZE*TRANSFERS];

static void LIBUSB_CALL xfr_cb_out(struct libusb_transfer *transfer )
{
  memcpy(wbuf+count_out*EP_SIZE, transfer->buffer, EP_SIZE);
}

static void LIBUSB_CALL xfr_cb_in(struct libusb_transfer *transfer )
{
  int *completed = transfer->user_data;
  memcpy(rbuf+count_in*EP_SIZE, transfer->buffer, EP_SIZE);

  count_in++;  // one transfer complete
  if (count_in < TRANSFERS)
    *completed = 1;
}

int main(int argc, char **argv)
{
  const struct libusb_version *version;
  int completed = 0;
  size_t length = 64;
  int n;
  int m;
  int rc;

  /* Get libusb version */
  version = libusb_get_version();
  fprintf(stderr, "libusb version: %d.%d.%d.%d\n", version->major, version->minor, version->micro, version->nano);

  /* Initialize libusb */
  rc = libusb_init(NULL);
  if (rc < 0)
  {
    fprintf(stderr, "Error Initializing libusb: %s\n", libusb_error_name(rc));
    exit(1);
  }

  /* Set debugging output to max level */
  libusb_set_debug(NULL, 3);

  /* Look for a specific device and open it */
  handle = libusb_open_device_with_vid_pid(NULL, VENDOR_ID, PRODUCT_ID);
  if (!handle)
  {
    fprintf(stderr, "Error finding USB device\n");
    goto out;
  }

  /* claim interface */
  rc = libusb_claim_interface(handle, 0);
  if (rc < 0)
  {
    fprintf(stderr, "Error claiming interface.\n");
    goto out;
  }

  /* fill the buffer with incrementing data */
  for (n = 0; n < TRANSFERS; n++)
  {
    for (m = 0; m < EP_SIZE; m++)
    {
      wbuf_tmp[m+n*EP_SIZE] = m+n;
    }
  }

  struct libusb_transfer *transfer;
  transfer = libusb_alloc_transfer(0);
  libusb_fill_bulk_transfer(transfer, handle, ep_out, wbuf_tmp, EP_SIZE*TRANSFERS, xfr_cb_out, NULL, TIMEOUT);
  libusb_submit_transfer(transfer);

  transfer = libusb_alloc_transfer(0);
  libusb_fill_bulk_transfer(transfer, handle, ep_in, rbuf_tmp, EP_SIZE*TRANSFERS, xfr_cb_in, &completed, TIMEOUT);
  libusb_submit_transfer(transfer);

  /* Handle Events */
  while (!completed)
  {
    rc = libusb_handle_events_completed(NULL, &completed);
    if (rc != LIBUSB_SUCCESS)
    {
      fprintf(stderr, "Transfer Error: %s\n", libusb_error_name(rc));
      break;
    }
  }

  fprintf(stderr, "completed\n");

  int res;
  res = memcmp(rbuf, wbuf, sizeof(wbuf));
  if (res != 0)
    fprintf(stderr, "miscompare\n");
  else
    fprintf(stderr, "success\n");

  //* Release the interface */
  libusb_release_interface(handle, 0);

  /* Close the device handle */
  if (handle)
    libusb_close(handle);

out:
  if (handle)
  {
    libusb_close(handle);
  }
  libusb_exit(NULL);

  return rc;
}

第二种方法:~755.9 MB/s(~6.047 Mb/s)

include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include </usr/include/libusb-1.0/libusb.h>

/* Specify VENDOR_ID and PRODUCT_ID for device */
#define VENDOR_ID   0x0471
#define PRODUCT_ID  0x3630

/* Define number of bytes to transfer */
#define EP_SIZE 64                       // bytes
#define TRANSFERS 1024*768*3/EP_SIZE     // number of transfers
#define TIMEOUT 10*1000                  // milliseconds

/* Use a global variable to keep the device handle */
static struct libusb_device_handle *handle = NULL;

/* count variables */
unsigned int count = 0;
unsigned int count_in = 0;
unsigned int count_out = 0;

/* The Endpoint addresses are hard-coded.  You should use libusb -v to find
 * the values corresponding to device
 */
static int ep_in  = 0x82;
static int ep_out = 0x01;

/* Write and Read buffers */
unsigned char wbuf[EP_SIZE*TRANSFERS];
unsigned char *wbuf_tmp;
unsigned char rbuf[EP_SIZE*TRANSFERS];
unsigned char rbuf_tmp[EP_SIZE*TRANSFERS];

static void LIBUSB_CALL xfr_cb_out(struct libusb_transfer *transfer )
{
  memcpy(wbuf+count_out*EP_SIZE, transfer->buffer, EP_SIZE);

  count_out++;  // one transfer complete
  if (count_out < TRANSFERS)
  {
    transfer->buffer = ++wbuf_tmp;
    libusb_submit_transfer(transfer);
  }
}

static void LIBUSB_CALL xfr_cb_in(struct libusb_transfer *transfer )
{
  int *completed = transfer->user_data;
  memcpy(rbuf+count_in*EP_SIZE, transfer->buffer, EP_SIZE);

  count_in++;  // one transfer complete
  if (count_in < TRANSFERS)
    libusb_submit_transfer(transfer);
  else
    *completed = 1;
}

int main(int argc, char **argv)
{
  const struct libusb_version *version;
  int completed = 0;
  size_t length = 64;
  int n;
  int rc;

  /* Get libusb version */
  version = libusb_get_version();
  fprintf(stderr, "libusb version: %d.%d.%d.%d\n", version->major, version->minor, version->micro, version->nano);

  /* Initialize libusb */
  rc = libusb_init(NULL);
  if (rc < 0)
  {
    fprintf(stderr, "Error Initializing libusb: %s\n", libusb_error_name(rc));
    exit(1);
  }

  /* Set debugging output to max level */
  libusb_set_debug(NULL, 3);

  /* Look for a specific device and open it */
  handle = libusb_open_device_with_vid_pid(NULL, VENDOR_ID, PRODUCT_ID);
  if (!handle)
  {
    fprintf(stderr, "Error finding USB device\n");
    goto out;
  }

  /* claim interface */
  rc = libusb_claim_interface(handle, 0);
  if (rc < 0)
  {
    fprintf(stderr, "Error claiming interface.\n");
    goto out;
  }

  /* allocate memory */
  wbuf_tmp = malloc(length*TRANSFERS);

  /* fill the buffer with incrementing data */
  for (n = 0; n < EP_SIZE*TRANSFERS; n++)
  {
    wbuf_tmp[n] = n;
  }

  struct libusb_transfer *transfer;
  transfer = libusb_alloc_transfer(0);
  libusb_fill_bulk_transfer(transfer, handle, ep_out, wbuf_tmp, EP_SIZE, xfr_cb_out, NULL, TIMEOUT);
  libusb_submit_transfer(transfer);

  transfer = libusb_alloc_transfer(0);
  libusb_fill_bulk_transfer(transfer, handle, ep_in, rbuf_tmp, EP_SIZE, xfr_cb_in, &completed, TIMEOUT);
  libusb_submit_transfer(transfer);

  /* Handle Events */
  while (!completed)
  {
    rc = libusb_handle_events_completed(NULL, &completed);
    if (rc != LIBUSB_SUCCESS)
    {
      fprintf(stderr, "Transfer Error: %s\n", libusb_error_name(rc));
      break;
    }
  }

  fprintf(stderr, "completed\n");

  int res;
  res = memcmp(rbuf, wbuf, sizeof(wbuf));
  if (res != 0)
    fprintf(stderr, "miscompare\n");
  else
    fprintf(stderr, "success\n");

  //* Release the interface */
  libusb_release_interface(handle, 0);

  /* Close the device handle */
  if (handle)
    libusb_close(handle);

out:
  if (handle)
  {
    libusb_close(handle);
  }
  libusb_exit(NULL);

  return rc;
}