faasm · Shillaker · Mar 15, 2022 · Mar 3, 2022 · Mar 3, 2022 · Mar 4, 2022
diff --git a/clients/cpp b/clients/cpp
diff --git a/docs/source/development.md b/docs/source/development.md
@@ -219,55 +219,47 @@ interfering with the tests:
 docker-compose down
 ```
 
-Start the distributed tests server:
+Make sure your local setup is built, along with the distributed tests:
 
 ```bash
-./deploy/dist-test/dev_server.sh
-```
+# Enter CLI container
+./bin/cli.sh faasm
 
-Then you need to make sure all the functions are up to date:
+# Build local environment
+inv dev.tools
 
-```bash
-./deploy/dist-test/upload.sh
+# Build dist tests
+inv dev.cc dist_tests dev.cc dist_test_server
 ```
 
-Build and run the tests:
+Outside the container, start the distributed tests server, then upload all the
+functions:
 
 ```bash
-# Enter the Faasm CLI container
-./bin/cli.sh faasm
-
-# Build and run the tests
-inv dev.cc dist_tests
-dist_tests
-```
+# Start server
+./deploy/dist-test/dev_server.sh
 
-To rebuild the server (inside the Faasm CLI container):
+# Upload everything
+./deploy/dist-test/upload.sh
 
-```bash
-inv dev.cc dist_test_server
+# Run the tests
+./deploy/dist-test/run.sh
 ```
 
-Then from outside the container, you can restart the server:
+You can then rebuild and rerun from inside the container:
 
 ```bash
-./deploy/dist-test/dev_server.sh restart
-```
+# Build tests and the server
+inv dev.cc dist_tests dev.cc dist_test_server
 
-### Replicating CI
+# Run
+dist_tests
+```
 
-To run the distributed tests as if in CI:
+If changing the server, you need to restart from outside the container:
 
 ```bash
-# Clear up
-docker-compose down
-
-# Set up
-./deploy/dist-test/build.sh
-./deploy/dist-test/upload.sh
-
-# Run once through
-./deploy/dist-test/run.sh
+./deploy/dist-test/dev_server.sh restart
 ```
 
 ## Building outside of the container

diff --git a/faabric b/faabric
diff --git a/include/faaslet/Faaslet.h b/include/faaslet/Faaslet.h
@@ -34,6 +34,8 @@ class Faaslet final : public faabric::scheduler::Executor
 
     void setMemorySize(size_t newSize) override;
 
+    size_t getMaxMemorySize() override;
   private:
     std::string localResetSnapshotKey;
 

diff --git a/src/faaslet/Faaslet.cpp b/src/faaslet/Faaslet.cpp
@@ -140,6 +140,11 @@ void Faaslet::setMemorySize(size_t newSize)
     module->setMemorySize(newSize);
 }
 
