Change flush to execute once per host

include/faabric/scheduler/ExecutorFactory.h

@@ -10,6 +10,8 @@ class ExecutorFactory
     virtual ~ExecutorFactory(){};
 
     virtual std::shared_ptr<Executor> createExecutor(faabric::Message& msg) = 0;
+
+    virtual void flushHost();
 };
 
 void setExecutorFactory(std::shared_ptr<ExecutorFactory> fac);

include/faabric/scheduler/Scheduler.h

@@ -36,8 +36,6 @@ class Executor
 
     void finish();
 
-    virtual void flush();
-
     virtual void reset(faabric::Message& msg);
 
     virtual int32_t executeTask(

src/scheduler/Executor.cpp

@@ -311,8 +311,6 @@ int32_t Executor::executeTask(int threadPoolIdx,
 
 void Executor::postFinish() {}
 
-void Executor::flush() {}
-
 void Executor::reset(faabric::Message& msg) {}
 
 faabric::util::SnapshotData Executor::snapshot()

src/scheduler/ExecutorFactory.cpp

@@ -1,9 +1,15 @@
 #include <faabric/scheduler/ExecutorFactory.h>
+#include <faabric/util/logging.h>
 
 namespace faabric::scheduler {
 
 static std::shared_ptr<ExecutorFactory> _factory;
 
+void ExecutorFactory::flushHost()
+{
+    SPDLOG_WARN("Using default flush method");
+}
+
 void setExecutorFactory(std::shared_ptr<ExecutorFactory> fac)
 {
     _factory = fac;

src/scheduler/FunctionCallServer.cpp

@@ -66,9 +66,6 @@ void FunctionCallServer::recvFlush(faabric::transport::Message& body)
 
     // Clear the scheduler
     scheduler.flushLocally();
-
-    // Reset the scheduler
-    scheduler.reset();
 }
 
 void FunctionCallServer::recvExecuteFunctions(faabric::transport::Message& body)
@@ -85,7 +82,7 @@ void FunctionCallServer::recvUnregister(faabric::transport::Message& body)
     PARSE_MSG(faabric::UnregisterRequest, body.data(), body.size())
 
     std::string funcStr = faabric::util::funcToString(msg.function(), false);
-    SPDLOG_INFO("Unregistering host {} for {}", msg.host(), funcStr);
+    SPDLOG_DEBUG("Unregistering host {} for {}", msg.host(), funcStr);
 
     // Remove the host from the warm set
     scheduler.removeRegisteredHost(msg.host(), msg.function());

src/scheduler/Scheduler.cpp

@@ -677,15 +677,11 @@ void Scheduler::flushLocally()
     SPDLOG_INFO("Flushing host {}",
                 faabric::util::getSystemConfig().endpointHost);
 
-    // Call flush on all executors
-    for (auto& p : executors) {
-        for (auto& e : p.second) {
-            e->flush();
-        }
-    }
-
     // Reset this scheduler
     reset();
+
+    // Flush the host
+    getExecutorFactory()->flushHost();
 }
 
 void Scheduler::setFunctionResult(faabric::Message& msg)

tests/test/scheduler/test_function_client_server.cpp

@@ -19,7 +19,7 @@
 
 #define TEST_TIMEOUT_MS 500
 
-using namespace scheduler;
+using namespace faabric::scheduler;
 
 namespace tests {
 class ClientServerFixture
@@ -30,6 +30,7 @@ class ClientServerFixture
   protected:
     FunctionCallServer server;
     FunctionCallClient cli;
+    std::shared_ptr<DummyExecutorFactory> executorFactory;
 
   public:
     ClientServerFixture()
@@ -39,15 +40,15 @@ class ClientServerFixture
         usleep(1000 * TEST_TIMEOUT_MS);
 
         // Set up executor
-        std::shared_ptr<faabric::scheduler::ExecutorFactory> fac =
-          std::make_shared<faabric::scheduler::DummyExecutorFactory>();
-        faabric::scheduler::setExecutorFactory(fac);
+        executorFactory = std::make_shared<DummyExecutorFactory>();
+        setExecutorFactory(executorFactory);
     }
 
     ~ClientServerFixture()
     {
         cli.close();
         server.stop();
+        executorFactory->reset();
     }
 };
 
@@ -126,6 +127,9 @@ TEST_CASE_METHOD(ClientServerFixture,
                  "Test sending flush message",
                  "[scheduler]")
 {
+    // Check no flushes to begin with
+    REQUIRE(executorFactory->getFlushCount() == 0);
+
     // Set up some state
     faabric::state::State& state = faabric::state::getGlobalState();
     state.getKV("demo", "blah", 10);
@@ -156,6 +160,10 @@ TEST_CASE_METHOD(ClientServerFixture,
 
     // Check state has been cleared
     REQUIRE(state.getKVCount() == 0);
+
+    // Check the flush hook has been called
+    int flushCount = executorFactory->getFlushCount();
+    REQUIRE(flushCount == 1);
 }
 
 TEST_CASE_METHOD(ClientServerFixture,

tests/utils/DummyExecutorFactory.cpp

@@ -1,11 +1,28 @@
 #include "DummyExecutorFactory.h"
 #include "DummyExecutor.h"
 
+#include <faabric/util/logging.h>
+
 namespace faabric::scheduler {
 
 std::shared_ptr<Executor> DummyExecutorFactory::createExecutor(
   faabric::Message& msg)
 {
     return std::make_shared<DummyExecutor>(msg);
 }
+
+int DummyExecutorFactory::getFlushCount()
+{
+    return flushCount;
+}
+
+void DummyExecutorFactory::flushHost()
+{
+    flushCount++;
+}
+
+void DummyExecutorFactory::reset()
+{
+    flushCount = 0;
+}
 }

tests/utils/DummyExecutorFactory.h

@@ -6,7 +6,17 @@ namespace faabric::scheduler {
 
 class DummyExecutorFactory : public ExecutorFactory
 {
+  public:
+    void reset();
+
+    int getFlushCount();
+
   protected:
     std::shared_ptr<Executor> createExecutor(faabric::Message& msg) override;
+
+    void flushHost() override;
+
+  private:
+    int flushCount = 0;
 };
 }