liz/flipperzero-firmware
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[FL-2839] Furi stream buffer (#1834)

Browse Source 
* Core: stream buffer
* stream buffer: API and usage
* stream buffer: documentation
* stream buffer: more documentation
* Furi: fix spelling

Co-authored-by: Aleksandr Kutuzov <alleteam@gmail.com>
This commit is contained in:
Sergey Gavrilov
2022-10-07 22:27:11 +10:00
committed by GitHub
parent d1843c0094
commit 38a82a1907
21 changed files with 403 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
lib/subghz/subghz_file_encoder_worker.c
View File

@@ -1,5 +1,4 @@
#include "subghz_file_encoder_worker.h"
#include <stream_buffer.h>
#include <toolbox/stream/stream.h>
#include <flipper_format/flipper_format.h>
@@ -11,7 +10,7 @@
struct SubGhzFileEncoderWorker {
FuriThread* thread;
StreamBufferHandle_t stream;
FuriStreamBuffer* stream;
Storage* storage;
FlipperFormat* flipper_format;
@@ -48,7 +47,7 @@ void subghz_file_encoder_worker_add_level_duration(
if(res) {
instance->level = !instance->level;
xStreamBufferSend(instance->stream, &duration, sizeof(int32_t), 100);
furi_stream_buffer_send(instance->stream, &duration, sizeof(int32_t), 100);
} else {
FURI_LOG_E(TAG, "Invalid level in the stream");
}
@@ -83,10 +82,7 @@ LevelDuration subghz_file_encoder_worker_get_level_duration(void* context) {
furi_assert(context);
SubGhzFileEncoderWorker* instance = context;
int32_t duration;
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
int ret = xStreamBufferReceiveFromISR(
instance->stream, &duration, sizeof(int32_t), &xHigherPriorityTaskWoken);
portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
int ret = furi_stream_buffer_receive(instance->stream, &duration, sizeof(int32_t), 0);
if(ret == sizeof(int32_t)) {
LevelDuration level_duration = {.level = LEVEL_DURATION_RESET};
if(duration < 0) {
@@ -137,7 +133,7 @@ static int32_t subghz_file_encoder_worker_thread(void* context) {
} while(0);
while(res && instance->worker_running) {
size_t stream_free_byte = xStreamBufferSpacesAvailable(instance->stream);
size_t stream_free_byte = furi_stream_buffer_spaces_available(instance->stream);
if((stream_free_byte / sizeof(int32_t)) >= SUBGHZ_FILE_ENCODER_LOAD) {
if(stream_read_line(stream, instance->str_data)) {
furi_string_trim(instance->str_data);
@@ -183,7 +179,7 @@ SubGhzFileEncoderWorker* subghz_file_encoder_worker_alloc() {
furi_thread_set_stack_size(instance->thread, 2048);
furi_thread_set_context(instance->thread, instance);
furi_thread_set_callback(instance->thread, subghz_file_encoder_worker_thread);
instance->stream = xStreamBufferCreate(sizeof(int32_t) * 2048, sizeof(int32_t));
instance->stream = furi_stream_buffer_alloc(sizeof(int32_t) * 2048, sizeof(int32_t));
instance->storage = furi_record_open(RECORD_STORAGE);
instance->flipper_format = flipper_format_file_alloc(instance->storage);
@@ -199,7 +195,7 @@ SubGhzFileEncoderWorker* subghz_file_encoder_worker_alloc() {
void subghz_file_encoder_worker_free(SubGhzFileEncoderWorker* instance) {
furi_assert(instance);
vStreamBufferDelete(instance->stream);
furi_stream_buffer_free(instance->stream);
furi_thread_free(instance->thread);
furi_string_free(instance->str_data);
@@ -215,7 +211,7 @@ bool subghz_file_encoder_worker_start(SubGhzFileEncoderWorker* instance, const c
furi_assert(instance);
furi_assert(!instance->worker_running);
xStreamBufferReset(instance->stream);
furi_stream_buffer_reset(instance->stream);
furi_string_set(instance->file_path, file_path);
instance->worker_running = true;
furi_thread_start(instance->thread);

37
lib/subghz/subghz_tx_rx_worker.c
View File

@@ -1,6 +1,5 @@
#include "subghz_tx_rx_worker.h"
#include <stream_buffer.h>
#include <furi.h>
#define TAG "SubGhzTxRxWorker"
@@ -13,8 +12,8 @@
struct SubGhzTxRxWorker {
FuriThread* thread;
StreamBufferHandle_t stream_tx;
StreamBufferHandle_t stream_rx;
FuriStreamBuffer* stream_tx;
FuriStreamBuffer* stream_rx;
volatile bool worker_running;
volatile bool worker_stoping;
@@ -30,9 +29,9 @@ struct SubGhzTxRxWorker {
bool subghz_tx_rx_worker_write(SubGhzTxRxWorker* instance, uint8_t* data, size_t size) {
furi_assert(instance);
bool ret = false;
size_t stream_tx_free_byte = xStreamBufferSpacesAvailable(instance->stream_tx);
size_t stream_tx_free_byte = furi_stream_buffer_spaces_available(instance->stream_tx);
if(size && (stream_tx_free_byte >= size)) {
if(xStreamBufferSend(
if(furi_stream_buffer_send(
instance->stream_tx, data, size, SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF) ==
size) {
ret = true;
@@ -43,12 +42,12 @@ bool subghz_tx_rx_worker_write(SubGhzTxRxWorker* instance, uint8_t* data, size_t
size_t subghz_tx_rx_worker_available(SubGhzTxRxWorker* instance) {
furi_assert(instance);
return xStreamBufferBytesAvailable(instance->stream_rx);
return furi_stream_buffer_bytes_available(instance->stream_rx);
}
size_t subghz_tx_rx_worker_read(SubGhzTxRxWorker* instance, uint8_t* data, size_t size) {
furi_assert(instance);
return xStreamBufferReceive(instance->stream_rx, data, size, 0);
return furi_stream_buffer_receive(instance->stream_rx, data, size, 0);
}
void subghz_tx_rx_worker_set_callback_have_read(
@@ -148,11 +147,11 @@ static int32_t subghz_tx_rx_worker_thread(void* context) {
while(instance->worker_running) {
//transmit
size_tx = xStreamBufferBytesAvailable(instance->stream_tx);
size_tx = furi_stream_buffer_bytes_available(instance->stream_tx);
if(size_tx > 0 && !timeout_tx) {
timeout_tx = 10; //20ms
if(size_tx > SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE) {
xStreamBufferReceive(
furi_stream_buffer_receive(
instance->stream_tx,
&data,
SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE,
@@ -160,20 +159,20 @@ static int32_t subghz_tx_rx_worker_thread(void* context) {
subghz_tx_rx_worker_tx(instance, data, SUBGHZ_TXRX_WORKER_MAX_TXRX_SIZE);
} else {
//todo checking that he managed to write all the data to the TX buffer
xStreamBufferReceive(
furi_stream_buffer_receive(
instance->stream_tx, &data, size_tx, SUBGHZ_TXRX_WORKER_TIMEOUT_READ_WRITE_BUF);
subghz_tx_rx_worker_tx(instance, data, size_tx);
}
} else {
//recive
if(subghz_tx_rx_worker_rx(instance, data, size_rx)) {
if(xStreamBufferSpacesAvailable(instance->stream_rx) >= size_rx[0]) {
if(furi_stream_buffer_spaces_available(instance->stream_rx) >= size_rx[0]) {
if(instance->callback_have_read &&
xStreamBufferBytesAvailable(instance->stream_rx) == 0) {
furi_stream_buffer_bytes_available(instance->stream_rx) == 0) {
callback_rx = true;
}
//todo checking that he managed to write all the data to the RX buffer
xStreamBufferSend(
furi_stream_buffer_send(
instance->stream_rx,
&data,
size_rx[0],
@@ -208,9 +207,9 @@ SubGhzTxRxWorker* subghz_tx_rx_worker_alloc() {
furi_thread_set_context(instance->thread, instance);
furi_thread_set_callback(instance->thread, subghz_tx_rx_worker_thread);
instance->stream_tx =
xStreamBufferCreate(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
furi_stream_buffer_alloc(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
instance->stream_rx =
xStreamBufferCreate(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
furi_stream_buffer_alloc(sizeof(uint8_t) * SUBGHZ_TXRX_WORKER_BUF_SIZE, sizeof(uint8_t));
instance->status = SubGhzTxRxWorkerStatusIDLE;
instance->worker_stoping = true;
@@ -221,8 +220,8 @@ SubGhzTxRxWorker* subghz_tx_rx_worker_alloc() {
void subghz_tx_rx_worker_free(SubGhzTxRxWorker* instance) {
furi_assert(instance);
furi_assert(!instance->worker_running);
vStreamBufferDelete(instance->stream_tx);
vStreamBufferDelete(instance->stream_rx);
furi_stream_buffer_free(instance->stream_tx);
furi_stream_buffer_free(instance->stream_rx);
furi_thread_free(instance->thread);
free(instance);
@@ -232,8 +231,8 @@ bool subghz_tx_rx_worker_start(SubGhzTxRxWorker* instance, uint32_t frequency) {
furi_assert(instance);
furi_assert(!instance->worker_running);
bool res = false;
xStreamBufferReset(instance->stream_tx);
xStreamBufferReset(instance->stream_rx);
furi_stream_buffer_reset(instance->stream_tx);
furi_stream_buffer_reset(instance->stream_rx);
instance->worker_running = true;

18
lib/subghz/subghz_worker.c
View File

@@ -1,13 +1,12 @@
#include "subghz_worker.h"
#include <stream_buffer.h>
#include <furi.h>
#define TAG "SubGhzWorker"
struct SubGhzWorker {
FuriThread* thread;
StreamBufferHandle_t stream;
FuriStreamBuffer* stream;
volatile bool running;
volatile bool overrun;
@@ -30,16 +29,14 @@ struct SubGhzWorker {
void subghz_worker_rx_callback(bool level, uint32_t duration, void* context) {
SubGhzWorker* instance = context;
BaseType_t xHigherPriorityTaskWoken = pdFALSE;
LevelDuration level_duration = level_duration_make(level, duration);
if(instance->overrun) {
instance->overrun = false;
level_duration = level_duration_reset();
}
size_t ret = xStreamBufferSendFromISR(
instance->stream, &level_duration, sizeof(LevelDuration), &xHigherPriorityTaskWoken);
size_t ret =
furi_stream_buffer_send(instance->stream, &level_duration, sizeof(LevelDuration), 0);
if(sizeof(LevelDuration) != ret) instance->overrun = true;
portYIELD_FROM_ISR(xHigherPriorityTaskWoken);
}
/** Worker callback thread
@@ -52,8 +49,8 @@ static int32_t subghz_worker_thread_callback(void* context) {
LevelDuration level_duration;
while(instance->running) {
int ret =
xStreamBufferReceive(instance->stream, &level_duration, sizeof(LevelDuration), 10);
int ret = furi_stream_buffer_receive(
instance->stream, &level_duration, sizeof(LevelDuration), 10);
if(ret == sizeof(LevelDuration)) {
if(level_duration_is_reset(level_duration)) {
FURI_LOG_E(TAG, "Overrun buffer");
@@ -97,7 +94,8 @@ SubGhzWorker* subghz_worker_alloc() {
furi_thread_set_context(instance->thread, instance);
furi_thread_set_callback(instance->thread, subghz_worker_thread_callback);
instance->stream = xStreamBufferCreate(sizeof(LevelDuration) * 4096, sizeof(LevelDuration));
instance->stream =
furi_stream_buffer_alloc(sizeof(LevelDuration) * 4096, sizeof(LevelDuration));
//setting filter
instance->filter_running = true;
@@ -109,7 +107,7 @@ SubGhzWorker* subghz_worker_alloc() {
void subghz_worker_free(SubGhzWorker* instance) {
furi_assert(instance);
vStreamBufferDelete(instance->stream);
furi_stream_buffer_free(instance->stream);
furi_thread_free(instance->thread);
free(instance);