+size_t Faaslet::getMaxMemorySize()
+{
+    return MAX_WASM_MEM;
+}
+
 std::string Faaslet::getLocalResetSnapshotKey()
 {
     return localResetSnapshotKey;

diff --git a/src/wavm/faasm.cpp b/src/wavm/faasm.cpp
@@ -654,16 +654,40 @@ WAVM_DEFINE_INTRINSIC_FUNCTION(env,
                                I32 reduceOp,
                                int currentBatch)
 {
+    // Here we have two scenarios, the second of which differs in behaviour when
+    // we're in single host mode:
+    //
+    // 1. We're being notified of a reduction variable in an *upcoming* batch of
+    // threads. This means the snapshot doesn't yet exist, and we don't know
+    // whether it will be in single host mode or not. Therefore, we always keep
+    // the merge region in a list.
+    //
+    // 2. We're being notified of a reduction variable in the *current* batch of
+    // threads. If this is in single host mode, we can ignore it, as there won't
+    // be any snapshotting at all, otherwise we add it to the snapshot.
+
+    bool isCurrentBatch = currentBatch == 1;
+    bool isSingleHost = ExecutorContext::get()->getBatchRequest()->singlehost();
+
+    // Here we can ignore if we're in the current batch, and it's in single host
+    // mode.
+    if (isCurrentBatch && isSingleHost) {
+        SPDLOG_DEBUG("S - sm_reduce - {} {} {} {} (ignored, single host)",
+                     varPtr,
+                     varType,
+                     reduceOp,
+                     currentBatch);
+        return;
+    }
+
     SPDLOG_DEBUG(
       "S - sm_reduce - {} {} {} {}", varPtr, varType, reduceOp, currentBatch);
 
     auto dataType = extractSnapshotDataType(varType);
     faabric::util::SnapshotMergeOperation mergeOp =
       extractSnapshotMergeOp(reduceOp);
 
-    bool isCurrentBatch = currentBatch == 1;
     faabric::Message* msg = &ExecutorContext::get()->getMsg();
-
     SPDLOG_DEBUG("Registering reduction variable {}-{} for {} {}",
                  varPtr,
                  varPtr + dataType.first,

diff --git a/src/wavm/mpi.cpp b/src/wavm/mpi.cpp
@@ -18,10 +18,10 @@ using namespace faabric::scheduler;
 using namespace WAVM;
 
 #define MPI_FUNC(str)                                                          \
-    SPDLOG_DEBUG("MPI-{} {}", executingContext.getRank(), str);
+    SPDLOG_TRACE("MPI-{} {}", executingContext.getRank(), str);
 
 #define MPI_FUNC_ARGS(formatStr, ...)                                          \
-    SPDLOG_DEBUG("MPI-{} " formatStr, executingContext.getRank(), __VA_ARGS__);
+    SPDLOG_TRACE("MPI-{} " formatStr, executingContext.getRank(), __VA_ARGS__);
 
 namespace wasm {
 static thread_local faabric::scheduler::MpiContext executingContext;
@@ -413,7 +413,7 @@ WAVM_DEFINE_INTRINSIC_FUNCTION(env,
                                I32 datatype,
                                I32 countPtr)
 {
-    SPDLOG_DEBUG("S - MPI_Get_count {} {} {}", statusPtr, datatype, countPtr);
+    SPDLOG_TRACE("S - MPI_Get_count {} {} {}", statusPtr, datatype, countPtr);
 
     MPI_Status* status =
       &Runtime::memoryRef<MPI_Status>(ctx->memory, statusPtr);

diff --git a/tests/dist/state/test_state.cpp b/tests/dist/state/test_state.cpp
@@ -0,0 +1,41 @@
+#include <catch2/catch.hpp>
+
+#include "fixtures.h"
+
+#include <faabric/scheduler/Scheduler.h>
+#include <faabric/util/logging.h>
+#include <faabric/util/string_tools.h>
+
+namespace tests {
+
+TEST_CASE_METHOD(DistTestsFixture,
+                 "Test running distributed Pi estimate",
+                 "[scheduler]")
+{
+    int nWorkers = 20;
+
+    // Make sure only half can be executed on this host
+    int nLocalSlots = 10;
+    faabric::HostResources res;
+    res.set_slots(nLocalSlots);
+    sch.setThisHostResources(res);
+
+    // Set up the message
+    std::shared_ptr<faabric::BatchExecuteRequest> req =
+      faabric::util::batchExecFactory("demo", "pi", 1);
+    faabric::Message& msg = req->mutable_messages()->at(0);
+    msg.set_inputdata(std::to_string(nWorkers));
+
+    // Call the functions
+    sch.callFunctions(req);
+
+    // Check it's successful
+    faabric::Message result =
+      sch.getFunctionResult(msg.id(), functionCallTimeout);
+    REQUIRE(result.returnvalue() == 0);
+
+    // Get the estimate (check one dp)
+    std::string outputData = msg.outputdata();
+    REQUIRE(faabric::util::startsWith(outputData, "Pi estimate: 3.1"));
+}
+}
diff --git a/tests/dist/threads/test_openmp.cpp b/tests/dist/threads/test_openmp.cpp
@@ -3,12 +3,14 @@
 #include "fixtures.h"
 
 #include <faabric/scheduler/Scheduler.h>
+#include <faabric/util/string_tools.h>
+
+#define PI_FUNCTION "pi_faasm"
 
 namespace tests {
-// 10/02/22 Broken by latest Faabric
 TEST_CASE_METHOD(DistTestsFixture,
                  "Test OpenMP across hosts",
-                 "[.][threads][openmp]")
+                 "[threads][openmp]")
 {
     conf.overrideCpuCount = 6;
 
@@ -26,6 +28,8 @@ TEST_CASE_METHOD(DistTestsFixture,
 
     SECTION("Repeated reduce") { function = "repeated_reduce"; }
 
+    SECTION("Pi estimation") { function = PI_FUNCTION; }
+
     // Set up the message
     std::shared_ptr<faabric::BatchExecuteRequest> req =
       faabric::util::batchExecFactory("omp", function, 1);
@@ -48,5 +52,11 @@ TEST_CASE_METHOD(DistTestsFixture,
     // Check other host is registered
     std::set<std::string> expectedRegisteredHosts = { getDistTestWorkerIp() };
     REQUIRE(sch.getFunctionRegisteredHosts(msg) == expectedRegisteredHosts);
+
+    // Check specific results
+    if (function == PI_FUNCTION) {
+        REQUIRE(
+          faabric::util::startsWith(result.outputdata(), "Pi estimate: 3.1"));
+    }
 }
 }
diff --git a/tests/test/faaslet/test_mpi.cpp b/tests/test/faaslet/test_mpi.cpp
@@ -8,6 +8,7 @@
 #include <faabric/util/environment.h>
 #include <faabric/util/func.h>
 #include <faabric/util/logging.h>
+#include <faabric/util/string_tools.h>
 
 namespace tests {
 
@@ -16,12 +17,12 @@ class MPIFuncTestFixture
   , public MpiBaseTestFixture
 {
   public:
-    void checkMpiFunc(const char* funcName)
+    faabric::Message checkMpiFunc(const char* funcName)
     {
         // Note: we don't `set_mpiworldsize` here, so all tests run with the
         // default MPI world size (5). Some tests will fail if we change this.
         faabric::Message msg = faabric::util::messageFactory("mpi", funcName);
-        execFuncWithPool(msg, true, 10000);
+        faabric::Message result = execFuncWithPool(msg, true, 10000);
 
         // Check all other functions were successful
         auto& sch = faabric::scheduler::getScheduler();
@@ -32,13 +33,14 @@ class MPIFuncTestFixture
                 continue;
             }
 
-            const faabric::Message& result =
-              sch.getFunctionResult(messageId, 1);
+            faabric::Message result = sch.getFunctionResult(messageId, 1);
 
             if (result.returnvalue() != 0) {
                 FAIL(fmt::format("Message ID {} failed", messageId));
             }
         }
+
+        return result;
     }
 };
 
@@ -132,4 +134,11 @@ TEST_CASE_METHOD(MPIFuncTestFixture, "Test MPI async", "[mpi]")
 {
     checkMpiFunc("mpi_isendrecv");
 }
+
+TEST_CASE_METHOD(MPIFuncTestFixture, "Test MPI Pi", "[mpi]")
+{
+    faabric::Message result = checkMpiFunc("mpi_pi");
+    std::string output = result.outputdata();
+    REQUIRE(faabric::util::startsWith(output, "Pi estimate: 3.1"));
+}
 }
diff --git a/tests/test/faaslet/test_state.cpp b/tests/test/faaslet/test_state.cpp
@@ -6,6 +6,7 @@
 #include <faabric/runner/FaabricMain.h>
 #include <faabric/util/bytes.h>
 #include <faabric/util/environment.h>
+#include <faabric/util/string_tools.h>
 #include <faaslet/Faaslet.h>
 
 using namespace faaslet;
@@ -122,4 +123,13 @@ TEST_CASE_METHOD(StateFuncTestFixture, "Test appended state", "[state]")
     faabric::Message& call = req->mutable_messages()->at(0);
     execFuncWithPool(call);
 }
+
+TEST_CASE_METHOD(StateFuncTestFixture, "Test Pi estimate", "[state]")
+{
+    auto req = setUpContext("demo", "pi");
+    faabric::Message& call = req->mutable_messages()->at(0);
+    faabric::Message result = execFuncWithPool(call);
+    std::string output = result.outputdata();
+    REQUIRE(faabric::util::startsWith(output, "Pi estimate: 3.1"));
+}
 }
diff --git a/tests/test/wasm/test_openmp.cpp b/tests/test/wasm/test_openmp.cpp
@@ -10,6 +10,7 @@
 #include <faabric/util/config.h>
 #include <faabric/util/environment.h>
 #include <faabric/util/func.h>
+#include <faabric/util/string_tools.h>
 
 // Longer timeout to allow longer-running functions to finish even when doing
 // trace-level logging
@@ -27,16 +28,19 @@ class OpenMPTestFixture
 
     ~OpenMPTestFixture() {}
 
-    void doOmpTestLocal(const std::string& function)
+    std::string doOmpTestLocal(const std::string& function)
     {
         faabric::Message msg = faabric::util::messageFactory("omp", function);
-        execFuncWithPool(msg, false, OMP_TEST_TIMEOUT_MS);
+        faabric::Message result =
+          execFuncWithPool(msg, false, OMP_TEST_TIMEOUT_MS);
+
+        return result.outputdata();
     }
 };
 
 TEST_CASE_METHOD(OpenMPTestFixture,
                  "Test OpenMP static for scheduling",
-                 "[.][wasm][openmp]")
+                 "[wasm][openmp]")
 {
     doOmpTestLocal("for_static_schedule");
 }
