A large number of performance changes (related to faasm#156)

CMakeLists.txt

@@ -105,8 +105,9 @@ add_dependencies(faabric pistache_ext spdlog_ext)
 target_link_libraries(faabric PUBLIC
     faabricmpi
     hiredis
-    boost_system
-    boost_filesystem
+    Boost::system
+    Boost::filesystem
+    Boost::Boost
     zstd::libzstd_static
     ${PISTACHE_LIBRARY}
     ${PROTOBUF_LIBRARY}

cmake/ExternalProjects.cmake

@@ -2,6 +2,7 @@ include(FindGit)
 find_package(Git)
 include (ExternalProject)
 include (FetchContent)
+find_package (Threads REQUIRED)
 
 # Protobuf
 set(PROTOBUF_LIBRARY ${CMAKE_INSTALL_PREFIX}/lib/libprotobuf.so)
@@ -116,6 +117,47 @@ FetchContent_MakeAvailable(zstd_ext)
 target_include_directories(libzstd_static INTERFACE $<BUILD_INTERFACE:${zstd_ext_SOURCE_DIR}/lib>)
 add_library(zstd::libzstd_static ALIAS libzstd_static)
 
+# Boost libraries, the header-only ones
+FetchContent_Declare(boost_ext
+    URL "https://boostorg.jfrog.io/artifactory/main/release/1.77.0/source/boost_1_77_0.tar.bz2"
+    URL_HASH "SHA256=fc9f85fc030e233142908241af7a846e60630aa7388de9a5fafb1f3a26840854"
+)
+FetchContent_GetProperties(boost_ext)
+if(NOT boost_ext_POPULATED)
+  FetchContent_Populate(boost_ext)
+endif()
+add_library(Boost INTERFACE)
+target_compile_definitions(Boost INTERFACE
+    BOOST_BEAST_USE_STD_STRING_VIEW
+    BOOST_ASIO_NO_DEPRECATED
+    BOOST_ASIO_NO_TS_EXECUTORS
+    BOOST_ASIO_NO_DEFAULT_LINKED_LIBS
+)
+target_include_directories(Boost INTERFACE ${boost_ext_SOURCE_DIR})
+target_link_libraries(Boost INTERFACE Threads::Threads)
+target_compile_features(Boost INTERFACE cxx_std_17)
+add_library(Boost::Boost ALIAS Boost)
+# Header-only aliases
+add_library(Boost::atomic ALIAS Boost)
+add_library(Boost::core ALIAS Boost)
+add_library(Boost::assert ALIAS Boost)
+add_library(Boost::config ALIAS Boost)
+add_library(Boost::container_hash ALIAS Boost)
+add_library(Boost::detail ALIAS Boost)
+add_library(Boost::io ALIAS Boost)
+add_library(Boost::iterator ALIAS Boost)
+add_library(Boost::smart_ptr ALIAS Boost)
+add_library(Boost::system ALIAS Boost)
+add_library(Boost::type_traits ALIAS Boost)
+add_library(Boost::predef ALIAS Boost)
+add_library(Boost::Boost ALIAS Boost)
+add_subdirectory(${boost_ext_SOURCE_DIR}/libs/filesystem ${boost_ext_BINARY_DIR}/libs/filesystem EXCLUDE_FROM_ALL)
+
+install(
+    DIRECTORY ${boost_ext_SOURCE_DIR}/boost
+    DESTINATION include
+)
+
 # ZeroMQ
 set(ZEROMQ_LIBRARY ${CMAKE_INSTALL_PREFIX}/lib/libzmq.so)
 ExternalProject_Add(libzeromq_ext
@@ -129,7 +171,7 @@ ExternalProject_Get_Property(libzeromq_ext SOURCE_DIR)
 set(LIBZEROMQ_INCLUDE_DIR ${SOURCE_DIR})
 ExternalProject_Add(cppzeromq_ext
     GIT_REPOSITORY "https://github.com/zeromq/cppzmq.git"
-    GIT_TAG "v4.7.1"
+    GIT_TAG "v4.8.0"
     CMAKE_CACHE_ARGS "-DCPPZMQ_BUILD_TESTS:BOOL=OFF"
         "-DCMAKE_INSTALL_PREFIX:STRING=${CMAKE_INSTALL_PREFIX}"
 )

examples/CMakeLists.txt

@@ -11,22 +11,25 @@ set(CMAKE_CXX_EXTENSIONS OFF)
 # the installed Faabric library as an external user would.
 # -----------------------------------------------
 set(FAABRIC_LIB_DIR "/build/faabric/install/lib")
+set(FAABRIC_INCLUDE_DIR "/build/faabric/install/include")
 
 function(add_example example_name)
     add_executable(${example_name} ${example_name}.cpp)
 
-    target_link_libraries(${example_name}
+    target_link_libraries(${example_name} PRIVATE
         ${FAABRIC_LIB_DIR}/libfaabric.so
         ${FAABRIC_LIB_DIR}/libfaabricmpi.so
         ${FAABRIC_LIB_DIR}/libprotobuf.so
         ${FAABRIC_LIB_DIR}/libpistache.so
         ${FAABRIC_LIB_DIR}/libzmq.so
-        boost_system
-        boost_filesystem
         hiredis
         pthread
     )
 
+    target_include_directories(${example_name} PRIVATE
+        ${FAABRIC_INCLUDE_DIR}
+    )
+
     set(ALL_EXAMPLES ${ALL_EXAMPLES} ${example_name} PARENT_SCOPE)
 endfunction()
 

examples/server.cpp

@@ -1,7 +1,9 @@
-#include <faabric/endpoint/FaabricEndpoint.h>
+#include <faabric/endpoint/Endpoint.h>
+#include <faabric/endpoint/FaabricEndpointHandler.h>
 #include <faabric/runner/FaabricMain.h>
 #include <faabric/scheduler/ExecutorFactory.h>
 #include <faabric/transport/context.h>
+#include <faabric/util/config.h>
 #include <faabric/util/logging.h>
 
 using namespace faabric::scheduler;
@@ -50,7 +52,11 @@ int main()
 
     // Start endpoint (will also have multiple threads)
     SPDLOG_INFO("Starting endpoint");
-    faabric::endpoint::FaabricEndpoint endpoint;
+    const auto& config = faabric::util::getSystemConfig();
+    faabric::endpoint::Endpoint endpoint(
+      config.endpointPort,
+      config.endpointNumThreads,
+      std::make_shared<faabric::endpoint::FaabricEndpointHandler>());
     endpoint.start();
 
     SPDLOG_INFO("Shutting down endpoint");

include/faabric/endpoint/Endpoint.h

@@ -1,28 +1,53 @@
 #pragma once
 
+#include <functional>
+#include <memory>
+
 #include <faabric/proto/faabric.pb.h>
+#include <faabric/util/asio.h>
 #include <faabric/util/config.h>
-#include <pistache/endpoint.h>
-#include <pistache/http.h>
 
 namespace faabric::endpoint {
+namespace detail {
+struct EndpointState;
+}
+
+struct HttpRequestContext
+{
+    asio::io_context& ioc;
+    asio::any_io_executor executor;
+    std::function<void(faabric::util::BeastHttpResponse&&)> sendFunction;
+};
+
+class HttpRequestHandler
+{
+  public:
+    virtual void onRequest(HttpRequestContext&& ctx,
+                           faabric::util::BeastHttpRequest&& request) = 0;
+};
+
 class Endpoint
 {
   public:
-    Endpoint();
+    Endpoint() = delete;
+    Endpoint(const Endpoint&) = delete;
+    Endpoint(Endpoint&&) = delete;
+    Endpoint& operator=(const Endpoint&) = delete;
+    Endpoint& operator=(Endpoint&&) = delete;
+    virtual ~Endpoint();
 
-    Endpoint(int port, int threadCount);
+    Endpoint(int port,
+             int threadCount,
+             std::shared_ptr<HttpRequestHandler> requestHandlerIn);
 
     void start(bool awaitSignal = true);
 
     void stop();
 
-    virtual std::shared_ptr<Pistache::Http::Handler> getHandler() = 0;
-
   private:
-    int port = faabric::util::getSystemConfig().endpointPort;
-    int threadCount = faabric::util::getSystemConfig().endpointNumThreads;
-
-    Pistache::Http::Endpoint httpEndpoint;
+    int port;
+    int threadCount;
+    std::unique_ptr<detail::EndpointState> state;
+    std::shared_ptr<HttpRequestHandler> requestHandler;
 };
 }

include/faabric/endpoint/FaabricEndpointHandler.h

@@ -1,23 +1,24 @@
 #pragma once
 
+#include <faabric/endpoint/Endpoint.h>
 #include <faabric/proto/faabric.pb.h>
-#include <pistache/http.h>
 
 namespace faabric::endpoint {
-class FaabricEndpointHandler : public Pistache::Http::Handler
+class FaabricEndpointHandler final
+  : public HttpRequestHandler
+  , public std::enable_shared_from_this<FaabricEndpointHandler>
 {
   public:
-    HTTP_PROTOTYPE(FaabricEndpointHandler)
-
-    void onTimeout(const Pistache::Http::Request& request,
-                   Pistache::Http::ResponseWriter writer) override;
-
-    void onRequest(const Pistache::Http::Request& request,
-                   Pistache::Http::ResponseWriter response) override;
-
-    std::pair<int, std::string> handleFunction(const std::string& requestStr);
+    void onRequest(HttpRequestContext&& ctx,
+                   faabric::util::BeastHttpRequest&& request) override;
 
   private:
-    std::pair<int, std::string> executeFunction(faabric::Message& msg);
+    void executeFunction(HttpRequestContext&& ctx,
+                         faabric::util::BeastHttpResponse&& partialResponse,
+                         faabric::Message&& msg);
+
+    void onFunctionResult(HttpRequestContext&& ctx,
+                          faabric::util::BeastHttpResponse&& partialResponse,
+                          faabric::Message& msg);
 };
 }

include/faabric/mpi-native/MpiExecutor.h

@@ -1,6 +1,7 @@
 #pragma once
 
-#include <faabric/endpoint/FaabricEndpoint.h>
+#include <faabric/endpoint/Endpoint.h>
+#include <faabric/endpoint/FaabricEndpointHandler.h>
 #include <faabric/scheduler/ExecutorFactory.h>
 #include <faabric/scheduler/Scheduler.h>
 

include/faabric/scheduler/FunctionCallApi.h

@@ -9,5 +9,6 @@ enum FunctionCalls
     Unregister = 3,
     GetResources = 4,
     SetThreadResult = 5,
+    DirectResult = 6,
 };
 }

include/faabric/scheduler/FunctionCallClient.h

@@ -47,6 +47,8 @@ class FunctionCallClient : public faabric::transport::MessageEndpointClient
     void executeFunctions(
       const std::shared_ptr<faabric::BatchExecuteRequest> req);
 
+    void sendDirectResult(faabric::Message msg);
+
     void unregister(faabric::UnregisterRequest& req);
 
   private:

include/faabric/scheduler/FunctionCallServer.h

@@ -32,5 +32,7 @@ class FunctionCallServer final
     void recvExecuteFunctions(const uint8_t* buffer, size_t bufferSize);
 
     void recvUnregister(const uint8_t* buffer, size_t bufferSize);
+
+    void recvDirectResult(const uint8_t* buffer, size_t bufferSize);
 };
 }