Route callFunctions through the planner (#345)

* skeleton for the planner work

* scheduler: fix compilation issues

* tests: fix compilation issues

* tests: planner, proto, redis, and state tests working

* tests: util tests running

* tests: some mpi tests working

* tests: mpi tests working

* planner: update endpoint to use new callbatch method

* tests: towards migration test working

* debugging migration tests

* tests: migration tests in the scheduler passing

* tests: scheduler tests passing

* nits: run clang-format

* git: bump minor version

* dist-tests: all mpi tests (minus migration) working

* dist-tests: mpi function migration tests running

* dist-tests: all mpi dist tests working

* mpi: remove obsolete migration check period

* dist-tests: dist tests passing commenting out remote threads

* nits: run clang format

* tests: fix failing json test

* tests: fix failing mpi test

* nits: run clang format

* planner: add client/server method to get scheduling decision and add tests

* tests: remove unnecessary mpi sleep between tests

* threads: remove register/deregister and use a common result management via planner (all tests passing locally without sanitisers)

* tests: fix typo in tests

* batch: allow different messages to have different function names

* tests: set enough resources for executor tests

* nit: run clang format

* tasks: overwrite cpu count in tests

* tasks: override cpu count in tests

* cleaup + fix env. variable setting

* set environment variable to a string

* tests: fix failing tests

* tests: add enough slots for executor reaping test

* scheduler: remove unnecessary accounting structures

* mpi: clean-up todos and comments

* tests: wait for mpi messages to be executed when counting the number of messages sent to avoid race conditions

* tests: more mpi message waiting

* mpi: remove comment

* planner: clean-up

* threads: re-introduce remote threads (most tests and dist-tests passing, still need to uncomment more)

* dist-tests: re-introduce all remote threads tests

* tests: fix executor tests failing after planner resource managing update

* tests: fix scheduler tests after planner changes

* tests: fix snapshot tests after changes

* tests: fix util tests

* migration: don't release slots when setting a migration message result, as we already release them when making the decision

* dist-tests: fix multiple mpi world migration too

* dist-tests: fix race condition in mpi dist test

* executor: add test using executeThreads

* planner-cli: remove callFunctions signature with hints

* scheduler: clean-up includes in main scheduler file

* snapshot: clean-up and make clear that we currently don't delete snapshots

* self-review: cleanup

* more self-review

* planner: add log level to print state

* scheduler: use defined timeouts instead of hardcoded numbers

* tests: remove race in mpi test

* planner: add info logging around migration opportunities

* dist-tests: make migration test less flaky

Author: csegarragonz

.env

@@ -1,4 +1,4 @@
-FAABRIC_VERSION=0.6.1
-FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.6.1
+FAABRIC_VERSION=0.7.0
+FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.7.0
 COMPOSE_PROJECT_NAME=faabric-dev
 CONAN_CACHE_MOUNT_SOURCE=./conan-cache/

.github/workflows/tests.yml

@@ -23,7 +23,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -36,7 +36,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -50,7 +50,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -73,7 +73,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -113,7 +113,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -167,7 +167,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.6.1
+      image: faasm.azurecr.io/faabric:0.7.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}

VERSION

@@ -1 +1 @@
-0.6.1
+0.7.0

include/faabric/batch-scheduler/BatchScheduler.h

@@ -7,10 +7,11 @@
 
 #define DO_NOT_MIGRATE -98
 #define DO_NOT_MIGRATE_DECISION                                                \
-    SchedulingDecision(DO_NOT_MIGRATE, DO_NOT_MIGRATE)
+    faabric::batch_scheduler::SchedulingDecision(DO_NOT_MIGRATE, DO_NOT_MIGRATE)
 #define NOT_ENOUGH_SLOTS -99
 #define NOT_ENOUGH_SLOTS_DECISION                                              \
-    SchedulingDecision(NOT_ENOUGH_SLOTS, NOT_ENOUGH_SLOTS)
+    faabric::batch_scheduler::SchedulingDecision(NOT_ENOUGH_SLOTS,             \
+                                                 NOT_ENOUGH_SLOTS)
 
 namespace faabric::batch_scheduler {
 
@@ -70,7 +71,7 @@ class BatchScheduler
       std::shared_ptr<faabric::BatchExecuteRequest> req);
 
     virtual std::shared_ptr<SchedulingDecision> makeSchedulingDecision(
-      const HostMap& hostMap,
+      HostMap& hostMap,
       const InFlightReqs& inFlightReqs,
       std::shared_ptr<faabric::BatchExecuteRequest> req) = 0;
 
@@ -111,7 +112,7 @@ class BatchScheduler
       std::shared_ptr<SchedulingDecision> decisionB) = 0;
 
     virtual std::vector<Host> getSortedHosts(
-      const HostMap& hostMap,
+      HostMap& hostMap,
       const InFlightReqs& inFlightReqs,
       std::shared_ptr<faabric::BatchExecuteRequest> req,
       const DecisionType& decisionType) = 0;

