WIP Donald ug

.vscode/settings.json

@@ -0,0 +1,49 @@
+{
+<<<<<<< HEAD
+    "editor.tokenColorCustomizations": {
+        "[*Light*]": {
+            "textMateRules": [
+                {
+                    "scope": "ref.matchtext",
+                    "settings": {
+                        "foreground": "#000"
+                    }
+                }
+            ]
+        },
+        "[*Dark*]": {
+            "textMateRules": [
+                {
+                    "scope": "ref.matchtext",
+                    "settings": {
+                        "foreground": "#fff"
+                    }
+                }
+            ]
+        },
+        "textMateRules": [
+            {
+                "scope": "googletest.failed",
+                "settings": {
+                    "foreground": "#f00"
+                }
+            },
+            {
+                "scope": "googletest.passed",
+                "settings": {
+                    "foreground": "#0f0"
+                }
+            },
+            {
+                "scope": "googletest.run",
+                "settings": {
+                    "foreground": "#0f0"
+                }
+            }
+        ]
+=======
+    "files.associations": {
+        "array": "cpp"
+>>>>>>> f2f01849eae98603aa3f1223f9af265743b91e5d
+    }
+}

CMakeLists.txt

@@ -106,6 +106,7 @@ add_subdirectory(src/mpi)
 add_subdirectory(src/proto)
 add_subdirectory(src/redis)
 add_subdirectory(src/runner)
+add_subdirectory(src/loadbalance)
 add_subdirectory(src/scheduler)
 add_subdirectory(src/snapshot)
 add_subdirectory(src/state)
@@ -130,6 +131,7 @@ add_library(faabric
         $<TARGET_OBJECTS:scheduler_obj>
         $<TARGET_OBJECTS:snapshot_obj>
         $<TARGET_OBJECTS:state_obj>
+        $<TARGET_OBJECTS:loadbalance_obj>
         $<TARGET_OBJECTS:transport_obj>
         $<TARGET_OBJECTS:util_obj>
         )

include/faabric/loadbalance/LoadBalancePolicy.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <set>
+#include <string>
+#include <vector>
+#include <faabric/scheduler/Scheduler.h>
+
+class LoadBalancePolicy
+{
+    public:
+        virtual std::vector<std::pair<std::string, faabric::HostResources>> dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources) = 0;
+};
+
+class FaasmDefaultPolicy : public LoadBalancePolicy
+{
+    public:
+        std::vector<std::pair<std::string, faabric::HostResources>> dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources) override;
+};
+
+class LeastLoadAveragePolicy : public LoadBalancePolicy
+{
+    public:
+        std::vector<std::pair<std::string, faabric::HostResources>> dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources) override;
+};
+
+class MostSlotsPolicy : public LoadBalancePolicy
+{
+    public:
+        std::vector<std::pair<std::string, faabric::HostResources>> dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources) override;
+};

include/faabric/scheduler/Scheduler.h

@@ -8,6 +8,7 @@
 #include <faabric/snapshot/SnapshotRegistry.h>
 #include <faabric/transport/PointToPointBroker.h>
 #include <faabric/util/PeriodicBackgroundThread.h>
+#include <faabric/util/system_metrics.h>
 #include <faabric/util/asio.h>
 #include <faabric/util/clock.h>
 #include <faabric/util/config.h>
@@ -277,7 +278,11 @@ class Scheduler
 
     inline void setFunctionResult(const faabric::Message& msg)
     {
-        setFunctionResult(std::make_unique<faabric::Message>(msg));
+      try {
+          setFunctionResult(std::make_unique<faabric::Message>(msg));
+      } catch (const std::exception& e) {
+          SPDLOG_ERROR("[Scheduler.h] Failed to set function result: {}", e.what());
+      }
     }
 
     faabric::Message getFunctionResult(unsigned int messageId,
@@ -462,6 +467,8 @@ class Scheduler
       faabric::util::SchedulingTopologyHint topologyHint,
       std::shared_ptr<void> extraData);
 
