Add tests for concurrent reading/writing; bugfixes

system/inc/system_ledger.h

@@ -198,6 +198,8 @@ void ledger_set_callbacks(ledger_instance* ledger, const ledger_callbacks* callb
  * Attach application-specific data to a ledger instance.
  *
  * If a destructor callback is provided, it will be invoked when the ledger instance is destroyed.
+ * The calling code is responsible for destroying the old application data if it was already set for
+ * this ledger instance.
  *
  * @param ledger Ledger instance.
  * @param app_data Application data.

system/src/system_ledger_internal.cpp

@@ -497,7 +497,7 @@ int Ledger::writerClosed(const LedgerInfo& info, LedgerWriteSource src, int temp
         // Create a new file in "staged". Note that in this case, "current" can't be replaced with
         // the new data even if nobody is reading it, as otherwise it would get overwritten with
         // outdated data if the device resets before the staged directory is cleaned up
-        CHECK(getStagedDirPath(destPath, sizeof(destPath), name_));
+        CHECK(getStagedFilePath(destPath, sizeof(destPath), name_, tempSeqNum));
         newStagedFile = true;
     }
     CHECK_FS(lfs_rename(fs.instance(), srcPath, destPath));

user/tests/wiring/ledger/ledger.cpp

@@ -1,6 +1,8 @@
 #include "application.h"
 #include "unit-test/unit-test.h"
 
+#include "scope_guard.h"
+
 namespace {
 
 struct CountingCallback {
@@ -47,7 +49,7 @@ test(02_initial_state) {
     assertEqual(ledger.dataSize(), 0);
 }
 
-test(03_replace) {
+test(03_replace_data) {
     {
         auto ledger = Particle.ledger("test");
         LedgerData d = { { "a", 1 }, { "b", 2 }, { "c", 3 } };
@@ -56,12 +58,12 @@ test(03_replace) {
     }
     {
         auto ledger = Particle.ledger("test");
-        LedgerData d = ledger.get();
+        auto d = ledger.get();
         assertTrue((d == LedgerData{ { "a", 1 }, { "b", 2 }, { "c", 3 } }));
     }
 }
 
-test(04_merge) {
+test(04_merge_data) {
     {
         auto ledger = Particle.ledger("test");
         LedgerData d = { { "d", 4 }, { "f", 6 } };
@@ -71,12 +73,93 @@ test(04_merge) {
     }
     {
         auto ledger = Particle.ledger("test");
-        LedgerData d = ledger.get();
+        auto d = ledger.get();
         assertTrue((d == LedgerData{ { "d", 4 }, { "e", 5 }, { "f", 6 } }));
     }
 }
 
-test(05_sync_callback) {
+test(05_concurrent_writing) {
+    // Open two ledger streams for writing
+    ledger_instance* lr = nullptr;
+    assertEqual(ledger_get_instance(&lr, "test", nullptr), 0);
+    SCOPE_GUARD({
+        ledger_release(lr, nullptr);
+    });
+    ledger_stream* ls1 = nullptr;
+    assertEqual(ledger_open(&ls1, lr, LEDGER_STREAM_MODE_WRITE, nullptr), 0);
+    NAMED_SCOPE_GUARD(g1, {
+        ledger_close(ls1, 0, nullptr);
+    });
+    ledger_stream* ls2 = nullptr;
+    assertEqual(ledger_open(&ls2, lr, LEDGER_STREAM_MODE_WRITE, nullptr), 0);
+    NAMED_SCOPE_GUARD(g2, {
+        ledger_close(ls2, 0, nullptr);
+    });
+
+    // Write some data to both streams
+    String s = "{\"a\":1,\"b\":2,\"c\":3}";
+    assertEqual(ledger_write(ls1, s.c_str(), s.length(), nullptr), s.length());
+    s = "{\"d\":4,\"e\":5,\"f\":6}";
+    assertEqual(ledger_write(ls2, s.c_str(), s.length(), nullptr), s.length());
+
+    // Close the first stream and check that the data can be read back
+    g1.dismiss();
+    assertEqual(ledger_close(ls1, 0, nullptr), 0);
+    auto ledger = Particle.ledger("test");
+    auto d = ledger.get();
+    assertTrue((d == LedgerData{ { "a", 1 }, { "b", 2 }, { "c", 3 } }));
+
+    // Close the second stream and check that the data can be read back
+    g2.dismiss();
+    assertEqual(ledger_close(ls2, 0, nullptr), 0);
+    d = ledger.get();
+    assertTrue((d == LedgerData{ { "d", 4 }, { "e", 5 }, { "f", 6 } }));
+}
+
+test(06_concurrent_reading_and_writing) {
+    // Set some initial data
+    auto ledger = Particle.ledger("test");
+    LedgerData d = { { "a", 1 }, { "b", 2 }, { "c", 3 } };
+    assertEqual(ledger.set(d), 0);
+
+    // Open the ledger for reading
+    ledger_instance* lr = nullptr;
+    assertEqual(ledger_get_instance(&lr, "test", nullptr), 0);
+    SCOPE_GUARD({
+        ledger_release(lr, nullptr);
+    });
+    ledger_stream* ls = nullptr;
+    assertEqual(ledger_open(&ls, lr, LEDGER_STREAM_MODE_READ, nullptr), 0);
+    SCOPE_GUARD({
+        ledger_close(ls, 0, nullptr);
+    });
+
+    // Set some new data
+    d = { { "d", 4 }, { "e", 5 }, { "f", 6 } };
+    assertEqual(ledger.set(d), 0);
+
+    // Proceed to read the data
+    String s;
+    OutputStringStream ss(s);
+    char buf[128];
+    for (;;) {
+        int r = ledger_read(ls, buf, sizeof(buf), nullptr);
+        if (r == SYSTEM_ERROR_END_OF_STREAM) {
+            break;
+        }
+        assertMore(r, 0);
+        ss.write((uint8_t*)buf, r);
+    }
+
+    // The reader opened before the modification should see the original data
+    assertTrue(s == "{\"a\":1,\"b\":2,\"c\":3}");
+
+    // The reader opened after the modification should see the actual data
+    d = ledger.get();
+    assertTrue((d == LedgerData{ { "d", 4 }, { "e", 5 }, { "f", 6 } }));
+}
+
+test(07_set_sync_callback) {
     auto ledger = Particle.ledger("test");
 
     // Register a functor callback
@@ -88,7 +171,7 @@ test(05_sync_callback) {
     ledger.onSync(cb);
     assertEqual(CountingCallback::instanceCount, 0);
 
-    // Register a functor callback again callback
+    // Register a functor callback again
     ledger.onSync(CountingCallback());
     assertEqual(CountingCallback::instanceCount, 1);
 
@@ -97,7 +180,7 @@ test(05_sync_callback) {
     assertEqual(CountingCallback::instanceCount, 0);
 }
 
-test(06_purge) {
+test(08_purge) {
     // Remove the test ledger files
     assertEqual(ledger_purge("test", nullptr), 0);
 }

wiring/src/spark_wiring_cloud.cpp

@@ -132,6 +132,7 @@ Ledger CloudClass::ledger(const char* name) {
     int r = ledger_get_instance(&instance, name, nullptr);
     if (r < 0) {
         LOG(ERROR, "ledger_get_instance() failed: %d", r);
+        return Ledger();
     }
     return Ledger(instance, false /* addRef */);
 }

wiring/src/spark_wiring_ledger.cpp

@@ -110,13 +110,14 @@ int setSyncCallback(ledger_instance* ledger, CallbackT&& callback) {
         appData = newAppData.get();
     } else if (!callback) {
         ledger_set_callbacks(ledger, nullptr, nullptr); // Clear the callback
-        ledger_set_app_data(ledger, nullptr, nullptr, nullptr); // Destroy the app data
+        ledger_set_app_data(ledger, nullptr, nullptr, nullptr); // Clear the app data
+        delete appData; // Destroy the app data
         ledger_release(ledger, nullptr); // See below
         return 0;
     }
     appData->onSync = std::move(callback);
     if (newAppData) {
-        ledger_set_app_data(ledger, newAppData.release(), destroyAppData, nullptr); // Transfer ownership over the app data
+        ledger_set_app_data(ledger, newAppData.release(), destroyAppData, nullptr); // Transfer ownership over the app data to the system
         ledger_callbacks callbacks = {};
         callbacks.version = LEDGER_API_VERSION;
         callbacks.sync = syncCallbackSystem;