[brag 0]

Следующая задачка. Есть 2 шины I2C и SPI. К каждой подключено по 5 устройств. Для простоты рассмотрим 2 из них - I2C дисплей и SPI-флешка. Нужно реализовать такую программу, чтобы при ошибке инициализации дисплея была сделана соответствующая запись в лог-файл на флешке. При ошибке инициализации флешки необходимо вывести уведомление на дисплей. Если ни то ни другое не удалось - сжечь прибор :))) Инициализация этих обеих устройств должна выполняться параллельно - так, чтобы одна шина не простаивала, когда идет обмен на другой. Объем данных на шинах в процессе инициализации каждого устройства большой и занимает много времени. Процедуры инициализации сложные. Но вдаваться в подробности мы тут не будем, у нас уже это все давно реализовано  в двух вариантах - блок и нонблок.

Асинхронным способом решение простейшее:

Spoiler

struct State {
  bool display_error;
  bool flash_error;
};

static struct State g_state;

void main(void) {
  g_state.display_error = false;
  g_state.flash_error = false;
  
  display_init(&do_nothing, &on_display_init_error);
  flash_init(&do_nothing, &on_flash_init_error);
}

static void on_display_init_error(void) {
  g_state.display_error = true;
  flash_write_log("Display init error");
  fsm();
}

static void on_flash_init_error(void) {
  g_state.flash_error = true;
  display_draw_text("Flash init error");
  fsm();
}

static void fsm(void) {
  if(g_state.display_error && g_state.flash_error) {
    burn_fxxxng_device();
  }
}

 

Даже думать боюсь, как бы такое решение выглядело в линейном (чисто блокинг - тредовом) коде

[amaora 20]

20 hours ago, brag said:

А вот с этого места хотелось бы по-подробнее. Смотрите, есть UART(регистры rx_dat, tx_dat и 2 прерывания: rx_full, tx_empty), и есть блокирующая ОС, что делаем дальше? Как будем организовывить работу с UART с тредов?

В обработчиках прерываний перекладываем из регистров в очередь и обратно.

 

20 hours ago, brag said:

Как эта задача будет получать запросы и как передавать дальше на обработку? Как будет вести учет свободных тредов?

Читать входящие сетевые пакеты из очереди (куда их накладывает драйвер в прерывании), фильтровать если требуется. Если увидит GET запрос, то формирует структуру со всеми необходимыми данными и отправляет в одну из очередей к тредам-обработчикам. Можно проверять, сколько сообщений в уже есть в очереди к каждому треду-обработчику, если у всех очереди полны, то выдавать короткий ответ или не выдавать ничего.

 

20 hours ago, brag said:

Ок. Какие конкретно нужны очереди, какого размера и сколько штук? Чтобы гарантировано хватило и программа не заглючила ни при каких обстоятельствах? То есть чтобы ни один пакет не был утерян, чтобы ни одна АЦП выборка не была утеряна.

Чтобы "не заглючила" надо обрабатывать все переполнения очередей (явно определить поведение в этом случае) или иметь гарантии, что переполнение невозможно.

 

20 hours ago, brag said:

Именно последний способ применен у меня в коде, его можно использовать и в тредовом подходе.

Это самый простой способ.

[brag 0]

2 hours ago, amaora said:

В обработчиках прерываний перекладываем из регистров в очередь и обратно.

Как заведомо знать, что размера очереди хватит? Без блокировок посчитать легко, с ОС и блокировками куда сложнее, ведь может быть такая ситуация, что тред в блоке, а данные в очередь летят.

Но это еще не все. Доступ к UART-TX у нас конкурентный! Из 5 потоков. Как будем разруливать эту ситуацию? Очередь из фреймов(какого размера?)? Или может быть мютексы и блокировать порт(по крайней мере передающую его часть) на все время передачи фрейма? Ведь фрейм должен быть целостный, промаркированный (чтобы сетевой драйвер смог распознать какому соединению он принадлежит) и с контрольной суммой(чтобы сетевой стек мог отбрасывать кривые фреймы). Как по мне, тут ад уже начался.

