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

* 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: あく <[email protected]>

.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;

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

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;
 }

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.

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,14 +64,27 @@ 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;
 
     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;

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.