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

[FL-2693] RW buffered streams (#1523)

Browse Source 
* Add write methods for stream cache
* Fix logical error
* Implement write cache for buffered file streams
* Minor code refactoring
* Less ugly code
* Better read() implementation
* Intermediate implementation
* Fix logical error
* Code cleanup
* Update FFF comments
* Fix logical error
* Github: rsync with delete

Co-authored-by: あく <alleteam@gmail.com>
This commit is contained in:
Georgii Surkov
2022-08-03 19:43:14 +03:00
committed by GitHub
parent 6499597586
commit 135fbd294b
6 changed files with 156 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
lib/toolbox/stream/buffered_file_stream.c
View File

@@ -1,6 +1,5 @@
#include "buffered_file_stream.h"
#include "core/check.h"
#include "stream_i.h"
#include "file_stream.h"
#include "stream_cache.h"
@@ -9,6 +8,7 @@ typedef struct {
Stream stream_base;
Stream* file_stream;
StreamCache* cache;
bool sync_pending;
} BufferedFileStream;
static void buffered_file_stream_free(BufferedFileStream* stream);
@@ -27,6 +27,9 @@ static bool buffered_file_stream_delete_and_insert(
StreamWriteCB write_callback,
const void* ctx);
static bool buffered_file_stream_flush(BufferedFileStream* stream);
static bool buffered_file_stream_unread(BufferedFileStream* stream);
const StreamVTable buffered_file_stream_vtable = {
.free = (StreamFreeFn)buffered_file_stream_free,
.eof = (StreamEOFFn)buffered_file_stream_eof,
@@ -39,13 +42,12 @@ const StreamVTable buffered_file_stream_vtable = {
.delete_and_insert = (StreamDeleteAndInsertFn)buffered_file_stream_delete_and_insert,
};
static bool buffered_file_stream_unread(BufferedFileStream* stream);
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->sync_pending = false;
stream->stream_base.vtable = &buffered_file_stream_vtable;
return (Stream*)stream;
@@ -58,7 +60,6 @@ bool buffered_file_stream_open(
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);
}
@@ -67,7 +68,22 @@ 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);
bool success = false;
do {
if(!(stream->sync_pending ? buffered_file_stream_flush(stream) :
buffered_file_stream_unread(stream)))
break;
if(!file_stream_close(stream->file_stream)) break;
success = true;
} while(false);
return success;
}
bool buffered_file_stream_sync(Stream* _stream) {
furi_assert(_stream);
BufferedFileStream* stream = (BufferedFileStream*)_stream;
furi_check(stream->stream_base.vtable == &buffered_file_stream_vtable);
return stream->sync_pending ? buffered_file_stream_flush(stream) : true;
}
FS_Error buffered_file_stream_get_error(Stream* _stream) {
@@ -85,10 +101,23 @@ static void buffered_file_stream_free(BufferedFileStream* stream) {
}
static bool buffered_file_stream_eof(BufferedFileStream* stream) {
return stream_cache_at_end(stream->cache) && stream_eof(stream->file_stream);
bool ret;
const bool file_stream_eof = stream_eof(stream->file_stream);
const bool cache_at_end = stream_cache_at_end(stream->cache);
if(!stream->sync_pending) {
ret = file_stream_eof && cache_at_end;
} else {
const size_t remaining_size =
stream_size(stream->file_stream) - stream_tell(stream->file_stream);
ret = stream_cache_size(stream->cache) >=
(remaining_size ? cache_at_end : file_stream_eof);
}
return ret;
}
static void buffered_file_stream_clean(BufferedFileStream* stream) {
// Not syncing because data will be deleted anyway
stream->sync_pending = false;
stream_cache_drop(stream->cache);
stream_clean(stream->file_stream);
}
@@ -97,7 +126,7 @@ static bool buffered_file_stream_seek(
BufferedFileStream* stream,
int32_t offset,
StreamOffset offset_type) {
bool success = false;
bool success = true;
int32_t new_offset = offset;
if(offset_type == StreamOffsetFromCurrent) {
@@ -108,47 +137,71 @@ static bool buffered_file_stream_seek(
}
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;
if(stream->sync_pending) {
success = buffered_file_stream_sync((Stream*)stream);
} else {
stream_cache_drop(stream->cache);
}
if(success) {
success = stream_seek(stream->file_stream, new_offset, offset_type);
}
}
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);
size_t pos = stream_tell(stream->file_stream) + stream_cache_pos(stream->cache);
if(!stream->sync_pending) {
pos -= stream_cache_size(stream->cache);
}
return pos;
}
static size_t buffered_file_stream_size(BufferedFileStream* stream) {
return stream_cache_size(stream->cache) + stream_size(stream->file_stream);
size_t size = stream_size(stream->file_stream);
if(stream->sync_pending) {
const size_t remaining_size = size - stream_tell(stream->file_stream);
const size_t cache_size = stream_cache_size(stream->cache);
if(cache_size > remaining_size) {
size += (cache_size - remaining_size);
}
}
return size;
}
static size_t
buffered_file_stream_write(BufferedFileStream* stream, const uint8_t* data, size_t size) {
size_t need_to_write = size;
do {
if(!buffered_file_stream_unread(stream)) break;
need_to_write -= stream_write(stream->file_stream, data, size);
if(!stream->sync_pending) {
if(!buffered_file_stream_unread(stream)) break;
}
while(need_to_write) {
stream->sync_pending = true;
need_to_write -=
stream_cache_write(stream->cache, data + (size - need_to_write), need_to_write);
if(need_to_write) {
stream->sync_pending = false;
if(!stream_cache_flush(stream->cache, stream->file_stream)) break;
}
}
} while(false);
return size - need_to_write;
}
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;
if(stream->sync_pending) {
if(!buffered_file_stream_flush(stream)) break;
}
if(!stream_cache_fill(stream->cache, stream->file_stream)) break;
}
}
return size - need_to_read;
}
@@ -157,19 +210,43 @@ static bool buffered_file_stream_delete_and_insert(
size_t delete_size,
StreamWriteCB write_callback,
const void* ctx) {
return buffered_file_stream_unread(stream) &&
stream_delete_and_insert(stream->file_stream, delete_size, write_callback, ctx);
bool success = false;
do {
if(!(stream->sync_pending ? buffered_file_stream_flush(stream) :
buffered_file_stream_unread(stream)))
break;
if(!stream_delete_and_insert(stream->file_stream, delete_size, write_callback, ctx)) break;
success = true;
} while(false);
return success;
}
// Write the cache into the underlying stream and adjust seek position
static bool buffered_file_stream_flush(BufferedFileStream* stream) {
bool success = false;
do {
const int32_t offset = stream_cache_size(stream->cache) - stream_cache_pos(stream->cache);
if(!stream_cache_flush(stream->cache, stream->file_stream)) break;
if(offset > 0) {
if(!stream_seek(stream->file_stream, -offset, StreamOffsetFromCurrent)) break;
}
success = true;
} while(false);
stream->sync_pending = false;
return success;
}
// Drop read cache and adjust the underlying stream seek position
static bool buffered_file_stream_unread(BufferedFileStream* stream) {
bool success = true;
const size_t cache_size = stream_cache_size(stream->cache);
const size_t cache_pos = stream_cache_pos(stream->cache);
if(cache_pos < cache_size) {
const int32_t offset = cache_size - cache_pos;
success = stream_seek(stream->file_stream, -offset, StreamOffsetFromCurrent);
if(cache_size > 0) {
const size_t cache_pos = stream_cache_pos(stream->cache);
if(cache_pos < cache_size) {
const int32_t offset = cache_size - cache_pos;
success = stream_seek(stream->file_stream, -offset, StreamOffsetFromCurrent);
}
stream_cache_drop(stream->cache);
}
stream_cache_drop(stream->cache);
return success;
}