В асинхронной модели этих всех проблем нет. Нет потоков, нет потоковой конкуренции, нет проблем. Сериализация операций над портом осуществляется тривиальной intrusive-очередью(это та, которая никогда не переполняется) + такой же тривиальной fsm:

Spoiler

struct IntrusiveQueue q;
enum State state;

void on_queue_becoming_non_empty(struct IntrusiveQueue *q){
  switch(state){
    case Ready:
      state = Busy;
      struct *op = queue_pop(q);
      op->run();
      break;
  }
}

void on_op_complete(void){
  switch(state){
    case Busy:
      struct *op = queue_pop(&q);
      if(op){
        op->run();
      }else{
        state = Ready;
      }
      break;
  }
}

 

 

2 hours ago, amaora said:

Читать входящие сетевые пакеты из очереди (куда их накладывает драйвер в прерывании), фильтровать если требуется. Если увидит GET запрос, то формирует структуру со всеми необходимыми данными и отправляет в одну из очередей к тредам-обработчикам.

Итого имеем две потенциальные блокировки. Рассчитать требуемый размер первой очереди на случай заполненной второй довольно непросто. Остается только лупануть очереди с огромным запасом и надеяться, что этого хватит ))

2 hours ago, amaora said:

Чтобы "не заглючила" надо обрабатывать все переполнения очередей (явно определить поведение в этом случае) или иметь гарантии, что переполнение невозможно.

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

2 hours ago, amaora said:

Это самый простой способ.

Да и он асинхронный. блокирующим способом потенциально реализовать можно, но это будет мрак.

Но это еще цветочки. Настоящий бич тредового подхода - таймауты. Без введения асинхронных костылей в тредовую модель таймаут реализовать линейным кодом невозможно в принципе - тред не умеет одновременно делать sleep и read. А при введении таких костылей(например unix-сигналы) или просто kill_thread(что тоже по сути является асинхронным сигналом\прерыванием) программа перестает быть линейной и начинается ад.

[brag 0]

4 hours ago, brag said:

А при введении таких костылей(например unix-сигналы) или просто kill_thread(что тоже по сути является асинхронным сигналом\прерыванием) программа перестает быть линейной и начинается ад.

Тривиальная задача для асинхронного программирования(при чем без всяких вытеснений и приоритетов - проблема гонок отсутствует полностью, никаких критических секций не нужно) и вообще нерешаемая для линейного(пусть и многопоточного):

Есть 2 SPI-шины на каждой по 3 устройства. Нужно переписать файл объемом 100мб из одного из устройств на шине 1 в любое другое на шине 2.

Программа должна уметь копировать файлы одновременно по всем 3м устройствам в произвольном порядке. То есть одновременных операций копирования может быть 3.

Количество оперативной памяти ограничено и равно 32кб для всей программы.

На каждую операцию чтения и записи должен быть установлен таймаут. Также таймаут должен быть установлен на всю операцию копирования - не скопировались все 100мб за 10 минут - все, таймаут. При срабатывании любого таймаута необходимо освободить шину\шины (поднять соответствующие CS).

Нужно печатать в uart прогресс копирования от 1 до 100 % строчками вида: "Copy from SPI1.dev1 to SPI2.dev3 in progress 5%", строчки должны появляться каждый 1%. Пропускной способности uart для этого достаточно. При таймауте или любой другой ошибке записи\чтения тоже нужно отправить соответствующее сообщение в лог.

Копирование нужно выполнять максимально эффективно, то есть так, чтобы ни одна из шин не простаивала, если это физически возможно(скорость чтения равна скорости записи). В том числе и во время печатания в uart.

Также нужна возможность отмены операций пользователем с корректным освобождением шин.

[amaora 20]

On 11/16/2021 at 2:59 AM, brag said:

Есть 2 SPI-шины на каждой по 3 устройства. Нужно переписать файл объемом 100мб из одного из устройств на шине 1 в любое другое на шине 2.

