[FL-2655, FL-2650] Buffered file streams (#1424)

* Initial buffered file stream implementation
* Fix logical errors
* Fix more logical errors
* Minimally working implementation
* Adapt infrared unit tests for buffered streams
* Increase read buffer size from 512 to 1K
* Correct naming and formatting
* More code improvements
* Allow passing access and open modes for buffered streams
* Implement tests for buffered streams
* Better file and method names
* Add comments and correct formatting
* Use buffered streams in Infrared
* Fix compilation error
This commit is contained in:
Georgii Surkov 2022-07-22 19:00:25 +03:00 committed by GitHub
parent ec57dd310a
commit 16e598b2c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 501 additions and 11 deletions

View File

@ -51,9 +51,9 @@ bool infrared_brute_force_calculate_messages(InfraredBruteForce* brute_force) {
bool success = false;
Storage* storage = furi_record_open("storage");
FlipperFormat* ff = flipper_format_file_alloc(storage);
FlipperFormat* ff = flipper_format_buffered_file_alloc(storage);
success = flipper_format_file_open_existing(ff, brute_force->db_filename);
success = flipper_format_buffered_file_open_existing(ff, brute_force->db_filename);
if(success) {
string_t signal_name;
string_init(signal_name);
@ -95,8 +95,9 @@ bool infrared_brute_force_start(
if(*record_count) {
Storage* storage = furi_record_open("storage");
brute_force->ff = flipper_format_file_alloc(storage);
success = flipper_format_file_open_existing(brute_force->ff, brute_force->db_filename);
brute_force->ff = flipper_format_buffered_file_alloc(storage);
success =
flipper_format_buffered_file_open_existing(brute_force->ff, brute_force->db_filename);
if(!success) {
flipper_format_free(brute_force->ff);
brute_force->ff = NULL;

View File

@ -140,13 +140,13 @@ bool infrared_remote_store(InfraredRemote* remote) {
bool infrared_remote_load(InfraredRemote* remote, string_t path) {
Storage* storage = furi_record_open("storage");
FlipperFormat* ff = flipper_format_file_alloc(storage);
FlipperFormat* ff = flipper_format_buffered_file_alloc(storage);
string_t buf;
string_init(buf);
FURI_LOG_I(TAG, "load file: \'%s\'", string_get_cstr(path));
bool success = flipper_format_file_open_existing(ff, string_get_cstr(path));
bool success = flipper_format_buffered_file_open_existing(ff, string_get_cstr(path));
if(success) {
uint32_t version;

View File

@ -22,7 +22,7 @@ static void infrared_test_alloc() {
test = malloc(sizeof(InfraredTest));
test->decoder_handler = infrared_alloc_decoder();
test->encoder_handler = infrared_alloc_encoder();
test->ff = flipper_format_file_alloc(storage);
test->ff = flipper_format_buffered_file_alloc(storage);
string_init(test->file_path);
}
@ -52,7 +52,8 @@ static bool infrared_test_prepare_file(const char* protocol_name) {
do {
uint32_t format_version;
if(!flipper_format_file_open_existing(test->ff, string_get_cstr(test->file_path))) break;
if(!flipper_format_buffered_file_open_existing(test->ff, string_get_cstr(test->file_path)))
break;
if(!flipper_format_read_header(test->ff, file_type, &format_version)) break;
if(string_cmp_str(file_type, "IR tests file") || format_version != 1) break;
success = true;
@ -230,7 +231,7 @@ static void infrared_test_run_encoder(InfraredProtocol protocol, uint32_t test_i
test->ff, string_get_cstr(buf), &expected_timings, &expected_timings_count),
"Failed to load raw signal from file");
flipper_format_file_close(test->ff);
flipper_format_buffered_file_close(test->ff);
string_clear(buf);
uint32_t j = 0;
@ -280,7 +281,7 @@ static void infrared_test_run_encoder_decoder(InfraredProtocol protocol, uint32_
test->ff, string_get_cstr(buf), &input_messages, &input_messages_count),
"Failed to load messages from file");
flipper_format_file_close(test->ff);
flipper_format_buffered_file_close(test->ff);
string_clear(buf);
for(uint32_t message_counter = 0; message_counter < input_messages_count; ++message_counter) {
@ -343,7 +344,7 @@ static void infrared_test_run_decoder(InfraredProtocol protocol, uint32_t test_i
infrared_test_load_messages(test->ff, string_get_cstr(buf), &messages, &messages_count),
"Failed to load messages from file");
flipper_format_file_close(test->ff);
flipper_format_buffered_file_close(test->ff);
string_clear(buf);
InfraredMessage message_decoded_check_local;

View File

@ -2,6 +2,7 @@
#include <toolbox/stream/stream.h>
#include <toolbox/stream/string_stream.h>
#include <toolbox/stream/file_stream.h>
#include <toolbox/stream/buffered_file_stream.h>
#include <storage/storage.h>
#include "../minunit.h"
@ -282,6 +283,13 @@ MU_TEST(stream_composite_test) {
mu_check(file_stream_open(stream, "/ext/filestream.str", FSAM_READ_WRITE, FSOM_CREATE_ALWAYS));
MU_RUN_TEST_1(stream_composite_subtest, stream);
stream_free(stream);
// test buffered file stream
stream = buffered_file_stream_alloc(storage);
mu_check(buffered_file_stream_open(
stream, "/ext/filestream.str", FSAM_READ_WRITE, FSOM_CREATE_ALWAYS));
MU_RUN_TEST_1(stream_composite_subtest, stream);
stream_free(stream);
furi_record_close("storage");
}
@ -366,13 +374,101 @@ MU_TEST(stream_split_test) {
mu_check(file_stream_open(stream, "/ext/filestream.str", FSAM_READ_WRITE, FSOM_CREATE_ALWAYS));
MU_RUN_TEST_1(stream_split_subtest, stream);
stream_free(stream);
// test buffered stream
stream = buffered_file_stream_alloc(storage);
mu_check(buffered_file_stream_open(
stream, "/ext/filestream.str", FSAM_READ_WRITE, FSOM_CREATE_ALWAYS));
MU_RUN_TEST_1(stream_split_subtest, stream);
stream_free(stream);
furi_record_close("storage");
}
MU_TEST(stream_buffered_large_file_test) {
string_t input_data;
string_t output_data;
string_init(input_data);
string_init(output_data);
Storage* storage = furi_record_open("storage");
// generate test data consisting of several identical lines
const size_t data_size = 4096;
const size_t line_size = strlen(stream_test_data);
const size_t rep_count = data_size / line_size + 1;
for(size_t i = 0; i < rep_count; ++i) {
string_cat_printf(input_data, "%s\n", stream_test_data);
}
// write test data to file
Stream* stream = buffered_file_stream_alloc(storage);
mu_check(buffered_file_stream_open(
stream, "/ext/filestream.str", FSAM_READ_WRITE, FSOM_CREATE_ALWAYS));
mu_assert_int_eq(0, stream_size(stream));
mu_assert_int_eq(string_size(input_data), stream_write_string(stream, input_data));
mu_assert_int_eq(string_size(input_data), stream_size(stream));
const size_t substr_start = 8;
const size_t substr_len = 11;
mu_check(stream_seek(stream, substr_start, StreamOffsetFromStart));
mu_assert_int_eq(substr_start, stream_tell(stream));
// copy one substring from test data
char test_substr[substr_len + 1];
memset(test_substr, 0, substr_len + 1);
memcpy(test_substr, stream_test_data + substr_start, substr_len);
char buf[substr_len + 1];
memset(buf, 0, substr_len + 1);
// read substring
mu_assert_int_eq(substr_len, stream_read(stream, (uint8_t*)buf, substr_len));
mu_assert_string_eq(test_substr, buf);
memset(buf, 0, substr_len + 1);
// forward seek to cause a cache miss
mu_check(stream_seek(
stream, (line_size + 1) * (rep_count - 1) - substr_len, StreamOffsetFromCurrent));
// read same substring from a different line
mu_assert_int_eq(substr_len, stream_read(stream, (uint8_t*)buf, substr_len));
mu_assert_string_eq(test_substr, buf);
memset(buf, 0, substr_len + 1);
// backward seek to cause a cache miss
mu_check(stream_seek(
stream, -((line_size + 1) * (rep_count - 1) + substr_len), StreamOffsetFromCurrent));
mu_assert_int_eq(substr_len, stream_read(stream, (uint8_t*)buf, substr_len));
mu_assert_string_eq(test_substr, buf);
// read the whole file
mu_check(stream_rewind(stream));
string_t tmp;
string_init(tmp);
while(stream_read_line(stream, tmp)) {
string_cat(output_data, tmp);
}
string_clear(tmp);
// check against generated data
mu_assert_int_eq(string_size(input_data), string_size(output_data));
mu_check(string_equal_p(input_data, output_data));
mu_check(stream_eof(stream));
stream_free(stream);
furi_record_close("storage");
string_clear(input_data);
string_clear(output_data);
}
MU_TEST_SUITE(stream_suite) {
MU_RUN_TEST(stream_write_read_save_load_test);
MU_RUN_TEST(stream_composite_test);
MU_RUN_TEST(stream_split_test);
MU_RUN_TEST(stream_buffered_large_file_test);
}
int run_minunit_test_stream() {

View File

@ -2,6 +2,7 @@
#include <toolbox/stream/stream.h>
#include <toolbox/stream/string_stream.h>
#include <toolbox/stream/file_stream.h>
#include <toolbox/stream/buffered_file_stream.h>
#include "flipper_format.h"
#include "flipper_format_i.h"
#include "flipper_format_stream.h"
@ -36,11 +37,24 @@ FlipperFormat* flipper_format_file_alloc(Storage* storage) {
return flipper_format;
}
FlipperFormat* flipper_format_buffered_file_alloc(Storage* storage) {
FlipperFormat* flipper_format = malloc(sizeof(FlipperFormat));
flipper_format->stream = buffered_file_stream_alloc(storage);
flipper_format->strict_mode = false;
return flipper_format;
}
bool flipper_format_file_open_existing(FlipperFormat* flipper_format, const char* path) {
furi_assert(flipper_format);
return file_stream_open(flipper_format->stream, path, FSAM_READ_WRITE, FSOM_OPEN_EXISTING);
}
bool flipper_format_buffered_file_open_existing(FlipperFormat* flipper_format, const char* path) {
furi_assert(flipper_format);
return buffered_file_stream_open(
flipper_format->stream, path, FSAM_READ_WRITE, FSOM_OPEN_EXISTING);
}
bool flipper_format_file_open_append(FlipperFormat* flipper_format, const char* path) {
furi_assert(flipper_format);
@ -87,6 +101,11 @@ bool flipper_format_file_close(FlipperFormat* flipper_format) {
return file_stream_close(flipper_format->stream);
}
bool flipper_format_buffered_file_close(FlipperFormat* flipper_format) {
furi_assert(flipper_format);
return buffered_file_stream_close(flipper_format->stream);
}
void flipper_format_free(FlipperFormat* flipper_format) {
furi_assert(flipper_format);
stream_free(flipper_format->stream);

View File

@ -115,6 +115,12 @@ FlipperFormat* flipper_format_string_alloc();
*/
FlipperFormat* flipper_format_file_alloc(Storage* storage);
/**
* Allocate FlipperFormat as file, buffered read-only mode.
* @return FlipperFormat* pointer to a FlipperFormat instance
*/
FlipperFormat* flipper_format_buffered_file_alloc(Storage* storage);
/**
* Open existing file.
* Use only if FlipperFormat allocated as a file.
@ -124,6 +130,15 @@ FlipperFormat* flipper_format_file_alloc(Storage* storage);
*/
bool flipper_format_file_open_existing(FlipperFormat* flipper_format, const char* path);
/**
* Open existing file, read-only with buffered read operations.
* Use only if FlipperFormat allocated as a file.
* @param flipper_format Pointer to a FlipperFormat instance
* @param path File path
* @return True on success
*/
bool flipper_format_buffered_file_open_existing(FlipperFormat* flipper_format, const char* path);
/**
* Open existing file for writing and add values to the end of file.
* Use only if FlipperFormat allocated as a file.
@ -159,6 +174,14 @@ bool flipper_format_file_open_new(FlipperFormat* flipper_format, const char* pat
*/
bool flipper_format_file_close(FlipperFormat* flipper_format);
/**
* Closes the file, use only if FlipperFormat allocated as a buffered file.
* @param flipper_format
* @return true
* @return false
*/
bool flipper_format_buffered_file_close(FlipperFormat* flipper_format);
/**
* Free FlipperFormat.
* @param flipper_format Pointer to a FlipperFormat instance

View File

@ -0,0 +1,155 @@
#include "buffered_file_stream.h"
#include "stream_i.h"
#include "file_stream.h"
#include "stream_cache.h"
typedef struct {
Stream stream_base;
Stream* file_stream;
StreamCache* cache;
} BufferedFileStream;
static void buffered_file_stream_free(BufferedFileStream* stream);
static bool buffered_file_stream_eof(BufferedFileStream* stream);
static void buffered_file_stream_clean(BufferedFileStream* stream);
static bool
buffered_file_stream_seek(BufferedFileStream* stream, int32_t offset, StreamOffset offset_type);
static size_t buffered_file_stream_tell(BufferedFileStream* stream);
static size_t buffered_file_stream_size(BufferedFileStream* stream);
static size_t
buffered_file_stream_write(BufferedFileStream* stream, const uint8_t* data, size_t size);
static size_t buffered_file_stream_read(BufferedFileStream* stream, uint8_t* data, size_t size);
static bool buffered_file_stream_delete_and_insert(
BufferedFileStream* stream,
size_t delete_size,
StreamWriteCB write_callback,
const void* ctx);
const StreamVTable buffered_file_stream_vtable = {
.free = (StreamFreeFn)buffered_file_stream_free,
.eof = (StreamEOFFn)buffered_file_stream_eof,
.clean = (StreamCleanFn)buffered_file_stream_clean,
.seek = (StreamSeekFn)buffered_file_stream_seek,
.tell = (StreamTellFn)buffered_file_stream_tell,
.size = (StreamSizeFn)buffered_file_stream_size,
.write = (StreamWriteFn)buffered_file_stream_write,
.read = (StreamReadFn)buffered_file_stream_read,
.delete_and_insert = (StreamDeleteAndInsertFn)buffered_file_stream_delete_and_insert,
};
Stream* buffered_file_stream_alloc(Storage* storage) {
BufferedFileStream* stream = malloc(sizeof(BufferedFileStream));
stream->file_stream = file_stream_alloc(storage);
stream->cache = stream_cache_alloc();
stream->stream_base.vtable = &buffered_file_stream_vtable;
return (Stream*)stream;
}
bool buffered_file_stream_open(
Stream* _stream,
const char* path,
FS_AccessMode access_mode,
FS_OpenMode open_mode) {
furi_assert(_stream);
BufferedFileStream* stream = (BufferedFileStream*)_stream;
stream_cache_drop(stream->cache);
furi_check(stream->stream_base.vtable == &buffered_file_stream_vtable);
return file_stream_open(stream->file_stream, path, access_mode, open_mode);
}
bool buffered_file_stream_close(Stream* _stream) {
furi_assert(_stream);
BufferedFileStream* stream = (BufferedFileStream*)_stream;
furi_check(stream->stream_base.vtable == &buffered_file_stream_vtable);
return file_stream_close(stream->file_stream);
}
FS_Error buffered_file_stream_get_error(Stream* _stream) {
furi_assert(_stream);
BufferedFileStream* stream = (BufferedFileStream*)_stream;
furi_check(stream->stream_base.vtable == &buffered_file_stream_vtable);
return file_stream_get_error(stream->file_stream);
}
static void buffered_file_stream_free(BufferedFileStream* stream) {
furi_assert(stream);
stream_free(stream->file_stream);
stream_cache_free(stream->cache);
free(stream);
}
static bool buffered_file_stream_eof(BufferedFileStream* stream) {
return stream_cache_at_end(stream->cache) && stream_eof(stream->file_stream);
}
static void buffered_file_stream_clean(BufferedFileStream* stream) {
stream_cache_drop(stream->cache);
stream_clean(stream->file_stream);
}
static bool buffered_file_stream_seek(
BufferedFileStream* stream,
int32_t offset,
StreamOffset offset_type) {
bool success = false;
int32_t new_offset = offset;
if(offset_type == StreamOffsetFromCurrent) {
new_offset -= stream_cache_seek(stream->cache, offset);
if(new_offset < 0) {
new_offset -= (int32_t)stream_cache_size(stream->cache);
}
}
if((new_offset != 0) || (offset_type != StreamOffsetFromCurrent)) {
stream_cache_drop(stream->cache);
success = stream_seek(stream->file_stream, new_offset, offset_type);
} else {
success = true;
}
return success;
}
static size_t buffered_file_stream_tell(BufferedFileStream* stream) {
return stream_tell(stream->file_stream) + stream_cache_pos(stream->cache) -
stream_cache_size(stream->cache);
}
static size_t buffered_file_stream_size(BufferedFileStream* stream) {
return stream_cache_size(stream->cache) + stream_size(stream->file_stream);
}
static size_t
buffered_file_stream_write(BufferedFileStream* stream, const uint8_t* data, size_t size) {
stream_cache_drop(stream->cache);
return stream_write(stream->file_stream, data, size);
}
static size_t buffered_file_stream_read(BufferedFileStream* stream, uint8_t* data, size_t size) {
size_t need_to_read = size;
while(need_to_read) {
need_to_read -=
stream_cache_read(stream->cache, data + (size - need_to_read), need_to_read);
if(need_to_read) {
if(!stream_cache_fill(stream->cache, stream->file_stream)) {
break;
}
}
}
return size - need_to_read;
}
static bool buffered_file_stream_delete_and_insert(
BufferedFileStream* stream,
size_t delete_size,
StreamWriteCB write_callback,
const void* ctx) {
stream_cache_drop(stream->cache);
return stream_delete_and_insert(stream->file_stream, delete_size, write_callback, ctx);
}

View File

@ -0,0 +1,47 @@
#pragma once
#include <stdlib.h>
#include <storage/storage.h>
#include "stream.h"
#ifdef __cplusplus
extern "C" {
#endif
/**
* Allocate a file stream with buffered read operations
* @return Stream*
*/
Stream* buffered_file_stream_alloc(Storage* storage);
/**
* Opens an existing file or creates a new one.
* @param stream pointer to file stream object.
* @param path path to file
* @param access_mode access mode from FS_AccessMode
* @param open_mode open mode from FS_OpenMode
* @return success flag. You need to close the file even if the open operation failed.
*/
bool buffered_file_stream_open(
Stream* stream,
const char* path,
FS_AccessMode access_mode,
FS_OpenMode open_mode);
/**
* Closes the file.
* @param stream
* @return true
* @return false
*/
bool buffered_file_stream_close(Stream* stream);
/**
* Retrieves the error id from the file object
* @param stream pointer to stream object.
* @return FS_Error error id
*/
FS_Error buffered_file_stream_get_error(Stream* stream);
#ifdef __cplusplus
}
#endif

View File

@ -0,0 +1,71 @@
#include "stream_cache.h"
#define STREAM_CACHE_MAX_SIZE 1024U
struct StreamCache {
uint8_t data[STREAM_CACHE_MAX_SIZE];
size_t data_size;
size_t position;
};
StreamCache* stream_cache_alloc() {
StreamCache* cache = malloc(sizeof(StreamCache));
cache->data_size = 0;
cache->position = 0;
return cache;
}
void stream_cache_free(StreamCache* cache) {
furi_assert(cache);
cache->data_size = 0;
cache->position = 0;
free(cache);
}
void stream_cache_drop(StreamCache* cache) {
cache->data_size = 0;
cache->position = 0;
}
bool stream_cache_at_end(StreamCache* cache) {
furi_assert(cache->data_size >= cache->position);
return cache->data_size == cache->position;
}
size_t stream_cache_size(StreamCache* cache) {
return cache->data_size;
}
size_t stream_cache_pos(StreamCache* cache) {
return cache->position;
}
size_t stream_cache_fill(StreamCache* cache, Stream* stream) {
const size_t size_read = stream_read(stream, cache->data, STREAM_CACHE_MAX_SIZE);
cache->data_size = size_read;
cache->position = 0;
return size_read;
}
size_t stream_cache_read(StreamCache* cache, uint8_t* data, size_t size) {
furi_assert(cache->data_size >= cache->position);
const size_t size_read = MIN(size, cache->data_size - cache->position);
if(size_read > 0) {
memcpy(data, cache->data + cache->position, size_read);
cache->position += size_read;
}
return size_read;
}
int32_t stream_cache_seek(StreamCache* cache, int32_t offset) {
furi_assert(cache->data_size >= cache->position);
int32_t actual_offset = 0;
if(offset > 0) {
actual_offset = MIN(cache->data_size - cache->position, (size_t)offset);
} else if(offset < 0) {
actual_offset = -MIN(cache->position, (size_t)abs(offset));
}
cache->position += actual_offset;
return actual_offset;
}

View File

@ -0,0 +1,77 @@
#pragma once
#include "stream.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct StreamCache StreamCache;
/**
* Allocate stream cache.
* @return StreamCache* pointer to a StreamCache instance
*/
StreamCache* stream_cache_alloc();
/**
* Free stream cache.
* @param cache Pointer to a StreamCache instance
*/
void stream_cache_free(StreamCache* cache);
/**
* Drop the cache contents and set it to initial state.
* @param cache Pointer to a StreamCache instance
*/
void stream_cache_drop(StreamCache* cache);
/**
* Determine if the internal cursor is at end the end of cached data.
* @param cache Pointer to a StreamCache instance
* @return True if cursor is at end, otherwise false.
*/
bool stream_cache_at_end(StreamCache* cache);
/**
* Get the current size of cached data.
* @param cache Pointer to a StreamCache instance
* @return Size of cached data.
*/
size_t stream_cache_size(StreamCache* cache);
/**
* Get the internal cursor position.
* @param cache Pointer to a StreamCache instance
* @return Cursor position inside the cache.
*/
size_t stream_cache_pos(StreamCache* cache);
/**
* Load the cache with new data from a stream.
* @param cache Pointer to a StreamCache instance
* @param stream Pointer to a Stream instance
* @return Size of newly cached data.
*/
size_t stream_cache_fill(StreamCache* cache, Stream* stream);
/**
* Read cached data and advance the internal cursor.
* @param cache Pointer to a StreamCache instance.
* @param data Pointer to a data buffer. Must be initialized.
* @param size Maximum size in bytes to read from the cache.
* @return Actual size that was read.
*/
size_t stream_cache_read(StreamCache* cache, uint8_t* data, size_t size);
/**
* Move the internal cursor relatively to its current position.
* @param cache Pointer to a StreamCache instance.
* @param offset Cursor offset.
* @return Actual cursor offset. Equal to offset parameter on hit.
*/
int32_t stream_cache_seek(StreamCache* cache, int32_t offset);
#ifdef __cplusplus
}
#endif