Added LibFabric dependency

CMakeLists.txt

@@ -41,6 +41,8 @@ option(BUILD_PYTHON_DOCS "Uses sphinx to build the Python documentaion" OFF)
 option(BUILD_CONSUMER_TOOLS "Build consumer tools" OFF)
 option(BUILD_EXAMPLES "Build examples" OFF)
 
+option(ENABLE_LIBFABRIC "Enables LibFabric support for RDMA transfers" OFF)
+
 set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/CMakeModules/)
 
 set (ASAPO_CXX_COMMON_INCLUDE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/common/cpp/include)
@@ -61,7 +63,16 @@ if ("${Python_EXECUTABLE}" STREQUAL "")
 endif()
 message (STATUS "Using Python: ${Python_EXECUTABLE}")
 
-
+IF(ENABLE_LIBFABRIC)
+    find_package(LibFabric)
+    if(NOT LIBFABRIC_LIBRARY)
+        message(FATAL_ERROR "Did not find libfabric")
+    endif()
+    message(STATUS "LibFabric support enabled")
+    message(STATUS "LIB_FABRIC: Path: ${LIBFABRIC_LIBRARY} Include: ${LIBFABRIC_INCLUDE_DIR}")
+    SET_PROPERTY(GLOBAL PROPERTY ASAPO_COMMON_FABRIC_LIBRARIES ${ASAPO_COMMON_IO_LIBRARIES} libfabric)
+    add_definitions(-DLIBFABRIC_ENABLED)
+ENDIF()
 
 # format sources
 include(astyle)

CMakeModules/FindLibFabric.cmake

+# FindLibFabric
+# -------------
+#
+# Tries to find LibFabric on the system
+#
+# Available variables
+# LIBFABRIC_LIBRARY         - Path to the library
+# LIBFABRIC_INCLUDE_DIR     - Path to the include dir
+
+cmake_minimum_required(VERSION 2.6)
+
+find_path(LIBFABRIC_INCLUDE_DIR fabric.h)
+find_library(LIBFABRIC_LIBRARY fabric)
+
+mark_as_advanced(LIBFABRIC_INCLUDE_DIR LIBFABRIC_LIBRARY)

common/cpp/CMakeLists.txt

@@ -12,6 +12,8 @@ add_subdirectory(src/logger)
 
 add_subdirectory(src/request)
 
+add_subdirectory(src/asapo_fabric)
+
 if(BUILD_MONGODB_CLIENTLIB)
     add_subdirectory(src/database)
 endif()

common/cpp/include/asapo_fabric/asapo_fabric.h

+#ifndef ASAPO_FABRIC_H
+#define ASAPO_FABRIC_H
+
+#include <cstdint>
+#include <string>
+#include <memory>
+#include <common/error.h>
+
+namespace asapo { namespace fabric {
+        typedef uint64_t FabricAddress;
+        typedef uint64_t FabricMessageId;
+
+#pragma pack(push, 1)
+        struct MemoryRegionDetails {
+            uint64_t addr;
+            uint64_t length;
+            uint64_t key;
+        };
+#pragma pack(pop)
+
+        class FabricMemoryRegion {
+        public:
+            virtual ~FabricMemoryRegion() = default;
+            virtual MemoryRegionDetails* GetDetails() = 0;
+        };
+
+        class FabricContext {
+        public:
+            /// If this function is not called, the default timeout is 5000 ms
+            virtual void SetRequestTimeout(uint64_t msTimeout) = 0;
+
+            virtual std::string GetAddress() const = 0;
+
+            virtual std::unique_ptr<FabricMemoryRegion> ShareMemoryRegion(void* src, size_t size, Error* error) = 0;
+
+            virtual void Send(FabricAddress dstAddress, FabricMessageId messageId,
+                              const void* src, size_t size, Error* error) = 0;
+
+            virtual void Recv(FabricAddress srcAddress, FabricMessageId messageId,
+                              void* dst, size_t size, Error* error) = 0;
+
+            virtual void RdmaWrite(FabricAddress dstAddress,
+                                   MemoryRegionDetails* details, const void* buffer, size_t size,
+                                   Error* error) = 0;
+
+            // Since RdmaRead heavily impacts the performance we will not implement this
+            // virtual void RdmaRead(...) = 0;
+
+
+        };
+
+        class FabricClient : public FabricContext {
+        public:
+            virtual ~FabricClient() = default;
+
+            virtual FabricAddress AddServerAddress(const std::string& serverAddress, Error* error) = 0;
+        };
+
+        class FabricServer : public FabricContext {
+        public:
+            virtual ~FabricServer() = default;
+
+            virtual void RecvAny(FabricAddress* srcAddress, FabricMessageId* messageId, void* src, size_t size, Error* error) = 0;
+        };
+
+        class FabricFactory {
+        public:
+            virtual std::unique_ptr<FabricServer> CreateAndBindServer(Error* error) const = 0;
+
+            virtual std::unique_ptr<FabricClient> CreateClient(Error* error) const = 0;
+        };
+
+        std::unique_ptr<FabricFactory> GenerateDefaultFabricFactory();
+}}
+
+#endif //ASAPO_FABRIC_H

common/cpp/src/asapo_fabric/CMakeLists.txt

+set(TARGET_NAME asapo-fabric)
+
+include_directories(include)
+
+set(SOURCE_FILES asapo_fabric.cpp)
+
+IF(ENABLE_LIBFABRIC)
+	set(SOURCE_FILES ${SOURCE_FILES}
+		fabric_factory_impl.cpp
+		)
+ELSE()
+	set(SOURCE_FILES ${SOURCE_FILES}
+		fabric_factory_not_supported.cpp
+		)
+ENDIF()
+
+################################
+# Library
+################################
+
+add_library(${TARGET_NAME} STATIC ${SOURCE_FILES})
+
+target_include_directories(${TARGET_NAME} PUBLIC ${ASAPO_CXX_COMMON_INCLUDE_DIR})