Программа должна уметь копировать файлы одновременно по всем 3м устройствам в произвольном порядке. То есть одновременных операций копирования может быть 3.

Делаем уровень блочных устройств, примерно вот такой интерфейс,

int BLK_read(int devnum, u32_t addr, vois *buf, int size, int timeout);
int BLK_write(int devnum, u32_t addr, const void *buf, int size, int timeout);
int BLK_flush(int devnum);

задачи производящие копирование выглядят как-то так,

...

while (n < req_size) {
  
  rc = BLK_read(dev1, addr1, buf, sizeof(buf), 10);
  
  if (rc != 0)
    break;
  
  rc = BLK_write(dev2, addr2, buf, sizeof(buf), 10);
  
  if (rc != 0)
    break;
  
  addr1 += sizeof(buf);
  addr2 += sizeof(buf);
  n += sizeof(buf);
  
  if (1) {
  	printf("progress %i %%\n", n * 100 / req_size);
  }
}

if (rc != 0)
  log_printf("error: %s\n", strerr(rc));

rc = BLK_flush(dev2);

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

 

- BLK_write пишет данные в очередь записи и возвращает управление, либо блокируется (на время не более timeout) в ожидании свободного места в очереди. Конкурентная реализация не представляет сложности, можно писать в одно устройство из разных задач, функция является лишь обёрткой над отправкой в очередь (считаю это за базовый примитив синхронизации, как во FreeRTOS).

- BLK_read читает данные из кэша (если они там есть) и возвращает управление, либо формирует запрос чтения и блокируется (на время не более timeout) на ожидании события его исполнения. Наличие кэша не обязательно, а его конкурентная реализация не слишком проста.

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

 

На нижнем уровне драйвер SPI в двух экземплярах.

void irq_SPI1()
{
  spi_req	*req;
  
  if (dmastat == DMA_DONE) {
    
    gpio_relese();
    
    memcpy(req->rx_buf, dmabuf, req->size);
    semgive(req->sem);
  }
  
  if (1) {
    
    if (queueReceiveNonBlock(spitab[0], req) == 0) {
      
      memcpy(dmabuf, req->tx_buf, req->size);
      
      gpio_set(req->slave_select);
      dma_start(dmabuf, req->size);
    }
    else {
      disable_SPI1_irq();
    }
  }
}

void irq_SPI2() { ... }

int spi_xfer(int spidev, const void *tx_buf, void *rx_buf, int size)
{
  queue_t	*q = spitab[spidev];
  spi_req	req;
  
  req.tx_buf = tx_buf;
  req.rx_buf = rx_buf;
  req.size = size;
  req.sem = createsem();
  req.slave_select = sstab[spidev];
  
  queueSend(q, &req);
  enable_SPI*_irq();
  
  semtake(req.sem);
  deletesem(req.sem);
}

 

Теперь у драйвера блочного устройства с одной стороны две очереди (запросы записи и запросы чтения), а с другой стороны функция spi_xfer. Если следовать "тредовой концепции", то надо делать два треда, на обработку запросов записи и чтения. Запросы обрабатываются аналогично тому, как сделано в прерывании SPI.

void thread_SPI_flash_write()
{
  spi_xfer(...); // команды инициализации
  spi_xfer(...);
  
  mutex_op = createmutex();
  
  createthread(thread_SPI_flash_read);
  
  do {
    queueReceive(blk_qw, req);
    
    if (req->type == REQ_FLUSH) {
      
      semgive(req->sem);
    }
    else {
    
      semtake(mutex_op);
      
        spi_xfer(...); // команды начала записи (если надо все сделать не дергая CS то делаем один вызов xfer)
        spi_xfer(...);
        spi_xfer(spidev, req->buf, NULL, req->size);
      
      semgive(mutex_op);
      
      //semgive(req->sem); // завершения записи не ждут на верхнем уровне
    }
  }
  while (1);
}