include/faabric/batch-scheduler/BinPackScheduler.h

@@ -11,7 +11,7 @@ class BinPackScheduler final : public BatchScheduler
 {
   public:
     std::shared_ptr<SchedulingDecision> makeSchedulingDecision(
-      const HostMap& hostMap,
+      HostMap& hostMap,
       const InFlightReqs& inFlightReqs,
       std::shared_ptr<faabric::BatchExecuteRequest> req) override;
 
@@ -21,7 +21,7 @@ class BinPackScheduler final : public BatchScheduler
       std::shared_ptr<SchedulingDecision> decisionB) override;
 
     std::vector<Host> getSortedHosts(
-      const HostMap& hostMap,
+      HostMap& hostMap,
       const InFlightReqs& inFlightReqs,
       std::shared_ptr<faabric::BatchExecuteRequest> req,
       const DecisionType& decisionType) override;

include/faabric/batch-scheduler/SchedulingDecision.h

@@ -101,7 +101,7 @@ class SchedulingDecision
 
     std::set<std::string> uniqueHosts();
 
-    void print();
+    void print(const std::string& logLevel = "debug");
 };
 
 }

include/faabric/mpi/MpiWorld.h

@@ -201,9 +201,7 @@ class MpiWorld
 
     /* Function Migration */
 
-    void prepareMigration(
-      int thisRank,
-      std::shared_ptr<faabric::PendingMigrations> pendingMigrations);
+    void prepareMigration(int thisRank);
 
   private:
     int id = -1;
@@ -267,8 +265,5 @@ class MpiWorld
                 int count,
                 MPI_Status* status,
                 MPIMessage::MPIMessageType messageType = MPIMessage::NORMAL);
-
-    /* Function migration */
-    bool hasBeenMigrated = false;
 };
 }

include/faabric/planner/Planner.h

@@ -1,8 +1,10 @@
 #pragma once
 
+#include <faabric/batch-scheduler/SchedulingDecision.h>
 #include <faabric/planner/PlannerState.h>
 #include <faabric/planner/planner.pb.h>
 #include <faabric/proto/faabric.pb.h>
+#include <faabric/snapshot/SnapshotRegistry.h>
 
 #include <shared_mutex>
 
@@ -31,7 +33,7 @@ class Planner
     void printConfig() const;
 
     // ----------
-    // Util
+    // Util public API
     // ----------
 
     bool reset();
@@ -64,6 +66,12 @@ class Planner
     std::shared_ptr<faabric::BatchExecuteRequestStatus> getBatchResults(
       int32_t appId);
 
+    std::shared_ptr<faabric::batch_scheduler::SchedulingDecision>
+    getSchedulingDecision(std::shared_ptr<BatchExecuteRequest> req);
+
+    std::shared_ptr<faabric::batch_scheduler::SchedulingDecision> callBatch(
+      std::shared_ptr<BatchExecuteRequest> req);
+
   private:
     // There's a singleton instance of the planner running, but it must allow
     // concurrent requests
@@ -72,12 +80,31 @@ class Planner
     PlannerState state;
     PlannerConfig config;
 
+    // Snapshot registry to distribute snapshots in THREADS requests
+    faabric::snapshot::SnapshotRegistry& snapshotRegistry;
+
+    // ----------
+    // Util private API
+    // ----------
+
     void flushHosts();
 
     void flushExecutors();
 
+    // ----------
+    // Host membership private API
+    // ----------
+
     // Check if a host's registration timestamp has expired
     bool isHostExpired(std::shared_ptr<Host> host, long epochTimeMs = 0);
+
+    // ----------
+    // Request scheduling private API
+    // ----------
+
+    void dispatchSchedulingDecision(
+      std::shared_ptr<faabric::BatchExecuteRequest> req,
+      std::shared_ptr<faabric::batch_scheduler::SchedulingDecision> decision);
 };
 
 Planner& getPlanner();

include/faabric/planner/PlannerApi.h

@@ -13,5 +13,7 @@ enum PlannerCalls
     // Scheduling calls
     SetMessageResult = 8,
     GetMessageResult = 9,
+    GetSchedulingDecision = 10,
+    CallBatch = 11,
 };
 }

include/faabric/planner/PlannerClient.h

@@ -1,6 +1,8 @@
 #pragma once
 
+#include <faabric/batch-scheduler/SchedulingDecision.h>
 #include <faabric/planner/planner.pb.h>
+#include <faabric/snapshot/SnapshotClient.h>
 #include <faabric/transport/MessageEndpointClient.h>
 #include <faabric/util/PeriodicBackgroundThread.h>
 