common/cpp/src/asapo_fabric/asapo_fabric.cpp

+#include <asapo_fabric/asapo_fabric.h>
+
+#ifdef LIBFABRIC_ENABLED
+#include "fabric_factory_impl.h"
+#else
+#include "fabric_factory_not_supported.h"
+#endif
+
+using namespace asapo::fabric;
+
+std::unique_ptr<FabricFactory> asapo::fabric::GenerateDefaultFabricFactory() {
+#ifdef LIBFABRIC_ENABLED
+    return std::unique_ptr<FabricFactory>(new FabricFactoryImpl());
+#else
+    return std::unique_ptr<FabricFactory>(new FabricFactoryNotSupported());
+#endif
+}

common/cpp/src/asapo_fabric/fabric_factory_impl.cpp

+#include "fabric_factory_impl.h"
+#include "fabric_internal_impl_common.h"
+#include <rdma/fabric.h>
+
+using namespace asapo::fabric;
+
+std::string fi_version_string(uint32_t version) {
+    return std::to_string(FI_MAJOR(version)) + "." + std::to_string(FI_MINOR(version));
+}
+
+bool FabricFactoryImpl::HasValidVersion(Error* error) const {
+    auto current_version = fi_version();
+
+    if (FI_VERSION_LT(current_version, EXPECTED_FI_VERSION)) {
+        std::string found_version_str = fi_version_string(current_version);
+        std::string expected_version_str = fi_version_string(EXPECTED_FI_VERSION);
+
+        std::string errorText = "LibFabric outdated.";
+        errorText += " (Found " + found_version_str + " but expected at least " + expected_version_str + ")";
+
+        *error = TextError(errorText);
+        return false;
+    }
+
+    return true;
+}
+
+std::unique_ptr<FabricClient>
+FabricFactoryImpl::CreateClient(Error* error) const {
+    if (!HasValidVersion(error)) {
+        return nullptr;
+    }
+
+    *error = TextError("This build of ASAPO does not support LibFabric.");
+    return nullptr;
+}
+
+std::unique_ptr<FabricServer> FabricFactoryImpl::CreateAndBindServer(Error* error) const {
+    if (!HasValidVersion(error)) {
+        return nullptr;
+    }
+
+    *error = TextError("This build of ASAPO does not support LibFabric.");
+    return nullptr;
+}

common/cpp/src/asapo_fabric/fabric_factory_impl.h

+#include <asapo_fabric/asapo_fabric.h>
+
+#ifndef ASAPO_FABRIC_FACTORY_IMPL_H
+#define ASAPO_FABRIC_FACTORY_IMPL_H
+
+namespace asapo { namespace fabric {
+    class FabricFactoryImpl : public FabricFactory {
+        bool HasValidVersion(Error* error) const;
+
+        std::unique_ptr<FabricServer> CreateAndBindServer(Error* error) const override;
+
+        std::unique_ptr<FabricClient> CreateClient(Error* error) const override;
+    };
+}}
+
+#endif //ASAPO_FABRIC_FACTORY_IMPL_H

common/cpp/src/asapo_fabric/fabric_factory_not_supported.cpp

+#include "fabric_factory_not_supported.h"
+
+using namespace asapo::fabric;
+
+std::unique_ptr<FabricServer> asapo::fabric::FabricFactoryNotSupported::CreateAndBindServer(Error* error) const {
+    *error = TextError("This build of ASAPO does not support LibFabric.");
+    return nullptr;
+}
+
+std::unique_ptr<FabricClient> asapo::fabric::FabricFactoryNotSupported::CreateClient(Error* error) const {
+    *error = TextError("This build of ASAPO does not support LibFabric.");
+    return nullptr;
+}
+

common/cpp/src/asapo_fabric/fabric_factory_not_supported.h

+#ifndef ASAPO_FABRIC_FACTORY_NOT_SUPPORTED_H
+#define ASAPO_FABRIC_FACTORY_NOT_SUPPORTED_H
+
+#include <asapo_fabric/asapo_fabric.h>
+
+namespace asapo { namespace fabric {
+    class FabricFactoryNotSupported : public FabricFactory {
+        std::unique_ptr<FabricServer> CreateAndBindServer(Error* error) const override;
+
+        std::unique_ptr<FabricClient> CreateClient(Error* error) const override;
+    };
+}}
+
+#endif //ASAPO_FABRIC_FACTORY_NOT_SUPPORTED_H

common/cpp/src/asapo_fabric/fabric_internal_impl_common.h

+#ifndef ASAPO_FABRIC_INTERNAL_IMPL_COMMON_H
+#define ASAPO_FABRIC_INTERNAL_IMPL_COMMON_H
+
+/*
+ * This file contains common features used in ASAPO's integration of libfabric.
+ * Only include this file into *.cpp files, never in *.h files
+ */
+
+#ifndef EXPECTED_FI_VERSION
+#define EXPECTED_FI_VERSION FI_VERSION(1, 9)
+#endif
+
+#pragma pack(push, 1)
+struct HandshakePayload {
+    // Hostnames can be up to 256 Bytes long. We also need to store the port number.
+    char hostnameAndPort[512];
+};
+
+#pragma pack(pop)
+
+#define TODO_UNKNOWN_ADDRESS (FI_ADDR_NOTAVAIL - 1)
+
+#endif //ASAPO_FABRIC_INTERNAL_IMPL_COMMON_H