void thread_SPI_flash_read()
{
  do {
    queueReceive(blk_qr, req);
    
    semtake(mutex_op);
    
      spi_xfer(...); // команды начала чтения
      spi_xfer(...);
      spi_xfer(spidev, NULL, req->buf, req->size);
    
    semgive(mutex_op);

    semgive(req->sem);
  }
  while (1);
}

 

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

Изменено 17 ноября, 2021 пользователем amaora

[brag 0]

4 hours ago, amaora said:

Делаем уровень блочных устройств, примерно вот такой интерфейс,

int BLK_read(int devnum, u32_t addr, vois *buf, int size, int timeout);
int BLK_write(int devnum, u32_t addr, const void *buf, int size, int timeout);
int BLK_flush(int devnum);

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

То есть можно реализовать отдельно простейшие BLK_read BLK_write и BLK_cancel (отмена текущей операции). Потом отдельно реализовать op_timeout. И соединить это в кучу на следующем слое. Программа получается модульная, состоящая из простых кубиков, а не сложных огромных монстров.

4 hours ago, amaora said:

если надо то можно считать общее время, и выходить если оно превышено.

Tочность такого таймаута будет крайне низкая - с джиттером зависимым от  BLK_read timeout. А надо выходить сразу, не дожидаясь BLK_read timeout.

Кроме того не реализована логика печатания в лог с шагом 1% ;) Размер блока  не может быть 1мб, поскольку у нас всего 32кб оперативки.

 

4 hours ago, amaora said:

Так же и выход по запросу пользователя, не мгновенно, но ждать не дольше указанного таймаута операций чтения/записи.

А надо мгновенно(в разумных пределах конечно). BLK_read timeout - 5-10 секунд. таймаут на весь файл 10 минут. По запросу пользователя необходимо корректно прервать операцию так, чтобы у пользователя не возникло желания нажать на кнопку "Stop" еще раз, на практике это 100-200мс, не больше.

4 hours ago, amaora said:

функция является лишь обёрткой над отправкой в очередь

То есть оберткой над асинхронным драйвером :)) Иными словами львиную и самую сложную долю задачи пришлось таки решать асинхронно.

4 hours ago, amaora said:

Надеюсь, по этому псевдокоду все понятно, многие мелкие детали не стал расписывать.

Да все понятно, но в коде нет таймаута! А это ключевой момент в данной задаче. И чтобы его добавить, Вам понадобится править самый нижний уровень, тобышь spi_xfer. И реализовать такую правку будет непросто, честно говоря я даже не знаю как, давно спрыгнул с блока, программирую только асинк, как под PC, смартфоны, серверы, так и под embedded. По идее надо сначала дождаться semtake, затем проверить статус асинхронного таймера, остановить таймер, и на основе этих данных пихнуть в очередь либо ResulutOK, либо TimeoutError.

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

И это еще без реализации BLK_read, BLK_write. А туда еще как минимум 2 очереди.

И все равно задача не решена, поскольку "BLK_read timeout - 5-10 секунд. таймаут на весь файл 10 минут. По запросу пользователя необходимо корректно прервать операцию так, чтобы у пользователя не возникло желания нажать на кнопку "Stop" еще раз, на практике это 100-200мс, не больше."

[amaora 20]

1 hour ago, brag said:

Да все понятно, но в коде нет таймаута! А это ключевой момент в данной задаче. И чтобы его добавить, Вам понадобится править самый нижний уровень, тобышь spi_xfer.

Ну я по привычке подразумевал очереди FreeRTOS, таймаут у них встроенный, никаких дополнительных таймеров не надо, только тот по которому работает диспетчер. Это базовый примитив синхронизации. Семафоры и мьютексы там тоже через очередь реализуются, и так же наследуют таймауты. Если про внутреннюю реализацию, то видимо да, на каждом вызове диспетчера по таймеру проверяется не истекло ли время у одной из заблокированных задач, но это внутренность RTOS.

 