+    std::set<std::string> applyLoadBalancedPolicy(std::vector<std::string> hosts);
+
     std::shared_ptr<Executor> claimExecutor(const faabric::MessageInBatch& msg);
 
     std::vector<std::string> getUnregisteredHosts(const std::string& user,

include/faabric/util/config.h

@@ -32,6 +32,11 @@ class SystemConfig
     bool isStorageNode;
     int noSingleHostOptimisations;
 
+    std::string load_balance_policy;
+    double offload_cpu_threshold;
+    double offload_ram_threshold;
+    double offload_load_avg_threshold;
+
     // Worker-related timeouts
     int globalMessageTimeout;
     int boundTimeout;

include/faabric/util/queue.h

@@ -54,22 +54,29 @@ class Queue
             SPDLOG_ERROR("Invalid queue timeout: {} <= 0", timeoutMs);
             throw std::runtime_error("Invalid queue timeout");
         }
-
-        while (mq.empty()) {
+
+        while (mq.size() == 0) {
+            SPDLOG_DEBUG("Queue is empty... waiting for dequeue");
             std::cv_status returnVal = enqueueNotifier.wait_for(
               lock, std::chrono::milliseconds(timeoutMs));
+
+            SPDLOG_DEBUG("Queue has been notified");
 
             // Work out if this has returned due to timeout expiring
             if (returnVal == std::cv_status::timeout) {
                 throw QueueTimeoutException("Timeout waiting for dequeue");
             }
         }
 
-        T value = std::move(mq.front());
-        mq.pop();
-        emptyNotifier.notify_one();
-
-        return value;
+        try {
+            T value = std::move(mq.front());
+            mq.pop();
+            emptyNotifier.notify_one();
+            return value;
+        } catch (std::exception& e) {
+            SPDLOG_ERROR("Caught exception when dequeueing: {}", e.what());
+            throw;
+        }
     }
 
     T* peek(long timeoutMs = 0)

include/faabric/util/system_metrics.h

@@ -0,0 +1,37 @@
+#pragma once 
+
+#include <cstdint>
+#include <string>
+#include <vector>
+#include <fstream>
+#include <sstream>
+#include <iostream>
+#include <stdexcept>
+#include <thread>
+#include <chrono>
+
+namespace faabric::util {
+    struct UtilisationStats
+    {
+      double cpu_utilisation;
+      double ram_utilisation;
+      double load_average;
+    };
+
+    struct CPUStats
+    {
+      long totalCpuTime;
+      long idleCpuTime;
+    };
+
+    struct MemStats
+    {
+      uint64_t total;
+      uint64_t available;
+    };
+
+    UtilisationStats getSystemUtilisation();
+    CPUStats getCPUUtilisation();
+    double getMemoryUtilisation();
+    double getLoadAverage();
+}

src/endpoint/FaabricEndpoint.cpp

@@ -78,7 +78,27 @@ class HttpConnection : public std::enable_shared_from_this<HttpConnection>
                                 stream.get_executor(),
                                 std::bind_front(&HttpConnection::sendResponse,
                                                 this->shared_from_this()) };
-        handler->onRequest(std::move(hrc), std::move(msg));
+        try {
+          handler->onRequest(std::move(hrc), std::move(msg));
+        } catch (std::exception& e) {
+          SPDLOG_ERROR("Error handling HTTP request: {}", e.what());
+          faabric::util::BeastHttpResponse response;
+          response.result(beast::http::status::internal_server_error);
+          response.body() = e.what();
+          sendResponse(std::move(response));
+        } catch (boost::system::system_error& e) {
+          SPDLOG_ERROR("Error handling HTTP request: {}", e.what());
+          faabric::util::BeastHttpResponse response;
+          response.result(beast::http::status::internal_server_error);
+          response.body() = e.what();
+          sendResponse(std::move(response));
+        } catch(...) {
+          SPDLOG_ERROR("Error handling HTTP request: unknown exception");
+          faabric::util::BeastHttpResponse response;
+          response.result(beast::http::status::internal_server_error);
+          response.body() = "Unknown error";
+          sendResponse(std::move(response));
+        }
     }
 
     void onRead(beast::error_code ec, size_t bytesTransferred)

src/endpoint/FaabricEndpointHandler.cpp

