Add sanity-check for BatchExecuteRequests (#331)

* utils: move batch utils to different source file

* util: fix compilation errors after moving batchExec to a different header file

* tests: add regression tests

* dist-tests: fix compilation

include/faabric/scheduler/Scheduler.h

@@ -12,7 +12,6 @@
 #include <faabric/util/clock.h>
 #include <faabric/util/config.h>
 #include <faabric/util/dirty.h>
-#include <faabric/util/func.h>
 #include <faabric/util/memory.h>
 #include <faabric/util/queue.h>
 #include <faabric/util/scheduling.h>

include/faabric/util/batch.h

@@ -0,0 +1,14 @@
+#pragma once
+
+#include <faabric/proto/faabric.pb.h>
+
+namespace faabric::util {
+std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory();
+
+std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
+  const std::string& user,
+  const std::string& function,
+  int count = 1);
+
+bool isBatchExecRequestValid(std::shared_ptr<faabric::BatchExecuteRequest> ber);
+}

include/faabric/util/func.h

@@ -31,13 +31,6 @@ std::shared_ptr<faabric::Message> messageFactoryShared(
 faabric::Message messageFactory(const std::string& user,
                                 const std::string& function);
 
-std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory();
-
-std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
-  const std::string& user,
-  const std::string& function,
-  int count = 1);
-
 std::string resultKeyFromMessageId(unsigned int mid);
 
 std::string statusKeyFromMessageId(unsigned int mid);

src/endpoint/FaabricEndpointHandler.cpp

@@ -2,6 +2,7 @@
 
 #include <faabric/redis/Redis.h>
 #include <faabric/scheduler/Scheduler.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/json.h>
 #include <faabric/util/logging.h>
 #include <faabric/util/timing.h>

src/mpi/MpiWorld.cpp

@@ -3,8 +3,8 @@
 #include <faabric/scheduler/Scheduler.h>
 #include <faabric/transport/macros.h>
 #include <faabric/util/ExecGraph.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/environment.h>
-#include <faabric/util/func.h>
 #include <faabric/util/gids.h>
 #include <faabric/util/macros.h>
 #include <faabric/util/scheduling.h>

src/scheduler/Scheduler.cpp

@@ -9,9 +9,9 @@
 #include <faabric/snapshot/SnapshotClient.h>
 #include <faabric/snapshot/SnapshotRegistry.h>
 #include <faabric/transport/PointToPointBroker.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/concurrent_map.h>
 #include <faabric/util/environment.h>
-#include <faabric/util/func.h>
 #include <faabric/util/locks.h>
 #include <faabric/util/logging.h>
 #include <faabric/util/memory.h>

src/util/CMakeLists.txt

@@ -3,6 +3,7 @@ faabric_lib(util
     ExecGraph.cpp
     PeriodicBackgroundThread.cpp
     barrier.cpp
+    batch.cpp
     bytes.cpp
     config.cpp
     clock.cpp

src/util/batch.cpp

@@ -0,0 +1,55 @@
+#include <faabric/util/batch.h>
+#include <faabric/util/func.h>
+#include <faabric/util/gids.h>
+
+namespace faabric::util {
+std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory()
+{
+    auto req = std::make_shared<faabric::BatchExecuteRequest>();
+    req->set_id(generateGid());
+    return req;
+}
+
+std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
+  const std::string& user,
+  const std::string& function,
+  int count)
+{
+    auto req = batchExecFactory();
+
+    // Force the messages to have the same app ID than the BER
+    int appId = req->id();
+    for (int i = 0; i < count; i++) {
+        *req->add_messages() = messageFactory(user, function);
+        req->mutable_messages()->at(i).set_appid(appId);
+    }
+
+    return req;
+}
+
+bool isBatchExecRequestValid(std::shared_ptr<faabric::BatchExecuteRequest> ber)
+{
+    if (ber == nullptr) {
+        return false;
+    }
+
+    // An empty BER (thus invalid) will have 0 messages and an id of 0
+    if (ber->messages_size() <= 0 && ber->id() == 0) {
+        return false;
+    }
+
+    std::string user = ber->messages(0).user();
+    std::string func = ber->messages(0).function();
+    int appId = ber->messages(0).appid();
+
+    for (int i = 1; i < ber->messages_size(); i++) {
+        auto msg = ber->messages(i);
+        if (msg.user() != user || msg.function() != func ||
+            msg.appid() != appId) {
+            return false;
+        }
+    }
+
+    return true;
+}
+}

src/util/func.cpp

@@ -47,30 +47,6 @@ std::string buildAsyncResponse(const faabric::Message& msg)
     return std::to_string(msg.id());
 }
 
-std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory()
-{
-    auto req = std::make_shared<faabric::BatchExecuteRequest>();
-    req->set_id(faabric::util::generateGid());
-    return req;
-}
-
-std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
-  const std::string& user,
-  const std::string& function,
-  int count)
-{
-    auto req = batchExecFactory();
-
-    // Force the messages to have the same app ID
-    uint32_t appId = faabric::util::generateGid();
-    for (int i = 0; i < count; i++) {
-        *req->add_messages() = messageFactory(user, function);
-        req->mutable_messages()->at(i).set_appid(appId);
-    }
-
-    return req;
-}
-
 std::shared_ptr<faabric::Message> messageFactoryShared(
   const std::string& user,
   const std::string& function)