И незачем задавать 5-10 секунд, если известно, что операция чтения/записи укладывается в 10-50 мс с запасом. Я бы больше беспокоился о том по каким правилам планируется какая задача следующей сможет протолкнуть своё сообщение в очередь (если все хотят доступ к одному устройству или по одному SPI), и что делать, если хочется задать приоритеты воода-вывода для некоторых устройств или некоторых задач верхнего уровня. Но это все тоже решаемо, если такие сложности необходимы.

 

1 hour ago, brag said:

Мало того, Вам пришлось в драйвер устройства(точнее во все драйвера!) добавить поддержку таймаута(естественно асинхронным способом ибо иначе реализовать это невозможно).

По таймауту отвалится только отправка запроса на уровень ниже (если очередь забита и не освобождается) и более ничего, то есть запрос не дойдёт до задач thread_SPI_flash_xx. На уровне spi_xfer тоже можно сделать таймауты и перезапуск, но это для обработки ошибок и случаев неприхода прерывания от DMA, то есть это будут уже дела драйвера, а не верхнего уровня.

 

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

 

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

 

Изменено 17 ноября, 2021 пользователем amaora

[brag 0]

2 minutes ago, amaora said:

Ну я по привычке подразумевал очереди FreeRTOS, таймаут у них встроенный, никаких дополнительных таймеров не надо, только тот по которому работает диспетчер. Это базовый примитив синхронизации. Семафоры и мьютексы там тоже через очередь реализуются, и так же наследуют таймауты. Если про внутреннюю реализацию, то видимо да, на каждом вызове диспетчера по таймеру проверяется не истекло ли время у одной из заблокированных задач, но это внутренность RTOS.

Ок. Хороший ос ;) Но как быть с общим таймаутом на несколько(сотен) как параллельных, так и последовательных операций?

6 minutes ago, amaora said:

И незачем задавать 5-10 секунд, если известно, что операция чтения/записи укладывается в 10-50 мс с запасом.

Потому что такой накопитель. Некоторые блоки могут читаться очень медленно с кучей повторов, а другие моментально с первого раза да на полной скорости, все помним компакт-диски или дискеты ;)) Да и сетевые соединения тоже себя так могут вести. Такая задача просто. Таймаут на одну операцию 5 сек, на группу операций - 10 минут.

17 minutes ago, amaora said:

По таймауту отвалится только отправка запроса на уровень ниже (если очередь забита и не освобождается) и более ничего, то есть запрос не дойдёт до задач thread_SPI_flash_xx. На уровне spi_xfer тоже можно сделать таймауты и перезапуск, но это для обработки ошибок и случаев неприхода прерывания от DMA, то есть это будут уже дела драйвера, а не верхнего уровня.

Ок, согласен. Но это сработает, если есть поддержка нужного функционала со стороны OC. ОС должны быть уже реализованы все возможности, которые потребуются пользователю. Всякие там waitForMultipleObjects,как в винде, семафоры с таймаутами  итд. Тот же printf, у него нет таймаута, а если понадобится?

Хорошо, когда готовое есть. Можно вообще ничего не программировать, просто купить(скачать) и все ;) А когда его нет, приходится выдумывать что-то. Вот это что-то без асинка сделать нереально на практике. Нужно все больше и больше потоков, а их синхронизация - это боль.  Не даром же в эпоху очень мощного железа появляются такие вещи, как nodejs. Можно ведь было по старинке на php с обычным printf серваки клепать. Ресурсы ведь позволяют(а позволяют ли?)

51 minutes ago, amaora said:

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

Если ресурсов хватает на контекст, стек и очереди для примитив синхронизации, а так же на очереди, в которые будут падать данные, когда printf висит, значит их с лихвой хватит на то, чтобы просто увеличить очередь printf и забыть. Бонусом можно печатать с прерываний и любого другого места. Ведь printf же не просто печатает какой-то хардкод, он должен где-то брать данные, которые тоже должны где-то хранится и как-то этому printf передаваться. А если хардкод, то можно просто генернуть цепочку асинк-вызовов каким-нибудь скриптом ))

