FAQ: Как написать двойную буферизацию ввода с АЦП?

Двойная буферизация (применительно к потокам данных от АЦП или на ЦАП) – это алгоритм, позволяющий обрабатывать непрерывный поток данных порциями, в том числе при наличии случайных задержек в выполнении программы (например, на обычном компьютере с обычной операционной системой). Подробнее см. буферизация потоков данных.

Ниже рассматривается двойная буферизация ввода (чтения) данных – например, от АЦП. Для вывода (записи) всё аналогично, только буфер сначала заполняется, а потом передается в устройство; перед началом работы буфер должен быть заполнен первой порцией данных.

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

Образно можно представить себе поток данных как трубу, из которой непрерывно льется вода. Мы берём два ведра и, набрав одно полное ведро, сразу подставляем пустое и относим полное на грядку. Потом возвращаемся с пустым ведром, и когда закончит наполняться ведро, оставленное в прошлый раз, меняем их местами. И так далее. При этом вёдра должны быть достаточно большими (или поток воды достаточно маленьким), чтобы мы успевали отнести воду, пока наполняется второе ведро, даже если нас что-то немного задержит по дороге.

Ниже приведен упрощённый пример двойной буферизации чтения с использованием асинхронного (overlapped) ввода-вывода через WinAPI (предполагается, что устройство открыто как файл через HANDLE: так можно работать, например, со многими внешними модулями АЦПЦАП производства Л Кард).

#define BLOCK_LEN 8192              /* количество элементов в блоке */
typedef char block_elem_t;          /* элемент блока может быть любого типа */
typedef void (*data_callback_t)(block_elem_t* data, size_t data_size);

int read_stream
    (
    HANDLE h_device,                /* хендл устройства */
    HANDLE h_stop_event,            /* хендл события, останавливающего сбор данных */
    data_callback_t data_callback   /* указатель на функцию обработки данных */
    )
    {
    block_elem_t data_buf[2][BLOCK_LEN];
    int idx = 1, error_flag = 0;
    HANDLE h_event[2]; 
    OVERLAPPED ov[2];    
    DWORD transfer_size;
    
    /* Инициализация переменных */
    h_event[0] = CreateEvent(NULL, TRUE, FALSE, NULL);
    h_event[1] = CreateEvent(NULL, TRUE, FALSE, NULL);
    if (!h_event[0] || !h_event[1])
        {
        if (h_event[1]) CloseHandle(h_event[1]);
        if (h_event[0]) CloseHandle(h_event[0]);
        return 0; /* ошибка */
        }

    /* Постановка в очередь первого запроса чтения данных */
    ZeroMemory(&ov[0], sizeof(ov[0]));
    ov[0].hEvent = h_event[0];
    if (!ReadFile(h_device, data_buf[0], sizeof(data_buf[0]), NULL, &ov[0])
        && (GetLastError() != ERROR_IO_PENDING))
        return 0; /* ошибка чтения */

    /* Запуск АЦП (подразумевается какая-то функция управления устройством) */
    start_adc(h_device);

    while (!error_flag)
        { 
        HANDLE wait_list[2];
        DWORD wait_result;

        /* Постановка в очередь следующего запроса */
        ZeroMemory(&ov[idx], sizeof(ov[idx]));
        ov[idx].hEvent = h_event[idx];
        if (!ReadFile(h_device, data_buf[idx], sizeof(data_buf[idx]), NULL, &ov[idx])
            && (GetLastError() != ERROR_IO_PENDING))
            {
            error_flag = 1; /* ошибка чтения */
            break;
            } 

        idx ^= 1;

        /* Ожидание окончания предыдущего чтения или прихода команды stop */
        wait_list[0] = h_stop_event;
        wait_list[1] = h_event[idx];
        wait_result = WaitForMultipleObjects(2, wait_list, FALSE, INFINITE);
        if (WAIT_OBJECT_0 == wait_result)
            break; /* Пришла команда stop */ 
        if (WAIT_OBJECT_0 + 1 != wait_result)
            {
            error_flag = 1; /* ошибка WaitFor... */
            break;
            }
        if (!GetOverlappedResult(h_device, &ov[idx], &transfer_size, TRUE))
            {
            error_flag = 1; /* ошибка чтения */
            break;
            }

        /* Обработка данных */
        data_callback(data_buf[idx], transfer_size); 
        } /* while */

    /* Остановка АЦП (подразумевается какая-то функция управления устройством) */
    stop_adc(h_device);

    /* Завершение работы */
    CancelIo(h_device);
    /* После CancelIo следует делать GetOverlappedResult с bWait == TRUE */
    GetOverlappedResult(h_device, &ov[idx], &transfer_size, TRUE);
    GetOverlappedResult(h_device, &ov[idx ^ 1], &transfer_size, TRUE);
    CloseHandle(h_event[1]);
    CloseHandle(h_event[0]);
    return !error_flag;
    }