@@ -89,8 +89,22 @@ void FaabricEndpointHandler::onRequest(
             response.result(beast::http::status::ok);
             response.body() = std::string("Flush sent");
         } else {
-            executeFunction(
-              std::move(ctx), std::move(response), std::move(msg));
+            try
+            {
+                executeFunction(std::move(ctx), std::move(response), std::move(msg));
+            }
+            catch (const std::exception& e)
+            {
+                SPDLOG_ERROR("Caught exception in FaabricEndpointHandler::onRequest: {}", e.what());
+                response.result(beast::http::status::internal_server_error);
+                response.body() = std::string("Caught exception: ") + e.what();
+                ctx.sendFunction(std::move(response));
+            } catch (faabric::util::FaabricException& e) {
+                SPDLOG_ERROR("Caught FaabricException in FaabricEndpointHandler::onRequest: {}", e.what());
+                response.result(beast::http::status::internal_server_error);
+                response.body() = std::string("Caught exception: ") + e.what();
+                ctx.sendFunction(std::move(response));
+            }
             return;
         }
     }
@@ -174,7 +188,10 @@ void FaabricEndpointHandler::onFunctionResult(
                  faabric::util::funcToString(result, true));
 
     response.body() = result.outputdata();
+    SPDLOG_DEBUG("Worker thread {} sending response", gettid());
     return ctx.sendFunction(std::move(response));
+    SPDLOG_DEBUG("Worker thread {} response sent", gettid());
+    // We're done with this request
 }
 
 }

src/loadbalance/CMakeLists.txt

@@ -0,0 +1,7 @@
+faabric_lib(loadbalance
+    FaasmDefaultPolicy.cpp
+    LeastLoadAveragePolicy.cpp
+    MostSlotsPolicy.cpp)
+
+
+target_link_libraries(loadbalance PRIVATE faabric::scheduler)

src/loadbalance/FaasmDefaultPolicy.cpp

@@ -0,0 +1,7 @@
+#include <faabric/loadbalance/LoadBalancePolicy.h>
+#include <stdexcept>
+
+std::vector<std::pair<std::string, faabric::HostResources>> FaasmDefaultPolicy::dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources)
+{
+    return host_resources;
+}

src/loadbalance/LeastLoadAveragePolicy.cpp

@@ -0,0 +1,12 @@
+#include <faabric/loadbalance/LoadBalancePolicy.h>
+#include <stdexcept>
+
+std::vector<std::pair<std::string, faabric::HostResources>> LeastLoadAveragePolicy::dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources)
+{
+    // Sort the vector by the load average in ascending order
+    std::sort(host_resources.begin(), host_resources.end(), [](const auto &a, const auto &b) {
+        return a.second.loadaverage() < b.second.loadaverage();
+    });
+
+    return host_resources;
+}

src/loadbalance/MostSlotsPolicy.cpp

@@ -0,0 +1,14 @@
+#include <faabric/loadbalance/LoadBalancePolicy.h>
+#include <stdexcept>
+
+std::vector<std::pair<std::string, faabric::HostResources>> MostSlotsPolicy::dispatch(std::vector<std::pair<std::string, faabric::HostResources>>& host_resources)
+{
+    // Sort the vector by the number of available slots in descending order
+    std::sort(host_resources.begin(), host_resources.end(), [](const auto &a, const auto &b) {
+        int available_a = a.second.slots() - a.second.usedslots();
+        int available_b = b.second.slots() - b.second.usedslots();
+        return available_a > available_b;
+    });
+
+    return host_resources;
+}

src/proto/faabric.proto

@@ -47,6 +47,7 @@ message BatchExecuteRequest {
 message HostResources {
     int32 slots = 1;
     int32 usedSlots = 2;
+    double loadAverage = 3;
 }
 
 message UnregisterRequest {
@@ -72,6 +73,12 @@ message FunctionStatusResponse {
     FunctionStatus status = 1;
 }
 
+message NodeUtilisationResponse {
+    double cpu_utilisation = 1;
+    double mem_utilisation = 2;
+    double load_avg = 3;
+}
+
 // ---------------------------------------------
 // MPI
 // ---------------------------------------------

src/redis/Redis.cpp

@@ -678,7 +678,7 @@ UniqueRedisReply Redis::dequeueBase(const std::string& queueName, int timeoutMs)
     // Check if we got anything
     if (reply == nullptr || reply->type == REDIS_REPLY_NIL) {
         std::string msg =
-          fmt::format("No response from Redis dequeue in {}ms for queue {}",
+          fmt::format("d from Redis dequeue in {}ms for queue {}",
                       timeoutMs,
                       queueName);
         throw RedisNoResponseException(msg);