А так согласен, если хочется линейного кода, то да, надо делать.

[brag 0]

Вот асинхронная реализация. 80 строк чисто кода(без учета пустых). Примерно столько же понадобилось для тредовой реализации. Только тут учтены все таймауты и конкуренция.

Работа с SPI скопирована из Вашей, но семафор ожидания заменен на асинхронный post(тупо добавление  в очередь на исполнение после выходa из всех прерываний).

Реализация BLK_Read/BLK_write будет аналогичная spi_xfer - закидывание в очередь аргументов. С последующим выполнением

	spi_xfer(...); // команды начала чтения
on_xfer1_complete(){
    spi_xfer(...);
}
on_xfer2_complete(){
    spi_xfer(spidev, NULL, req->buf, req->size);
}

BLK_Cancel - соответственно удаление соответствующей записи из очереди, если она там есть. Если очередь реализована на circular doubly linked lists, то такая проверка происходит автоматически, дополнительный if не нужен и вся операция занимает 6 Thmub-инструкций - 2 чтения и 4 записи.

Spoiler

enum State state;
struct NonIntrusiveQueue<uint8_t[SIZEOF_BUF],2> queue;
struct Timer t1, t2, t3; 

void copy_start(void) {
    timer_start(&t1, &on_timeout, 10min);
    timer_start(&t2, &on_timeout, 5sec);
    BLK_read(dev1, addr1, queue_tail(&queue), SIZEOF_BUF, &on_read_complete, &on_error);
}

void copy_cancel(void) {
    // это нормально останавиливать незапущенный таймер и отменять незапущенную операцию. Внутри реализаций заложена такая возможность.
    BLK_cancel(dev1);
    BLK_cancel(dev2);
    timer_stop(&t1);
    timer_stop(&t2);
    timer_stop(&t3);
}

static void on_read_complete(void) {
    addr1 += SIZEOF_BUF;
    n1 += SIZEOF_BUF;
    queue_inc_tail(&queue);
    fsm();
}

static void on_write_complete(void) {
    addr2 += SIZEOF_BUF;
    n2 += SIZEOF_BUF;
    n_printf += SZEOF_BUF;
    queue_inc_head(&queue);
    fsm();
}

static void fsm(void) {
    uint8_t *tail = queue_tail(s->queue);
    if(tail && n1 < req_size) {
        // перезапуск таймера тоже поддерживается
        timer_start(&t2, &on_timeout, 5sec);
        BLK_read(dev1, addr1, tail, SIZEOF_BUF, &on_read_complete, &on_error);
    }       

    uint8_t *head = queue_head(s->queue);
    if(tail && n2 < req_size) {
        timer_start(&t3, &on_timeout, 5sec);
        BLK_write(dev2, addr2, head, SIZEOF_BUF, &on_write_complete, &on_error);
    }       

    if(n_printf >= req_size/100) {
        n_printf = 0;
        printf("progress %i %%\n", n * 100 / req_size);
    }       
}

static void on_error(int rc) {
    cancel();
    printf("error: %s\n", strerr(rc));
}

void irq_SPI1(void) {
    spi_req *req;
  
    if (dmastat == DMA_DONE) {
        gpio_relese();
        memcpy(req->rx_buf, dmabuf, req->size);
        disable_SPI1_irq();
        sst_post(&req->event);
    }
  
    if (1) {
        if (queueReceiveNonBlock(spitab[0], req) == 0) {
            memcpy(dmabuf, req->tx_buf, req->size);
            gpio_set(req->slave_select);
            dma_start(dmabuf, req->size);
        } else {
            disable_SPI1_irq();
        }
    }   
}

void spi_xfer(int spidev, const void *tx_buf, void *rx_buf, SST_Event *event) {
    queue_t *q = spitab[spidev];
    spi_req req;

    req.event = event;
    req.rx_buf = rx_buf;
    req.tx_buf = tx_buf;
    req.size = size;
    req.slave_select = sstab[spidev];
    queueSend(q, &req);
    enable_SPI*_irq();
}