@@ -83,7 +87,7 @@ TEST_CASE_METHOD(OpenMPTestFixture, "Test OpenMP reduction", "[wasm][openmp]")
 
 TEST_CASE_METHOD(OpenMPTestFixture,
                  "Test a mix of OpenMP constructs",
-                 "[.][wasm][openmp]")
+                 "[wasm][openmp]")
 {
     doOmpTestLocal("reduction_integral");
 }
@@ -110,10 +114,13 @@ TEST_CASE_METHOD(OpenMPTestFixture,
 }
 
 TEST_CASE_METHOD(OpenMPTestFixture,
-                 "Test OpenMP Pi calculation",
+                 "Test OpenMP Pi calculation using libfaasm",
                  "[wasm][openmp]")
 {
-    doOmpTestLocal("pi_faasm");
+    std::string output = doOmpTestLocal("pi_faasm");
+
+    // Just check Pi to one dp
+    REQUIRE(faabric::util::startsWith(output, "Pi estimate: 3.1"));
 }
 
 TEST_CASE_METHOD(OpenMPTestFixture,
@@ -149,7 +156,7 @@ TEST_CASE_METHOD(OpenMPTestFixture,
     doOmpTestLocal("repeated_reduce");
 }
 
-TEST_CASE_METHOD(OpenMPTestFixture, "Test OpenMP atomic", "[.][wasm][openmp]")
+TEST_CASE_METHOD(OpenMPTestFixture, "Test OpenMP atomic", "[wasm][openmp]")
 {
     doOmpTestLocal("simple_atomic");
 }
+4 −4		func/omp/repeated_reduce.cpp
+0 −3		func/omp/simple_atomic.cpp
+2 −0		libfaasm/faasm/shared_mem.h
+6 −6		include/faabric/scheduler/Scheduler.h
+1 −1		include/faabric/snapshot/SnapshotApi.h
+14 −12		include/faabric/snapshot/SnapshotClient.h
+5 −2		include/faabric/snapshot/SnapshotServer.h
+30 −3		include/faabric/util/snapshot.h
+3 −2		src/flat/faabric.fbs
+11 −25		src/scheduler/Executor.cpp
+24 −30		src/scheduler/Scheduler.cpp
+44 −47		src/snapshot/SnapshotClient.cpp
+56 −41		src/snapshot/SnapshotServer.cpp
+14 −3		src/util/snapshot.cpp
+23 −7		tests/test/scheduler/test_executor.cpp
+23 −6		tests/test/scheduler/test_scheduler.cpp
+38 −50		tests/test/snapshot/test_snapshot_client_server.cpp
+29 −0		tests/test/util/test_snapshot.cpp