MPI Standalone Implementation - First Bits

.github/workflows/tests.yml

@@ -105,6 +105,10 @@ jobs:
         run: inv dev.cc faabric --shared
       - name: "Install Faabric shared library"
         run: inv dev.install faabric --shared
+      - name: "Build MPI native library"
+        run: inv dev.cc faabricmpi_native --shared
+      - name: "Install MPI native library"
+        run: inv dev.install faabricmpi_native --shared
       - name: "Build examples"
         run: inv examples
       - name: "Run example to check"

CMakeLists.txt

@@ -56,6 +56,7 @@ endfunction()
 add_subdirectory(src/endpoint)
 add_subdirectory(src/executor)
 add_subdirectory(src/mpi)
+add_subdirectory(src/mpi-native)
 add_subdirectory(src/proto)
 add_subdirectory(src/redis)
 add_subdirectory(src/scheduler)

README.md

@@ -110,3 +110,9 @@ inv container.push
 inv container.build --push
 ```
 
+## Additional documentation
+
+More detail on some key features and implementations can be found below:
+
+- [Native MPI builds](docs/native_mpi.md) - run native applications against
+Faabric's MPI library.

docker-compose.yml

@@ -9,6 +9,7 @@ services:
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
       - /usr/bin/docker:/usr/bin/docker
+      - /usr/local/bin/docker-compose:/usr/bin/docker-compose
       - .:/code/faabric
       - ./build:/build/faabric
     working_dir: /code/faabric

docs/native_mpi.md

@@ -0,0 +1,51 @@
+# Native MPI execution in Faabric
+
+Faabric supports linking MPI binaries against our custom MPI implementation
+used in [Faasm](https://github.com/faasm/faasm). This way, you can test the
+compliance of your MPI application with our API (a subset of the standard)
+without the burden of cross-compiling to WebAssembly.
+
+To run native MPI applications you need to: compile the dynamic library, and
+slightly modify the original soource code.
+
+## Compiling the library
+
+Compilation should be smooth if you are running our recommended containerised
+[development environment](../README.md). You may access the container running
+`./bin/cli.sh`.
+
+Then, to compile the library:
+```bash
+inv dev.cmake --shared
+inv dev.cc faabricmpi_native --shared
+inv dev.install faabricmpi_native --shared
+```
+
+## Adapting the source code
+
+## Running the binary
+
+To run an example, run this command _outside_ the container:
+```bash
+# The --clean flag re-creates _all_ containers
+inv mpi.execute mpi_helloworld --clean --np 5
+```
+
+To clean the cluster and set the development one again:
+```bash
+inv mpi.clean
+```
+
+Using the `--force` flag will recreate _all_ containers hence finishing any
+sessions you may have open:
+```bash
+inv mpi.clean --force
+```
+
+## Debugging
+
+If at some point you reach an unstable state of the cluster, stop it completely
+using:
+```bash
+docker-compose down
+```

examples/CMakeLists.txt

@@ -17,6 +17,7 @@ function(add_example example_name)
     target_link_libraries(${example_name}
         faabric
         faabricmpi
+        faabricmpi_native
         protobuf
         pthread
         pistache
@@ -31,6 +32,7 @@ function(add_example example_name)
 endfunction()
 
 add_example(check)
+add_example(mpi_helloworld)
 add_example(server)
 
 add_custom_target(all_examples DEPENDS ${ALL_EXAMPLES})

examples/mpi_helloworld.cpp

@@ -0,0 +1,62 @@
+#include <faabric/util/logging.h>
+
+#include <faabric/mpi-native/MpiExecutor.h>
+#include <faabric/mpi/mpi.h>
+
+int main(int argc, char** argv)
+{
+    auto logger = faabric::util::getLogger();
+    auto& scheduler = faabric::scheduler::getScheduler();
+    auto& conf = faabric::util::getSystemConfig();
+
+    // Global configuration
+    conf.maxNodes = 1;
+    conf.maxNodesPerFunction = 1;
+
+    bool __isRoot;
+    int __worldSize;
+    if (argc < 2) {
+        logger->debug("Non-root process started");
+        __isRoot = false;
+    } else if (argc < 3) {
+        logger->error("Root process started without specifying world size!");
+        return 1;
+    } else {
+        logger->debug("Root process started");
+        __worldSize = std::stoi(argv[2]);
+        __isRoot = true;
+        logger->debug("MPI World Size: {}", __worldSize);
+    }
+
+    // Pre-load message to bootstrap execution
+    if (__isRoot) {
+        faabric::Message msg = faabric::util::messageFactory("mpi", "exec");
+        msg.set_mpiworldsize(__worldSize);
+        scheduler.callFunction(msg);
+    }
+
+    {
+        faabric::executor::SingletonPool p;
+        p.startPool(false);
+    }
+
+    return 0;
+}
+
+int faabric::executor::mpiFunc()
+{
+    auto logger = faabric::util::getLogger();
+    logger->info("Hello world from Faabric MPI Main!");
+
+    MPI_Init(NULL, NULL);
+
+    int rank, worldSize;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    MPI_Comm_size(MPI_COMM_WORLD, &worldSize);
+
+    logger->info("Hello faabric from process {} of {}", rank + 1, worldSize);
+
+    MPI_Finalize();
+
+    return 0;
+}

include/faabric/endpoint/Endpoint.h

@@ -5,6 +5,9 @@
 #include <pistache/endpoint.h>
 #include <pistache/http.h>
 
+#include <memory>
+#include <thread>
+
 namespace faabric::endpoint {
 class Endpoint
 {
@@ -13,12 +16,19 @@ class Endpoint
 
     Endpoint(int port, int threadCount);
 
-    void start();
+    void start(bool background = true);
+
+    void doStart();
+
+    void stop();
 
     virtual std::shared_ptr<Pistache::Http::Handler> getHandler() = 0;
 
   private:
+    bool isBackground;
     int port = faabric::util::getSystemConfig().endpointPort;
     int threadCount = faabric::util::getSystemConfig().endpointNumThreads;
+    std::thread servingThread;
+    std::unique_ptr<Pistache::Http::Endpoint> httpEndpoint;
 };
 }

include/faabric/executor/FaabricExecutor.h

@@ -40,6 +40,8 @@ class FaabricExecutor
                                bool success,
                                const std::string& errorMsg);
 
+    virtual void postFinishCall();
+
     virtual void postFinish();
 
     bool _isBound = false;

include/faabric/executor/FaabricPool.h

@@ -14,7 +14,7 @@ class FaabricPool
 
     void startFunctionCallServer();
 
-    void startThreadPool();
+    void startThreadPool(bool background = true);
 
     void startStateServer();
 

include/faabric/mpi-native/MpiExecutor.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include <faabric/endpoint/FaabricEndpoint.h>
+#include <faabric/executor/FaabricExecutor.h>
+#include <faabric/executor/FaabricPool.h>
+#include <faabric/scheduler/Scheduler.h>
+#include <faabric/util/logging.h>
+
+namespace faabric::executor {
+class MpiExecutor final : public faabric::executor::FaabricExecutor
+{
+  public:
+    explicit MpiExecutor();
+
+    bool doExecute(faabric::Message& msg) override;
+
+    void postFinishCall() override;
+
+    void postFinish() override;
+};
+
+class SingletonPool : public faabric::executor::FaabricPool
+{
+  public:
+    SingletonPool();
+
+    ~SingletonPool();
+
+    void startPool(bool background);
+
+  protected:
+    std::unique_ptr<FaabricExecutor> createExecutor(int threadIdx) override
+    {
+        return std::make_unique<MpiExecutor>();
+    }
+
+  private:
+    faabric::endpoint::FaabricEndpoint endpoint;
+    faabric::scheduler::Scheduler& scheduler;
+};
+
+extern faabric::Message* executingCall;
+extern int __attribute__((weak)) mpiFunc();
+}

src/endpoint/Endpoint.cpp

@@ -11,11 +11,23 @@ Endpoint::Endpoint(int portIn, int threadCountIn)
   , threadCount(threadCountIn)
 {}
 
-void Endpoint::start()
+void Endpoint::start(bool background)
 {
-    const std::shared_ptr<spdlog::logger>& logger = faabric::util::getLogger();
+    auto logger = faabric::util::getLogger();
+    this->isBackground = background;
 
-    logger->info("Starting HTTP endpoint");
+    if (background) {
+        logger->debug("Starting HTTP endpoint in background thread");
+        servingThread = std::thread([this] { doStart(); });
+    } else {
+        logger->debug("Starting HTTP endpoint in this thread");
+        doStart();
+    }
+}
+
+void Endpoint::doStart()
+{
+    auto logger = faabric::util::getLogger();
 
     // Set up signal handler
     sigset_t signals;
@@ -35,23 +47,41 @@ void Endpoint::start()
                   .backlog(256)
                   .flags(Pistache::Tcp::Options::ReuseAddr);
 
-    Pistache::Http::Endpoint httpEndpoint(addr);
-    httpEndpoint.init(opts);
+    httpEndpoint = std::make_unique<Pistache::Http::Endpoint>(addr);
+    httpEndpoint->init(opts);
 
     // Configure and start endpoint
-    httpEndpoint.setHandler(this->getHandler());
-    httpEndpoint.serveThreaded();
-
-    // Wait for a signal
-    logger->info("Awaiting signal");
-    int signal = 0;
-    int status = sigwait(&signals, &signal);
-    if (status == 0) {
-        logger->info("Received signal: {}", signal);
-    } else {
-        logger->info("Sigwait return value: {}", signal);
+    httpEndpoint->setHandler(this->getHandler());
+    httpEndpoint->serveThreaded();
+
+    // Wait for a signal if running in foreground (this blocks execution)
+    if (!isBackground) {
+        logger->info("Awaiting signal");
+        int signal = 0;
+        int status = sigwait(&signals, &signal);
+        if (status == 0) {
+            logger->info("Received signal: {}", signal);
+        } else {
+            logger->info("Sigwait return value: {}", signal);
+        }
+
+        stop();
     }
+}
+
+void Endpoint::stop()
+{
+    // Shut down the endpoint
+    auto logger = faabric::util::getLogger();
+
+    logger->debug("Shutting down HTTP endpoint");
+    this->httpEndpoint->shutdown();
 
-    httpEndpoint.shutdown();
+    if (isBackground) {
+        logger->debug("Waiting for HTTP endpoint background thread");
+        if (servingThread.joinable()) {
+            servingThread.join();
+        }
+    }
 }
 }

src/executor/FaabricExecutor.cpp

@@ -91,6 +91,9 @@ void FaabricExecutor::finishCall(faabric::Message& msg,
 
     // Increment the execution counter
     executionCount++;
+
+    // Hook
+    this->postFinishCall();
 }
 
 void FaabricExecutor::run()
@@ -202,6 +205,8 @@ void FaabricExecutor::preFinishCall(faabric::Message& call,
                                     const std::string& errorMsg)
 {}
 
+void FaabricExecutor::postFinishCall() {}
+
 void FaabricExecutor::postFinish() {}
 
 void FaabricExecutor::flush() {}

src/executor/FaabricPool.cpp

@@ -39,7 +39,7 @@ void FaabricPool::startStateServer()
     stateServer.start();
 }
 
-void FaabricPool::startThreadPool()
+void FaabricPool::startThreadPool(bool background)
 {
     const std::shared_ptr<spdlog::logger>& logger = faabric::util::getLogger();
     logger->info("Starting worker thread pool");
@@ -65,7 +65,11 @@ void FaabricPool::startThreadPool()
                   createExecutor(threadIdx);
 
                 // Worker will now run for a long time
-                executor.get()->run();
+                try {
+                    executor.get()->run();
+                } catch (faabric::util::ExecutorFinishedException& e) {
+                    this->_shutdown = true;
+                }
 
                 // Handle thread finishing
                 threadTokenPool.releaseToken(executor.get()->threadIdx);
@@ -82,6 +86,11 @@ void FaabricPool::startThreadPool()
 
         // Will die gracefully at this point
     });
+
+    // Make the call to startThreadPool blocking
+    if (!background) {
+        poolThread.join();
+    }
 }
 
 void FaabricPool::reset()