planner: implement compaction planner policy

include/faabric/batch-scheduler/BatchScheduler.h

@@ -120,4 +120,6 @@ class BatchScheduler
 std::shared_ptr<BatchScheduler> getBatchScheduler();
 
 void resetBatchScheduler();
+
+void resetBatchScheduler(const std::string& newMode);
 }

include/faabric/batch-scheduler/CompactScheduler.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <faabric/batch-scheduler/BatchScheduler.h>
+#include <faabric/batch-scheduler/SchedulingDecision.h>
+#include <faabric/util/batch.h>
+
+namespace faabric::batch_scheduler {
+
+// This batch scheduler behaves in the same way than BinPack for NEW and
+// SCALE_CHANGE requests, but for DIST_CHANGE requests it tries to compact
+// to the fewest number of VMs.
+class CompactScheduler final : public BatchScheduler
+{
+  public:
+    std::shared_ptr<SchedulingDecision> makeSchedulingDecision(
+      HostMap& hostMap,
+      const InFlightReqs& inFlightReqs,
+      std::shared_ptr<faabric::BatchExecuteRequest> req) override;
+
+  private:
+    bool isFirstDecisionBetter(
+      std::shared_ptr<SchedulingDecision> decisionA,
+      std::shared_ptr<SchedulingDecision> decisionB) override;
+
+    std::vector<Host> getSortedHosts(
+      HostMap& hostMap,
+      const InFlightReqs& inFlightReqs,
+      std::shared_ptr<faabric::BatchExecuteRequest> req,
+      const DecisionType& decisionType) override;
+};
+}

include/faabric/planner/Planner.h

@@ -33,6 +33,8 @@ class Planner
 
     void printConfig() const;
 
+    void setPolicy(const std::string& newPolicy);
+
     // ----------
     // Util public API
     // ----------

src/batch-scheduler/BatchScheduler.cpp

@@ -1,5 +1,6 @@
 #include <faabric/batch-scheduler/BatchScheduler.h>
 #include <faabric/batch-scheduler/BinPackScheduler.h>
+#include <faabric/batch-scheduler/CompactScheduler.h>
 #include <faabric/util/config.h>
 #include <faabric/util/logging.h>
 
@@ -20,7 +21,9 @@ std::shared_ptr<BatchScheduler> getBatchScheduler()
 
     if (mode == "bin-pack") {
         batchScheduler = std::make_shared<BinPackScheduler>();
-    } else {
+    } else if (mode == "compact") {
+        batchScheduler = std::make_shared<CompactScheduler>();
+    }else {
         SPDLOG_ERROR("Unrecognised batch scheduler mode: {}", mode);
         throw std::runtime_error("Unrecognised batch scheduler mode");
     }
@@ -33,6 +36,15 @@ void resetBatchScheduler()
     batchScheduler = nullptr;
 }
 
+void resetBatchScheduler(const std::string& newMode)
+{
+    resetBatchScheduler();
+
+    faabric::util::getSystemConfig().batchSchedulerMode = newMode;
+
+    getBatchScheduler();
+}
+
 DecisionType BatchScheduler::getDecisionType(
   const InFlightReqs& inFlightReqs,
   std::shared_ptr<faabric::BatchExecuteRequest> req)

src/planner/Planner.cpp

@@ -132,6 +132,14 @@ void Planner::printConfig() const
     SPDLOG_INFO("HTTP_SERVER_THREADS        {}", config.numthreadshttpserver());
 }
 
+void Planner::setPolicy(const std::string& newPolicy)
+{
+    // Acquire lock to prevent any changes in state whilst we change the policy
+    faabric::util::FullLock lock(plannerMx);
+
+    faabric::batch_scheduler::resetBatchScheduler(newPolicy);
+}
+
 bool Planner::reset()
 {
     SPDLOG_INFO("Resetting planner");

src/planner/PlannerEndpointHandler.cpp

@@ -267,7 +267,6 @@ void PlannerEndpointHandler::onRequest(
             return ctx.sendFunction(std::move(response));
         }
         case faabric::planner::HttpMessage_Type_PRELOAD_SCHEDULING_DECISION: {
-            // foo bar
             // in: BatchExecuteRequest
             // out: none
             SPDLOG_DEBUG(
@@ -308,6 +307,26 @@ void PlannerEndpointHandler::onRequest(
 
             return ctx.sendFunction(std::move(response));
         }
+        case faabric::planner::HttpMessage_Type_SET_POLICY: {
+            SPDLOG_DEBUG(
+              "Planner received SET_POLICY request");
+
+            std::string newPolicy = msg.payloadjson();
+
+            try {
+                faabric::planner::getPlanner().setPolicy(newPolicy);
+            } catch (std::exception& e) {
+                response.result(beast::http::status::bad_request);
+                response.body() = std::string("Unrecognised policy name: " + newPolicy);
+                return ctx.sendFunction(std::move(response));
+            }
+
+            // Prepare the response
+            response.result(beast::http::status::ok);
+            response.body() = std::string("Policy set correctly");
+
+            return ctx.sendFunction(std::move(response));
+        }
         default: {
             SPDLOG_ERROR("Unrecognised message type {}", msg.type());
             response.result(beast::http::status::bad_request);

src/planner/planner.proto

@@ -45,6 +45,7 @@ message HttpMessage {
         EXECUTE_BATCH = 10;
         EXECUTE_BATCH_STATUS = 11;
         PRELOAD_SCHEDULING_DECISION = 12;
+        SET_POLICY = 13;
     }
 
     Type type = 1 [json_name = "http_type"];
@@ -55,6 +56,7 @@ message HttpMessage {
     // - EXECUTE_BATCH_STATUS: where the body is a BER too
     // - PRELOAD_SCHEDULING_DECISION: where the body is a
     //     faabric::BatchExecuteRequest
+    // - SET_POLICY: where the body is a string with the new planner policy
     string payloadJson = 2 [json_name = "payload"];
 }