@@ -31,12 +33,14 @@ class KeepAliveThread : public faabric::util::PeriodicBackgroundThread
 };
 
 /*
- * Local state associated with the current host, used to cache results and
- * avoid unnecessary interactions with the planner server.
+ * Local state associated with the current host, used to store useful state
+ * like cached results to unnecessary interactions with the planner server.
  */
 struct PlannerCache
 {
     std::unordered_map<uint32_t, MessageResultPromisePtr> plannerResults;
+    // Keeps track of the snapshots that have been pushed to the planner
+    std::set<std::string> pushedSnapshots;
 };
 
 /*
@@ -79,10 +83,19 @@ class PlannerClient final : public faabric::transport::MessageEndpointClient
     faabric::Message getMessageResult(const faabric::Message& msg,
                                       int timeoutMs);
 
+    faabric::batch_scheduler::SchedulingDecision callFunctions(
+      std::shared_ptr<faabric::BatchExecuteRequest> req);
+
+    faabric::batch_scheduler::SchedulingDecision getSchedulingDecision(
+      std::shared_ptr<faabric::BatchExecuteRequest> req);
+
   private:
     std::mutex plannerCacheMx;
     PlannerCache cache;
 
+    // Snapshot client for the planner snapshot server
+    std::shared_ptr<faabric::snapshot::SnapshotClient> snapshotClient;
+
     faabric::Message doGetMessageResult(
       std::shared_ptr<faabric::Message> msgPtr,
       int timeoutMs);

include/faabric/planner/PlannerServer.h

@@ -34,6 +34,12 @@ class PlannerServer final : public faabric::transport::MessageEndpointServer
     std::unique_ptr<google::protobuf::Message> recvGetMessageResult(
       std::span<const uint8_t> buffer);
 
+    std::unique_ptr<google::protobuf::Message> recvGetSchedulingDecision(
+      std::span<const uint8_t> buffer);
+
+    std::unique_ptr<google::protobuf::Message> recvCallBatch(
+      std::span<const uint8_t> buffer);
+
   private:
     faabric::planner::Planner& planner;
 };

include/faabric/planner/PlannerState.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <faabric/batch-scheduler/BatchScheduler.h>
 #include <faabric/planner/planner.pb.h>
 #include <faabric/proto/faabric.pb.h>
 
@@ -23,5 +24,8 @@ struct PlannerState
     // Map holding the hosts that have registered interest in getting an app
     // result
     std::map<int, std::vector<std::string>> appResultWaiters;
+
+    // Map keeping track of the requests that are in-flight
+    faabric::batch_scheduler::InFlightReqs inFlightReqs;
 };
 }

include/faabric/scheduler/FunctionCallApi.h

@@ -6,9 +6,6 @@ enum FunctionCalls
     NoFunctionCall = 0,
     ExecuteFunctions = 1,
     Flush = 2,
-    Unregister = 3,
-    GetResources = 4,
-    PendingMigrations = 5,
-    SetMessageResult = 6,
+    SetMessageResult = 3,
 };
 }

include/faabric/scheduler/FunctionCallClient.h

@@ -20,21 +20,9 @@ std::vector<
   std::pair<std::string, std::shared_ptr<faabric::BatchExecuteRequest>>>
 getBatchRequests();
 
-std::vector<std::pair<std::string, faabric::EmptyRequest>>
-getResourceRequests();
-
-std::vector<std::pair<std::string, std::shared_ptr<faabric::PendingMigrations>>>
-getPendingMigrationsRequests();
-
-std::vector<std::pair<std::string, faabric::UnregisterRequest>>
-getUnregisterRequests();
-
 std::vector<std::pair<std::string, std::shared_ptr<faabric::Message>>>
 getMessageResults();
 
-void queueResourceResponse(const std::string& host,
-                           faabric::HostResources& res);
-
 void clearMockRequests();
 
 // -----------------------------------
@@ -52,14 +40,8 @@ class FunctionCallClient : public faabric::transport::MessageEndpointClient
 
     void sendFlush();
 
-    faabric::HostResources getResources();
-
-    void sendPendingMigrations(std::shared_ptr<faabric::PendingMigrations> req);
-
     void executeFunctions(std::shared_ptr<faabric::BatchExecuteRequest> req);
 
-    void unregister(faabric::UnregisterRequest& req);
-
     void setMessageResult(std::shared_ptr<faabric::Message> msg);
 };
 

include/faabric/scheduler/FunctionCallServer.h

@@ -23,16 +23,8 @@ class FunctionCallServer final
     std::unique_ptr<google::protobuf::Message> recvFlush(
       std::span<const uint8_t> buffer);
 
-    std::unique_ptr<google::protobuf::Message> recvGetResources(
-      std::span<const uint8_t> buffer);
-
-    std::unique_ptr<google::protobuf::Message> recvPendingMigrations(
-      std::span<const uint8_t> buffer);
-
     void recvExecuteFunctions(std::span<const uint8_t> buffer);
 
-    void recvUnregister(std::span<const uint8_t> buffer);
-
     void recvSetMessageResult(std::span<const uint8_t> buffer);
 };
 }