tests/dist/mpi/mpi_native.cpp

@@ -7,6 +7,7 @@
 #include <faabric/scheduler/ExecutorContext.h>
 #include <faabric/scheduler/Scheduler.h>
 #include <faabric/util/ExecGraph.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/compare.h>
 #include <faabric/util/config.h>
 #include <faabric/util/logging.h>

tests/dist/scheduler/functions.cpp

@@ -9,9 +9,9 @@
 #include <faabric/proto/faabric.pb.h>
 #include <faabric/scheduler/Scheduler.h>
 #include <faabric/transport/PointToPointBroker.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/bytes.h>
 #include <faabric/util/config.h>
-#include <faabric/util/func.h>
 #include <faabric/util/gids.h>
 #include <faabric/util/logging.h>
 #include <faabric/util/memory.h>

tests/dist/transport/functions.cpp

@@ -7,8 +7,8 @@
 #include <faabric/proto/faabric.pb.h>
 #include <faabric/scheduler/Scheduler.h>
 #include <faabric/transport/PointToPointBroker.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/bytes.h>
-#include <faabric/util/func.h>
 #include <faabric/util/gids.h>
 #include <faabric/util/scheduling.h>
 #include <faabric/util/string_tools.h>

tests/test/util/test_batch.cpp

@@ -0,0 +1,76 @@
+#include <catch2/catch.hpp>
+
+#include <faabric/util/batch.h>
+
+using namespace faabric::util;
+
+namespace tests {
+TEST_CASE("Test batch exec factory", "[util]")
+{
+    int nMessages = 4;
+    std::shared_ptr<faabric::BatchExecuteRequest> req =
+      batchExecFactory("demo", "echo", nMessages);
+
+    REQUIRE(req->messages().size() == nMessages);
+
+    REQUIRE(req->id() > 0);
+
+    // Expect all messages to have the same app ID by default
+    int appId = req->messages().at(0).appid();
+    REQUIRE(appId > 0);
+
+    for (const auto& m : req->messages()) {
+        REQUIRE(m.appid() == appId);
+        REQUIRE(m.user() == "demo");
+        REQUIRE(m.function() == "echo");
+    }
+}
+
+TEST_CASE("Test batch. exec request sanity checks")
+{
+    int nMessages = 4;
+    std::shared_ptr<faabric::BatchExecuteRequest> ber =
+      batchExecFactory("demo", "echo", nMessages);
+    bool isBerValid;
+
+    // A null BER is invalid
+    SECTION("Null BER")
+    {
+        isBerValid = false;
+        ber = nullptr;
+    }
+
+    // An empty BER is invalid
+    SECTION("Empty BER")
+    {
+        isBerValid = false;
+        ber = std::make_shared<faabric::BatchExecuteRequest>();
+    }
+
+    // An appId mismatch between the messages deems a BER invalid
+    SECTION("App ID mismatch")
+    {
+        isBerValid = false;
+        ber->mutable_messages(1)->set_appid(1337);
+    }
+
+    // A user mismatch between the messages deems a BER invalid
+    SECTION("User mismatch")
+    {
+        isBerValid = false;
+        ber->mutable_messages(1)->set_user("foo");
+    }
+
+    // A function mismatch between the messages deems a BER invalid
+    SECTION("Function mismatch")
+    {
+        isBerValid = false;
+        ber->mutable_messages(1)->set_function("foo");
+    }
+
+    // BERs constructed with the default factory are valid
+    SECTION("Valid BER") { isBerValid = true; }
+
+    REQUIRE(isBerValid == isBatchExecRequestValid(ber));
+}
+}

tests/test/util/test_func.cpp

@@ -30,27 +30,6 @@ TEST_CASE("Test message factory shared", "[util]")
     REQUIRE(!msg->resultkey().empty());
 }
 
-TEST_CASE("Test batch exec factory", "[util]")
-{
-    int nMessages = 4;
-    std::shared_ptr<faabric::BatchExecuteRequest> req =
-      faabric::util::batchExecFactory("demo", "echo", nMessages);
-
-    REQUIRE(req->messages().size() == nMessages);
-
-    REQUIRE(req->id() > 0);
-
-    // Expect all messages to have the same app ID by default
-    int appId = req->messages().at(0).appid();
-    REQUIRE(appId > 0);
-
-    for (const auto& m : req->messages()) {
-        REQUIRE(m.appid() == appId);
-        REQUIRE(m.user() == "demo");
-        REQUIRE(m.function() == "echo");
-    }
-}
-
 TEST_CASE("Test adding ids to message", "[util]")
 {
     faabric::Message msgA;

tests/utils/fixtures.h

@@ -22,9 +22,9 @@
 #include <faabric/transport/PointToPointBroker.h>
 #include <faabric/transport/PointToPointClient.h>
 #include <faabric/transport/PointToPointServer.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/dirty.h>
 #include <faabric/util/environment.h>
-#include <faabric/util/func.h>
 #include <faabric/util/json.h>
 #include <faabric/util/latch.h>
 #include <faabric/util/memory.h>