From 135fbd294bcc9e67eda18230df7547c1ef347645 Mon Sep 17 00:00:00 2001 From: Georgii Surkov <37121527+gsurkov@users.noreply.github.com> Date: Wed, 3 Aug 2022 19:43:14 +0300 Subject: [PATCH] [FL-2693] RW buffered streams (#1523) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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: あく --- .github/workflows/build.yml | 2 +- lib/flipper_format/flipper_format.h | 4 +- lib/toolbox/stream/buffered_file_stream.c | 131 +++++++++++++++++----- lib/toolbox/stream/buffered_file_stream.h | 14 ++- lib/toolbox/stream/stream_cache.c | 23 +++- lib/toolbox/stream/stream_cache.h | 17 +++ 6 files changed, 156 insertions(+), 35 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index b491c600..a1ae875a 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -113,7 +113,7 @@ jobs: run: | echo "${{ secrets.RSYNC_DEPLOY_KEY }}" > deploy_key; chmod 600 ./deploy_key; - rsync -avzP --mkpath \ + rsync -avzP --delete --mkpath \ -e 'ssh -p ${{ secrets.RSYNC_DEPLOY_PORT }} -i ./deploy_key' \ artifacts/ ${{ secrets.RSYNC_DEPLOY_USER }}@${{ secrets.RSYNC_DEPLOY_HOST }}:"${{ secrets.RSYNC_DEPLOY_BASE_PATH }}${{steps.names.outputs.artifacts-path}}/"; rm ./deploy_key; diff --git a/lib/flipper_format/flipper_format.h b/lib/flipper_format/flipper_format.h index 09928c18..6163dee0 100644 --- a/lib/flipper_format/flipper_format.h +++ b/lib/flipper_format/flipper_format.h @@ -116,7 +116,7 @@ FlipperFormat* flipper_format_string_alloc(); FlipperFormat* flipper_format_file_alloc(Storage* storage); /** - * Allocate FlipperFormat as file, buffered read-only mode. + * Allocate FlipperFormat as file, buffered mode. * @return FlipperFormat* pointer to a FlipperFormat instance */ FlipperFormat* flipper_format_buffered_file_alloc(Storage* storage); @@ -131,7 +131,7 @@ FlipperFormat* flipper_format_buffered_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. + * Open existing file, buffered mode. * Use only if FlipperFormat allocated as a file. * @param flipper_format Pointer to a FlipperFormat instance * @param path File path diff --git a/lib/toolbox/stream/buffered_file_stream.c b/lib/toolbox/stream/buffered_file_stream.c index 2f2359a0..3b20a391 100644 --- a/lib/toolbox/stream/buffered_file_stream.c +++ b/lib/toolbox/stream/buffered_file_stream.c @@ -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; } diff --git a/lib/toolbox/stream/buffered_file_stream.h b/lib/toolbox/stream/buffered_file_stream.h index e6ad7209..54917b6a 100644 --- a/lib/toolbox/stream/buffered_file_stream.h +++ b/lib/toolbox/stream/buffered_file_stream.h @@ -19,7 +19,7 @@ Stream* buffered_file_stream_alloc(Storage* storage); * @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. + * @return True on success, False on failure. You need to close the file even if the open operation failed. */ bool buffered_file_stream_open( Stream* stream, @@ -29,12 +29,18 @@ bool buffered_file_stream_open( /** * Closes the file. - * @param stream - * @return true - * @return false + * @param stream pointer to file stream object. + * @return True on success, False on failure. */ bool buffered_file_stream_close(Stream* stream); +/** + * Forces write from cache to the underlying file. + * @param stream pointer to file stream object. + * @return True on success, False on failure. + */ +bool buffered_file_stream_sync(Stream* stream); + /** * Retrieves the error id from the file object * @param stream pointer to stream object. diff --git a/lib/toolbox/stream/stream_cache.c b/lib/toolbox/stream/stream_cache.c index 164ac466..f5e147df 100644 --- a/lib/toolbox/stream/stream_cache.c +++ b/lib/toolbox/stream/stream_cache.c @@ -46,6 +46,14 @@ size_t stream_cache_fill(StreamCache* cache, Stream* stream) { return size_read; } +bool stream_cache_flush(StreamCache* cache, Stream* stream) { + const size_t size_written = stream_write(stream, cache->data, cache->data_size); + const bool success = (size_written == cache->data_size); + cache->data_size = 0; + cache->position = 0; + return success; +} + 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); @@ -56,6 +64,19 @@ size_t stream_cache_read(StreamCache* cache, uint8_t* data, size_t size) { return size_read; } +size_t stream_cache_write(StreamCache* cache, const uint8_t* data, size_t size) { + furi_assert(cache->data_size >= cache->position); + const size_t size_written = MIN(size, STREAM_CACHE_MAX_SIZE - cache->position); + if(size_written > 0) { + memcpy(cache->data + cache->position, data, size_written); + cache->position += size_written; + if(cache->position > cache->data_size) { + cache->data_size = cache->position; + } + } + return size_written; +} + int32_t stream_cache_seek(StreamCache* cache, int32_t offset) { furi_assert(cache->data_size >= cache->position); int32_t actual_offset = 0; @@ -63,7 +84,7 @@ int32_t stream_cache_seek(StreamCache* cache, int32_t offset) { 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)); + actual_offset = MAX(-((int32_t)cache->position), offset); } cache->position += actual_offset; diff --git a/lib/toolbox/stream/stream_cache.h b/lib/toolbox/stream/stream_cache.h index 20c18d80..f61e5e8f 100644 --- a/lib/toolbox/stream/stream_cache.h +++ b/lib/toolbox/stream/stream_cache.h @@ -55,6 +55,14 @@ size_t stream_cache_pos(StreamCache* cache); */ size_t stream_cache_fill(StreamCache* cache, Stream* stream); +/** + * Write as much cached data as possible to a stream. + * @param cache Pointer to a StreamCache instance + * @param stream Pointer to a Stream instance + * @return True on success, False on failure. + */ +bool stream_cache_flush(StreamCache* cache, Stream* stream); + /** * Read cached data and advance the internal cursor. * @param cache Pointer to a StreamCache instance. @@ -64,6 +72,15 @@ size_t stream_cache_fill(StreamCache* cache, Stream* stream); */ size_t stream_cache_read(StreamCache* cache, uint8_t* data, size_t size); +/** + * Write to cached data and advance the internal cursor. + * @param cache Pointer to a StreamCache instance. + * @param data Pointer to a data buffer. + * @param size Maximum size in bytes to write to the cache. + * @return Actual size that was written. + */ +size_t stream_cache_write(StreamCache* cache, const uint8_t* data, size_t size); + /** * Move the internal cursor relatively to its current position. * @param cache Pointer to a StreamCache instance.